Agent Health issue for OpenMetrics metrics dropped by configured limit

cilium/tests/test_unit.py

@@ -91,6 +91,30 @@ def test_v2_both_endpoints_build_two_scrapers():
     assert endpoints == ["http://agent/metrics", "http://op/metrics"]
 
 
+def test_v2_metric_limit_issue_uses_generated_scraper_endpoints(datadog_agent):
+    agent_endpoint = "http://z-agent/metrics"
+    operator_endpoint = "http://a-operator/metrics"
+    check = CiliumCheckV2(
+        "cilium",
+        {},
+        [{"agent_endpoint": agent_endpoint, "operator_endpoint": operator_endpoint}],
+    )
+    check._parse_config()
+    check.configure_scrapers()
+
+    check._on_metric_limit_state(True, 20, 5)
+
+    [issue] = datadog_agent._sent_reported_issues["cilium"]
+    assert "openmetrics_endpoint" not in check.instance
+    assert issue["id"] == "openmetrics-dropped-config:659a0d48b0fa9caf"
+    assert issue["title"] == "Dropping 15 of 20 OpenMetrics metrics"
+    assert issue["extra"]["endpoints"] == [operator_endpoint, agent_endpoint]
+    assert "2 configured endpoints" in issue["description"]
+    assert "totals cover the complete check run" in issue["description"]
+    assert agent_endpoint in issue["description"]
+    assert operator_endpoint in issue["description"]
+
+
 def test_v2_agent_scraper_uses_agent_metrics_not_operator():
     check = CiliumCheckV2("cilium", {}, [{"agent_endpoint": "http://agent/metrics"}])
     check._parse_config()

datadog_checks_base/changelog.d/24819.added

@@ -0,0 +1 @@
+Add an Agent Health issue when OpenMetrics checks drop metrics at the configured limit.

datadog_checks_base/datadog_checks/base/checks/base.py

@@ -1661,6 +1661,15 @@ def run(self):
                 else:
                     self.check(instance)
 
+                if self.metric_limiter:
+                    try:
+                        reached_limit = self.metric_limiter.reached_limit
+                        observed_count = self.metric_limiter.count
+                        limit = self.metric_limiter.limit
+                        self._on_metric_limit_state(reached_limit, observed_count, limit)
+                    except Exception:
+                        self.log.debug('Error handling metric limit state', exc_info=True)
+
             error_report = ''
         except Exception as e:
             message = self.sanitize(str(e))
@@ -1682,6 +1691,10 @@ def run(self):
 
         return error_report
 
+    def _on_metric_limit_state(self, reached_limit: bool, observed_count: int, limit: int) -> None:
+        """Called once per run for checks with an active metric limiter."""
+        pass
+
     def run_check_initializations(self):
         while self.check_initializations:
             initialization = self.check_initializations.popleft()
@@ -1877,6 +1890,8 @@ def load_config(yaml_str: str) -> Any:
     # Remediation *Remediation `protobuf:"bytes,11,opt,name=remediation,proto3" json:"remediation,omitempty"`
     # // Tags are additional labels for the issue
     # Tags []string `protobuf:"bytes,12,rep,name=tags,proto3" json:"tags,omitempty"`
+    # // IssueType snake_case version of issue name
+    # IssueType string `protobuf:"bytes,14,opt,name=issue_type,json=issueType,proto3" json:"issue_type,omitempty"`
 
     # Remediation should be a dict with the following keys:
     # - summary: str
