o `۷i@sddlZddlZddlZddlZddlZddlZddlZddlmZm Z ddl m Z m Z m Z mZmZmZddlmZddlmZddlmZmZddlmZmZmZmZddlmZdd l m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z(m)Z*m+Z,dd l-m.Z.m/Z/m0Z0m1Z2ddl3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9m:Z:ddl;mm?Z?e@eAZBdZCdZDeEdZFGddde#ZGe dgdZHde1fddZIGdddZJGd d!d!ZKGd"d#d#ZLGd$d%d%ZMGd&d'd'ejNZOdS)(N) defaultdict namedtuple)AnyDictListSetTupleUnion)MetricDescriptorType) ValueDouble) aggregationmeasure)CountAggregationDataDistributionAggregationDataLastValueAggregationDataSumAggregationData) StatsExporter) StatsRecorder)View) ViewManager)tag_keytag_map tag_value)CounterMetricFamilyGaugeMetricFamilyHistogramMetricFamilyMetric build_address)env_bool)WORKER_ID_TAG_KEYMetricCardinality) GcsClient)r)_is_invalid_metric_nameRAY_WORKER_TIMEOUT_SCOREz [^a-zA-Z0-9]c@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(P/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/_private/metrics_agent.py Jsz"Gauge.__init__..) r# ValueErrormeasure_module MeasureInt_measure _descriptionrr r LastValueAggregation_view)selfname descriptionunitr'r(r(r-__init__@s  zGauge.__init__cC|jSN)r2r6r(r(r-r Oz Gauge.measurecCr;r<)r5r=r(r(r-viewSr>z Gauge.viewcCs|jjSr<)r r7r=r(r(r-r7Wsz Gauge.namecCr;r<)r3r=r(r(r-r8[r>zGauge.descriptionN) __name__ __module__ __qualname____doc__rstrr:propertyr r?r7r8r(r(r(r-r&9s   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_descriptorr7 startswithRE_NON_ALPHANUMSsub timeseriespointsHasFieldrKbucket_optionsexplicitboundslen)rIseriespoint dist_value bucket_boundsr(r(r-fix_grpc_metriccs    r[c @seZdZdedededeefddZeddZed d Zed d Z ed dZ eddZ ddZ de defddZdefddZdS)OpencensusProxyMetricr7descr9 label_keyscCs"||_||_||_||_i|_dS)z>Represents the OpenCensus metrics that will be proxy exported.N)_name_desc_unit _label_keys_data)r6r7r]r9r^r(r(r-r:s  zOpencensusProxyMetric.__init__cCr;r<)r_r=r(r(r-r7r>zOpencensusProxyMetric.namecCr;r<)r`r=r(r(r-r]r>zOpencensusProxyMetric.desccCr;r<)rar=r(r(r-r9r>zOpencensusProxyMetric.unitcCr;r<)rbr=r(r(r-r^r>z OpencensusProxyMetric.label_keyscCr;r<rcr=r(r(r-datar>zOpencensusProxyMetric.datacCs&t|jdkottt|jtS)z8Check if the metric is a distribution aggreation metric.r)rVrc isinstancenextitervaluesrr=r(r(r- is_distribution_aggregation_datasz6OpencensusProxyMetric.is_distribution_aggregation_data label_valuesrecCs||j|<dS)zAdd the data to the metric. Args: label_values: The label values of the metric. data: The data to be added. Nrd)r6rkrer(r(r-add_dataszOpencensusProxyMetric.add_datarIc 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<rH)r+valr(r(r- sz/OpencensusProxyMetric.record..cSg|]}|jqSr()count)r+bucketr(r(r-r.z0OpencensusProxyMetric.record..zSummary is not supported)rPrVtuplerkrQrLtyper CUMULATIVE_INT64r int64_valueCUMULATIVE_DOUBLErr double_value GAUGE_DOUBLErCUMULATIVE_DISTRIBUTIONrKbucketsrSrTrUrsumrqsum_of_squared_deviationr/rc) r6rIrPrWlabelsrXrerYcounts_per_bucketrZr(r(r-recordsB      zOpencensusProxyMetric.recordN)r@rArBrDrr:rEr7r]r9r^rerjrrrlrrr(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)r6rr(r(r-r:s  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. cSrpr()key)r+ label_keyr(r(r-r. rsz$Component.record..N) rrrr[rLr7r^rr\r8r9r)r6rrI descriptorr7r^r(r(r-rs    zComponent.recordN) r@rArBrDr:rErr\rrrrrr(r(r(r-rs rc@seZdZddedefddZd deedefd d Zd d Z d ededeedede e j de deeeefddfddZdeeeeefdeeeeffddZdeedeefddZddZdS)!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)r6rrr(r(r-r:s 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)r6rrrr(r(r-r5s  "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)r6stale_componentsstale_component_idsr componentelapsedr(r(r-clean_stale_componentsCs   $z/OpenCensusProxyCollector.clean_stale_components metric_namemetric_descriptionr^ metric_unitsrkagg_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. rJcSsg|]}|r|ndqS)r()r+tvr(r(r-r.zszBOpenCensusProxyCollector.to_prometheus_metrics..)r7 documentationr9rr)rrHN)r7rr_totalz(DEPRECATED, use z_total metric instead) z+Inf)rr| sum_valuezunsupported aggregation type )rlockedrrVrfrgetr add_metric count_datarsum_datarendswithrrrrUsorted enumeraterrDrr}rrHr/ru)r6rrr^rrkrrrrIr| cum_countiiboundrrr(r(r-to_prometheus_metricsZs             z.OpenCensusProxyCollector.to_prometheus_metricsdatasc Cst|dksJ|d}t|trtttdd|DSt|tr-ttdd|DSt|tr>tttdd|DStdt|dtdtdtd |d ) NrcSrpr(rmr+rer(r(r-r.rszCOpenCensusProxyCollector._aggregate_metric_data..cSrpr()rrr(r(r-r.rscSrpr()rrr(r(r-r.rszUnsupported aggregation type z. Supported types are z, z.Got .) rVrfrr r}rrr/ru)r6rsampler(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)r7r]r9r^)rgrhr r^indexrlistrerrr\r7r]r9rlr) r6rrIworker_id_label_indexlabel_value_to_datarkreaggregated_metricrr(r(r-'_aggregate_with_recommended_cardinalitys:       z@OpenCensusProxyCollector._aggregate_with_recommended_cardinalityc cs |jhg}tt}t}|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)rrrr!get_cardinality_levelrrir RECOMMENDEDrjr7rextendrrerrr]r^r9) r6open_cencus_metricsto_lower_cardinalitycardinality_levelrrIrprometheus_metrics_maprkrerr(r(r-collect9sR    .z OpenCensusProxyCollector.collect)rr<)r@rArBrDintr:rrrrrtag_value_moduleTagValuerrPrometheusMetricrr rrrrr\rrr(r(r(r-rsF     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)r6rrrr(r(r-r:us    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: ) rrrGrHr' _record_gauge Exceptionrerrorr7)r6r global_tagsrrGrHr'er(r(r-record_and_exports&""zMetricsAgent.record_and_exportrGrHr'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 )r7rr register_viewr?addrnew_measurement_maptag_map_moduleTagMaprr)r*r/rrrrinsertmeasure_float_putr r) r6rGrHr'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_metricsr6rrr(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@rArBrrrr:rrFrr&floatdictrrrDrrrr(r(r(r-rts 1 rcsPeZdZdZfddZddZddZdd Zd d Zd d Z ddZ 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. csVtjjj|dddd}||_tjjj|||_d|_ g|_ t |_ tdS)NTF)allow_cluster_id_nilfetch_cluster_id_if_nil)ray_rayletGcsClientOptionscreate gcs_address_privatestate_initialize_global_statetemp_dir&default_service_discovery_flush_period latest_service_discovery_contentrRLock _content_locksuperr:)r6rrgcs_client_options __class__r(r-r:s z)PrometheusServiceDiscoveryWriter.__init__cCs0|j |jWdS1swYdS)z3Return the latest stored service discovery content.N)r r r=r(r(r-$get_latest_service_discovery_contents$zEPrometheusServiceDiscoveryWriter.get_latest_service_discovery_contentcCst}dd|D}t|jd}|dd}|r!||d|dd}|r1||ddd i|d g}|j ||_Wdn1sJwYt |S) z4Return the content for Prometheus service discovery.cSs*g|]}|ddurt|d|dqS)aliveTNodeManagerAddressMetricsExportPortr)r+noder(r(r-r.s  zOPrometheusServiceDiscoveryWriter.get_file_discovery_content..)addresssAutoscalerMetricsAddressNzutf-8sDashboardMetricsAddressjobr)rtargets) rnodesr"rinternal_kv_getrdecoder r jsondumps)r6rmetrics_export_addresses gcs_clientautoscaler_addrdashboard_addrcontentr(r(r-get_file_discovery_contents     z;PrometheusServiceDiscoveryWriter.get_file_discovery_contentcCsT|}t|d}||Wdn1swYt||dS)Nw)get_temp_file_nameopenwriter"rreplaceget_target_file_name)r6temp_file_name json_filer(r(r-r&*s  z&PrometheusServiceDiscoveryWriter.writecCstj|jtjjjSr<)rpathjoinrrr ray_constants!PROMETHEUS_SERVICE_DISCOVERY_FILEr=r(r(r-r(7s z5PrometheusServiceDiscoveryWriter.get_target_file_namecCstj|jddtjjjS)Nz{}_{}tmp) rr+r,rrrrr-r.r=r(r(r-r$<s  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)r6rr(r(r-runDs   z$PrometheusServiceDiscoveryWriter.run) r@rArBrCr:rr"r&r(r$r4 __classcell__r(r(rr-rs  r)Prloggingrrerrr1 collectionsrrtypingrrrrrr +opencensus.metrics.export.metric_descriptorr opencensus.metrics.export.valuer opencensus.statsr r r0!opencensus.stats.aggregation_datarrrropencensus.stats.base_exporterropencensus.stats.stats_recorderropencensus.stats.viewropencensus.stats.view_managerropencensus.tagsrr)rrrrprometheus_client.corerrrrrrray._common.network_utilsrray._private.ray_constantsr)ray._private.telemetry.metric_cardinalityr r! ray._rayletr"ray.core.generated.metrics_pb2ray.util.metricsr# getLoggerr@rr$rcompilerNr&rFr[r\rrrThreadrr(r(r(r-sR              '&^.a