add DO queries to postgres check (DataDog#23123)

* Add data_observability config to Postgres

* Add async job

* Ddev update

* Fmt

* initial move to postgres

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* fix ai issues

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* move to postgres check

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* code review updates

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* Fix test_config and test_config_defaults for DATA_OBSERVABILITY feature

Add DATA_OBSERVABILITY to the expected feature set in
test_initialize_features_enabled_and_disabled and add
the data_observability defaults entry to EXPECTED_DEFAULTS
in test_config_defaults.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* Strip trailing semicolons before subquery wrapping

Queries with a trailing semicolon would produce invalid SQL when wrapped
in SELECT * FROM (...) _dd_row_limit LIMIT N. Strip trailing semicolons
and whitespace from the query string before constructing the wrapper.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* Switch query_status gauge to query_executions counter with status tag

Replace gauge(query_status, 1/0) with count(query_executions, 1, status:success/error).
A counter allows computing failure rate over time; a gauge hides flakiness because
a single successful check in a window obscures preceding failures.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* Remove subquery wrapping; rely on fetchmany for row cap

The SELECT * FROM (...) _dd_row_limit LIMIT N wrapper broke on real RC
queries which have trailing semicolons or trailing -- Datadog {...} comments
appended by the Go handler. Both cause syntax errors inside the subquery.

fetchmany(MAX_RESULT_ROWS) is a sufficient memory safety net on the Python
side. Executing the query as-is avoids all SQL manipulation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* address code review

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* fmt

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* explanation

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* process grouped queries by db, use per-db conn

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

* run with newer ddev

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>

---------

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>
Co-authored-by: Seth Samuel <seth.samuel@datadoghq.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: 3 people

postgres/assets/configuration/spec.yaml

@@ -28,7 +28,6 @@ files:
     - name: port
       description: The port to use when connecting to PostgreSQL.
       fleet_configurable: true
-      formats: ["port"]
       value:
         type: integer
         display_default: 5432
@@ -543,6 +542,103 @@ files:
         type: boolean
         example: false
         display_default: false
+    - name: data_observability
+      fleet_configurable: true
+      description: |
+        Configure the Data Observability async job, which executes monitoring
+        queries delivered via Remote Configuration.
+      options:
+        - name: enabled
+          fleet_configurable: true
+          description: Enable the Data Observability job.
+          value:
+            type: boolean
+            example: false
+            display_default: false
+        - name: collection_interval
+          fleet_configurable: true
+          description: |
+            How often (in seconds) the job wakes up to check which queries are
+            due. Individual queries have their own interval_seconds. This value
+            must be less than or equal to the smallest interval_seconds across
+            all configured queries (i.e. the least common denominator), otherwise
+            queries with short intervals will be delayed.
+          value:
+            type: number
+            example: 10
+            display_default: 10
+        - name: run_sync
+          hidden: true
+          description: Run the job synchronously (for testing).
+          value:
+            type: boolean
+            example: false
+            display_default: false
+        - name: config_id
+          hidden: true
+          description: |
+            Unique identifier of the Remote Configuration payload that populated
+            this data_observability block. Set automatically by the Datadog Agent
+            Go RC handler when it delivers query configuration; do not set this
+            field manually. It is forwarded as-is on every emitted event so that
+            the intake can correlate results back to the originating RC config.
+          value:
+            type: string
+        - name: queries
+          hidden: true
+          description: |
+            List of monitor queries to execute. Populated automatically by the
+            Datadog Agent Go RC handler when Remote Configuration delivers a
+            DO_QUERY_ACTIONS payload; do not set this field manually. Each entry
+            describes one SQL monitoring query along with its scheduling
+            parameters and entity metadata.
+          value:
+            type: array
+            items:
+              type: object
+              required:
+                - monitor_id
+                - dbname
+                - query
+                - interval_seconds
+                - entity
+              properties:
+              - name: monitor_id
+                type: integer
+              - name: dbname
+                type: string
+              - name: type
+                type: string
+              - name: query
+                type: string
+              - name: interval_seconds
+                type: integer
+              - name: entity
+                type: object
+                required:
+                  - platform
+                  - account
+                  - database
+                  - schema
+                  - table
+                properties:
+                - name: platform
+                  type: string
+                - name: account
+                  type: string
+                - name: database
+                  type: string
+                - name: schema
+                  type: string
+                - name: table
+                  type: string
+              - name: custom_sql_select_fields
+                type: object
+                properties:
+                - name: metric_config_id
+                  type: integer
+                - name: entity_id
+                  type: string
     - name: pg_stat_statements_view
       description: |
         Set this value if you want to define a custom view or function to allow the datadog user to query the

postgres/changelog.d/23123.added

@@ -0,0 +1 @@
+Add Data Observability async job for executing monitoring queries delivered via Remote Configuration.

postgres/datadog_checks/postgres/config.py

@@ -121,6 +121,10 @@ def build_config(check: PostgreSql) -> Tuple[InstanceConfig, ValidationResult]:
             "dbm": instance.get(
                 'dbm', instance.get('deep_database_monitoring', defaults.instance_dbm())
             ),  # Deprecated, use `dbm` instead
+            "data_observability": {
+                **dict_defaults.instance_data_observability().model_dump(),
+                **(instance.get('data_observability') or {}),
+            },
             "custom_metrics": map_custom_metrics(
                 instance.get('custom_metrics', [])
             ),  # Deprecated, use `custom_queries` instead
@@ -431,6 +435,7 @@ def apply_features(config: InstanceConfig, validation_result: ValidationResult):
     validation_result.add_feature(FeatureKey.QUERY_METRICS, config.query_metrics.enabled and config.dbm)
     validation_result.add_feature(FeatureKey.COLLECT_SETTINGS, config.collect_settings.enabled and config.dbm)
     validation_result.add_feature(FeatureKey.COLLECT_SCHEMAS, config.collect_schemas.enabled and config.dbm)
+    validation_result.add_feature(FeatureKey.DATA_OBSERVABILITY, config.data_observability.enabled)
 
 
 METRIC_TYPES = {

postgres/datadog_checks/postgres/config_models/dict_defaults.py

@@ -132,3 +132,13 @@ def instance_locks_idle_in_transaction():
         collection_interval=300,
         max_rows=100,
     )
+
+
+def instance_data_observability():
+    return instance.DataObservability(
+        enabled=False,
+        collection_interval=10,
+        run_sync=False,
+        config_id=None,
+        queries=(),
+    )

postgres/datadog_checks/postgres/config_models/instance.py

@@ -111,6 +111,53 @@ class CustomQuery(BaseModel):
     tags: Optional[tuple[str, ...]] = None
 
 
+class CustomSqlSelectFields(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        frozen=True,
+    )
+    entity_id: Optional[str] = None
+    metric_config_id: Optional[int] = None
+
+
+class Entity(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        frozen=True,
+    )
+    account: str
+    database: str
+    platform: str
+    schema_: str = Field(..., alias='schema')
+    table: str
+
+
+class Query(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        frozen=True,
+    )
+    custom_sql_select_fields: Optional[CustomSqlSelectFields] = None
+    dbname: str
+    entity: Entity
+    interval_seconds: int
+    monitor_id: int
+    query: str
+    type: Optional[str] = None
+
+
+class DataObservability(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        frozen=True,
+    )
+    collection_interval: Optional[float] = None
+    config_id: Optional[str] = None
+    enabled: Optional[bool] = None
+    queries: Optional[tuple[Query, ...]] = None
+    run_sync: Optional[bool] = None
+
+
 class DatabaseAutodiscovery(BaseModel):
     model_config = ConfigDict(
         arbitrary_types_allowed=True,
@@ -263,6 +310,7 @@ class InstanceConfig(BaseModel):
     custom_metrics: Optional[tuple[MappingProxyType[str, Any], ...]] = None
     custom_queries: Optional[tuple[CustomQuery, ...]] = None
     data_directory: Optional[str] = None
+    data_observability: Optional[DataObservability] = None
     database_autodiscovery: Optional[DatabaseAutodiscovery] = None
     database_identifier: Optional[DatabaseIdentifier] = None
     database_instance_collection_interval: Optional[float] = None

postgres/datadog_checks/postgres/data/conf.yaml.example

@@ -434,6 +434,25 @@ instances:
     #
     # dbm: false
 
+    ## Configure the Data Observability async job, which executes monitoring
+    ## queries delivered via Remote Configuration.
+    #
+    # data_observability:
+
+        ## @param enabled - boolean - optional - default: false
+        ## Enable the Data Observability job.
+        #
+        # enabled: false
+
+        ## @param collection_interval - number - optional - default: 10
+        ## How often (in seconds) the job wakes up to check which queries are
+        ## due. Individual queries have their own interval_seconds. This value
+        ## must be less than or equal to the smallest interval_seconds across
+        ## all configured queries (i.e. the least common denominator), otherwise
+        ## queries with short intervals will be delayed.
+        #
+        # collection_interval: 10
+
     ## @param pg_stat_statements_view - string - optional - default: pg_stat_statements
     ## Set this value if you want to define a custom view or function to allow the datadog user to query the
     ## `pg_stat_statements` table, which is useful for restricting the permissions given to the datadog agent.