diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/prometheus_client/metrics_core.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/prometheus_client/metrics_core.py | 418 |
1 files changed, 418 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/prometheus_client/metrics_core.py b/.venv/lib/python3.12/site-packages/prometheus_client/metrics_core.py new file mode 100644 index 00000000..7226d920 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/prometheus_client/metrics_core.py @@ -0,0 +1,418 @@ +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, + )) |