Add configuration discovery infrastructure to datadog_checks_base.

Extends AgentCheck with discover_config / generate_configs class methods
for auto-detecting working configurations from service metadata. Adds
Discovery, Service, and Port data types to datadog_checks.base.utils.discovery,
a probe module (_suppress_discovery_side_effects, _try_discovery_candidate,
run_discovery) that trial-runs candidates in isolation, and a _package_name
helper in base.py to eliminate repeated module-string parsing.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: vitkyrka

datadog_checks_base/changelog.d/23963.added

@@ -0,0 +1 @@
+Add helpers for generating and validating discovered Autodiscovery configurations.

datadog_checks_base/datadog_checks/base/checks/base.py

@@ -10,6 +10,7 @@
 import os
 import re
 from collections import deque
+from collections.abc import Iterable
 from os.path import basename
 from pathlib import Path
 from typing import (
@@ -65,6 +66,7 @@
     import unicodedata as _module_unicodedata
 
     from datadog_checks.base.utils.diagnose import Diagnosis
+    from datadog_checks.base.utils.discovery import Service
     from datadog_checks.base.utils.http import RequestsWrapper
     from datadog_checks.base.utils.metadata import MetadataManager
 
@@ -179,6 +181,33 @@ def __init_subclass__(cls, *args, **kwargs):
         except Exception:
             return cls
 
+    @classmethod
+    def generate_configs(cls, service: Service) -> Iterable[dict[str, Any]]:
+        """
+        Yield candidate full configurations for service discovery.
+
+        Integrations can opt into config discovery by declaring a discovery
+        stanza in their spec and generating config_models.discovery.
+        """
+        from datadog_checks.base.utils.discovery.probe import generated_discovery_candidates
+
+        return generated_discovery_candidates(cls, service)
+
+    @classmethod
+    def discover_config(cls, service_json: str) -> str:
+        """
+        Return discovered configurations for an AD service payload.
+
+        The Agent calls this classmethod through rtloader. Candidate configs
+        are generated by ``generate_configs`` and accepted only when the real
+        check can run against their metric instances successfully. Returns the
+        first accepted candidate only (first-match-wins); remaining candidates
+        are not evaluated.
+        """
+        from datadog_checks.base.utils.discovery.probe import run_discovery
+
+        return run_discovery(cls, service_json)
+
     def __init__(self, *args, **kwargs):
         # type: (*Any, **Any) -> None
         """
@@ -524,9 +553,8 @@ def check_version(self):
         Return the dynamically detected integration version.
         """
         if not hasattr(self, '_check_version'):
-            # 'datadog_checks.<PACKAGE>.<MODULE>...'
-            module_parts = self.__module__.split('.')
-            package_path = '.'.join(module_parts[:2])
+            pkg = _package_name(self.__module__)
+            package_path = f'datadog_checks.{pkg}' if pkg else self.__module__
             package = importlib.import_module(package_path)
 
             # Provide a default just in case
@@ -545,8 +573,8 @@ def _get_package_dir(self) -> Path:
         used by :attr:`check_version`.
         """
         if not hasattr(self, '_package_dir'):
-            module_parts = self.__module__.split('.')
-            package_path = '.'.join(module_parts[:2])
+            pkg = _package_name(self.__module__)
+            package_path = f'datadog_checks.{pkg}' if pkg else self.__module__
             package = importlib.import_module(package_path)
             if package.__file__ is not None:
                 self._package_dir = Path(package.__file__).parent
@@ -613,9 +641,8 @@ def log_typos_in_options(self, user_config, models_config, level):
 
     def load_configuration_models(self, package_path=None):
         if package_path is None:
-            # 'datadog_checks.<PACKAGE>.<MODULE>...'
-            module_parts = self.__module__.split('.')
-            package_path = '{}.config_models'.format('.'.join(module_parts[:2]))
+            pkg = _package_name(self.__module__)
+            package_path = f'datadog_checks.{pkg}.config_models' if pkg else f'{self.__module__}.config_models'
         if self._config_model_shared is None:
             shared_config = copy.deepcopy(self.init_config)
             context = self._get_config_model_context(shared_config)
@@ -752,7 +779,7 @@ def submit_histogram_bucket(
         if hostname is None:
             hostname = ''
 
-        aggregator.submit_histogram_bucket(
+        self._agg().submit_histogram_bucket(
             self,
             self.check_id,
             self._format_namespace(name, raw),
@@ -770,28 +797,28 @@ def database_monitoring_query_sample(self, raw_event):
         if raw_event is None:
             return
 
-        aggregator.submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-samples")
+        self._agg().submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-samples")
 
     def database_monitoring_query_metrics(self, raw_event):
         # type: (str) -> None
         if raw_event is None:
             return
 
-        aggregator.submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-metrics")
+        self._agg().submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-metrics")
 
     def database_monitoring_query_activity(self, raw_event):
         # type: (str) -> None
         if raw_event is None:
             return
 
-        aggregator.submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-activity")
+        self._agg().submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-activity")
 
     def database_monitoring_metadata(self, raw_event):
         # type: (str) -> None
         if raw_event is None:
             return
 
-        aggregator.submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-metadata")
+        self._agg().submit_event_platform_event(self, self.check_id, to_native_string(raw_event), "dbm-metadata")
 
     def event_platform_event(self, raw_event, event_track_type):
         # type: (str | bytes, str) -> None
@@ -810,7 +837,7 @@ def event_platform_event(self, raw_event, event_track_type):
             raw_event = bytes(raw_event)
         elif not isinstance(raw_event, bytes):
             raw_event = to_native_string(raw_event)
-        aggregator.submit_event_platform_event(self, self.check_id, raw_event, event_track_type)
+        self._agg().submit_event_platform_event(self, self.check_id, raw_event, event_track_type)
 
     def submit_generic_resource(self, *, type, key, fields, include, seen_at=None, expire_at=None):
         # type: (str, str, dict | None, dict, int | None, int | None) -> None
@@ -974,6 +1001,10 @@ def _metric_excluded(self, metric_name):
 
         return self.exclude_metrics_pattern.search(metric_name) is not None
 
+    def _agg(self) -> Any:
+        """Return the active aggregator: proxy during a discovery probe, module singleton otherwise."""
+        return getattr(self, '_discovery_aggregator', None) or aggregator
+
     def _submit_metric(
         self, mtype, name, value, tags=None, hostname=None, device_name=None, raw=False, flush_first_value=False
     ):
@@ -1012,7 +1043,7 @@ def _submit_metric(
             self.warning(err_msg)
             return
 
-        aggregator.submit_metric(self, self.check_id, mtype, name, value, tags, hostname, flush_first_value)
+        self._agg().submit_metric(self, self.check_id, mtype, name, value, tags, hostname, flush_first_value)
 
     def gauge(self, name, value, tags=None, hostname=None, device_name=None, raw=False):
         # type: (str, float, Sequence[str], str, str, bool) -> None
@@ -1233,7 +1264,7 @@ def service_check(self, name, status, tags=None, hostname=None, message=None, ra
 
         message = self.sanitize(message)
 
-        aggregator.submit_service_check(
+        self._agg().submit_service_check(
             self, self.check_id, self._format_namespace(name, raw), status, tags, hostname, message
         )
 
@@ -1649,7 +1680,7 @@ def event(self, event):
         if self.__NAMESPACE__:
             event.setdefault('source_type_name', self.__NAMESPACE__)
 
-        aggregator.submit_event(self, self.check_id, event)
+        self._agg().submit_event(self, self.check_id, event)
 
     def _normalize_tags_type(self, tags, device_name=None, metric_name=None):
         # type: (Sequence[Union[None, str, bytes]], str, str) -> List[str]
@@ -1759,3 +1790,10 @@ def load_config(yaml_str: str) -> Any:
             raise ValueError(f'Failed to load config: {stderr.decode("utf-8", errors="replace")}')
 
         return _parse_ast_config(stdout.strip().decode('utf-8'))
+
+
+def _package_name(module: str) -> str | None:
+    parts = module.split('.')
+    if len(parts) >= 2 and parts[0] == 'datadog_checks':
+        return parts[1]
+    return None

datadog_checks_base/datadog_checks/base/utils/discovery/__init__.pyi

@@ -1,6 +1,6 @@
 # (C) Datadog, Inc. 2025-present
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
-from .discovery import Discovery
+from .discovery import Discovery, Port, Service, candidate_ports, from_ports
 
-__all__ = ['Discovery']
+__all__ = ['Discovery', 'Port', 'Service', 'candidate_ports', 'from_ports']

datadog_checks_base/datadog_checks/base/utils/discovery/discovery.py

@@ -1,10 +1,30 @@
 # (C) Datadog, Inc. 2023-present
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
+from collections.abc import Iterable, Iterator
+from dataclasses import dataclass, field
+
 from .cache import Cache
 from .filter import Filter
 
 
+@dataclass(frozen=True)
+class Port:
+    """An Autodiscovery-exposed port on a service."""
+
+    number: int
+    name: str = ""
+
+
+@dataclass(frozen=True)
+class Service:
+    """An Autodiscovery-discovered service instance."""
+
+    id: str
+    host: str
+    ports: tuple[Port, ...] = field(default_factory=tuple)
+
+
 class Discovery:
     def __init__(
         self,
@@ -21,3 +41,25 @@ def __init__(
     def get_items(self):
         items = self._cache.get_items()
         return self._filter.get_items(items)
+
+
+def candidate_ports(service: Service, hints: Iterable[int]) -> Iterator[Port]:
+    """Yield hinted ports first, then remaining service ports."""
+    by_number = {port.number: port for port in service.ports}
+    seen: set[int] = set()
+
+    for hint in hints:
+        if hint in by_number and hint not in seen:
+            seen.add(hint)
+            yield by_number[hint]
+
+    for port in service.ports:
+        if port.number not in seen:
+            seen.add(port.number)
+            yield port
+
+
+def from_ports(service: Service, *, port_hints: Iterable[int]) -> Iterator[dict[str, Port]]:
+    """Yield a ``{'port': Port}`` render context per candidate port."""
+    for port in candidate_ports(service, port_hints):
+        yield {'port': port}