o `jk(@s`ddlZddlZddlZddlZddlmZmZmZmZddl m Z e e Z GdddZdS)N)AnonMeasurable KafkaMetric MetricConfig MetricName)Sensorc@seZdZdZ  dddZeddZedd Zd d d Zd dZ de j dfddZ ddZ d!ddZddZddZddZGdddZddZdS)"Metricsa A registry of sensors and metrics. A metric is a named, numerical measurement. A sensor is a handle to record numerical measurements as they occur. Each Sensor has zero or more associated metrics. For example a Sensor might represent message sizes and we might associate with this sensor a metric for the average, maximum, or other statistics computed off the sequence of message sizes that are recorded by the sensor. Usage looks something like this: # set up metrics: metrics = Metrics() # the global repository of metrics and sensors sensor = metrics.sensor('message-sizes') metric_name = MetricName('message-size-avg', 'producer-metrics') sensor.add(metric_name, Avg()) metric_name = MetricName('message-size-max', 'producer-metrics') sensor.add(metric_name, Max()) # as messages are sent we record the sizes sensor.record(message_size); NFcst_|p t_i_i_i_|pg_jD]}| gqd_ |r<fdd}tj |d}d|_ | dddtfd d d S) a Create a metrics repository with a default config, given metric reporters and the ability to expire eligible sensors Arguments: default_config (MetricConfig, optional): The default config reporters (list of AbstractMetricsReporter, optional): The metrics reporters enable_expiration (bool, optional): true if the metrics instance can garbage collect inactive sensors, false otherwise Fcs*jstdjjrdSdS)N)_closedtimesleepExpireSensorTaskrunselfrS/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/metrics/metrics.py expire_loop;s  z%Metrics.__init__..expire_loop)targetTcountzkafka-metrics-countz"total number of registered metricscs tjSN)len_metrics)confignowrrrGs z"Metrics.__init__..N) threadingRLock_lockr_config_sensorsr_children_sensors _reportersinitr Threaddaemonstart add_metric metric_namer)rdefault_config reportersenable_expirationreporterrmetrics_schedulerrrr__init__#s&       zMetrics.__init__cC|jSr)rrrrrrIszMetrics.configcCr.)zT Get all the metrics currently maintained and indexed by metricName )rrrrrmetricsMszMetrics.metricscCs(t|jj}||p it||||S)a Create a MetricName with the given name, group, description and tags, plus default tags specified in the metric configuration. Tag in tags takes precedence if the same tag key is specified in the default metric configuration. Arguments: name (str): The name of the metric group (str): logical group name of the metrics to which this metric belongs description (str, optional): A human-readable description to include in the metric tags (dict, optionals): additional key/value attributes of the metric )dictrtagsupdater)rnamegroup descriptionr2 combined_tagsrrrr'Ts zMetrics.metric_namecCs|std|j|dS)z Get the sensor with the given name if it exists Arguments: name (str): The name of the sensor Returns: Sensor: The sensor or None if no such sensor exists zname must be non-emptyN) ValueErrorrget)rr4rrr get_sensorhs zMetrics.get_sensorcCs||}|r |S|jB||}|sEt||||p|j|}||j|<|r?|D]}|j|}|s9g}||j|<||q(t d||WdS1sQwYdS)a Get or create a sensor with the given unique name and zero or more parent sensors. All parent sensors will receive every value recorded with this sensor. Arguments: name (str): The name of the sensor config (MetricConfig, optional): A default configuration to use for this sensor for metrics that don't have their own config inactive_sensor_expiration_time_seconds (int, optional): If no value if recorded on the Sensor for this duration of time, it is eligible for removal parents (list of Sensor): The parent sensors Returns: Sensor: The sensor that is created zAdded sensor with name %sN) r:rrrrr r9appendloggerdebug)rr4r'inactive_sensor_expiration_time_secondsparentssensorparentchildrenrrrr@vs(       $zMetrics.sensorc Cs|ji|j|}|rTd}|j.|j|d}|r8||kr8|jD]}||jq"td||j |d}Wdn1sBwY|r\|D]}| |j qKWddSWddSWddS1sowYdS)z Remove a sensor (if it exists), associated metrics and its children. Arguments: name (str): The name of the sensor to be removed NzRemoved sensor with name %s) rrr9popr/ remove_metricr'r<r=r remove_sensorr4)rr4r@ child_sensorsvalmetric child_sensorrrrrEs,      "zMetrics.remove_sensorcCs t|||p|j}||dS)a Add a metric to monitor an object that implements measurable. This metric won't be associated with any sensor. This is a way to expose existing values as metrics. Arguments: metricName (MetricName): The name of the metric measurable (AbstractMeasurable): The measurable that will be measured by this metric config (MetricConfig, optional): The configuration to use when measuring this measurable N)rrregister_metric)rr' measurablerrHrrrr&szMetrics.add_metriccCsV|j|j|d}|r|jD]}||q|WdS1s$wYdS)aw Remove a metric if it exists and return it. Return None otherwise. If a metric is removed, `metric_removal` will be invoked for each reporter. Arguments: metric_name (MetricName): The name of the metric Returns: KafkaMetric: the removed `KafkaMetric` or None if no such metric exists N)rrrCr!metric_removal)rr'rHr+rrrrDs   $zMetrics.remove_metriccCsL|j|t|j|j|WddS1swYdS)zAdd a MetricReporterN)rr"listr/valuesr!r;rr+rrr add_reporters"zMetrics.add_reportercCsj|j(|j|jvrtd|jf||j|j<|jD]}||qWddS1s.wYdS)Nz@A metric named "%s" already exists, cannot register another one.)rr'r/r8r! metric_change)rrHr+rrrrJs    "zMetrics.register_metricc@seZdZdZeddZdS)zMetrics.ExpireSensorTaskz This iterates over every Sensor and triggers a remove_sensor if it has expired. Package private for testing c Csdt|j}|D]&\}}|j|r td|||Wdn1s*wYq dS)NzRemoving expired sensor %s)rMritemsr has_expiredr<r=rE)r/rRr4r@rrrr s    zMetrics.ExpireSensorTask.runN)__name__ __module__ __qualname____doc__ staticmethodr rrrrr sr cCs(|jD]}|q|jd|_dS)zClose this metrics repository.TN)r!closerclearr rOrrrrYs    z Metrics.close)NNF)r0Nr)rTrUrVrWr-propertyrr/r'r:sysmaxsizer@rEr&rDrPrJr rYrrrrr s, &    (  r)loggingr\r r kafka.metricsrrrrkafka.metrics.statsr getLoggerrTr<rrrrrs