Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/prometheus_client/metrics_core.py: 43%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

154 statements  

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