Ensure shared properties on all DBM payloads

datadog_checks_base/datadog_checks/base/checks/db.py

@@ -4,24 +4,39 @@
 
 from abc import abstractmethod
 
+from datadog_checks.base.utils.db.utils import default_json_event_encoding
+from datadog_checks.base.utils.serialization import json
+
 from . import AgentCheck
 
 
 class DatabaseCheck(AgentCheck):
-    def database_monitoring_query_sample(self, raw_event: str):
-        self.event_platform_event(raw_event, "dbm-samples")
+    def database_monitoring_query_sample(self, raw_event: dict):
+        self.set_event_platform_properties(raw_event, "dbm-samples")
+        self.event_platform_event(json.dumps(raw_event, default=default_json_event_encoding), "dbm-samples")
+
+    def database_monitoring_query_metrics(self, raw_event: dict):
+        self.set_event_platform_properties(raw_event, "dbm-metrics")
+        self.event_platform_event(json.dumps(raw_event, default=default_json_event_encoding), "dbm-metrics")
 
-    def database_monitoring_query_metrics(self, raw_event: str):
-        self.event_platform_event(raw_event, "dbm-metrics")
+    def database_monitoring_query_activity(self, raw_event: dict):
+        self.set_event_platform_properties(raw_event, "dbm-activity")
+        self.event_platform_event(json.dumps(raw_event, default=default_json_event_encoding), "dbm-activity")
 
-    def database_monitoring_query_activity(self, raw_event: str):
-        self.event_platform_event(raw_event, "dbm-activity")
+    def database_monitoring_metadata(self, raw_event: dict):
+        self.set_event_platform_properties(raw_event, "dbm-metadata")
+        self.event_platform_event(json.dumps(raw_event, default=default_json_event_encoding), "dbm-metadata")
 
-    def database_monitoring_metadata(self, raw_event: str):
-        self.event_platform_event(raw_event, "dbm-metadata")
+    def database_monitoring_health(self, raw_event: dict):
+        self.set_event_platform_properties(raw_event, "dbm-health")
+        self.event_platform_event(json.dumps(raw_event, default=default_json_event_encoding), "dbm-health")
 
-    def database_monitoring_health(self, raw_event: str):
-        self.event_platform_event(raw_event, "dbm-health")
+    def set_event_platform_properties(self, raw_event: dict, track: str):
+        # Ensure all events have shared properties
+        raw_event["track"] = track
+        raw_event["database_instance"] = self.database_identifier
+        raw_event["dbms"] = self.dbms
+        raw_event["dbms_version"] = self.dbms_version
 
     @property
     @abstractmethod

datadog_checks_base/datadog_checks/base/utils/db/schemas.py

@@ -7,8 +7,6 @@
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, TypedDict
 
-from datadog_checks.base.utils.serialization import json
-
 from .utils import now_ms
 
 if TYPE_CHECKING:
@@ -143,7 +141,7 @@ def maybe_flush(self, is_last_payload):
                 # For the last payload, we need to include the total number of payloads collected
                 # This is used for snapshotting to ensure that all payloads have been received
                 event["collection_payloads_count"] = self._collection_payloads_count
-            self._check.database_monitoring_metadata(json.dumps(event))
+            self._check.database_monitoring_metadata(event)
 
             self._queued_rows = []
 

mysql/datadog_checks/mysql/activity.py

@@ -221,13 +221,7 @@ def _collect_activity(self):
             rows = self._get_activity(cursor)
             rows = self._normalize_rows(rows)
             event = self._create_activity_event(rows, tags)
-            payload = json.dumps(event, default=self._json_event_encoding)
-            self._check.database_monitoring_query_activity(payload)
-            self._check.histogram(
-                "dd.mysql.activity.collect_activity.payload_size",
-                len(payload),
-                tags=tags + self._check._get_debug_tags(),
-            )
+            self._check.database_monitoring_query_activity(event)
 
     def _should_collect_blocking_queries(self):
         # type: () -> bool

mysql/datadog_checks/mysql/metadata.py

@@ -20,9 +20,7 @@
 from datadog_checks.base import is_affirmative
 from datadog_checks.base.utils.db.utils import (
     DBMAsyncJob,
-    default_json_event_encoding,
 )
-from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 
 # default pg_settings collection interval in seconds
@@ -180,4 +178,4 @@ def report_mysql_metadata(self):
             "cloud_metadata": self._config.cloud_metadata,
             "metadata": settings,
         }
