D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
self
/
root
/
opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
prometheus_client
/
Filename :
metrics_core.py
back
Copy
import re from typing import Dict, List, Optional, Sequence, Tuple, Union from .samples import Exemplar, Sample, Timestamp METRIC_TYPES = ( 'counter', 'gauge', 'summary', 'histogram', 'gaugehistogram', 'unknown', 'info', 'stateset', ) METRIC_NAME_RE = re.compile(r'^[a-zA-Z_:][a-zA-Z0-9_:]*$') METRIC_LABEL_NAME_RE = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$') RESERVED_METRIC_LABEL_NAME_RE = re.compile(r'^__.*$') class Metric: """A single metric family and its samples. This is intended only for internal use by the instrumentation client. Custom collectors should use GaugeMetricFamily, CounterMetricFamily and SummaryMetricFamily instead. """ def __init__(self, name: str, documentation: str, typ: str, unit: str = ''): if unit and not name.endswith("_" + unit): name += "_" + unit if not METRIC_NAME_RE.match(name): raise ValueError('Invalid metric name: ' + name) self.name: str = name self.documentation: str = documentation self.unit: str = unit if typ == 'untyped': typ = 'unknown' if typ not in METRIC_TYPES: raise ValueError('Invalid metric type: ' + typ) self.type: str = typ self.samples: List[Sample] = [] def add_sample(self, name: str, labels: Dict[str, str], value: float, timestamp: Optional[Union[Timestamp, float]] = None, exemplar: Optional[Exemplar] = None) -> None: """Add a sample to the metric. Internal-only, do not use.""" self.samples.append(Sample(name, labels, value, timestamp, exemplar)) def __eq__(self, other: object) -> bool: return (isinstance(other, Metric) and self.name == other.name and self.documentation == other.documentation and self.type == other.type and self.unit == other.unit and self.samples == other.samples) def __repr__(self) -> str: return "Metric({}, {}, {}, {}, {})".format( self.name, self.documentation, self.type, self.unit, self.samples, ) def _restricted_metric(self, names): """Build a snapshot of a metric with samples restricted to a given set of names.""" samples = [s for s in self.samples if s[0] in names] if samples: m = Metric(self.name, self.documentation, self.type) m.samples = samples return m return None class UnknownMetricFamily(Metric): """A single unknown metric and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, value: Optional[float] = None, labels: Optional[Sequence[str]] = None, unit: str = '', ): Metric.__init__(self, name, documentation, 'unknown', unit) if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels: Sequence[str], value: float, timestamp: Optional[Union[Timestamp, float]] = None) -> None: """Add a metric to the metric family. Args: labels: A list of label values value: The value of the metric. """ self.samples.append(Sample(self.name, dict(zip(self._labelnames, labels)), value, timestamp)) # For backward compatibility. UntypedMetricFamily = UnknownMetricFamily class CounterMetricFamily(Metric): """A single counter and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, value: Optional[float] = None, labels: Optional[Sequence[str]] = None, created: Optional[float] = None, unit: str = '', ): # Glue code for pre-OpenMetrics metrics. if name.endswith('_total'): name = name[:-6] Metric.__init__(self, name, documentation, 'counter', unit) if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value, created) def add_metric(self, labels: Sequence[str], value: float, created: Optional[float] = None, timestamp: Optional[Union[Timestamp, float]] = None, ) -> None: """Add a metric to the metric family. Args: labels: A list of label values value: The value of the metric created: Optional unix timestamp the child was created at. """ self.samples.append(Sample(self.name + '_total', dict(zip(self._labelnames, labels)), value, timestamp)) if created is not None: self.samples.append(Sample(self.name + '_created', dict(zip(self._labelnames, labels)), created, timestamp)) class GaugeMetricFamily(Metric): """A single gauge and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, value: Optional[float] = None, labels: Optional[Sequence[str]] = None, unit: str = '', ): Metric.__init__(self, name, documentation, 'gauge', unit) if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels: Sequence[str], value: float, timestamp: Optional[Union[Timestamp, float]] = None) -> None: """Add a metric to the metric family. Args: labels: A list of label values value: A float """ self.samples.append(Sample(self.name, dict(zip(self._labelnames, labels)), value, timestamp)) class SummaryMetricFamily(Metric): """A single summary and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, count_value: Optional[int] = None, sum_value: Optional[float] = None, labels: Optional[Sequence[str]] = None, unit: str = '', ): Metric.__init__(self, name, documentation, 'summary', unit) if (sum_value is None) != (count_value is None): raise ValueError('count_value and sum_value must be provided together.') if labels is not None and count_value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) # The and clause is necessary only for typing, the above ValueError will raise if only one is set. if count_value is not None and sum_value is not None: self.add_metric([], count_value, sum_value) def add_metric(self, labels: Sequence[str], count_value: int, sum_value: float, timestamp: Optional[Union[float, Timestamp]] = None ) -> None: """Add a metric to the metric family. Args: labels: A list of label values count_value: The count value of the metric. sum_value: The sum value of the metric. """ self.samples.append(Sample(self.name + '_count', dict(zip(self._labelnames, labels)), count_value, timestamp)) self.samples.append(Sample(self.name + '_sum', dict(zip(self._labelnames, labels)), sum_value, timestamp)) class HistogramMetricFamily(Metric): """A single histogram and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, buckets: Optional[Sequence[Union[Tuple[str, float], Tuple[str, float, Exemplar]]]] = None, sum_value: Optional[float] = None, labels: Optional[Sequence[str]] = None, unit: str = '', ): Metric.__init__(self, name, documentation, 'histogram', unit) if sum_value is not None and buckets is None: raise ValueError('sum value cannot be provided without buckets.') if labels is not None and buckets is not None: raise ValueError('Can only specify at most one of buckets and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if buckets is not None: self.add_metric([], buckets, sum_value) def add_metric(self, labels: Sequence[str], buckets: Sequence[Union[Tuple[str, float], Tuple[str, float, Exemplar]]], sum_value: Optional[float], timestamp: Optional[Union[Timestamp, float]] = None) -> None: """Add a metric to the metric family. Args: labels: A list of label values buckets: A list of lists. Each inner list can be a pair of bucket name and value, or a triple of bucket name, value, and exemplar. The buckets must be sorted, and +Inf present. sum_value: The sum value of the metric. """ for b in buckets: bucket, value = b[:2] exemplar = None if len(b) == 3: exemplar = b[2] # type: ignore self.samples.append(Sample( self.name + '_bucket', dict(list(zip(self._labelnames, labels)) + [('le', bucket)]), value, timestamp, exemplar, )) # Don't include sum and thus count if there's negative buckets. if float(buckets[0][0]) >= 0 and sum_value is not None: # +Inf is last and provides the count value. self.samples.append( Sample(self.name + '_count', dict(zip(self._labelnames, labels)), buckets[-1][1], timestamp)) self.samples.append( Sample(self.name + '_sum', dict(zip(self._labelnames, labels)), sum_value, timestamp)) class GaugeHistogramMetricFamily(Metric): """A single gauge histogram and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, buckets: Optional[Sequence[Tuple[str, float]]] = None, gsum_value: Optional[float] = None, labels: Optional[Sequence[str]] = None, unit: str = '', ): Metric.__init__(self, name, documentation, 'gaugehistogram', unit) if labels is not None and buckets is not None: raise ValueError('Can only specify at most one of buckets and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if buckets is not None: self.add_metric([], buckets, gsum_value) def add_metric(self, labels: Sequence[str], buckets: Sequence[Tuple[str, float]], gsum_value: Optional[float], timestamp: Optional[Union[float, Timestamp]] = None, ) -> None: """Add a metric to the metric family. Args: labels: A list of label values buckets: A list of pairs of bucket names and values. The buckets must be sorted, and +Inf present. gsum_value: The sum value of the metric. """ for bucket, value in buckets: self.samples.append(Sample( self.name + '_bucket', dict(list(zip(self._labelnames, labels)) + [('le', bucket)]), value, timestamp)) # +Inf is last and provides the count value. self.samples.extend([ Sample(self.name + '_gcount', dict(zip(self._labelnames, labels)), buckets[-1][1], timestamp), # TODO: Handle None gsum_value correctly. Currently a None will fail exposition but is allowed here. Sample(self.name + '_gsum', dict(zip(self._labelnames, labels)), gsum_value, timestamp), # type: ignore ]) class InfoMetricFamily(Metric): """A single info and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, value: Optional[Dict[str, str]] = None, labels: Optional[Sequence[str]] = None, ): Metric.__init__(self, name, documentation, 'info') if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels: Sequence[str], value: Dict[str, str], timestamp: Optional[Union[Timestamp, float]] = None, ) -> None: """Add a metric to the metric family. Args: labels: A list of label values value: A dict of labels """ self.samples.append(Sample( self.name + '_info', dict(dict(zip(self._labelnames, labels)), **value), 1, timestamp, )) class StateSetMetricFamily(Metric): """A single stateset and its samples. For use by custom collectors. """ def __init__(self, name: str, documentation: str, value: Optional[Dict[str, bool]] = None, labels: Optional[Sequence[str]] = None, ): Metric.__init__(self, name, documentation, 'stateset') if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels: Sequence[str], value: Dict[str, bool], timestamp: Optional[Union[Timestamp, float]] = None, ) -> None: """Add a metric to the metric family. Args: labels: A list of label values value: A dict of string state names to booleans """ labels = tuple(labels) for state, enabled in sorted(value.items()): v = (1 if enabled else 0) self.samples.append(Sample( self.name, dict(zip(self._labelnames + (self.name,), labels + (state,))), v, timestamp, ))