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,
))