Emit telemetry when OpenMetrics reaches its metric limit

datadog_checks_base/changelog.d/24577.added

@@ -0,0 +1 @@
+Add best-effort OpenMetrics metric-limit Agent telemetry when ``max_returned_metrics`` is reached.

datadog_checks_base/datadog_checks/base/checks/base.py

@@ -171,6 +171,7 @@ class except the `check` method but sometimes it might be useful for a Check to
     # be sent to the aggregator, the rest are dropped. The state is reset after each run.
     # See https://github.com/DataDog/integrations-core/pull/2093 for more information.
     DEFAULT_METRIC_LIMIT = 0
+    OPENMETRICS_MAX_RETURNED_METRICS_TELEMETRY = False
 
     # Allow tracing for classic integrations
     def __init_subclass__(cls, *args, **kwargs):
@@ -391,10 +392,30 @@ def _get_metric_limiter(self, name, instance=None):
             # See Performance Optimizations in this package's README.md.
             from datadog_checks.base.utils.limiter import Limiter
 
-            return Limiter(name, 'metrics', limit, self.warning)
+            on_limit_reached = None
+            if self.OPENMETRICS_MAX_RETURNED_METRICS_TELEMETRY:
+                on_limit_reached = self._emit_openmetrics_max_returned_metrics_reached
+
+            return Limiter(name, 'metrics', limit, self.warning, on_limit_reached_func=on_limit_reached)
 
         return None
 
+    def _emit_openmetrics_max_returned_metrics_reached(self):
+        emit_agent_telemetry_with_labels = getattr(datadog_agent, 'emit_agent_telemetry_with_labels', None)
+        if not callable(emit_agent_telemetry_with_labels):
+            return
+
+        try:
+            emit_agent_telemetry_with_labels(
+                "openmetrics",
+                "max_returned_metrics_reached",
+                1,
+                "counter",
+                {"check_name": self.name},
+            )
+        except Exception:
+            self.log.debug("Unable to emit OpenMetrics max_returned_metrics Agent telemetry", exc_info=True)
+
     def _get_metric_limit(self, instance=None):
         # type: (InstanceType) -> int
         if instance is None:
@@ -1617,8 +1638,13 @@ def run(self):
                     self.metric_limiter.reset()
 
                     tags = self.get_debug_metric_tags()
-                    for metric_name, value in debug_metrics:
-                        self.gauge(metric_name, value, tags=tags, raw=True)
+                    on_limit_reached = self.metric_limiter.on_limit_reached
+                    self.metric_limiter.on_limit_reached = None
+                    try:
+                        for metric_name, value in debug_metrics:
+                            self.gauge(metric_name, value, tags=tags, raw=True)
+                    finally:
+                        self.metric_limiter.on_limit_reached = on_limit_reached
 
                 self.metric_limiter.reset()
 

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

@@ -56,6 +56,7 @@ class OpenMetricsBaseCheck(OpenMetricsScraperMixin, AgentCheck):
     """
 
     DEFAULT_METRIC_LIMIT = 2000
+    OPENMETRICS_MAX_RETURNED_METRICS_TELEMETRY = True
 
     HTTP_CONFIG_REMAPPER = {
         'ssl_verify': {'name': 'tls_verify'},

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

@@ -41,6 +41,7 @@ class OpenMetricsBaseCheckV2(AgentCheck):
     """
 
     DEFAULT_METRIC_LIMIT = 2000
+    OPENMETRICS_MAX_RETURNED_METRICS_TELEMETRY = True
 
     METRICS_MAP: tuple[MetricsMapping, ...] = ()
     """YAML files with metric name mappings to load automatically.

datadog_checks_base/datadog_checks/base/utils/limiter.py

@@ -11,14 +11,16 @@ class Limiter(object):
     that can be set by an instance.
     """
 
-    def __init__(self, check_name, object_name, object_limit, warning_func=None):
+    def __init__(self, check_name, object_name, object_limit, warning_func=None, on_limit_reached_func=None):
         """
         :param check_name: name of the check using this limiter
         :param object_name: (plural) name of counted objects for warning wording
         :param object_limit: maximum number of objects to accept before limiting
         :param warning_func: callback function, called with a string when limit is exceeded
+        :param on_limit_reached_func: callback function, called when limit is first exceeded
         """
         self.warning = warning_func
+        self.on_limit_reached = on_limit_reached_func
         self.name = object_name
         self.limit = object_limit
         self.check_name = check_name
@@ -70,6 +72,8 @@ def is_reached(self, uid=None):
                     "Check %s exceeded limit of %s %s, ignoring next ones", self.check_name, self.limit, self.name
                 )
             self.reached_limit = True
