1import re
2from typing import Dict, List, Optional, Sequence, Tuple, Union
3
4from .samples import Exemplar, Sample, Timestamp
5
6METRIC_TYPES = (
7 'counter', 'gauge', 'summary', 'histogram',
8 'gaugehistogram', 'unknown', 'info', 'stateset',
9)
10METRIC_NAME_RE = re.compile(r'^[a-zA-Z_:][a-zA-Z0-9_:]*$')
11METRIC_LABEL_NAME_RE = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
12RESERVED_METRIC_LABEL_NAME_RE = re.compile(r'^__.*$')
13
14
15class Metric:
16 """A single metric family and its samples.
17
18 This is intended only for internal use by the instrumentation client.
19
20 Custom collectors should use GaugeMetricFamily, CounterMetricFamily
21 and SummaryMetricFamily instead.
22 """
23
24 def __init__(self, name: str, documentation: str, typ: str, unit: str = ''):
25 if unit and not name.endswith("_" + unit):
26 name += "_" + unit
27 if not METRIC_NAME_RE.match(name):
28 raise ValueError('Invalid metric name: ' + name)
29 self.name: str = name
30 self.documentation: str = documentation
31 self.unit: str = unit
32 if typ == 'untyped':
33 typ = 'unknown'
34 if typ not in METRIC_TYPES:
35 raise ValueError('Invalid metric type: ' + typ)
36 self.type: str = typ
37 self.samples: List[Sample] = []
38
39 def add_sample(self, name: str, labels: Dict[str, str], value: float, timestamp: Optional[Union[Timestamp, float]] = None, exemplar: Optional[Exemplar] = None) -> None:
40 """Add a sample to the metric.
41
42 Internal-only, do not use."""
43 self.samples.append(Sample(name, labels, value, timestamp, exemplar))
44
45 def __eq__(self, other: object) -> bool:
46 return (isinstance(other, Metric)
47 and self.name == other.name
48 and self.documentation == other.documentation
49 and self.type == other.type
50 and self.unit == other.unit
51 and self.samples == other.samples)
52
53 def __repr__(self) -> str:
54 return "Metric({}, {}, {}, {}, {})".format(
55 self.name,
56 self.documentation,
57 self.type,
58 self.unit,
59 self.samples,
60 )
61
62 def _restricted_metric(self, names):
63 """Build a snapshot of a metric with samples restricted to a given set of names."""
64 samples = [s for s in self.samples if s[0] in names]
65 if samples:
66 m = Metric(self.name, self.documentation, self.type)
67 m.samples = samples
68 return m
69 return None
70
71
72class UnknownMetricFamily(Metric):
73 """A single unknown metric and its samples.
74 For use by custom collectors.
75 """
76
77 def __init__(self,
78 name: str,
79 documentation: str,
80 value: Optional[float] = None,
81 labels: Optional[Sequence[str]] = None,
82 unit: str = '',
83 ):
84 Metric.__init__(self, name, documentation, 'unknown', unit)
85 if labels is not None and value is not None:
86 raise ValueError('Can only specify at most one of value and labels.')
87 if labels is None:
88 labels = []
89 self._labelnames = tuple(labels)
90 if value is not None:
91 self.add_metric([], value)
92
93 def add_metric(self, labels: Sequence[str], value: float, timestamp: Optional[Union[Timestamp, float]] = None) -> None:
94 """Add a metric to the metric family.
95 Args:
96 labels: A list of label values
97 value: The value of the metric.
98 """
99 self.samples.append(Sample(self.name, dict(zip(self._labelnames, labels)), value, timestamp))
100
101
102# For backward compatibility.
103UntypedMetricFamily = UnknownMetricFamily
104
105
106class CounterMetricFamily(Metric):
107 """A single counter and its samples.
108
109 For use by custom collectors.
110 """
111
112 def __init__(self,
113 name: str,
114 documentation: str,
115 value: Optional[float] = None,
116 labels: Optional[Sequence[str]] = None,
117 created: Optional[float] = None,
118 unit: str = '',
119 ):
120 # Glue code for pre-OpenMetrics metrics.
121 if name.endswith('_total'):
122 name = name[:-6]
123 Metric.__init__(self, name, documentation, 'counter', unit)
124 if labels is not None and value is not None:
125 raise ValueError('Can only specify at most one of value and labels.')
126 if labels is None:
127 labels = []
128 self._labelnames = tuple(labels)
129 if value is not None:
130 self.add_metric([], value, created)
131
132 def add_metric(self,
133 labels: Sequence[str],
134 value: float,
135 created: Optional[float] = None,
136 timestamp: Optional[Union[Timestamp, float]] = None,
137 ) -> None:
138 """Add a metric to the metric family.
139
140 Args:
141 labels: A list of label values
142 value: The value of the metric
143 created: Optional unix timestamp the child was created at.
144 """
145 self.samples.append(Sample(self.name + '_total', dict(zip(self._labelnames, labels)), value, timestamp))
146 if created is not None:
147 self.samples.append(Sample(self.name + '_created', dict(zip(self._labelnames, labels)), created, timestamp))
148
149
150class GaugeMetricFamily(Metric):
151 """A single gauge and its samples.
152
153 For use by custom collectors.
154 """
155
156 def __init__(self,
157 name: str,
158 documentation: str,
159 value: Optional[float] = None,
160 labels: Optional[Sequence[str]] = None,
161 unit: str = '',
162 ):
163 Metric.__init__(self, name, documentation, 'gauge', unit)
164 if labels is not None and value is not None:
165 raise ValueError('Can only specify at most one of value and labels.')
166 if labels is None:
167 labels = []
168 self._labelnames = tuple(labels)
169 if value is not None:
170 self.add_metric([], value)
171
172 def add_metric(self, labels: Sequence[str], value: float, timestamp: Optional[Union[Timestamp, float]] = None) -> None:
173 """Add a metric to the metric family.
174
175 Args:
176 labels: A list of label values
177 value: A float
178 """
179 self.samples.append(Sample(self.name, dict(zip(self._labelnames, labels)), value, timestamp))
180
181
182class SummaryMetricFamily(Metric):
183 """A single summary and its samples.
184
185 For use by custom collectors.
186 """
187
188 def __init__(self,
189 name: str,
190 documentation: str,
191 count_value: Optional[int] = None,
192 sum_value: Optional[float] = None,
193 labels: Optional[Sequence[str]] = None,
194 unit: str = '',
195 ):
196 Metric.__init__(self, name, documentation, 'summary', unit)
197 if (sum_value is None) != (count_value is None):
198 raise ValueError('count_value and sum_value must be provided together.')
199 if labels is not None and count_value is not None:
200 raise ValueError('Can only specify at most one of value and labels.')
201 if labels is None:
202 labels = []
203 self._labelnames = tuple(labels)
204 # The and clause is necessary only for typing, the above ValueError will raise if only one is set.
205 if count_value is not None and sum_value is not None:
206 self.add_metric([], count_value, sum_value)
207
208 def add_metric(self,
209 labels: Sequence[str],
210 count_value: int,
211 sum_value: float,
212 timestamp:
213 Optional[Union[float, Timestamp]] = None
214 ) -> None:
215 """Add a metric to the metric family.
216
217 Args:
218 labels: A list of label values
219 count_value: The count value of the metric.
220 sum_value: The sum value of the metric.
221 """
222 self.samples.append(Sample(self.name + '_count', dict(zip(self._labelnames, labels)), count_value, timestamp))
223 self.samples.append(Sample(self.name + '_sum', dict(zip(self._labelnames, labels)), sum_value, timestamp))
224
225
226class HistogramMetricFamily(Metric):
227 """A single histogram and its samples.
228
229 For use by custom collectors.
230 """
231
232 def __init__(self,
233 name: str,
234 documentation: str,
235 buckets: Optional[Sequence[Union[Tuple[str, float], Tuple[str, float, Exemplar]]]] = None,
236 sum_value: Optional[float] = None,
237 labels: Optional[Sequence[str]] = None,
238 unit: str = '',
239 ):
240 Metric.__init__(self, name, documentation, 'histogram', unit)
241 if sum_value is not None and buckets is None:
242 raise ValueError('sum value cannot be provided without buckets.')
243 if labels is not None and buckets is not None:
244 raise ValueError('Can only specify at most one of buckets and labels.')
245 if labels is None:
246 labels = []
247 self._labelnames = tuple(labels)
248 if buckets is not None:
249 self.add_metric([], buckets, sum_value)
250
251 def add_metric(self,
252 labels: Sequence[str],
253 buckets: Sequence[Union[Tuple[str, float], Tuple[str, float, Exemplar]]],
254 sum_value: Optional[float],
255 timestamp: Optional[Union[Timestamp, float]] = None) -> None:
256 """Add a metric to the metric family.
257
258 Args:
259 labels: A list of label values
260 buckets: A list of lists.
261 Each inner list can be a pair of bucket name and value,
262 or a triple of bucket name, value, and exemplar.
263 The buckets must be sorted, and +Inf present.
264 sum_value: The sum value of the metric.
265 """
266 for b in buckets:
267 bucket, value = b[:2]
268 exemplar = None
269 if len(b) == 3:
270 exemplar = b[2] # type: ignore
271 self.samples.append(Sample(
272 self.name + '_bucket',
273 dict(list(zip(self._labelnames, labels)) + [('le', bucket)]),
274 value,
275 timestamp,
276 exemplar,
277 ))
278 # Don't include sum and thus count if there's negative buckets.
279 if float(buckets[0][0]) >= 0 and sum_value is not None:
280 # +Inf is last and provides the count value.
281 self.samples.append(
282 Sample(self.name + '_count', dict(zip(self._labelnames, labels)), buckets[-1][1], timestamp))
283 self.samples.append(
284 Sample(self.name + '_sum', dict(zip(self._labelnames, labels)), sum_value, timestamp))
285
286
287
288class GaugeHistogramMetricFamily(Metric):
289 """A single gauge histogram and its samples.
290
291 For use by custom collectors.
292 """
293
294 def __init__(self,
295 name: str,
296 documentation: str,
297 buckets: Optional[Sequence[Tuple[str, float]]] = None,
298 gsum_value: Optional[float] = None,
299 labels: Optional[Sequence[str]] = None,
300 unit: str = '',
301 ):
302 Metric.__init__(self, name, documentation, 'gaugehistogram', unit)
303 if labels is not None and buckets is not None:
304 raise ValueError('Can only specify at most one of buckets and labels.')
305 if labels is None:
306 labels = []
307 self._labelnames = tuple(labels)
308 if buckets is not None:
309 self.add_metric([], buckets, gsum_value)
310
311 def add_metric(self,
312 labels: Sequence[str],
313 buckets: Sequence[Tuple[str, float]],
314 gsum_value: Optional[float],
315 timestamp: Optional[Union[float, Timestamp]] = None,
316 ) -> None:
317 """Add a metric to the metric family.
318
319 Args:
320 labels: A list of label values
321 buckets: A list of pairs of bucket names and values.
322 The buckets must be sorted, and +Inf present.
323 gsum_value: The sum value of the metric.
324 """
325 for bucket, value in buckets:
326 self.samples.append(Sample(
327 self.name + '_bucket',
328 dict(list(zip(self._labelnames, labels)) + [('le', bucket)]),
329 value, timestamp))
330 # +Inf is last and provides the count value.
331 self.samples.extend([
332 Sample(self.name + '_gcount', dict(zip(self._labelnames, labels)), buckets[-1][1], timestamp),
333 # TODO: Handle None gsum_value correctly. Currently a None will fail exposition but is allowed here.
334 Sample(self.name + '_gsum', dict(zip(self._labelnames, labels)), gsum_value, timestamp), # type: ignore
335 ])
336
337
338class InfoMetricFamily(Metric):
339 """A single info and its samples.
340
341 For use by custom collectors.
342 """
343
344 def __init__(self,
345 name: str,
346 documentation: str,
347 value: Optional[Dict[str, str]] = None,
348 labels: Optional[Sequence[str]] = None,
349 ):
350 Metric.__init__(self, name, documentation, 'info')
351 if labels is not None and value is not None:
352 raise ValueError('Can only specify at most one of value and labels.')
353 if labels is None:
354 labels = []
355 self._labelnames = tuple(labels)
356 if value is not None:
357 self.add_metric([], value)
358
359 def add_metric(self,
360 labels: Sequence[str],
361 value: Dict[str, str],
362 timestamp: Optional[Union[Timestamp, float]] = None,
363 ) -> None:
364 """Add a metric to the metric family.
365
366 Args:
367 labels: A list of label values
368 value: A dict of labels
369 """
370 self.samples.append(Sample(
371 self.name + '_info',
372 dict(dict(zip(self._labelnames, labels)), **value),
373 1,
374 timestamp,
375 ))
376
377
378class StateSetMetricFamily(Metric):
379 """A single stateset and its samples.
380
381 For use by custom collectors.
382 """
383
384 def __init__(self,
385 name: str,
386 documentation: str,
387 value: Optional[Dict[str, bool]] = None,
388 labels: Optional[Sequence[str]] = None,
389 ):
390 Metric.__init__(self, name, documentation, 'stateset')
391 if labels is not None and value is not None:
392 raise ValueError('Can only specify at most one of value and labels.')
393 if labels is None:
394 labels = []
395 self._labelnames = tuple(labels)
396 if value is not None:
397 self.add_metric([], value)
398
399 def add_metric(self,
400 labels: Sequence[str],
401 value: Dict[str, bool],
402 timestamp: Optional[Union[Timestamp, float]] = None,
403 ) -> None:
404 """Add a metric to the metric family.
405
406 Args:
407 labels: A list of label values
408 value: A dict of string state names to booleans
409 """
410 labels = tuple(labels)
411 for state, enabled in sorted(value.items()):
412 v = (1 if enabled else 0)
413 self.samples.append(Sample(
414 self.name,
415 dict(zip(self._labelnames + (self.name,), labels + (state,))),
416 v,
417 timestamp,
418 ))