o iR@sRddlZddlmZddlmZddlZddlZddlZddlZddl Z ddl m Z ddl m Z ddl m Z ddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(ddl)m*Z*ddl+m,Z,ddl-m.Z.ddl/m0Z0ddl/m1Z1ddl2m3Z3ddl4m5Z5ddZ6 e*e7Z8dZ9d Z:d!Z; ee>fZ?Gd"d#d#e@ZAe d$d%e=fd&e>fd'e=fgZBe d(d)e=fd%e=fd&e>fd'e=fgZCe d*d+ee?eAffd,eeBe>ffd-eeCe>ffgZDGd.d/d/e,ZEGd0d1d1ZFGd2d3d3ZGd6d4d5ZHdS)7N) defaultdict)partial) NamedTuple)Optional)Union)compat) process_tags)register_on_exit_signal)DEFAULT_SERVICE_NAME)DDSketch)config)Lock)fnv1_64)fibonacci_backoff_with_jitter) __version__)packb)get_connection) get_hostname) get_logger)PeriodicService) _human_size)decode_var_int_64)encode_var_int_64) SchemaBuilder) SchemaSamplercCs t|dS)Nr)gzipcompress)payloadr Z/home/ubuntu/.local/lib/python3.10/site-packages/ddtrace/internal/datastreams/processor.py gzip_compress& r"zdd-pathway-ctxzdd-pathway-ctx-base64c@seZdZdZdZddZdS) PathwayStatszAggregated pathway statistics.)full_pathway_latency edge_latency payload_sizecCst|_t|_t|_dSN)r r&r'r(selfr r r!__init__Ks zPathwayStats.__init__N)__name__ __module__ __qualname____doc__ __slots__r,r r r r!r%Fs r% PartitionKeytopic partition cluster_idConsumerPartitionKeygroupBucket pathway_statslatest_produce_offsetslatest_commit_offsetscs*eZdZdZ    d4deedeededeffd d Z d5d ed ede edededededdfddZ d6ddZ d6ddZ de e fddZdeddfddZd7d d!Zdeeddfd"d#Zd$edd%fd&d'Zd$eeeefdd%fd(d)Zd8d*d+Zd9d,d-Zd.d/Zd0d1Zd2d3ZZS):DataStreamsProcessorzeDataStreamsProcessor for computing, collecting and submitting data stream stats to the Datadog Agent.N? agent_urlintervaltimeoutretry_attemptscs |dur ttdp d}tt|j|d|ptj|_d|_ d|j|j f|_ ||_ t |d|_ tdd|_t|_d |jd d d |_tt|_ttt|_t|_t|_d |_ i|_!t"|d|j#d|dd|j$|_%t&t't(|d|)dS)N_DD_TRACE_STATS_WRITER_INTERVALg$@)r@z/v0.1/pipeline_statsz%s%seAcSstttttttSr))r8rr%intr r r r!ssz/DataStreamsProcessor.__init__..pythonzapplication/msgpackr)zDatadog-Meta-LangzDatadog-Meta-Tracer-Versionz Content-TypezContent-EncodingTg-?gS?r)attempts initial_wait)obj)*floatosgetenvsuperr<r, agent_configtrace_agent_url _agent_url _endpoint_agent_endpoint_timeoutrE_bucket_size_nsr_bucketsr_version_headersr ensure_textr _hostnamer _get_servicer _servicer _lock threadinglocal_current_context_enabled_schema_samplersrr@ _flush_stats_flush_stats_with_backoffr r_atexitstart)r+r?r@rArB __class__r r!r,bs>   zDataStreamsProcessor.__init__r hash_value parent_hash edge_tagsnow_secedge_latency_secfull_pathway_latency_secr(returnc Cs|jsdSt|d}|j:|||j} d|||f} |j| j| } | j|| j || j || |j| j| <WddS1sKwYdS)a on_checkpoint_creation is called every time a new checkpoint is created on a pathway. It records the latency to the previous checkpoint in the pathway (edge latency), and the latency from the very first element in the pathway (full_pathway_latency) the pathway is hashed to reduce amount of information transmitted in headers. :param hash_value: hash of the pathway, it's a hash of the edge leading to this point, and the parent hash. :param parent_hash: hash of the previous step in the pathway :param edge_tags: all tags associated with the edge leading to this step in the pathway :param now_sec: current time :param edge_latency_sec: latency of the direct edge between the previous point in the pathway, and the current step :param full_pathway_latency_sec: latency from the very start of the pathway. :return: Nothing NrD,) rarEr]rUjoinrVr9r&addr'r() r+rirjrkrlrmrnr(now_nsbucket_time_nsaggr_keystatsr r r!on_checkpoint_creations    "z+DataStreamsProcessor.on_checkpoint_creationc Cstt|d}t|||}|j!|||j}t||j|j||j|j|<WddS1s3wYdSNrD)rEr2r]rUmaxrVr:) r+r3r4offsetrlr5rskeyrtr r r!track_kafka_produces  "z(DataStreamsProcessor.track_kafka_producec Csvt|d}t||||}|j!|||j} t||j| j||j| j|<WddS1s4wYdSry)rEr6r]rUrzrVr;) r+r7r3r4r{rlr5rsr|rtr r r!track_kafka_commits "z'DataStreamsProcessor.track_kafka_commitc Csfg}g}|jD]\}}g}g}|||jD](\}}|\} } } dd| dD| | |j|j|jd} || q|j D](\} }dd| j d| j dt | j g}| jri|d | j|||d qI|jD]$\}}d d|j dt |j g}|jr|d |j|||d qw|||j||d q |D]}|j|=q|S) z!Serialize and update the buckets.cSsg|]}t|qSr )rrY).0tagr r r! sz;DataStreamsProcessor._serialize_buckets..rp)EdgeTagsHash ParentHashPathwayLatency EdgeLatency PayloadSizeztype:kafka_commitzconsumer_group:ztopic:z partition:zkafka_cluster_id:)TagsValueztype:kafka_produce)StartDurationStatsBacklogs)rVitemsappendr9splitr&to_protor'r(r;r7r3strr4r5r:rU)r+serialized_bucketsserialized_bucket_keysrtbucketbucket_aggr_statsbacklogsru stat_aggrrkrirjserialized_bucket consumer_keyr{ commit_tags producer_key produce_tagsr|r r r!_serialize_bucketssV      z'DataStreamsProcessor._serialize_bucketsrcCszt|j|j}|d|j||j|}Wnty(tj d|j ddw|j dkr5t ddS|j dkrJt d|j |j ||j dSt d tt||j dS) NPOSTz9failed to submit pathway stats to the Datadog agent at %sTexc_infoizHDatadog agent does not support data streams monitoring. Upgrade to 7.34+izXfailed to send data stream stats payload, %s (%s) (%s) response from Datadog agent at %sz sent %s to %s)rrQrTrequestrRrX getresponse ExceptionlogdebugrSstatuserrorreasonreadrlen)r+rconnrespr r r!rcs(     z!DataStreamsProcessor._flush_statscCs|j |}Wdn1swY|s tddS|j|jd||jd}tjr6t tj|d<tj rAt tj |d<t j }rJ||d<t|}t|}z||WdStymtjd|jdd YdSw) Nz,No data streams reported. Skipping flushing.rG)Service TracerVersionLangrHostnameEnvVersion ProcessTagszHretry limit exceeded submitting pathway stats to the Datadog agent at %sTr)r]rrrr\rWrZr envrrYversionrprocess_tags_listrr"rdrrrS)r+serialized_stats raw_payloadp_tagsr compressedr r r!periodics:     zDataStreamsProcessor.periodiccCs|||dSr))rstop)r+rAr r r!shutdown2szDataStreamsProcessor.shutdowndataDataStreamsCtxc Csz4td|ddd}|dd}t|\}}t|\}}t||t|dt|d}||j_|WStttj fyE| YSw)N|s|St|tr|d}n|}t|}||}|SNutf-8)r isinstancerencodebase64 b64decoder)r+rbinary_pathwayencoded_pathwaydata_streams_contextr r r!decode_pathway_b64Cs    z'DataStreamsProcessor.decode_pathway_b64cCs&|st}t|d||}||j_|S)z type: (Optional[int]) -> DataStreamsCtx :param now_sec: optional start time of this path. Use for services like Kinesis which we aren't getting path information for. r)timerr`r)r+rlrr r r!rPs z DataStreamsProcessor.new_pathwaycCsh|st}t|jdr|jj}n|}||j_d|vr)|t|tt7}|j||||d|S)a@ type: (list[str], Optional[int], Optional[int]) -> DataStreamsCtx :param tags: a list of strings identifying the pathway and direction :param now_sec: The time in seconds to count as "now" when computing latencies :param payload_size: The size of the payload being sent in bytes rz direction:out)rlr(span) rhasattrr`rrr encode_b64PROPAGATION_KEY_BASE_64set_checkpoint)r+tagsrlr(rrr r r!r]s  z#DataStreamsProcessor.set_checkpointcC&td}|j|t}||SNi)rrb setdefaultr try_sampler+r3now_mssamplerr r r!try_sample_schemas  z&DataStreamsProcessor.try_sample_schemacCrr)rrbrr can_samplerr r r!can_sample_schemayrz&DataStreamsProcessor.can_sample_schemacCs t||Sr))r get_schema)r+ schema_nameiteratorr r r!rr#zDataStreamsProcessor.get_schema)NNr=r>)r)rx)roNr))NrN)r-r.r/r0rrrKrEr,listrwr}r~dictrbytesrcrrrrrrrrrr __classcell__r r rgr!r<_s\1 (  6   r<c @sbeZdZdededededdf ddZdefd d Zde fd d Z d dZ     dddZ dS)r processorripathway_start_seccurrent_edge_start_secroNcCsR||_||_||_||_ttt|_ ttj pd|_ d|_ d|_ ||_ dS)Nnonerxr)rrrhashrrYr r[r servicerprevious_directionclosest_opposite_direction_hash%closest_opposite_direction_edge_start)r+rrirrr r r!r,s zDataStreamsCtx.__init__cCs2td|jtt|jdtt|jdS)Nrr)rpackrrrErrr*r r r!rs zDataStreamsCtx.encodecCs |}t|}|d}|Sr)rr b64encodedecode)r+rrrr r r!rs  zDataStreamsCtx.encode_b64cCs\dd}||j||jtj}|D]}|||7}qt|}ttd|td|S)NcSs t|ddS)Nr)encoding)r)sr r r! get_bytesr#z/DataStreamsCtx._compute_hash..get_bytesr)rrrbase_hash_bytesrrr)r+rrjrbt node_hashr r r! _compute_hashs zDataStreamsCtx._compute_hashrc Cs|st}t|}d}|D] }|dr|}nq||jkr4|j|_|jdkr/||_||_n|j|_n ||_|j|_||_|rC||_|rH||_|j} | || } |r[| dt | t ||jd} t ||jd} | |_||_|j j| | ||| | |ddS)a type: (list[str], float, float, float) -> None :param tags: an list of tags identifying the pathway and direction :param now_sec: The time in seconds to count as "now" when computing latencies :param edge_start_sec_override: Use this to override the starting time of an edge :param pathway_start_sec_override: Use this to override the starting time of a pathway rxz direction:rz pathway.hashg)r(N)rsorted startswithrrrrrrr _set_tag_strrrzrrw) r+rrledge_start_sec_overridepathway_start_sec_overrider(r directionrrjrirmpathway_latency_secr r r!rsB      zDataStreamsCtx.set_checkpoint)NNNrN) r-r.r/r<rErKr,rrrrrrr r r r!rs* rc@sDeZdZdZedededdfddZedededefd d Z dS) DsmPathwayCodecz DsmPathwayCodec is responsible for: - encoding and injecting DSM pathway context into produced message headers - extracting and decoding DSM pathway context from consumed message headers rcarrierroNcCs(t|tr |r |js dS||t<dSr))rrrrr)rr r r r!rszDsmPathwayCodec.encodedata_streams_processorcCs|s|Sd}t|vr||t}n%t|vr9||t}|jdkr9z ||t}Wn ty8d}Ynw|s?|S|S)Nr)rrrPROPAGATION_KEYrrr)r rrr r r!rs   zDsmPathwayCodec.decode) r-r.r/r0 staticmethodrrrr<rr r r r!r s r c Cs\z|tWdSty-}ztjr"tdt|WYd}~dSWYd}~dSd}~ww)Nz-Failed to shutdown data streams processor: %s)rSHUTDOWN_TIMEOUTrr _data_streams_enabledrwarningrepr)rJer r r!re srer))Ir collectionsr functoolsrrrLrr^rtypingrrrddtrace.internalrrddtrace.internal.atexitr ddtrace.internal.constantsr ddtrace.internal.nativer ddtrace.internal.settings._agentr rO!ddtrace.internal.settings._configddtrace.internal.threadsr ddtrace.internal.utils.fnvrddtrace.internal.utils.retryrddtrace.versionr _encodingragentrhostnamerloggerrrrwriterrrrrschemas.schema_builderrschemas.schema_samplerrr"r-rrrrtuplerrEPathwayAggrKeyobjectr%r2r6r8r<rr rer r r r!s~                             &a%