-        self._check.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
+        self._check.database_monitoring_metadata(event)

mysql/datadog_checks/mysql/mysql.py

@@ -21,13 +21,11 @@
 from datadog_checks.base.utils.db.health import HealthEvent, HealthStatus
 from datadog_checks.base.utils.db.utils import (
     TagManager,
-    default_json_event_encoding,
     tracked_query,
 )
 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.mysql import aws
 from datadog_checks.mysql.cursor import CommenterCursor, CommenterDictCursor, CommenterSSCursor
 from datadog_checks.mysql.health import MySqlHealth
@@ -1416,7 +1414,7 @@ def _send_database_instance_metadata(self):
                 },
             }
             self._database_instance_emitted[self.database_identifier] = event
-            self.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
+            self.database_monitoring_metadata(event)
 
     def set_cluster_tags(self, db):
         if not self._config.replication_enabled:

mysql/datadog_checks/mysql/statement_samples.py

@@ -25,7 +25,6 @@
 from datadog_checks.base.utils.db.utils import (
     DBMAsyncJob,
     RateLimitingTTLCache,
-    default_json_event_encoding,
     obfuscate_sql_with_metadata,
 )
 from datadog_checks.base.utils.serialization import json
@@ -595,7 +594,7 @@ def _collect_statement_samples(self):
             self._tags + ["events_statements_table:{}".format(events_statements_table)] + self._check._get_debug_tags()
         )
         for e in events:
-            self._check.database_monitoring_query_sample(json.dumps(e, default=default_json_event_encoding))
+            self._check.database_monitoring_query_sample(e)
             submitted_count += 1
         self._check.histogram(
             "dd.mysql.collect_statement_samples.time",

mysql/datadog_checks/mysql/statements.py

@@ -15,7 +15,7 @@
 from datadog_checks.base.utils.common import to_native_string
 from datadog_checks.base.utils.db.sql import compute_sql_signature
 from datadog_checks.base.utils.db.statement_metrics import StatementMetrics
-from datadog_checks.base.utils.db.utils import DBMAsyncJob, default_json_event_encoding, obfuscate_sql_with_metadata
+from datadog_checks.base.utils.db.utils import DBMAsyncJob, obfuscate_sql_with_metadata
 from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 from datadog_checks.mysql.cursor import CommenterDictCursor
@@ -151,7 +151,7 @@ def collect_per_statement_metrics(self):
             # No rows to process, can skip the rest of the payload generation and avoid an empty payload
             return
         for event in self._rows_to_fqt_events(rows, tags):
-            self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
+            self._check.database_monitoring_query_sample(event)
         payload = {
             'host': self._check.resolved_hostname,
             'timestamp': time.time() * 1000,
@@ -164,7 +164,7 @@ def collect_per_statement_metrics(self):
             'service': self._config.service,
             'mysql_rows': rows,
         }
-        self._check.database_monitoring_query_metrics(json.dumps(payload, default=default_json_event_encoding))
+        self._check.database_monitoring_query_metrics(payload)
         self._check.gauge(
             "dd.mysql.collect_per_statement_metrics.rows",
             len(rows),

postgres/datadog_checks/postgres/metadata.py

@@ -3,7 +3,6 @@
 # Licensed under a 3-clause BSD style license (see LICENSE)
 from __future__ import annotations
 
-import json
 import re
 import time
 
@@ -22,7 +21,7 @@
 if TYPE_CHECKING:
     from datadog_checks.postgres import PostgreSql
 
-from datadog_checks.base.utils.db.utils import DBMAsyncJob, default_json_event_encoding
+from datadog_checks.base.utils.db.utils import DBMAsyncJob
 from datadog_checks.base.utils.tracking import tracked_method
 from datadog_checks.postgres.config_models import InstanceConfig
 
@@ -169,7 +168,7 @@ def report_postgres_extensions(self):
                 "cloud_metadata": self._check.cloud_metadata,
                 "metadata": self._extensions_cached,
             }
-            self._check.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
+            self._check.database_monitoring_metadata(event)
 
     @tracked_method(agent_check_getter=agent_check_getter)
     def _collect_postgres_extensions(self):
@@ -206,7 +205,7 @@ def report_postgres_metadata(self):
                 "cloud_metadata": self._check.cloud_metadata,
                 "metadata": self._pg_settings_cached,
             }