+            if self.on_limit_reached:
+                self.on_limit_reached()
             return True
         return False
 

datadog_checks_base/tests/base/checks/test_agent_check.py

@@ -1162,6 +1162,10 @@ def check(self, _):
             self.gauge('foo', i)
 
 
+class OpenMetricsTelemetryCheck(LimitedCheck):
+    OPENMETRICS_MAX_RETURNED_METRICS_TELEMETRY = True
+
+
 class TestLimits:
     def test_context_uid(self, aggregator):
         check = LimitedCheck()
@@ -1208,6 +1212,141 @@ def test_metric_limit_count(self, aggregator):
         assert len(check.get_warnings()) == 1
         assert len(aggregator.metrics("metric")) == 29
 
+    def test_openmetrics_metric_limit_telemetry(self, aggregator, monkeypatch):
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [{'max_returned_metrics': 2}])
+
+        check.gauge('metric', 0)
+        check.gauge('metric', 0)
+        emit_agent_telemetry.assert_not_called()
+        assert len(aggregator.metrics('metric')) == 2
+
+        check.gauge('metric', 0)
+        check.gauge('metric', 0)
+
+        emit_agent_telemetry.assert_called_once_with(
+            'openmetrics',
+            'max_returned_metrics_reached',
+            1,
+            'counter',
+            {'check_name': 'canonical_openmetrics'},
+        )
+        assert len(check.get_warnings()) == 1
+        assert len(aggregator.metrics('metric')) == 2
+
+    def test_openmetrics_metric_limit_telemetry_aggregates_same_check_name(self, monkeypatch):
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        first_check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [{'max_returned_metrics': 1}])
+        second_check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [{'max_returned_metrics': 1}])
+
+        first_check.gauge('metric', 0)
+        first_check.gauge('metric', 0)
+        second_check.gauge('metric', 0)
+        second_check.gauge('metric', 0)
+
+        assert emit_agent_telemetry.call_args_list == [
+            mock.call(
+                'openmetrics', 'max_returned_metrics_reached', 1, 'counter', {'check_name': 'canonical_openmetrics'}
+            ),
+            mock.call(
+                'openmetrics', 'max_returned_metrics_reached', 1, 'counter', {'check_name': 'canonical_openmetrics'}
+            ),
+        ]
+
+    def test_openmetrics_metric_limit_telemetry_uid_dedup(self, aggregator, monkeypatch):
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [{'max_returned_metrics': 2}])
+
+        for _ in range(5):
+            check.count('metric', 0, hostname='same-host')
+        assert len(aggregator.metrics('metric')) == 5
+        emit_agent_telemetry.assert_not_called()
+
+        check.count('metric', 0, hostname='second-host')
+        check.count('metric', 0, hostname='third-host')
+        check.count('metric', 0, hostname='fourth-host')
+
+        emit_agent_telemetry.assert_called_once_with(
+            'openmetrics',
+            'max_returned_metrics_reached',
+            1,
+            'counter',
+            {'check_name': 'canonical_openmetrics'},
+        )
+        assert len(check.get_warnings()) == 1
+        assert len(aggregator.metrics('metric')) == 6
+
+    def test_metric_limit_telemetry_skips_non_openmetrics(self, aggregator, monkeypatch):
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        check = LimitedCheck('generic_check', {}, [{'max_returned_metrics': 1}])
+
+        check.gauge('metric', 0)
+        check.gauge('metric', 0)
+
+        emit_agent_telemetry.assert_not_called()
+        assert len(check.get_warnings()) == 1
+        assert len(aggregator.metrics('metric')) == 1
+
+    def test_openmetrics_metric_limit_telemetry_skips_without_labeled_bridge(self, aggregator, monkeypatch):
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.delattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels', raising=False
+        )
+        monkeypatch.setattr('datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry', emit_agent_telemetry)
+        check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [{'max_returned_metrics': 1}])
+
+        check.gauge('metric', 0)
+        check.gauge('metric', 0)
+
+        emit_agent_telemetry.assert_not_called()
+        assert len(check.get_warnings()) == 1
+        assert len(aggregator.metrics('metric')) == 1
+
+    def test_openmetrics_metric_limit_telemetry_bridge_failure(self, aggregator, caplog, monkeypatch):
+        emit_agent_telemetry = mock.MagicMock(side_effect=Exception('bridge failed'))
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        caplog.set_level(logging.DEBUG)
+        check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [{'max_returned_metrics': 1}])
+
+        check.gauge('metric', 0)
+        check.gauge('metric', 0)
+        check.gauge('metric', 0)
+
+        emit_agent_telemetry.assert_called_once_with(
+            'openmetrics',
+            'max_returned_metrics_reached',
+            1,
+            'counter',
+            {'check_name': 'canonical_openmetrics'},
+        )
+        assert len(check.get_warnings()) == 1
+        assert len(aggregator.metrics('metric')) == 1
+        assert 'Unable to emit OpenMetrics max_returned_metrics Agent telemetry' in caplog.text
+
     def test_metric_limit_instance_config(self, aggregator):
         instances = [{"max_returned_metrics": 42}]
         check = AgentCheck("test", {}, instances)
