Add support for sending database_instance metadata (#15559)

* Add support for sending database_instance metadata

* Maybe fix test

* Add changelog

* Update CHANGELOG entry

* Update CHANGELOG entry using ddev tooling

* Fix some tests

* Set expected tags in test

* Fix sneaking typo

* Fix time package usage

* Proper fix for wrong time import

* Fix linting error

postgres/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+***Added***:
+
+* Add support for sending `database_instance` metadata ([#15559](https://github.com/DataDog/integrations-core/pull/15559))
+
 ## 14.1.0 / 2023-08-10
 
 ***Added***:

postgres/assets/configuration/spec.yaml

@@ -682,6 +682,16 @@ files:
         type: boolean
         example: false
         display_default: false
+    - name: database_instance_collection_interval
+      hidden: true
+      description: |
+        Set the database instance collection interval (in seconds). The database instance collection sends
+        basic information about the database instance along with a signal that it still exists. 
+        This collection does not involve any additional queries to the database.
+      value:
+        type: number
+        example: 1800
+        display_default: false
     - template: instances/default
       overrides:
         disable_generic_tags.hidden: False

postgres/datadog_checks/postgres/config.py

@@ -134,6 +134,7 @@ def __init__(self, instance):
         }
         self.log_unobfuscated_queries = is_affirmative(instance.get('log_unobfuscated_queries', False))
         self.log_unobfuscated_plans = is_affirmative(instance.get('log_unobfuscated_plans', False))
+        self.database_instance_collection_interval = instance.get('database_instance_collection_interval', 1800)
 
     def _build_tags(self, custom_tags):
         # Clean up tags in case there was a None entry in the instance

postgres/datadog_checks/postgres/config_models/defaults.py

@@ -48,6 +48,10 @@ def instance_data_directory():
     return '/usr/local/pgsql/data'
 
 
+def instance_database_instance_collection_interval():
+    return False
+
+
 def instance_dbm():
     return False
 

postgres/datadog_checks/postgres/config_models/instance.py

@@ -159,6 +159,7 @@ class InstanceConfig(BaseModel):
     custom_queries: Optional[tuple[MappingProxyType[str, Any], ...]] = None
     data_directory: Optional[str] = None
     database_autodiscovery: Optional[DatabaseAutodiscovery] = None
+    database_instance_collection_interval: Optional[float] = None
     dbm: Optional[bool] = None
     dbname: Optional[str] = None
     dbstrict: Optional[bool] = None

postgres/datadog_checks/postgres/metadata.py

@@ -17,6 +17,8 @@
 from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 
+from .util import payload_pg_version
+
 # default pg_settings collection interval in seconds
 DEFAULT_SETTINGS_COLLECTION_INTERVAL = 600
 DEFAULT_RESOURCES_COLLECTION_INTERVAL = 300
@@ -98,20 +100,14 @@ def report_postgres_metadata(self):
             "dbms": "postgres",
             "kind": "pg_settings",
             "collection_interval": self.collection_interval,
-            'dbms_version': self._payload_pg_version(),
+            'dbms_version': payload_pg_version(self._check.version),
             "tags": self._tags_no_db,
             "timestamp": time.time() * 1000,
             "cloud_metadata": self._config.cloud_metadata,
             "metadata": self._pg_settings_cached,
         }
         self._check.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
 
-    def _payload_pg_version(self):
-        version = self._check.version
-        if not version:
-            return ""
-        return 'v{major}.{minor}.{patch}'.format(major=version.major, minor=version.minor, patch=version.patch)
-
     @tracked_method(agent_check_getter=agent_check_getter)
     def _collect_postgres_settings(self):
         with self._check.get_main_db().cursor(row_factory=dict_row) as cursor:

postgres/datadog_checks/postgres/postgres.py

@@ -7,13 +7,18 @@
 from time import time
 
 import psycopg
+from cachetools import TTLCache
 from psycopg import ClientCursor
 from psycopg.rows import dict_row
 from six import iteritems
 
 from datadog_checks.base import AgentCheck
 from datadog_checks.base.utils.db import QueryExecutor
+from datadog_checks.base.utils.db.utils import (
+    default_json_event_encoding,
+)
 from datadog_checks.base.utils.db.utils import resolve_db_host as agent_host_resolver
+from datadog_checks.base.utils.serialization import json
 from datadog_checks.postgres import aws
 from datadog_checks.postgres.connections import MultiDatabaseConnectionPool
 from datadog_checks.postgres.discovery import PostgresAutodiscovery
@@ -29,6 +34,7 @@
 from datadog_checks.postgres.statement_samples import PostgresStatementSamples
 from datadog_checks.postgres.statements import PostgresStatementMetrics
 
+from .__about__ import __version__
 from .config import PostgresConfig
 from .util import (
     AWS_RDS_HOSTNAME_SUFFIX,
@@ -50,6 +56,7 @@
     DatabaseConfigurationError,  # noqa: F401
     fmt,
     get_schema_field,
+    payload_pg_version,
     warning_with_tags,
 )
 from .version_utils import V9, V9_2, V10, V13, V14, VersionUtils
@@ -89,6 +96,9 @@ def __init__(self, name, init_config, instances):
         self._config = PostgresConfig(self.instance)
         self.cloud_metadata = self._config.cloud_metadata
         self.tags = self._config.tags
+        # Keep a copy of the tags without the internal resource tags so they can be used for paths that don't
+        # go through the agent internal metrics submission processing those tags
+        self._non_internal_tags = copy.deepcopy(self.tags)
         self.set_resource_tags()
         self.pg_settings = {}
         self._warnings_by_code = {}
@@ -104,8 +114,12 @@ def __init__(self, name, init_config, instances):
         self.check_initializations.append(self.set_resolved_hostname_metadata)
         self.tags_without_db = [t for t in copy.copy(self.tags) if not t.startswith("db:")]
         self.autodiscovery = self._build_autodiscovery()
-
         self._dynamic_queries = None
+        # _database_instance_emitted: limit the collection and transmission of the database instance metadata
+        self._database_instance_emitted = TTLCache(
+            maxsize=1,
+            ttl=self._config.database_instance_collection_interval,
+        )  # type: TTLCache
 
     def _build_autodiscovery(self):
         if not self._config.discovery_config['enabled']:
@@ -814,6 +828,27 @@ def _report_warnings(self):
         for warning in messages:
             self.warning(warning)
 
+    def _send_database_instance_metadata(self):
+        if self.resolved_hostname not in self._database_instance_emitted:
+            event = {
+                "host": self.resolved_hostname,
+                "agent_version": datadog_agent.get_version(),
+                "dbms": "postgres",
+                "kind": "database_instance",
+                "collection_interval": self._config.database_instance_collection_interval,
+                'dbms_version': payload_pg_version(self.version),
+                'integration_version': __version__,
+                "tags": self._non_internal_tags,
+                "timestamp": time() * 1000,
+                "cloud_metadata": self._config.cloud_metadata,
+                "metadata": {
+                    "dbm": self._config.dbm_enabled,
+                    "connection_host": self._config.host,
+                },
+            }
+            self._database_instance_emitted[self.resolved_hostname] = event
+            self.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
+
     def check(self, _):
         tags = copy.copy(self.tags)
         # Collect metrics
@@ -837,7 +872,7 @@ def check(self, _):
                 self.metadata_samples.run_job_loop(tags)
             if self._config.collect_wal_metrics:
                 self._collect_wal_metrics(tags)
-
+            self._send_database_instance_metadata()
         except Exception as e:
             self.log.exception("Unable to collect postgres metrics.")
             self._clean_state()

postgres/datadog_checks/postgres/statements.py

@@ -18,7 +18,7 @@
 from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 
-from .util import DatabaseConfigurationError, warning_with_tags
+from .util import DatabaseConfigurationError, payload_pg_version, warning_with_tags
 from .version_utils import V9_4, V14
 
 try:
@@ -182,12 +182,6 @@ def run_job(self):
         self._tags_no_db = [t for t in self.tags if not t.startswith('db:')]
         self.collect_per_statement_metrics()
 
-    def _payload_pg_version(self):
-        version = self._check.version
-        if not version:
-            return ""
-        return 'v{major}.{minor}.{patch}'.format(major=version.major, minor=version.minor, patch=version.patch)
-
     @tracked_method(agent_check_getter=agent_check_getter)
     def collect_per_statement_metrics(self):
         # exclude the default "db" tag from statement metrics & FQT events because this data is collected from
@@ -206,7 +200,7 @@ def collect_per_statement_metrics(self):
                 'tags': self._tags_no_db,
                 'cloud_metadata': self._config.cloud_metadata,
                 'postgres_rows': rows,
-                'postgres_version': self._payload_pg_version(),
+                'postgres_version': payload_pg_version(self._check.version),
                 'ddagentversion': datadog_agent.get_version(),
                 "ddagenthostname": self._check.agent_hostname,
             }