-            self._check.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
+            self._check.database_monitoring_metadata(event)
 
         if (
             self._collect_schemas_enabled

postgres/datadog_checks/postgres/statement_samples.py

@@ -32,7 +32,6 @@
 from datadog_checks.base.utils.db.utils import (
     DBMAsyncJob,
     RateLimitingTTLCache,
-    default_json_event_encoding,
     obfuscate_sql_with_metadata,
 )
 from datadog_checks.base.utils.serialization import json
@@ -488,15 +487,13 @@ def _collect_statement_samples(self):
         submitted_count = 0
         if self._explain_plan_coll_enabled:
             for e in self._collect_plans(rows):
-                self._check.database_monitoring_query_sample(json.dumps(e, default=default_json_event_encoding))
+                self._check.database_monitoring_query_sample(e)
                 submitted_count += 1
 
         if collect_activity:
             active_connections = self._get_active_connections()
             activity_event = self._create_activity_event(rows, active_connections)
-            self._check.database_monitoring_query_activity(
-                json.dumps(activity_event, default=default_json_event_encoding)
-            )
+            self._check.database_monitoring_query_activity(activity_event)
             self._check.histogram(
                 "dd.postgres.collect_activity_snapshot.time",
                 (time.time() - start_time) * 1000,
@@ -631,7 +628,7 @@ def _row_to_raw_statement_event(self, row):
             },
         }
 
-        self._check.database_monitoring_query_sample(json.dumps(raw_query_event, default=default_json_event_encoding))
+        self._check.database_monitoring_query_sample(raw_query_event)
 
     def _can_explain_statement(self, obfuscated_statement):
         if obfuscated_statement.startswith('SELECT {}'.format(self._explain_function)):

postgres/datadog_checks/postgres/statements.py

@@ -274,7 +274,7 @@ def collect_per_statement_metrics(self):
             if not rows:
                 return
             for event in self._rows_to_fqt_events(rows):
-                self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
+                self._check.database_monitoring_query_sample(event)
 
             payload_wrapper = {
                 'host': self._check.reported_hostname,

sqlserver/datadog_checks/sqlserver/activity.py

@@ -13,7 +13,6 @@
 from datadog_checks.base.utils.db.utils import (
     DBMAsyncJob,
     RateLimitingTTLCache,
-    default_json_event_encoding,
     obfuscate_sql_with_metadata,
 )
 from datadog_checks.base.utils.serialization import json
@@ -490,13 +489,6 @@ def collect_activity(self):
                 normalized_rows = self._normalize_queries_and_filter_rows(rows, MAX_PAYLOAD_BYTES)
                 if self._collect_raw_query_statement:
                     for raw_statement_event in self._rows_to_raw_statement_events(normalized_rows):
-                        self._check.database_monitoring_query_sample(
-                            json.dumps(raw_statement_event, default=default_json_event_encoding)
-                        )
+                        self._check.database_monitoring_query_sample(raw_statement_event)
                 event = self._create_activity_event(normalized_rows, connections)
-                payload = json.dumps(event, default=default_json_event_encoding)
-                self._check.database_monitoring_query_activity(payload)
-
-        self._check.histogram(
-            "dd.sqlserver.activity.collect_activity.payload_size", len(payload), **self._check.debug_stats_kwargs()
-        )
+                self._check.database_monitoring_query_activity(event)

sqlserver/datadog_checks/sqlserver/deadlocks.py

@@ -6,8 +6,7 @@
 from time import time
 
 from datadog_checks.base.utils.db.sql import compute_sql_signature
-from datadog_checks.base.utils.db.utils import DBMAsyncJob, default_json_event_encoding, obfuscate_sql_with_metadata
-from datadog_checks.base.utils.serialization import json
+from datadog_checks.base.utils.db.utils import DBMAsyncJob, obfuscate_sql_with_metadata
 from datadog_checks.base.utils.tracking import tracked_method
 from datadog_checks.sqlserver.config import SQLServerConfig
 from datadog_checks.sqlserver.const import STATIC_INFO_ENGINE_EDITION, STATIC_INFO_VERSION
@@ -260,9 +259,7 @@ def collect_deadlocks(self):
         # Send payload only if deadlocks found
         if rows:
             deadlocks_event = self._create_deadlock_event(rows)
-            payload = json.dumps(deadlocks_event, default=default_json_event_encoding)
-            self._log.debug("Deadlocks payload: %s", str(payload))
-            self._check.database_monitoring_query_activity(payload)
+            self._check.database_monitoring_query_activity(deadlocks_event)
 
     def _create_deadlock_event(self, deadlock_rows):
         event = {

sqlserver/datadog_checks/sqlserver/metadata.py

@@ -6,9 +6,7 @@
 from datadog_checks.base import is_affirmative
 from datadog_checks.base.utils.db.utils import (
     DBMAsyncJob,
-    default_json_event_encoding,
 )
-from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 from datadog_checks.sqlserver.config import SQLServerConfig
 from datadog_checks.sqlserver.const import (
@@ -158,7 +156,7 @@ def report_sqlserver_metadata(self):
                     "cloud_metadata": self._check.cloud_metadata,
                     "metadata": settings_rows,
                 }
-                self._check.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
+                self._check.database_monitoring_metadata(event)
         self.collect_schemas()
 
     def collect_schemas(self):

sqlserver/datadog_checks/sqlserver/sqlserver.py

@@ -17,13 +17,11 @@
 from datadog_checks.base.utils.db.health import HealthEvent, HealthStatus
 from datadog_checks.base.utils.db.utils import (
     TagManager,
-    default_json_event_encoding,
     tracked_query,
 )
 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.sqlserver.activity import SqlserverActivity
 from datadog_checks.sqlserver.agent_history import SqlserverAgentHistory
 from datadog_checks.sqlserver.config import SQLServerConfig
@@ -1141,4 +1139,4 @@ def _send_database_instance_metadata(self):
                 },
             }
             self._database_instance_emitted[self.database_identifier] = event
