o bi:@sddlZddlZddlZddlZddlZddlZddlZddlmZm Z ddl m Z ddl m Z mZmZmZmZmZddlmZddlmZddlmZmZddlmZmZmZmZdd l m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z'dd l(m)Z*m+Z,m-Z.ddl/m0Z0m1Z1m2Z2m3Z4ddl5Z5ddl6m7Z7m8Z8ddl9m:Z:ddl;m3Z3ddle?Z@dZAdZBeCdZDdZEGddde%ZFe dgdZGde3fddZHGdddeIe ZJGd d!d!ZKGd"d#d#ZLGd$d%d%ZMGd&d'd'ZNGd(d)d)ejOZPdS)*N) defaultdict namedtuple)Enum)AnyDictListSetTupleUnion)MetricDescriptorType) ValueDouble) aggregationmeasure)CountAggregationDataDistributionAggregationDataLastValueAggregationDataSumAggregationData) StatsExporter) StatsRecorder)View) ViewManager)tag_keytag_map tag_value)CounterMetricFamilyGaugeMetricFamilyHistogramMetricFamilyMetric)RAY_METRIC_CARDINALITY_LEVELenv_bool) GcsClient)r)_is_invalid_metric_nameRAY_WORKER_TIMEOUT_SCOREz [^a-zA-Z0-9]WorkerIdc@sReZdZdZdeefddZeddZeddZ ed d Z ed d Z d S)GaugezGauge representation of opencensus view. This class is used to collect process metrics from the reporter agent. Cpp metrics should be collected in a different way. tagscCsXt|r td|dt||||_||_dd|D}t||||jt |_ dS)NzInvalid metric name: z. Metric will be discarded and data will not be collected or published. Metric names can only contain letters, numbers, _, and :. Metric names cannot start with numbers.cSsg|]}t|qS)tag_key_moduleTagKey).0tagr'r'N/home/ubuntu/.local/lib/python3.10/site-packages/ray/_private/metrics_agent.py Hsz"Gauge.__init__..) r! ValueErrormeasure_module MeasureInt_measure _descriptionrrr LastValueAggregation_view)selfname descriptionunitr&r'r'r,__init__>s  zGauge.__init__cC|jSN)r1r5r'r'r,rMz Gauge.measurecCr:r;)r4r<r'r'r,viewQr=z Gauge.viewcCs|jjSr;)rr6r<r'r'r,r6Usz Gauge.namecCr:r;)r2r<r'r'r,r7Yr=zGauge.descriptionN) __name__ __module__ __qualname____doc__rstrr9propertyrr>r6r7r'r'r'r,r%7s   r%Record)gaugevaluer&metriccCs||jjds dStd|jj|j_|jD]%}|jD]}|dr:|j}|j j j }t |dkr:|ddkr:d|d<qqdS)a Fix the inbound `opencensus.proto.metrics.v1.Metric` protos to make it acceptable by opencensus.stats.DistributionAggregationData. - metric name: gRPC OpenCensus metrics have names with slashes and dots, e.g. `grpc.io/client/server_latency`[1]. However Prometheus metric names only take alphanums,underscores and colons[2]. We santinize the name by replacing non-alphanum chars to underscore, like the official opencensus prometheus exporter[3]. - distribution bucket bounds: The Metric proto asks distribution bucket bounds to be > 0 [4]. However, gRPC OpenCensus metrics have their first bucket bound == 0 [1]. This makes the `DistributionAggregationData` constructor to raise Exceptions. This applies to all bytes and milliseconds (latencies). The fix: we update the initial 0 bounds to be 0.000_000_1. This will not affect the precision of the metrics, since we don't expect any less-than-1 bytes, or less-than-1-nanosecond times. [1] https://github.com/census-instrumentation/opencensus-specs/blob/master/stats/gRPC.md#units # noqa: E501 [2] https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels [3] https://github.com/census-instrumentation/opencensus-cpp/blob/50eb5de762e5f87e206c011a4f930adb1a1775b1/opencensus/exporters/stats/prometheus/internal/prometheus_utils.cc#L39 # noqa: E501 [4] https://github.com/census-instrumentation/opencensus-proto/blob/master/src/opencensus/proto/metrics/v1/metrics.proto#L218 # noqa: E501 zgrpc.io/N_distribution_valuergHz>) metric_descriptorr6 startswithRE_NON_ALPHANUMSsub timeseriespointsHasFieldrJbucket_optionsexplicitboundslen)rHseriespoint dist_value bucket_boundsr'r'r,fix_grpc_metricas    rZc@seZdZdZdZdZdS)MetricCardinalityLevelzCardinality level of the metric. This is used to determine the cardinality level of the metric. The cardinality level is used to determine the type of the metric. legacy recommendedN)r?r@rArBLEGACY RECOMMENDEDr'r'r'r,r[sr[c @seZdZdedededeefddZeddZed d Zed d Z ed dZ eddZ ddZ de defddZdefddZdS)OpencensusProxyMetricr6descr8 label_keyscCs"||_||_||_||_i|_dS)z>Represents the OpenCensus metrics that will be proxy exported.N)_name_desc_unit _label_keys_data)r5r6rar8rbr'r'r,r9s  zOpencensusProxyMetric.__init__cCr:r;)rcr<r'r'r,r6r=zOpencensusProxyMetric.namecCr:r;)rdr<r'r'r,rar=zOpencensusProxyMetric.desccCr:r;)rer<r'r'r,r8r=zOpencensusProxyMetric.unitcCr:r;)rfr<r'r'r,rbr=z OpencensusProxyMetric.label_keyscCr:r;rgr<r'r'r,datar=zOpencensusProxyMetric.datacCs&t|jdkottt|jtS)z8Check if the metric is a distribution aggreation metric.r)rUrg isinstancenextitervaluesrr<r'r'r, is_distribution_aggregation_datasz6OpencensusProxyMetric.is_distribution_aggregation_data label_valuesricCs||j|<dS)zAdd the data to the metric. Args: label_values: The label values of the metric. data: The data to be added. Nrh)r5rorir'r'r,add_dataszOpencensusProxyMetric.add_datarHc Cs|j}t|dkr dS|D]j}tdd|jD}|jD]Z}|jjtjkr+t |j }nF|jjtj kr9t t |j}n8|jjtjkrGtt |j}n*|jjtjkrm|j}dd|jD}|jjj} t|j|j|j|j|| }ntd||j|<qq dS)zzParse the Opencensus Protobuf and store the data. The data can be accessed via `data` API once recorded. rNcss|]}|jVqdSr;rG)r*valr'r'r, sz/OpencensusProxyMetric.record..cSg|]}|jqSr')count)r*bucketr'r'r,r-z0OpencensusProxyMetric.record..zSummary is not supported)rOrUtuplerorPrKtyper CUMULATIVE_INT64r int64_valueCUMULATIVE_DOUBLErr double_value GAUGE_DOUBLErCUMULATIVE_DISTRIBUTIONrJbucketsrRrSrTrsumrusum_of_squared_deviationr.rg) r5rHrOrVlabelsrWrirXcounts_per_bucketrYr'r'r,recordsB      zOpencensusProxyMetric.recordN)r?r@rArCrr9rDr6rar8rbrirnr rrprrr'r'r'r,r`s      r`c@sReZdZdefddZedeeeffddZeddZ d e e fd d Z d S) ComponentidcCs||_t|_i|_dS)zyRepresent a component that requests to proxy export metrics Args: id: Id of this component. N)rtime monotonic_last_reported_time_metrics)r5rr'r'r,r9s  zComponent.__init__returncCr:)zAReturn the metrics requested to proxy export from this component.)rr<r'r'r,metricsszComponent.metricscCr:r;)rr<r'r'r,last_reported_timer=zComponent.last_reported_timercCsnt|_|D]-}t||j}|j}dd|jD}||jvr,t||j |j ||j|<|j| |qdS)zParse the Opencensus protobuf and store metrics. Metrics can be accessed via `metrics` API for proxy export. Args: metrics: A list of Opencensus protobuf for proxy export. cSrtr')key)r* label_keyr'r'r,r-rwz$Component.record..N) rrrrZrKr6rbrr`r7r8r)r5rrH descriptorr6rbr'r'r,rs    zComponent.recordN) r?r@rArCr9rDrr`rrrrrr'r'r'r,rs rc@seZdZd#dedefddZd$deedefd d Zd d Z d ededeedede e j de deeeefddfddZdefddZdefddZdeeeeefdeeeeffddZdeedeefdd Zd!d"ZdS)%OpenCensusProxyCollector< namespacecomponent_timeout_scCs,t|_||_||_i|_tdd|_dS)a)Prometheus collector implementation for opencensus proxy export. Prometheus collector requires to implement `collect` which is invoked whenever Prometheus queries the endpoint. The class is thread-safe. Args: namespace: Prometheus namespace. RAY_EXPORT_COUNTER_AS_GAUGETN) threadingLock_components_lock_component_timeout_s _namespace _componentsr_export_counter_as_gauge)r5rrr'r'r,r9s z!OpenCensusProxyCollector.__init__Nr worker_id_hexcCs`|stn|}|j||jvrt||j|<|j||WddS1s)wYdS)a.Record the metrics reported from the component that reports it. Args: metrics: A list of opencensus protobuf to proxy export metrics. worker_id_hex: A worker id that reports these metrics. If None, it means they are reported from Raylet or GCS. N)GLOBAL_COMPONENT_KEYrrrr)r5rrrr'r'r,r>s  "zOpenCensusProxyCollector.recordcCs|j@g}g}|jD]\}}t|j}||jkr+||t d ||q |D] }||j |q.|WdS1sFwYdS)zClean up stale components. Stale means the component is dead or unresponsive. Stale components won't be reported to Prometheus anymore. zSMetrics from a worker ({}) is cleaned up due to timeout. Time since last report {}sN) rritemsrrrrappendloggerinfoformatpop)r5stale_componentsstale_component_idsr componentelapsedr'r'r,clean_stale_componentsLs   $z/OpenCensusProxyCollector.clean_stale_components metric_namemetric_descriptionrb metric_unitsroagg_data metrics_maprcCsh|jsJ|jd|}t|t|ksJ||fdd|D}t|trK||}|s?t||||d} | g}|||<|dj||j ddSt|t r||}|set|||d} | g}|||<|dj||j d|j su dS| d r} dSt|d krt|d |d ||d} || t|d ksJ|d j||j ddSt|tr|jt|jksJg} d} t|jD]\} } | |j| 7} t| | g}| |q| d|j g||}|st|||d} | g}|||<|dj|| |jddSt|tr+||}|st|||d} | g}|||<|dj||jddStdt|)ato_metric translate the data that OpenCensus create to Prometheus format, using Prometheus Metric object. This method is from Opencensus Prometheus Exporter. Args: metric_name: Name of the metric. metric_description: Description of the metric. label_keys: The fixed label keys of the metric. metric_units: Units of the metric. label_values: The values of `label_keys`. agg_data: `opencensus.stats.aggregation_data.AggregationData` object. Aggregated data that needs to be converted as Prometheus samples metrics_map: The converted metric is added to this map. rIcSsg|]}|r|ndqS)r')r*tvr'r'r,r-szBOpenCensusProxyCollector.to_prometheus_metrics..)r6 documentationr8rr)rrGN)r6rr_totalz(DEPRECATED, use z_total metric instead) z+Inf)rr sum_valuezunsupported aggregation type )rlockedrrUrjrgetr add_metric count_datarsum_datarendswithrrrrTsorted enumeraterrCrrrrGr.ry)r5rrrbrrorrrrHr cum_countiiboundrvr'r'r,to_prometheus_metricscs             z.OpenCensusProxyCollector.to_prometheus_metricscCstSr;)rlowerr<r'r'r,%_get_metric_cardinality_level_settingsz>OpenCensusProxyCollector._get_metric_cardinality_level_settingcCs(zt|WStytjYSw)zGet the cardinality level of the core metric. This is used to determine set of metric labels. Some high cardinality labels such as `WorkerId` and `Name` will be removed on low cardinality level. )r[rr.r^r<r'r'r,_get_metric_cardinality_levels   z6OpenCensusProxyCollector._get_metric_cardinality_leveldatasc Cst|dksJ|d}t|trtttdd|DSt|tr-ttdd|DSt|tr>tttdd|DStdt|dtdtdtd |d ) NrcSrtr'rqr*rir'r'r,r- rwzCOpenCensusProxyCollector._aggregate_metric_data..cSrtr')rrr'r'r,r- rwcSrtr')rrr'r'r,r-rwzUnsupported aggregation type z. Supported types are z, z.Got .) rUrjrr rrrr.ry)r5rsampler'r'r,_aggregate_metric_datas.    z/OpenCensusProxyCollector._aggregate_metric_dataper_worker_metricsc Cstt|d}|rt|jvr|S|jt}tt}|D]}|jD]\}}||d|||dd |q#qt |j |j |j |jd||j|ddd}|D] \}}||||qY|gS)aCollect per-worker metrics, aggregate them into per-node metrics and convert them to Prometheus format. Args: per_worker_metrics: A list of per-worker metrics for the same metric name. Returns: A list of per-node metrics for the same metric name, with the high cardinality labels removed and the values aggregated. Nr)r6rar8rb)rkrlWORKER_ID_TAG_KEYrbindexrlistrirrr`r6rar8rpr) r5rrHworker_id_label_indexlabel_value_to_dataroriaggregated_metricrr'r'r,'_aggregate_with_recommended_cardinalitys:       z@OpenCensusProxyCollector._aggregate_with_recommended_cardinalityc cs |jhg}tt}|}|jD]!}|jD]}|tjkr/| s/||j  |q| |qq|D] }| | |q:i}|D]}|jD]\}} ||j |j|j|j|| |qPqIWdn1sowY|D] } | D]}|Vq|qxdS)a,Collect fetches the statistics from OpenCensus and delivers them as Prometheus Metrics. Collect is invoked every time a prometheus.Gatherer is run for example when the HTTP endpoint is invoked by Prometheus. This method is required as a Prometheus Collector. N)rrrrrrmrr[r_rnr6rextendrrirrrarbr8) r5open_cencus_metricsto_lower_cardinalitycardinality_levelrrHrprometheus_metrics_maprorirr'r'r,collectPsR    .z OpenCensusProxyCollector.collect)rr;)r?r@rArCintr9rrrrr tag_value_moduleTagValuerrPrometheusMetricrrr[rr rrrrr`rrr'r'r'r,rsJ      7rc@seZdZ ddededefddZddeefdd Z d e d e d e fd dZ ddeedefddZddeedefddZddZdS) MetricsAgentN view_managerstats_recorderstats_exportercCslt|_||_||_||_d|_|jdurd|_n|j|t|jj j t t tdd|_t|_dS)a+A class to record and export metrics. The class exports metrics in 2 different ways. - Directly record and export metrics using OpenCensus. - Proxy metrics from other core components (e.g., raylet, GCS, core workers). This class is thread-safe. Nx)r)rr_lockrrrproxy_exporter_collectorregister_exporterroptionsrrosgetenvr"set_registered_views)r5rrrr'r'r,r9s    zMetricsAgent.__init__recordscCs|pi}|jW|js WddS|D]?}|j}|j}|j}z |||i||WqtyU}ztd|j d|d|d|d| WYd}~qd}~wwWddS1sawYdS)z7Directly record and export stats from the same process.NzFailed to record metric z with value z with tags z and global tags due to: ) rrrFrGr& _record_gauge Exceptionrerrorr6)r5r global_tagsrrFrGr&er'r'r,record_and_exports&""zMetricsAgent.record_and_exportrFrGr&c Cs|j|jvr|j|j|j|j|j}t }| D]W\}}zt |}Wnt yI} ztd|d|jd| | d} ~ wwzt|} Wn t yq} ztd|d|d|jd| | d} ~ ww||| q!||j|||dS)NzFailed to create tag key z for metric rzFailed to create tag value z for key )r6rr register_viewr>addrnew_measurement_maptag_map_moduleTagMaprr(r)r.rrrrinsertmeasure_float_putrr) r5rFrGr&measurement_maprrtag_valrrrr'r'r,rs6  zMetricsAgent._record_gaugerrcCsN|j|js WddSWdn1swY|||dS)a{Proxy export metrics specified by a Opencensus Protobuf. This API is used to export metrics emitted from core components. Args: metrics: A list of protobuf Metric defined from OpenCensus. worker_id_hex: The worker ID it proxies metrics export. None if the metric is not from a worker (i.e., raylet, GCS). N)rr_proxy_export_metricsr5rrr'r'r,proxy_export_metricss z!MetricsAgent.proxy_export_metricscCs|j||dSr;)rrrr'r'r,rsz"MetricsAgent._proxy_export_metricscCsL|j|js WddSWdn1swY|jdS)zClean dead worker's metrics. Worker metrics are cleaned up and won't be exported once it is considered as dead. This method has to be periodically called by a caller. N)rrrrr<r'r'r,clean_all_dead_worker_metricssz*MetricsAgent.clean_all_dead_worker_metricsr;)r?r@rArrrr9rrErr%floatdictrrrCrrrr'r'r'r,rs 1 rcsHeZdZdZfddZddZddZdd Zd d Zd d Z Z S) PrometheusServiceDiscoveryWriteraFA class to support Prometheus service discovery. It supports file-based service discovery. Checkout https://prometheus.io/docs/guides/file-sd/ for more details. Args: gcs_address: Gcs address for this cluster. temp_dir: Temporary directory used by Ray to store logs and metadata. csFtjjj|dddd}||_tjjj|||_d|_ t dS)NTF)allow_cluster_id_nilfetch_cluster_id_if_nil) ray_rayletGcsClientOptionscreate gcs_address_privatestate_initialize_global_statetemp_dir&default_service_discovery_flush_periodsuperr9)r5r r gcs_client_options __class__r'r,r9sz)PrometheusServiceDiscoveryWriter.__init__cCsxt}dd|D}t|jd}|dd}|r!||d|dd}|r1||dtdd i|d gS) z4Return the content for Prometheus service discovery.cSs,g|]}|ddurd|d|dqS)aliveTz{}:{}NodeManagerAddressMetricsExportPort)r)r*noder'r'r,r-&s  zOPrometheusServiceDiscoveryWriter.get_file_discovery_content..)addresssAutoscalerMetricsAddressNzutf-8sDashboardMetricsAddressjobr)rtargets) rnodesr r internal_kv_getrdecodejsondumps)r5rmetrics_export_addresses gcs_clientautoscaler_addrdashboard_addrr'r'r,get_file_discovery_content#s   z;PrometheusServiceDiscoveryWriter.get_file_discovery_contentcCsT|}t|d}||Wdn1swYt||dS)Nw)get_temp_file_nameopenwriter#rreplaceget_target_file_name)r5temp_file_name json_filer'r'r,r'6s  z&PrometheusServiceDiscoveryWriter.writecCstj|jtjjjSr;)rpathjoinr rr  ray_constants!PROMETHEUS_SERVICE_DISCOVERY_FILEr<r'r'r,r)Cs z5PrometheusServiceDiscoveryWriter.get_target_file_namecCstj|jddtjjjS)Nz{}_{}tmp) rr,r-r rrr r.r/r<r'r'r,r%Hs  z3PrometheusServiceDiscoveryWriter.get_temp_file_namec Cst z|Wn+ty2}ztd|tttd|WYd}~nd}~wwt |j q)NTz,Writing a service discovery file, {},failed.zError message: ) r'rrwarningrr) traceback format_excrsleepr)r5rr'r'r,runPs   z$PrometheusServiceDiscoveryWriter.run) r?r@rArBr9r#r'r)r%r5 __classcell__r'r'rr,r s   r)Qrloggingrrerrr2 collectionsrrenumrtypingrrrrr r +opencensus.metrics.export.metric_descriptorr opencensus.metrics.export.valuer opencensus.statsr rr/!opencensus.stats.aggregation_datarrrropencensus.stats.base_exporterropencensus.stats.stats_recorderropencensus.stats.viewropencensus.stats.view_managerropencensus.tagsrr(rrrrprometheus_client.corerrrrrrray._private.ray_constantsrr ray._rayletr ray.core.generated.metrics_pb2ray.util.metricsr! getLoggerr?rr"rcompilerMrr%rErZrCr[r`rrrThreadrr'r'r'r,sT             '& ^.o