postgres/datadog_checks/postgres/util.py

@@ -59,6 +59,12 @@ def get_schema_field(descriptors):
     raise CheckException("The descriptors are missing a schema field")
 
 
+def payload_pg_version(version):
+    if not version:
+        return ""
+    return 'v{major}.{minor}.{patch}'.format(major=version.major, minor=version.minor, patch=version.patch)
+
+
 fmt = PartialFormatter()
 
 AWS_RDS_HOSTNAME_SUFFIX = ".rds.amazonaws.com"

postgres/tests/test_metadata.py

@@ -33,7 +33,8 @@ def test_collect_metadata(integration_check, dbm_instance, aggregator):
     check = integration_check(dbm_instance)
     check.check(dbm_instance)
     dbm_metadata = aggregator.get_event_platform_events("dbm-metadata")
-    event = dbm_metadata[0]
+    event = next((e for e in dbm_metadata if e['kind'] == 'pg_settings'), None)
+    assert event is not None
     assert event['host'] == "stubbed.hostname"
     assert event['dbms'] == "postgres"
     assert event['kind'] == "pg_settings"

postgres/tests/test_pg_integration.py

@@ -10,6 +10,7 @@
 from semver import VersionInfo
 
 from datadog_checks.postgres import PostgreSql
+from datadog_checks.postgres.__about__ import __version__
 from datadog_checks.postgres.util import PartialFormatter, fmt
 
 from .common import (
@@ -610,6 +611,48 @@ def test_correct_hostname(dbm_enabled, reported_hostname, expected_hostname, agg
     )
 
 
+@pytest.mark.parametrize(
+    'dbm_enabled, reported_hostname',
+    [
+        (True, None),
+        (False, None),
+        (True, 'forced_hostname'),
+        (True, 'forced_hostname'),
+    ],
+)
+@pytest.mark.integration
+@pytest.mark.usefixtures('dd_environment')
+def test_database_instance_metadata(aggregator, dd_run_check, pg_instance, dbm_enabled, reported_hostname):
+    pg_instance['dbm'] = dbm_enabled
+    if reported_hostname:
+        pg_instance['reported_hostname'] = reported_hostname
+    expected_host = reported_hostname if reported_hostname else 'stubbed.hostname'
+    expected_tags = pg_instance['tags'] + ['port:{}'.format(pg_instance['port'])]
+    check = PostgreSql('test_instance', {}, [pg_instance])
+    dd_run_check(check)
+
+    dbm_metadata = aggregator.get_event_platform_events("dbm-metadata")
+    event = next((e for e in dbm_metadata if e['kind'] == 'database_instance'), None)
+    assert event is not None
+    assert event['host'] == expected_host
+    assert event['dbms'] == "postgres"
+    assert event['tags'].sort() == expected_tags.sort()
+    assert event['integration_version'] == __version__
+    assert event['collection_interval'] == 1800
+    assert event['metadata'] == {
+        'dbm': dbm_enabled,
+        'connection_host': pg_instance['host'],
+    }
+
+    # Run a second time and expect the metadata to not be emitted again because of the cache TTL
+    aggregator.reset()
+    dd_run_check(check)
+
+    dbm_metadata = aggregator.get_event_platform_events("dbm-metadata")
+    event = next((e for e in dbm_metadata if e['kind'] == 'database_instance'), None)
+    assert event is None
+
+
 def assert_state_clean(check):
     assert check.metrics_cache.instance_metrics is None
     assert check.metrics_cache.bgw_metrics is None

postgres/tests/test_statements.py

@@ -25,6 +25,7 @@
     StatementTruncationState,
 )
 from datadog_checks.postgres.statements import PG_STAT_STATEMENTS_METRICS_COLUMNS, PG_STAT_STATEMENTS_TIMING_COLUMNS
+from datadog_checks.postgres.util import payload_pg_version
 
 from .common import DB_NAME, HOST, PORT, PORT_REPLICA2, POSTGRES_VERSION
 from .utils import WaitGroup, _get_conn, _get_superconn, requires_over_10, run_one_check
@@ -100,15 +101,15 @@ def test_statement_metrics_version(integration_check, dbm_instance, version, exp
         check = integration_check(dbm_instance)
         check._version = version
         check._connect()
-        assert check.statement_metrics._payload_pg_version() == expected_payload_version
+        assert payload_pg_version(check.version) == expected_payload_version
     else:
         with mock.patch(
             'datadog_checks.postgres.postgres.PostgreSql.version', new_callable=mock.PropertyMock
         ) as patched_version:
             patched_version.return_value = None
             check = integration_check(dbm_instance)
             check._connect()
-            assert check.statement_metrics._payload_pg_version() == expected_payload_version
+            assert payload_pg_version(check.version) == expected_payload_version
 
 
 @pytest.mark.parametrize("dbstrict,ignore_databases", [(True, []), (False, ['dogs']), (False, [])])