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.
| 10 | |
| 11 | |
| 12 | class Metric: |
| 13 | """A single metric family and its samples. |
| 14 | |
| 15 | This is intended only for internal use by the instrumentation client. |
| 16 | |
| 17 | Custom collectors should use GaugeMetricFamily, CounterMetricFamily |
| 18 | and SummaryMetricFamily instead. |
| 19 | """ |
| 20 | |
| 21 | def __init__(self, name: str, documentation: str, typ: str, unit: str = ''): |
| 22 | if unit and not name.endswith("_" + unit): |
| 23 | name += "_" + unit |
| 24 | _validate_metric_name(name) |
| 25 | self.name: str = name |
| 26 | self.documentation: str = documentation |
| 27 | self.unit: str = unit |
| 28 | if typ == 'untyped': |
| 29 | typ = 'unknown' |
| 30 | if typ not in METRIC_TYPES: |
| 31 | raise ValueError('Invalid metric type: ' + typ) |
| 32 | self.type: str = typ |
| 33 | self.samples: List[Sample] = [] |
| 34 | |
| 35 | def add_sample(self, name: str, labels: Dict[str, str], value: float, timestamp: Optional[Union[Timestamp, float]] = None, exemplar: Optional[Exemplar] = None, native_histogram: Optional[NativeHistogram] = None) -> None: |
| 36 | """Add a sample to the metric. |
| 37 | |
| 38 | Internal-only, do not use.""" |
| 39 | self.samples.append(Sample(name, labels, value, timestamp, exemplar, native_histogram)) |
| 40 | |
| 41 | def __eq__(self, other: object) -> bool: |
| 42 | return (isinstance(other, Metric) |
| 43 | and self.name == other.name |
| 44 | and self.documentation == other.documentation |
| 45 | and self.type == other.type |
| 46 | and self.unit == other.unit |
| 47 | and self.samples == other.samples) |
| 48 | |
| 49 | def __repr__(self) -> str: |
| 50 | return "Metric({}, {}, {}, {}, {})".format( |
| 51 | self.name, |
| 52 | self.documentation, |
| 53 | self.type, |
| 54 | self.unit, |
| 55 | self.samples, |
| 56 | ) |
| 57 | |
| 58 | def _restricted_metric(self, names): |
| 59 | """Build a snapshot of a metric with samples restricted to a given set of names.""" |
| 60 | samples = [s for s in self.samples if s[0] in names] |
| 61 | if samples: |
| 62 | m = Metric(self.name, self.documentation, self.type) |
| 63 | m.samples = samples |
| 64 | return m |
| 65 | return None |
| 66 | |
| 67 | |
| 68 | class UnknownMetricFamily(Metric): |
no outgoing calls