Add new version of OpenMetrics base class

datadog_checks_base/datadog_checks/base/checks/openmetrics/base_check.py

@@ -7,6 +7,9 @@
 from .. import AgentCheck
 from .mixins import OpenMetricsScraperMixin
 
+if not PY2:
+    from .v2.base import OpenMetricsBaseCheckV2
+
 STANDARD_FIELDS = [
     'prometheus_url',
     'namespace',
@@ -60,6 +63,17 @@ class OpenMetricsBaseCheck(OpenMetricsScraperMixin, AgentCheck):
         'prometheus_timeout': {'name': 'timeout'},
     }
 
+    def __new__(cls, *args, **kwargs):
+        # Legacy signature
+        if kwargs or len(args) != 3 or not isinstance(args[2], list):
+            return super(OpenMetricsBaseCheck, cls).__new__(cls)
+
+        instance = args[2][0]
+        if not PY2 and 'openmetrics_endpoint' in instance:
+            return OpenMetricsBaseCheckV2(*args, **kwargs)
+        else:
+            return super(OpenMetricsBaseCheck, cls).__new__(cls)
+
     def __init__(self, *args, **kwargs):
         """
         The base class for any Prometheus-based integration.

datadog_checks_base/datadog_checks/base/checks/openmetrics/mixins.py

@@ -62,6 +62,8 @@ def create_scraper_configuration(self, instance=None):
 
         A default mixin configuration will be returned if there is no instance.
         """
+        if 'openmetrics_endpoint' in instance:
+            raise CheckException('The setting `openmetrics_endpoint` is only available for Agent version 7 or later')
 
         # We can choose to create a default mixin configuration for an empty instance
         if instance is None:

datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/__init__.py

@@ -0,0 +1,3 @@
+# (C) Datadog, Inc. 2020-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)

datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/base.py

@@ -0,0 +1,75 @@
+# (C) Datadog, Inc. 2020-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from collections import ChainMap
+from contextlib import contextmanager
+
+from ....errors import ConfigurationError
+from ... import AgentCheck
+from .scraper import OpenMetricsScraper
+
+
+class OpenMetricsBaseCheckV2(AgentCheck):
+    DEFAULT_METRIC_LIMIT = 2000
+
+    def __init__(self, name, init_config, instances):
+        super(OpenMetricsBaseCheckV2, self).__init__(name, init_config, instances)
+
+        # All desired scraper configurations, which subclasses can override as needed
+        self.scraper_configs = [self.instance]
+
+        # All configured scrapers keyed by the endpoint
+        self.scrapers = {}
+
+        self.check_initializations.append(self.configure_scrapers)
+
+    def check(self, _):
+        self.refresh_scrapers()
+
+        for endpoint, scraper in self.scrapers.items():
+            self.log.info('Scraping OpenMetrics endpoint: %s', endpoint)
+
+            with self.adopt_namespace(scraper.namespace):
+                scraper.scrape()
+
+    def configure_scrapers(self):
+        scrapers = {}
+
+        for config in self.scraper_configs:
+            endpoint = config.get('openmetrics_endpoint', '')
+            if not isinstance(endpoint, str):
+                raise ConfigurationError('The setting `openmetrics_endpoint` must be a string')
+            elif not endpoint:
+                raise ConfigurationError('The setting `openmetrics_endpoint` is required')
+
+            scrapers[endpoint] = self.create_scraper(config)
+
+        self.scrapers.clear()
+        self.scrapers.update(scrapers)
+
+    def create_scraper(self, config):
+        # Subclasses can override to return a custom scraper based on configuration
+        return OpenMetricsScraper(self, self.get_config_with_defaults(config))
+
+    def set_dynamic_tags(self, *tags):
+        for scraper in self.scrapers.values():
+            scraper.set_dynamic_tags(*tags)
+
+    def get_config_with_defaults(self, config):
+        return ChainMap(config, self.get_default_config())
+
+    def get_default_config(self):
+        return {}
+
+    def refresh_scrapers(self):
+        pass
+
+    @contextmanager
+    def adopt_namespace(self, namespace):
+        old_namespace = self.__NAMESPACE__
+
+        try:
+            self.__NAMESPACE__ = namespace or old_namespace
+            yield
+        finally:
+            self.__NAMESPACE__ = old_namespace

datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/labels.py

@@ -0,0 +1,148 @@
+# (C) Datadog, Inc. 2020-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from ....utils.common import no_op
+
+
+class LabelAggregator:
+    def __init__(self, check, config):
+        share_labels = config.get('share_labels', {})
+        if not isinstance(share_labels, dict):
+            raise TypeError('Setting `share_labels` must be a mapping')
+        elif not share_labels:
+            self.populate = no_op
+            return
+
+        self.metric_config = {}
+        for metric, config in share_labels.items():
+            data = self.metric_config[metric] = {}
+
+            if config is True:
+                continue
+            elif not isinstance(config, dict):
+                raise TypeError(f'Metric `{metric}` of setting `share_labels` must be a mapping or set to `true`')
+
+            for option_name in ('labels', 'match'):
+                if option_name not in config:
+                    continue
+
+                option = config[option_name]
+                if not isinstance(option, list):
+                    raise TypeError(
+                        f'Option `{option_name}` for metric `{metric}` of setting `share_labels` must be an array'
+                    )
+
+                for i, entry in enumerate(option, 1):
+                    if not isinstance(entry, str):
+                        raise TypeError(
+                            f'Entry #{i} of option `{option_name}` for metric `{metric}` '
+                            f'of setting `share_labels` must be a string'
+                        )
+
+                if option:
+                    data[option_name] = frozenset(option)
+
+        self.logger = check.log
+
+        self.label_sets = []
+
+        self.unconditional_labels = {}
+
+    def __call__(self, metrics):
+        # TODO: add new option to cache metrics until all configured ones are
+        # seen to avoid dependence on the order in which they are exposed
+        with self:
+            metric_config = self.metric_config.copy()
+
+            for metric in metrics:
+                if metric_config and metric.name in metric_config:
+                    self.collect(metric, metric_config.pop(metric.name))
+
+                yield metric
+
+    def collect(self, metric, config):
+        if 'match' in config:
+            matching_labels = config['match']
+            if 'labels' in config:
+                labels = config['labels']
+                for sample in metric.samples:
+                    label_set = set()
+                    shared_labels = {}
+
+                    for label, value in sample.labels.items():
+                        if label in matching_labels:
+                            label_set.add((label, value))
+
+                        if label in labels:
+                            shared_labels[label] = value
+
+                    self.label_sets.append((label_set, shared_labels))
+            else:
+                for sample in metric.samples:
+                    label_set = set()
+                    shared_labels = {}
+
+                    for label, value in sample.labels.items():
+                        if label in matching_labels:
+                            label_set.add((label, value))
+
+                        shared_labels[label] = value
+
+                    self.label_sets.append((label_set, shared_labels))
+        else:
+            if 'labels' in config:
+                labels = config['labels']
+                for sample in metric.samples:
+                    for label, value in sample.labels.items():
+                        if label in labels:
+                            self.unconditional_labels[label] = value
+            else:
+                for sample in metric.samples:
+                    for label, value in sample.labels.items():
+                        self.unconditional_labels[label] = value
+
+    def populate(self, labels):
+        label_set = set(labels.items())
+        labels.update(self.unconditional_labels)
+
+        for matching_label_set, shared_labels in self.label_sets:
+            if matching_label_set.issubset(label_set):
+                labels.update(shared_labels)
+
+    @property
+    def configured(self):
+        return self.populate is not no_op
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.label_sets.clear()
+        self.unconditional_labels.clear()
+
+
+def canonicalize_numeric_label(label):
+    # Prevent 0.0, see:
+    # https://github.com/OpenObservability/OpenMetrics/blob/master/specification/OpenMetrics.md#considerations-canonical-numbers
+    return float(label) or 0
+
+
+def normalize_labels_histogram(labels):
+    upper_bound = labels.get('le')
+    if upper_bound is not None:
+        labels['le'] = str(canonicalize_numeric_label(upper_bound))
+
+
+def normalize_labels_summary(labels):
+    quantile = labels.get('quantile')
+    if quantile is not None:
+        labels['quantile'] = str(canonicalize_numeric_label(quantile))
+
+
+def get_label_normalizer(metric_type):
+    if metric_type == 'histogram':
+        return normalize_labels_histogram
+    elif metric_type == 'summary':
+        return normalize_labels_summary
+    else:
+        return no_op