Add configuration discovery infrastructure to datadog_checks_base (#23963)

* 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>

* walrus

* comment

* Use pydantic BaseModel for Port and Service

Simplifies _parse_service_json: the Agent always sends well-typed JSON
(int number, string name), so manual field extraction and to_native_string
coercion are unnecessary — model_validate handles all of it.

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

* validate

* Remove from_ports from base; add discovery_strategy decorator

Move from_ports strategy logic to the dev-side codegen registry (Phase 2).
Base retains only the runtime primitives: Service, Port, candidate_ports,
and the probing harness. Add the discovery_strategy decorator for custom
(local:) strategies that run arbitrary integration code at Agent runtime.

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

* datadog_checks_base: match changelog text to phase 1 spec

Use the wording prescribed by the phase 1 plan: the Phase 1 changelog
describes the discovery runtime (Service/Port types, candidate_ports,
probing harness, discovery entry points), not the codegen/validation
tooling, which belongs to Phase 2.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Fix test mock to patch import_module where it is called

The test patched datadog_checks.base.checks.base.importlib.import_module,
but generate_configs() delegates to the probe module, which calls
probe.importlib.import_module. Patch that path instead so the mock is
actually exercised.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Remove __discovery_provides__ attribute and dead-code tests

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

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: vitkyrka

datadog_checks_base/changelog.d/23963.added

@@ -0,0 +1 @@
+Add configuration discovery runtime (Service/Port types, candidate_ports, probing harness, and discovery entry points).

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
         """
@@ -752,7 +781,7 @@ def submit_histogram_bucket(
         if hostname is None:
             hostname = ''
 
-        aggregator.submit_histogram_bucket(
+        self._aggregator().submit_histogram_bucket(
             self,
             self.check_id,
             self._format_namespace(name, raw),
@@ -770,28 +799,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._aggregator().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._aggregator().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._aggregator().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._aggregator().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 +839,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._aggregator().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 +1003,10 @@ def _metric_excluded(self, metric_name):
 
         return self.exclude_metrics_pattern.search(metric_name) is not None
 
+    def _aggregator(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 +1045,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._aggregator().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 +1266,7 @@ def service_check(self, name, status, tags=None, hostname=None, message=None, ra
 
         message = self.sanitize(message)
 
-        aggregator.submit_service_check(
+        self._aggregator().submit_service_check(
             self, self.check_id, self._format_namespace(name, raw), status, tags, hostname, message
         )
 
@@ -1649,7 +1682,7 @@ def event(self, event):
         if self.__NAMESPACE__:
             event.setdefault('source_type_name', self.__NAMESPACE__)
 
-        aggregator.submit_event(self, self.check_id, event)
+        self._aggregator().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]

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

@@ -1,6 +1,7 @@
 # (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 .strategies import discovery_strategy
 
-__all__ = ['Discovery']
+__all__ = ['Discovery', 'Port', 'Service', 'candidate_ports', 'discovery_strategy']

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

@@ -1,10 +1,33 @@
 # (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 pydantic import BaseModel, ConfigDict
+
 from .cache import Cache
 from .filter import Filter
 
 
+class Port(BaseModel):
+    """An Autodiscovery-exposed port on a service."""
+
+    model_config = ConfigDict(frozen=True)
+
+    number: int
+    name: str = ""
+
+
+class Service(BaseModel):
+    """An Autodiscovery-discovered service instance."""
+
+    model_config = ConfigDict(frozen=True)
+
+    id: str
+    host: str
+    ports: tuple[Port, ...] = ()
+
+
 class Discovery:
     def __init__(
         self,
@@ -21,3 +44,19 @@ 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