Collector for files for multi-process mode.
| 17 | |
| 18 | |
| 19 | class MultiProcessCollector: |
| 20 | """Collector for files for multi-process mode.""" |
| 21 | |
| 22 | def __init__(self, registry, path=None): |
| 23 | if path is None: |
| 24 | # This deprecation warning can go away in a few releases when removing the compatibility |
| 25 | if 'prometheus_multiproc_dir' in os.environ and 'PROMETHEUS_MULTIPROC_DIR' not in os.environ: |
| 26 | os.environ['PROMETHEUS_MULTIPROC_DIR'] = os.environ['prometheus_multiproc_dir'] |
| 27 | warnings.warn("prometheus_multiproc_dir variable has been deprecated in favor of the upper case naming PROMETHEUS_MULTIPROC_DIR", DeprecationWarning) |
| 28 | path = os.environ.get('PROMETHEUS_MULTIPROC_DIR') |
| 29 | if not path or not os.path.isdir(path): |
| 30 | raise ValueError('env PROMETHEUS_MULTIPROC_DIR is not set or not a directory') |
| 31 | self._path = path |
| 32 | if registry: |
| 33 | registry.register(self) |
| 34 | |
| 35 | @staticmethod |
| 36 | def merge(files, accumulate=True): |
| 37 | """Merge metrics from given mmap files. |
| 38 | |
| 39 | By default, histograms are accumulated, as per prometheus wire format. |
| 40 | But if writing the merged data back to mmap files, use |
| 41 | accumulate=False to avoid compound accumulation. |
| 42 | """ |
| 43 | metrics = MultiProcessCollector._read_metrics(files) |
| 44 | return MultiProcessCollector._accumulate_metrics(metrics, accumulate) |
| 45 | |
| 46 | @staticmethod |
| 47 | def _read_metrics(files): |
| 48 | metrics = {} |
| 49 | key_cache = {} |
| 50 | |
| 51 | def _parse_key(key): |
| 52 | val = key_cache.get(key) |
| 53 | if not val: |
| 54 | metric_name, name, labels, help_text = json.loads(key) |
| 55 | labels_key = tuple(sorted(labels.items())) |
| 56 | val = key_cache[key] = (metric_name, name, labels, labels_key, help_text) |
| 57 | return val |
| 58 | |
| 59 | for f in files: |
| 60 | parts = os.path.basename(f).split('_') |
| 61 | typ = parts[0] |
| 62 | try: |
| 63 | file_values = MmapedDict.read_all_values_from_file(f) |
| 64 | except FileNotFoundError: |
| 65 | if typ == 'gauge' and parts[1].startswith('live'): |
| 66 | # Files for 'live*' gauges can be deleted between the glob of collect |
| 67 | # and now (via a mark_process_dead call) so don't fail if |
| 68 | # the file is missing |
| 69 | continue |
| 70 | raise |
| 71 | for key, value, timestamp, _ in file_values: |
| 72 | metric_name, name, labels, labels_key, help_text = _parse_key(key) |
| 73 | |
| 74 | metric = metrics.get(metric_name) |
| 75 | if metric is None: |
| 76 | metric = Metric(metric_name, help_text, typ) |
no outgoing calls