-            self.database_monitoring_metadata(json.dumps(event, default=default_json_event_encoding))
+            self.database_monitoring_metadata(event)

sqlserver/datadog_checks/sqlserver/statements.py

@@ -17,10 +17,8 @@
 from datadog_checks.base.utils.db.utils import (
     DBMAsyncJob,
     RateLimitingTTLCache,
-    default_json_event_encoding,
     obfuscate_sql_with_metadata,
 )
-from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 from datadog_checks.sqlserver.config import SQLServerConfig
 from datadog_checks.sqlserver.utils import is_azure_sql_database
@@ -537,11 +535,11 @@ def collect_statement_metrics_and_plans(self):
                 if not rows:
                     return
                 for event in self._rows_to_fqt_events(rows):
-                    self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
+                    self._check.database_monitoring_query_sample(event)
                 payload = self._to_metrics_payload(rows, self._max_query_metrics)
-                self._check.database_monitoring_query_metrics(json.dumps(payload, default=default_json_event_encoding))
+                self._check.database_monitoring_query_metrics(payload)
                 for event in self._collect_plans(rows, cursor, deadline):
-                    self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
+                    self._check.database_monitoring_query_sample(event)
                     plans_submitted += 1
 
         self._check.count(

sqlserver/datadog_checks/sqlserver/stored_procedures.py

@@ -8,9 +8,7 @@
 from datadog_checks.base.utils.db.statement_metrics import StatementMetrics
 from datadog_checks.base.utils.db.utils import (
     DBMAsyncJob,
-    default_json_event_encoding,
 )
-from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 from datadog_checks.sqlserver.config import SQLServerConfig
 
@@ -176,7 +174,7 @@ def collect_procedure_metrics(self):
                     self.log.debug("collect_procedure_metrics: no rows returned")
                     return
                 payload = self._to_metrics_payload(rows, self._max_procedure_metrics)
-                self._check.database_monitoring_query_metrics(json.dumps(payload, default=default_json_event_encoding))
+                self._check.database_monitoring_query_metrics(payload)
 
     def run_job(self):
         self.collect_procedure_metrics()

sqlserver/datadog_checks/sqlserver/xe_collection/base.py

@@ -629,10 +629,7 @@ def run_job(self):
                     self._log.error(f"Error serializing batched payload for logging: {e}")
 
             # Send the batched payload
-            serialized_payload = json.dumps(batched_payload, default=default_json_event_encoding)
-            # Log payload size
-            self._log.debug(f"Batched {self.session_name} payload size: {len(serialized_payload)} bytes")
-            self._check.database_monitoring_query_activity(serialized_payload)
+            self._check.database_monitoring_query_activity(batched_payload)
 
         self._log.info(f"Found {len(events)} events from {self.session_name} session")