external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

Disable CLS and APM by default. v5.0.101

Browse source
This commit is contained in:
winlin 2022-12-03 18:35:41 +08:00
parent 8adb1693cc
commit e86e0c8999
20 changed files with 104 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

20
trunk/src/app/srs_app_edge.cpp
View file

@ -110,9 +110,11 @@ srs_error_t SrsEdgeRtmpUpstream::connect(SrsRequest* r, SrsLbRoundRobin* lb)
srs_utime_t sto = SRS_CONSTS_RTMP_PULSE;
sdk = new SrsSimpleRtmpClient(url, cto, sto);
#ifdef SRS_APM
// Create a client span and store it to an AMF0 propagator.
ISrsApmSpan* span_client = _srs_apm->inject(_srs_apm->span("edge-pull")->set_kind(SrsApmKindClient)->as_child(_srs_apm->load()), sdk->extra_args());
SrsAutoFree(ISrsApmSpan, span_client);
#endif
if ((err = sdk->connect()) != srs_success) {
return srs_error_wrap(err, "edge pull %s failed, cto=%dms, sto=%dms.", url.c_str(), srsu2msi(cto), srsu2msi(sto));
@ -393,7 +395,9 @@ SrsEdgeIngester::SrsEdgeIngester()
source = NULL;
edge = NULL;
req = NULL;
#ifdef SRS_APM
span_main_ = NULL;
#endif
upstream = new SrsEdgeRtmpUpstream("");
lb = new SrsLbRoundRobin();
@ -404,7 +408,9 @@ SrsEdgeIngester::~SrsEdgeIngester()
{
stop();
#ifdef SRS_APM
srs_freep(span_main_);
#endif
srs_freep(upstream);
srs_freep(lb);
srs_freep(trd);
@ -416,10 +422,12 @@ srs_error_t SrsEdgeIngester::initialize(SrsLiveSource* s, SrsPlayEdge* e, SrsReq
edge = e;
req = r;
#ifdef SRS_APM
// We create a dedicate span for edge ingester, and all players will link to this one.
// Note that we use a producer span and end it immediately.
srs_assert(!span_main_);
span_main_ = _srs_apm->span("edge")->set_kind(SrsApmKindProducer)->end();
#endif
return srs_success;
}
@ -458,11 +466,13 @@ string SrsEdgeIngester::get_curr_origin()
return lb->selected();
}
#ifdef SRS_APM
ISrsApmSpan* SrsEdgeIngester::span()
{
srs_assert(span_main_);
return span_main_;
}
#endif
// when error, edge ingester sleep for a while and retry.
#define SRS_EDGE_INGESTER_CIMS (3 * SRS_UTIME_SECONDS)
@ -471,10 +481,12 @@ srs_error_t SrsEdgeIngester::cycle()
{
srs_error_t err = srs_success;
#ifdef SRS_APM
// Save span from parent coroutine to current coroutine context, so that we can load if in this coroutine, for
// example to use it in SrsEdgeRtmpUpstream which use RTMP or FLV client to connect to upstream server.
_srs_apm->store(span_main_);
srs_assert(span_main_);
#endif
while (true) {
// We always check status first.
@ -483,18 +495,22 @@ srs_error_t SrsEdgeIngester::cycle()
return srs_error_wrap(err, "edge ingester");
}
#ifdef SRS_APM
srs_assert(span_main_);
ISrsApmSpan* start = _srs_apm->span("edge-start")->set_kind(SrsApmKindConsumer)->as_child(span_main_)->end();
srs_freep(start);
#endif
if ((err = do_cycle()) != srs_success) {
srs_warn("EdgeIngester: Ignore error, %s", srs_error_desc(err).c_str());
srs_freep(err);
}
#ifdef SRS_APM
srs_assert(span_main_);
ISrsApmSpan* stop = _srs_apm->span("edge-stop")->set_kind(SrsApmKindConsumer)->as_child(span_main_)->end();
srs_freep(stop);
#endif
// Check whether coroutine is stopped, see https://github.com/ossrs/srs/issues/2901
if ((err = trd->pull()) != srs_success) {
@ -771,10 +787,12 @@ srs_error_t SrsEdgeForwarder::start()
srs_utime_t sto = SRS_CONSTS_RTMP_TIMEOUT;
sdk = new SrsSimpleRtmpClient(url, cto, sto);
#ifdef SRS_APM
// Create a client span and store it to an AMF0 propagator.
// Note that we are able to load the span from coroutine context because in the same coroutine.
ISrsApmSpan* span_client = _srs_apm->inject(_srs_apm->span("edge-push")->set_kind(SrsApmKindClient)->as_child(_srs_apm->load()), sdk->extra_args());
SrsAutoFree(ISrsApmSpan, span_client);
#endif
if ((err = sdk->connect()) != srs_success) {
return srs_error_wrap(err, "sdk connect %s failed, cto=%dms, sto=%dms.", url.c_str(), srsu2msi(cto), srsu2msi(sto));
@ -962,6 +980,7 @@ srs_error_t SrsPlayEdge::on_client_play()
return srs_error_new(ERROR_RTMP_EDGE_PLAY_STATE, "state is stopping");
}
#ifdef SRS_APM
// APM bind client span to edge span, which fetch stream from upstream server.
// We create a new span to link the two span, because these two spans might be ended.
if (ingester->span() && _srs_apm->load()) {
@ -969,6 +988,7 @@ srs_error_t SrsPlayEdge::on_client_play()
ISrsApmSpan* to = _srs_apm->span("edge-link")->as_child(ingester->span())->link(from);
srs_freep(from); srs_freep(to);
}
#endif
return err;
}

4
trunk/src/app/srs_app_edge.hpp
View file

@ -143,7 +143,9 @@ private:
SrsCoroutine* trd;
SrsLbRoundRobin* lb;
SrsEdgeUpstream* upstream;
#ifdef SRS_APM
ISrsApmSpan* span_main_;
#endif
public:
SrsEdgeIngester();
virtual ~SrsEdgeIngester();
@ -152,8 +154,10 @@ public:
virtual srs_error_t start();
virtual void stop();
virtual std::string get_curr_origin();
#ifdef SRS_APM
// Get the current main span. Note that it might be NULL.
ISrsApmSpan* span();
#endif
// Interface ISrsReusableThread2Handler
public:
virtual srs_error_t cycle();

4
trunk/src/app/srs_app_hybrid.cpp
View file

@ -181,6 +181,7 @@ srs_error_t SrsHybridServer::initialize()
return srs_error_wrap(err, "dvr async");
}
#ifdef SRS_APM
// Initialize TencentCloud CLS object.
if ((err = _srs_cls->initialize()) != srs_success) {
return srs_error_wrap(err, "cls client");
@ -188,6 +189,7 @@ srs_error_t SrsHybridServer::initialize()
if ((err = _srs_apm->initialize()) != srs_success) {
return srs_error_wrap(err, "apm client");
}
#endif
// Register some timers.
timer20ms_->subscribe(clock_monitor_);
@ -395,6 +397,7 @@ srs_error_t SrsHybridServer::on_timer(srs_utime_t interval)
thread_desc.c_str(), free_desc.c_str(), objs_desc.c_str()
);
#ifdef SRS_APM
// Report logs to CLS if enabled.
if ((err = _srs_cls->report()) != srs_success) {
srs_warn("ignore cls err %s", srs_error_desc(err).c_str());
@ -406,6 +409,7 @@ srs_error_t SrsHybridServer::on_timer(srs_utime_t interval)
srs_warn("ignore apm err %s", srs_error_desc(err).c_str());
srs_freep(err);
}
#endif
return err;
}