@@ -1891,6 +1906,7 @@ def report_issue(
         self,
         id: str,
         issue_name: str,
+        issue_type: str,
         title: str = None,
         description: str = None,
         category: str = None,
@@ -1899,14 +1915,17 @@ def report_issue(
         remediation: dict = None,
         tags: list = None,
     ):
-        # Issue ID and Name are required
+        # Issue ID, Name, and Type are required
         if not id:
             raise ValueError("Issue ID is required")
         if not issue_name:
             raise ValueError("Issue Name is required")
+        if not issue_type:
+            raise ValueError("Issue Type is required")
         issue = {
             'id': id,
             'issue_name': issue_name,
+            'issue_type': issue_type,
             'title': title,
             'description': description,
             'category': category,

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

@@ -9,6 +9,7 @@
 from datadog_checks.base.errors import CheckException
 from datadog_checks.base.utils.tracing import traced_class
 
+from .metric_limit_issue import MetricLimitIssueReporter
 from .mixins import OpenMetricsScraperMixin
 
 STANDARD_FIELDS = [
@@ -88,6 +89,9 @@ def __init__(self, *args, **kwargs):
             default_namespace = legacy_kwargs_in_args[1]
 
         super(OpenMetricsBaseCheck, self).__init__(*args, **kwargs)
+        self.metric_limit_issue_reporter: MetricLimitIssueReporter = MetricLimitIssueReporter(
+            filter_option_text='metrics / ignore_metrics'
+        )
         self.config_map = {}
         self._http_handlers = {}
         self.default_instances = default_instances
@@ -141,6 +145,15 @@ def check(self, instance):
 
         self.process(scraper_config)
 
+    def _on_metric_limit_state(self, reached_limit: bool, observed_count: int, limit: int) -> None:
+        self.metric_limit_issue_reporter.handle(
+            self,
+            (self.instance.get('prometheus_url'),),
+            reached_limit,
+            observed_count,
+            limit,
+        )
+
     def get_scraper_config(self, instance):
         """
         Validates the instance configuration and creates a scraper configuration for a new instance.

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

@@ -0,0 +1,141 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from datadog_checks.base.checks import AgentCheck
+
+SEVERITY_HIGH_RATIO = 0.25
+SEVERITY_MEDIUM_RATIO = 0.05
+
+ISSUE_NAME = 'OpenMetrics Metrics Dropped By Configured Limit'
+ISSUE_TYPE = 'openmetrics_metrics_dropped_by_configured_limit'
+
+
+@dataclass
+class MetricLimitIssueReporter:
+    filter_option_text: str
+
+    def handle(
+        self,
+        check: AgentCheck,
+        endpoints: Iterable[str | None] | None,
+        reached_limit: bool,
+        observed_count: int,
+        limit: int,
+    ) -> None:
+        """Report or resolve one configured-limit issue for an OpenMetrics check run."""
+        endpoints = _normalize_endpoints(endpoints or ())
+        if not endpoints:
+            check.log.debug('Cannot handle the OpenMetrics metric limit state without an endpoint')
+            return
+
+        issue_id = _issue_id(check.hostname, check.name, endpoints, check.instance.get('namespace', ''))
+
+        if reached_limit:
+            dropped = max(0, observed_count - limit)
+            ratio = dropped / observed_count
+            title, description = _issue_text(check.name, endpoints, limit, observed_count, dropped)
+            check.report_issue(
+                id=issue_id,
+                issue_name=ISSUE_NAME,
+                issue_type=ISSUE_TYPE,
+                title=title,
+                description=description,
+                category='integration',
+                severity=_severity(check, ratio),
+                extra={
+                    'check_name': check.name,
+                    'endpoints': list(endpoints),
+                    'effective_limit': limit,
+                    'observed_contexts': observed_count,
+                    'dropped_contexts': dropped,
+                    'dropped_ratio': round(ratio, 4),
+                    'limit_is_default': limit == check.DEFAULT_METRIC_LIMIT,
+                },
+                remediation=_remediation(filter_option_text=self.filter_option_text),
+                tags=[f'integration:{check.name}', 'openmetrics', 'metric-limit'],
+            )
+            return
+
+        check.resolve_issue(issue_id)
+
+
+def _normalize_endpoints(endpoints: Iterable[str | None]) -> tuple[str, ...]:
+    return tuple(sorted({endpoint for endpoint in endpoints if endpoint}))
+
+
+def _issue_id(hostname: str, check_name: str, endpoints: tuple[str, ...], namespace: object) -> str:
+    # Keep the original identity for one endpoint while representing multiple endpoints structurally.
+    endpoint_identity: str | list[str] = endpoints[0] if len(endpoints) == 1 else list(endpoints)
+    identity = json.dumps((hostname, check_name, endpoint_identity, str(namespace)), separators=(',', ':'))
+    digest = hashlib.sha256(identity.encode('utf-8')).hexdigest()[:16]
+    return f'openmetrics-dropped-config:{digest}'
+
+
+def _issue_text(
+    check_name: str, endpoints: tuple[str, ...], limit: int, observed_count: int, dropped: int
+) -> tuple[str, str]:
+    endpoint_count = len(endpoints)
+    endpoint_noun = 'endpoint' if endpoint_count == 1 else 'endpoints'
+    title = f'Dropping {dropped} of {observed_count} OpenMetrics metrics'
+    description = (
+        f'The {check_name} check collected {observed_count} metric contexts from {endpoint_count} configured '
+        f'{endpoint_noun}: {", ".join(endpoints)}. These totals cover the complete check run. The check is configured '
+        f'to submit at most {limit} metric contexts per run, so the Agent submitted {limit} and discarded the '
+        f'remaining {dropped}.'
+    )
+    return title, description
+
+
+def _severity(check: AgentCheck, ratio: float) -> int:
+    if ratio >= SEVERITY_HIGH_RATIO:
+        return check.IssueSeverity['HIGH']
+    if ratio >= SEVERITY_MEDIUM_RATIO:
+        return check.IssueSeverity['MEDIUM']
+    return check.IssueSeverity['LOW']
+
+
+def _remediation(*, filter_option_text: str) -> dict[str, str | list[dict[str, int | str]]]:
+    return {
+        'summary': (
+            "Reduce what this endpoint sends to Datadog, or raise this instance's metric limit after checking the cost."
+        ),
+        'steps': [
+            {
+                'order': 1,
+                'text': (
+                    f'Decide what you actually need. Use {filter_option_text} on this instance to stop collecting '
+                    'series you do not query, alert on, or keep.'
+                ),
+            },
+            {
+                'order': 2,
+                'text': 'Only then raise max_returned_metrics on this instance to a value above the observed count.',
+            },
+            {
+                'order': 3,
+                'text': (
+                    'Verify: on the instance, set metric_contexts to true under the debug_metrics section. '
+                    'This publishes datadog.agent.metrics.contexts.total and '
+                    'datadog.agent.metrics.contexts.limit; confirm the total stays below the limit at peak. '
+                    'Consider a monitor at 80% of the limit.'
+                ),
+            },
+            {
+                'order': 4,
+                'text': (
+                    'Check the cost before you leave it: additional contexts are billable custom metrics and increase '
+                    'Agent memory.'
+                ),
+            },
+        ],
+    }

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

@@ -12,6 +12,7 @@
 from requests.exceptions import RequestException
 
 from datadog_checks.base.checks import AgentCheck
+from datadog_checks.base.checks.openmetrics.metric_limit_issue import MetricLimitIssueReporter
 from datadog_checks.base.errors import ConfigurationError
 from datadog_checks.base.utils.tracing import traced_class
 
@@ -63,6 +64,9 @@ def __init__(self, name, init_config, instances):
         When overriding, make sure to call this (the parent's) __init__ first!
         """
         super(OpenMetricsBaseCheckV2, self).__init__(name, init_config, instances)
+        self.metric_limit_issue_reporter: MetricLimitIssueReporter = MetricLimitIssueReporter(
+            filter_option_text='metrics / exclude_metrics'
+        )
 
         # All desired scraper configurations, which subclasses can override as needed
         self.scraper_configs = [self.instance]
@@ -96,6 +100,18 @@ def check(self, _):
                     self.log.error("There was an error scraping endpoint %s: %s", endpoint, str(e))
                     raise type(e)("There was an error scraping endpoint {}: {}".format(endpoint, e)) from None
 
+    def _on_metric_limit_state(self, reached_limit: bool, observed_count: int, limit: int) -> None:
+        # Use the actual configured scraper endpoint keys rather than the raw instance field, so
+        # integrations that synthesize ``scraper_configs`` from options such as ``agent_endpoint``
+        # (e.g. Cilium) still report drops against the endpoint that was scraped.
+        self.metric_limit_issue_reporter.handle(
+            self,
+            self.scrapers.keys(),
+            reached_limit,
+            observed_count,
+            limit,
+        )
+
     def configure_scrapers(self):
         """
         Creates a scraper configuration for each instance.