@@ -1294,6 +1433,114 @@ def test_debug_metrics_over_limit(self, aggregator, dd_run_check):
         aggregator.assert_metric('datadog.agent.metrics.contexts.limit', 3)
         aggregator.assert_metric('datadog.agent.metrics.contexts.total', 5)
 
+    @pytest.mark.parametrize('debug_metrics', [False, True])
+    def test_openmetrics_metric_limit_telemetry_once_per_run(
+        self, aggregator, dd_run_check, debug_metrics, monkeypatch
+    ):
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        instance = {'max_returned_metrics': 3}
+        if debug_metrics:
+            instance['debug_metrics'] = {'metric_contexts': True}
+        check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [instance])
+
+        dd_run_check(check)
+        dd_run_check(check)
+
+        assert emit_agent_telemetry.call_args_list == [
+            mock.call(
+                'openmetrics', 'max_returned_metrics_reached', 1, 'counter', {'check_name': 'canonical_openmetrics'}
+            ),
+            mock.call(
+                'openmetrics', 'max_returned_metrics_reached', 1, 'counter', {'check_name': 'canonical_openmetrics'}
+            ),
+        ]
+        assert len(aggregator.metrics('foo')) == 6
+        if debug_metrics:
+            aggregator.assert_metric('datadog.agent.metrics.contexts.limit', 3)
+            aggregator.assert_metric('datadog.agent.metrics.contexts.total', 5)
+
+    def test_openmetrics_metric_limit_telemetry_not_emitted_for_debug_metrics(
+        self, aggregator, dd_run_check, monkeypatch
+    ):
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        instance = {'debug_metrics': {'metric_contexts': True}, 'max_returned_metrics': 1}
+        check = OpenMetricsTelemetryCheck('canonical_openmetrics', {}, [instance])
+
+        dd_run_check(check)
+
+        emit_agent_telemetry.assert_called_once_with(
+            'openmetrics',
+            'max_returned_metrics_reached',
+            1,
+            'counter',
+            {'check_name': 'canonical_openmetrics'},
+        )
+        assert len(aggregator.metrics('foo')) == 1
+        aggregator.assert_metric('datadog.agent.metrics.contexts.limit', 1)
+
+    @pytest.mark.parametrize(
+        'openmetrics_version, instance',
+        [
+            pytest.param(
+                'v1',
+                {
+                    'prometheus_url': 'http://example.com/metrics',
+                    'namespace': 'test',
+                    'metrics': [],
+                    'max_returned_metrics': 1,
+                },
+                id='v1',
+            ),
+            pytest.param(
+                'v2',
+                {
+                    'openmetrics_endpoint': 'http://example.com/metrics',
+                    'namespace': 'test',
+                    'metrics': [],
+                    'max_returned_metrics': 1,
+                },
+                id='v2',
+            ),
+        ],
+    )
+    def test_openmetrics_base_classes_opt_in_to_metric_limit_telemetry(
+        self, aggregator, openmetrics_version, instance, monkeypatch
+    ):
+        from datadog_checks.base.checks.openmetrics.base_check import OpenMetricsBaseCheck
+        from datadog_checks.base.checks.openmetrics.v2.base import OpenMetricsBaseCheckV2
+
+        check_class = OpenMetricsBaseCheck if openmetrics_version == 'v1' else OpenMetricsBaseCheckV2
+        emit_agent_telemetry = mock.MagicMock()
+        monkeypatch.setattr(
+            'datadog_checks.base.checks.base.datadog_agent.emit_agent_telemetry_with_labels',
+            emit_agent_telemetry,
+            raising=False,
+        )
+        check = check_class('canonical_openmetrics', {}, [instance])
+
+        check.gauge('metric', 0)
+        check.gauge('metric', 0)
+
+        emit_agent_telemetry.assert_called_once_with(
+            'openmetrics',
+            'max_returned_metrics_reached',
+            1,
+            'counter',
+            {'check_name': 'canonical_openmetrics'},
+        )
+        assert len(check.get_warnings()) == 1
+        assert len(aggregator.metrics('metric')) == 1
+
 
 class TestCheckInitializations:
     def test_success_only_once(self):