2
trunk/src/app/srs_app_latest_version.cpp
View file

@ -166,11 +166,13 @@ void srs_build_features(stringstream& ss)
SRS_CHECK_FEATURE(security, ss);
SRS_CHECK_FEATURE2(_srs_config_by_env, "env", ss);
#ifdef SRS_APM
SRS_CHECK_FEATURE2(_srs_cls->enabled(), "cls", ss);
SRS_CHECK_FEATURE3(_srs_cls->nn_logs(), "logs", _srs_cls->nn_logs(), ss);
SRS_CHECK_FEATURE2(_srs_apm->enabled(), "apm", ss);
SRS_CHECK_FEATURE3(_srs_apm->nn_spans(), "spans", _srs_apm->nn_spans(), ss);
#endif
}
SrsLatestVersion::SrsLatestVersion()

40
trunk/src/app/srs_app_rtmp_conn.cpp
View file

@ -101,9 +101,11 @@ SrsRtmpConn::SrsRtmpConn(SrsServer* svr, srs_netfd_t c, string cip, int cport)
ip = cip;
port = cport;
create_time = srsu2ms(srs_get_system_time());
#ifdef SRS_APM
span_main_ = _srs_apm->dummy();
span_connect_ = _srs_apm->dummy();
span_client_ = _srs_apm->dummy();
#endif
trd = new SrsSTCoroutine("rtmp", this, _srs_context->get_id());
kbps = new SrsNetworkKbps();
@ -149,9 +151,11 @@ SrsRtmpConn::~SrsRtmpConn()
srs_freep(rtmp);
srs_freep(refer);
srs_freep(security);
#ifdef SRS_APM
srs_freep(span_main_);
srs_freep(span_connect_);
srs_freep(span_client_);
#endif
}
std::string SrsRtmpConn::desc()
@ -169,16 +173,22 @@ srs_error_t SrsRtmpConn::do_cycle()
{
srs_error_t err = srs_success;
#ifdef SRS_APM
// We should keep the root span to alive util connection closed.
// Note that we use producer and consumer span because RTMP connection is long polling connection.
// Note that we also store this span in coroutine context, so that edge could load it.
srs_freep(span_main_);
span_main_ = _srs_apm->span("rtmp")->set_kind(SrsApmKindServer)->attr("cip", ip)
->attr("cid", _srs_context->get_id().c_str());
#endif
#ifdef SRS_APM
srs_trace("RTMP client ip=%s:%d, fd=%d, trace=%s, span=%s", ip.c_str(), port, srs_netfd_fileno(stfd),
span_main_->format_trace_id(), span_main_->format_span_id()
);
#else
srs_trace("RTMP client ip=%s:%d, fd=%d", ip.c_str(), port, srs_netfd_fileno(stfd));
#endif
rtmp->set_recv_timeout(SRS_CONSTS_RTMP_TIMEOUT);
rtmp->set_send_timeout(SRS_CONSTS_RTMP_TIMEOUT);
@ -193,12 +203,14 @@ srs_error_t SrsRtmpConn::do_cycle()
srs_trace("RTMP proxy real client ip=%s", rips.c_str());
}
#ifdef SRS_APM
// Update the real IP of client, also set the HTTP fields.
span_main_->attr("rip", rip ? rips : ip)->attr("http.client_ip", rip ? rips : ip);
// The span for RTMP connecting to application.
srs_freep(span_connect_);
span_connect_ = _srs_apm->span("connect")->as_child(span_main_);
#endif
SrsRequest* req = info->req;
if ((err = rtmp->connect_app(req)) != srs_success) {
@ -239,9 +251,11 @@ srs_error_t SrsRtmpConn::do_cycle()
srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id);
}
#ifdef SRS_APM
// Load the span from the AMF0 object propagator.
// Note that we will update the trace id, so please make sure no spans are ended before this.
_srs_apm->extract(span_main_, req->args);
#endif
}
if ((err = service_cycle()) != srs_success) {
@ -414,8 +428,10 @@ srs_error_t SrsRtmpConn::service_cycle()
return srs_error_wrap(err, "rtmp: response connect app");
}
#ifdef SRS_APM
// Must be a connecting application span.
span_connect_->end();
#endif
if ((err = rtmp->on_bw_done()) != srs_success) {
return srs_error_wrap(err, "rtmp: on bw down");
@ -493,20 +509,24 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
srs_trace("client identified, type=%s, vhost=%s, app=%s, stream=%s, param=%s, duration=%dms",
srs_client_type_string(info->type).c_str(), req->vhost.c_str(), req->app.c_str(), req->stream.c_str(), req->param.c_str(), srsu2msi(req->duration));
#ifdef SRS_APM
// Start APM only when client is identified, because it might republish.
srs_freep(span_client_);
span_client_ = _srs_apm->span("client")->as_child(span_connect_)->attr("type", srs_client_type_string(info->type))
->attr("url", req->get_stream_url())->attr("http.url", req->get_stream_url());
// We store the span to coroutine context, for edge to load it.
_srs_apm->store(span_client_);
#endif
// discovery vhost, resolve the vhost from config
SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
if (parsed_vhost) {
req->vhost = parsed_vhost->arg0();
}
#ifdef SRS_APM
span_client_->attr("vhost", req->vhost)->attr("http.host", req->host)->attr("http.server_name", req->vhost)
->attr("http.target", srs_fmt("/%s/%s", req->app.c_str(), req->stream.c_str()));
#endif
if (req->schema.empty() || req->vhost.empty() || req->port == 0 || req->app.empty()) {
return srs_error_new(ERROR_RTMP_REQ_TCURL, "discovery tcUrl failed, tcUrl=%s, schema=%s, vhost=%s, port=%d, app=%s",
@ -582,10 +602,12 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
return srs_error_wrap(err, "rtmp: callback on play");
}
#ifdef SRS_APM
// Must be a client span.
span_client_->set_name("play")->end();
// We end the connection span because it's a producer and only trace the established.
span_main_->end();
#endif
err = playing(source);
http_hooks_on_stop();
@ -597,10 +619,12 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
return srs_error_wrap(err, "rtmp: start FMLE publish");
}
#ifdef SRS_APM
// Must be a client span.
span_client_->set_name("publish")->end();
// We end the connection span because it's a producer and only trace the established.
span_main_->end();
#endif
return publishing(source);
}
@ -609,10 +633,12 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
return srs_error_wrap(err, "rtmp: start HAIVISION publish");
}
#ifdef SRS_APM
// Must be a client span.
span_client_->set_name("publish")->end();
// We end the connection span because it's a producer and only trace the established.
span_main_->end();
#endif
return publishing(source);
}
@ -621,10 +647,12 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
return srs_error_wrap(err, "rtmp: start FLASH publish");
}
#ifdef SRS_APM
// Must be a client span.
span_client_->set_name("publish")->end();
// We end the connection span because it's a producer and only trace the established.
span_main_->end();
#endif
return publishing(source);
}
@ -786,9 +814,11 @@ srs_error_t SrsRtmpConn::do_playing(SrsLiveSource* source, SrsLiveConsumer* cons
srs_trace("start play smi=%dms, mw_sleep=%d, mw_msgs=%d, realtime=%d, tcp_nodelay=%d",
srsu2msi(send_min_interval), srsu2msi(mw_sleep), mw_msgs, realtime, tcp_nodelay);
#ifdef SRS_APM
ISrsApmSpan* span = _srs_apm->span("play-cycle")->set_kind(SrsApmKindProducer)->as_child(span_client_)
->attr("realtime", srs_fmt("%d", realtime))->end();
SrsAutoFree(ISrsApmSpan, span);
#endif
while (true) {
// when source is set to expired, disconnect it.
@ -833,10 +863,12 @@ srs_error_t SrsRtmpConn::do_playing(SrsLiveSource* source, SrsLiveConsumer* cons
(int)pprint->age(), count, kbps->get_send_kbps(), kbps->get_send_kbps_30s(), kbps->get_send_kbps_5m(),
kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m(), srsu2msi(mw_sleep), mw_msgs);
#ifdef SRS_APM
// TODO: Do not use pithy print for frame span.
ISrsApmSpan* sample = _srs_apm->span("play-frame")->set_kind(SrsApmKindConsumer)->as_child(span)
->attr("msgs", srs_fmt("%d", count))->attr("kbps", srs_fmt("%d", kbps->get_send_kbps_30s()));
srs_freep(sample);
#endif
}
if (count <= 0) {
@ -963,9 +995,11 @@ srs_error_t SrsRtmpConn::do_publishing(SrsLiveSource* source, SrsPublishRecvThre
srs_trace("start publish mr=%d/%d, p1stpt=%d, pnt=%d, tcp_nodelay=%d", mr, srsu2msi(mr_sleep), srsu2msi(publish_1stpkt_timeout), srsu2msi(publish_normal_timeout), tcp_nodelay);
}
#ifdef SRS_APM
ISrsApmSpan* span = _srs_apm->span("publish-cycle")->set_kind(SrsApmKindProducer)->as_child(span_client_)
->attr("timeout", srs_fmt("%d", srsu2msi(publish_normal_timeout)))->end();
SrsAutoFree(ISrsApmSpan, span);
#endif
int64_t nb_msgs = 0;
uint64_t nb_frames = 0;
@ -1015,10 +1049,12 @@ srs_error_t SrsRtmpConn::do_publishing(SrsLiveSource* source, SrsPublishRecvThre
kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m(), mr, srsu2msi(mr_sleep),
srsu2msi(publish_1stpkt_timeout), srsu2msi(publish_normal_timeout));
#ifdef SRS_APM
// TODO: Do not use pithy print for frame span.
ISrsApmSpan* sample = _srs_apm->span("publish-frame")->set_kind(SrsApmKindConsumer)->as_child(span)
->attr("msgs", srs_fmt("%" PRId64, nb_frames))->attr("kbps", srs_fmt("%d", kbps->get_recv_kbps_30s()));
srs_freep(sample);
#endif
}
}
@ -1543,6 +1579,7 @@ srs_error_t SrsRtmpConn::cycle()
// Serve the client.
err = do_cycle();
#ifdef SRS_APM
// Final APM span, parent is the last span, not the root span. Note that only client or server kind will be filtered
// for error or exception report.
ISrsApmSpan* span_final = _srs_apm->span("final")->set_kind(SrsApmKindServer)->as_child(span_client_);
@ -1550,6 +1587,7 @@ srs_error_t SrsRtmpConn::cycle()
if (srs_error_code(err) != 0) {
span_final->record_error(err)->set_status(SrsApmStatusError, srs_fmt("fail code=%d", srs_error_code(err)));
}
#endif
// Update statistic when done.
SrsStatistic* stat = SrsStatistic::instance();

