Metric collector registry. Collectors must have a no-argument method 'collect' that returns a list of Metric objects. The returned metrics should be consistent with the Prometheus exposition formats.
| 16 | |
| 17 | |
| 18 | class CollectorRegistry: |
| 19 | """Metric collector registry. |
| 20 | |
| 21 | Collectors must have a no-argument method 'collect' that returns a list of |
| 22 | Metric objects. The returned metrics should be consistent with the Prometheus |
| 23 | exposition formats. |
| 24 | """ |
| 25 | |
| 26 | def __init__(self, auto_describe: bool = False, target_info: Optional[Dict[str, str]] = None, |
| 27 | support_collectors_without_names: bool = False): |
| 28 | self._collector_to_names: Dict[Collector, List[str]] = {} |
| 29 | self._names_to_collectors: Dict[str, Collector] = {} |
| 30 | self._auto_describe = auto_describe |
| 31 | self._lock = Lock() |
| 32 | self._target_info: Optional[Dict[str, str]] = {} |
| 33 | self._support_collectors_without_names = support_collectors_without_names |
| 34 | self._collectors_without_names: List[Collector] = [] |
| 35 | self.set_target_info(target_info) |
| 36 | |
| 37 | def register(self, collector: Collector) -> None: |
| 38 | """Add a collector to the registry.""" |
| 39 | with self._lock: |
| 40 | names = self._get_names(collector) |
| 41 | duplicates = set(self._names_to_collectors).intersection(names) |
| 42 | if duplicates: |
| 43 | raise ValueError( |
| 44 | 'Duplicated timeseries in CollectorRegistry: {}'.format( |
| 45 | duplicates)) |
| 46 | for name in names: |
| 47 | self._names_to_collectors[name] = collector |
| 48 | self._collector_to_names[collector] = names |
| 49 | if self._support_collectors_without_names and not names: |
| 50 | self._collectors_without_names.append(collector) |
| 51 | |
| 52 | def unregister(self, collector: Collector) -> None: |
| 53 | """Remove a collector from the registry.""" |
| 54 | with self._lock: |
| 55 | for name in self._collector_to_names[collector]: |
| 56 | del self._names_to_collectors[name] |
| 57 | del self._collector_to_names[collector] |
| 58 | |
| 59 | def _get_names(self, collector): |
| 60 | """Get names of timeseries the collector produces and clashes with.""" |
| 61 | desc_func = None |
| 62 | # If there's a describe function, use it. |
| 63 | try: |
| 64 | desc_func = collector.describe |
| 65 | except AttributeError: |
| 66 | pass |
| 67 | # Otherwise, if auto describe is enabled use the collect function. |
| 68 | if not desc_func and self._auto_describe: |
| 69 | desc_func = collector.collect |
| 70 | |
| 71 | if not desc_func: |
| 72 | return [] |
| 73 | |
| 74 | result = [] |
| 75 | type_suffixes = { |