[DBMON-6785] add support for database_identifier templating to DatabaseCheck (#24250)

* Implement agent_hostname in DatabaseCheck

* Add changelog

* add support for database_identifier templating to DatabaseCheck

* Add changelog

Author: eric-weaver

datadog_checks_base/changelog.d/24250.added

@@ -0,0 +1 @@
+Provide a default `database_identifier` implementation on the `DatabaseCheck` base class that is built (and cached) from the `database_identifier_template` and `database_identifier_params` hooks, so integrations no longer need to reimplement the database identifier templating logic.

datadog_checks_base/datadog_checks/base/checks/db.py

@@ -3,6 +3,7 @@
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
 from abc import abstractmethod
+from string import Template
 
 from datadog_checks.base.agent import datadog_agent
 from datadog_checks.base.utils.db.utils import TagManager
@@ -14,6 +15,7 @@ class DatabaseCheck(AgentCheck):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._agent_hostname = None
+        self._database_identifier = None
         self.tag_manager = TagManager()
 
     def database_monitoring_query_sample(self, raw_event: str):
@@ -37,9 +39,68 @@ def reported_hostname(self) -> str | None:
         pass
 
     @property
-    @abstractmethod
     def database_identifier(self) -> str:
-        pass
+        """
+        The unique identifier for this database instance, used as the ``database_instance`` tag and
+        in DBM metadata payloads.
+
+        The value is built once (and cached) from :attr:`database_identifier_template` and
+        :attr:`database_identifier_params` via :meth:`_build_database_identifier`. Integrations
+        customize the result by overriding those two hooks instead of reimplementing the templating
+        logic.
+        """
+        if self._database_identifier is None:
+            self._database_identifier = self._build_database_identifier(
+                self.database_identifier_template,
+                self.database_identifier_params,
+            )
+        return self._database_identifier
+
+    @property
+    def database_identifier_template(self) -> str:
+        """
+        The ``string.Template``-style template used to build :attr:`database_identifier`.
+
+        Defaults to ``"$resolved_hostname"``. Integrations typically override this to return the
+        template from their configuration (e.g. ``self._config.database_identifier.template``).
+        """
+        return "$resolved_hostname"
+
+    @property
+    def database_identifier_params(self) -> dict:
+        """
+        Connection-derived values exposed to :attr:`database_identifier_template`.
+
+        These are applied after tags, so they take precedence over any tag of the same name.
+        Values are stringified by the template engine, so callers need not cast them. Defaults to an
+        empty mapping.
+        """
+        return {}
+
+    def _build_database_identifier(self, template: str, connection_params: dict | None = None) -> str:
+        """
+        Build a database identifier string from a template and the check's tags.
+
+        Each ``key:value`` tag is exposed to the template as a ``$key`` variable, with duplicate
+        keys joined by commas (after sorting tags for a stable ordering). The ``connection_params``
+        mapping is applied last so connection-derived values (e.g. ``resolved_hostname``, ``host``,
+        ``port``) take precedence over any tag of the same name.
+
+        :param template: A ``string.Template``-style template, e.g. ``"$resolved_hostname"``.
+        :param connection_params: Optional mapping of additional template variables.
+        :return: The substituted identifier. Unknown ``$variables`` are left intact.
+        """
+        tag_dict: dict[str, str] = {}
+        for tag in sorted(self.tags):
+            if ':' in tag:
+                key, value = tag.split(':', 1)
+                if key in tag_dict:
+                    tag_dict[key] += f",{value}"
+                else:
+                    tag_dict[key] = value
+        if connection_params:
+            tag_dict.update(connection_params)
+        return Template(template).safe_substitute(**tag_dict)
 
     @property
     def agent_hostname(self) -> str:

datadog_checks_base/tests/base/checks/test_database_check.py

@@ -3,6 +3,8 @@
 # Licensed under a 3-clause BSD style license (see LICENSE)
 from unittest import mock
 
+import pytest
+
 from datadog_checks.base.checks.db import DatabaseCheck
 from datadog_checks.base.stubs.datadog_agent import datadog_agent
 
@@ -12,18 +14,10 @@ class FakeDatabaseCheck(DatabaseCheck):
     def reported_hostname(self):
         return None
 
-    @property
-    def database_identifier(self):
-        return "test-db"
-
     @property
     def dbms_version(self):
         return "1.0.0"
 
-    @property
-    def tags(self):
-        return []
-
     @property
     def cloud_metadata(self):
         return {}
@@ -36,3 +30,62 @@ def test_agent_hostname_resolves_once_and_caches():
         assert check.agent_hostname == "my-agent-host"
         assert check.agent_hostname == "my-agent-host"
         assert get_hostname.call_count == 1
+
+
+@pytest.mark.parametrize(
+    ("tags", "template", "connection_params", "expected"),
+    [
+        pytest.param([], "$resolved_hostname", {"resolved_hostname": "my-host"}, "my-host", id="connection_params"),
+        pytest.param(["env:prod"], "$env", None, "prod", id="substitutes_tags"),
+        # Tags are sorted before merging, so the result is deterministic regardless of input order.
+        pytest.param(["team:b", "team:a"], "$team", None, "a,b", id="merges_duplicate_keys_sorted"),
+        pytest.param(["host:tag-host"], "$host", {"host": "conn-host"}, "conn-host", id="connection_params_override"),
+        # Tags without a ':' are not exposed as template variables.
+        pytest.param(["keyless"], "$keyless", None, "$keyless", id="ignores_keyless_tags"),
+        pytest.param([], "$missing", None, "$missing", id="unknown_variables_intact"),
+    ],
+)
+def test_build_database_identifier(tags, template, connection_params, expected):
+    check = FakeDatabaseCheck("test", {}, [{}])
+    check.tag_manager.set_tags_from_list(tags)
+    assert check._build_database_identifier(template, connection_params) == expected
+
+
+@pytest.mark.parametrize(
+    ("template", "params", "tags", "tags_after", "expected"),
+    [
+        # template=None / params=None delegate to the base hooks (default template "$resolved_hostname").
+        pytest.param(None, None, ["resolved_hostname:my-host"], None, "my-host", id="default_template_uses_tags"),
+        # Non-string param values are stringified by the template engine, so no casting is needed.
+        pytest.param("$host:$port", {"host": "db-host", "port": 5432}, [], None, "db-host:5432", id="overridden_hooks"),
+        # Connection params take precedence over tags of the same name.
+        pytest.param(
+            "$host:$port",
+            {"host": "db-host", "port": 5432},
+            ["host:tag-host", "port:1111"],
+            None,
+            "db-host:5432",
+            id="params_override_tags",
+        ),
+        # Mutating tags after first access has no effect: the identifier is built once and cached.
+        pytest.param(
+            None, None, ["resolved_hostname:first"], ["resolved_hostname:second"], "first", id="built_once_and_cached"
+        ),
+    ],
+)
+def test_database_identifier(template, params, tags, tags_after, expected):
+    class EmbeddedDatabaseCheck(FakeDatabaseCheck):
+        @property
+        def database_identifier_template(self):
+            return super().database_identifier_template if template is None else template
+
+        @property
+        def database_identifier_params(self):
+            return super().database_identifier_params if params is None else params
+
+    check = EmbeddedDatabaseCheck("test", {}, [{}])
+    check.tag_manager.set_tags_from_list(tags)
+    assert check.database_identifier == expected
+    if tags_after is not None:
+        check.tag_manager.set_tags_from_list(tags_after, replace=True)
+        assert check.database_identifier == expected