2
trunk/src/app/srs_app_statistic.cpp
View file

@ -624,6 +624,7 @@ void SrsStatistic::dumps_hints_kv(std::stringstream & ss)
}
}
#ifdef SRS_APM
void SrsStatistic::dumps_cls_summaries(SrsClsSugar* sugar)
{
if (!vhosts.empty()) {
@ -682,6 +683,7 @@ void SrsStatistic::dumps_cls_streams(SrsClsSugars* sugars)
}
}
}
#endif
SrsStatisticVhost* SrsStatistic::create_vhost(SrsRequest* req)
{

2
trunk/src/app/srs_app_statistic.hpp
View file

@ -208,10 +208,12 @@ public:
virtual srs_error_t dumps_clients(SrsJsonArray* arr, int start, int count);
// Dumps the hints about SRS server.
void dumps_hints_kv(std::stringstream & ss);
#ifdef SRS_APM
public:
// Dumps the CLS summary.
void dumps_cls_summaries(SrsClsSugar* sugar);
void dumps_cls_streams(SrsClsSugars* sugars);
#endif
private:
virtual SrsStatisticVhost* create_vhost(SrsRequest* req);
virtual SrsStatisticStream* create_stream(SrsStatisticVhost* vhost, SrsRequest* req);

2
trunk/src/app/srs_app_tencentcloud.cpp
View file

@ -5,6 +5,7 @@
//
#include <srs_app_tencentcloud.hpp>
#ifdef SRS_APM
#include <srs_kernel_error.hpp>
#include <srs_kernel_buffer.hpp>
@ -2280,4 +2281,5 @@ void SrsApmClient::snapshot(SrsOtelSpan* span)
{
spans_.push_back(span);
}
#endif

2
trunk/src/app/srs_app_tencentcloud.hpp
View file

@ -8,6 +8,7 @@
#define SRS_APP_TENCENTCLOUD_HPP
#include <srs_core.hpp>
#ifdef SRS_APM
#include <srs_kernel_buffer.hpp>
@ -533,4 +534,5 @@ public:
extern SrsApmClient* _srs_apm;
#endif
#endif

2
trunk/src/app/srs_app_threads.cpp
View file

@ -445,9 +445,11 @@ srs_error_t srs_global_initialize()
// Create global async worker for DVR.
_srs_dvr_async = new SrsAsyncCallWorker();
#ifdef SRS_APM
// Initialize global TencentCloud CLS object.
_srs_cls = new SrsClsClient();
_srs_apm = new SrsApmClient();
#endif
return err;
}