postgres data-observability - use cron to precisely schedule queries (DataDog#23529)

* postgres data-observability - use cron to precisely schedule queries

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

* review

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

* fix: apply ddev formatter

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

* Skip invalid queries at construction time

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

* remove lookback as configurable param

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

* Replace croniter with datadog_checks_base CronScheduler utility

Drop the third-party croniter==6.2.2 dependency and rewrite the cron
scheduling in PostgresDataObservability on top of the CronExpression /
CronScheduler utilities added to datadog_checks_base in DataDog#23741.

Changes:
- _filter_valid_queries: build a CronScheduler per monitor_id inside a
  try/except(ValueError, TypeError) instead of calling croniter.is_valid.
  Scheduler captures startup_lookback at construction time.
- _get_due_queries: collapse the entire cron branch to a single
  due_ticks(now) call; state registration, lookback recovery, tick
  detection, and advancement are all handled by CronScheduler.
- run_job: remove the post-fire croniter re-advance; due_ticks() already
  advanced the scheduler at poll time.
- Tests: replace _next_run[mid] accesses with _schedulers[mid].next_tick;
  update boundary test to reflect CronScheduler's inclusive (<=) lookback
  semantics vs the old strict (<) comparison.
- Remove croniter from agent_requirements.in, postgres/pyproject.toml,
  and LICENSE-3rdparty.csv; bump datadog-checks-base pin to >=37.39.0
  (first release shipping cron.py).

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

* Address code review: type hints, explicit state returns, lateness comment, error guard

- _filter_valid_queries: add Iterable[Query] parameter type; return
  (tuple[Query,...], dict[int,CronScheduler]) so callers own the
  assignment — removes the hidden self._schedulers side-effect. __init__
  now assigns both fields explicitly.
- _get_due_queries: remove setdefault seed for _last_execution; use
  .get() and treat None as first-sight. Scheduling-state mutations
  (seed and advance) are now consolidated in run_job.
- run_job: add comment distinguishing lateness (scheduling delay) from
  execution time; wrap emit_failures count in a nested try/except so a
  concurrent _shutdown() cannot turn the error handler into a job crash.
- Tests: merge _make_cron_lookback_check into _make_cron_check via
  optional window_seconds/monkeypatch params; collapse three identical
  lookback-window tests into one parametrized test_cron_startup_lookback_window_behavior.

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

* fmt

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

* code review fixes

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

* postgres DO: fix cron boundary miss and improve schedule error message

- due_ticks now probes now+0.001 so a first poll landing exactly on a
  cron tick boundary fires instead of skipping the cycle
- invalid cron schedule warning now includes the exception message and
  the monitor_id to help customers identify and fix the bad config
- regression test added for the exact-boundary case

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

* revert cron.py boundary fix (moved to postgres data_observability)

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

* fix cron exact-boundary detection without inflating lookback window

The previous +0.001 epsilon was passed directly to due_ticks(), which
caused CronScheduler to compare (now+0.001) - prev <= lookback, breaking
the inclusive-boundary test where now - prev == lookback exactly.

Instead, call due_ticks(now) normally and only on the first poll (before
next_tick is cached) probe previous_tick(before=now+0.001) to detect
whether now itself is a tick. This isolates the epsilon to the boundary
detection and leaves the lookback window comparison untouched.

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

* simplify cron boundary fix; update over-precise test

Use due_ticks(now + 0.001) — the simpler approach. The previous complex
wrapper was added to avoid breaking test_cron_startup_lookback_boundary_inclusive,
which was asserting now - prev == window exactly. That exact-equality check
belongs in test_cron.py (CronScheduler unit tests), not here. Updated the
test to use a clock 55s after the tick (clearly inside the 60s window),
keeping the meaningful assertion that recovery fires inside the window and
does not fire outside it.

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

---------

Signed-off-by: mobuchowski <maciej.obuchowski@datadoghq.com>
Signed-off-by: Maciej Obuchowski <maciej.obuchowski@datadoghq.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: mobuchowski

postgres/assets/configuration/spec.yaml

@@ -643,7 +643,6 @@ files:
                 - monitor_id
                 - dbname
                 - query
-                - interval_seconds
                 - entity
               properties:
               - name: monitor_id
@@ -655,7 +654,17 @@ files:
               - name: query
                 type: string
               - name: interval_seconds
+                description: |
+                  How often (in seconds) to run this query. Ignored when schedule is set
+                  (see schedule for the precedence rule).
                 type: integer
+              - name: schedule
+                description: |
+                  A standard 5-field cron expression (minute hour dom month dow) specifying
+                  when to run this query. When both schedule and interval_seconds are set,
+                  schedule wins and interval_seconds is ignored. If neither is set, the
+                  query is skipped at runtime with a warning.
+                type: string
               - name: entity
                 type: object
                 required:

postgres/changelog.d/23529.added

@@ -0,0 +1 @@
+Support cron `schedule` field for Data Observability queries.

postgres/datadog_checks/postgres/config_models/instance.py

@@ -172,9 +172,16 @@ class Query(BaseModel):
     custom_sql_select_fields: Optional[CustomSqlSelectFields] = None
     dbname: str
     entity: Entity
-    interval_seconds: int
+    interval_seconds: Optional[int] = Field(
+        None,
+        description='How often (in seconds) to run this query. Ignored when schedule is set\n(see schedule for the precedence rule).\n',
+    )
     monitor_id: int
     query: str
+    schedule: Optional[str] = Field(
+        None,
+        description='A standard 5-field cron expression (minute hour dom month dow) specifying\nwhen to run this query. When both schedule and interval_seconds are set,\nschedule wins and interval_seconds is ignored. If neither is set, the\nquery is skipped at runtime with a warning.\n',
+    )
     type: Optional[str] = None
 
 

postgres/datadog_checks/postgres/data_observability.py

@@ -5,10 +5,13 @@
 
 import json
 import time
-from typing import TYPE_CHECKING, Any
+from collections.abc import Iterable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Literal
 
 import psycopg
 
+from datadog_checks.base.utils.cron import CronScheduler
 from datadog_checks.base.utils.db.utils import DBMAsyncJob, default_json_event_encoding
 
 if TYPE_CHECKING:
@@ -17,15 +20,29 @@
 
 EVENT_TRACK_TYPE = 'do-query-results'
 
-# Cap the number of rows fetched per query to prevent unbounded memory usage.
 MAX_RESULT_ROWS = 10_000
 
+# After an agent start, run cron-scheduled queries whose scheduled time of execution
+# fell within this many seconds in the past. Recovers missed runs across short check
+# restarts (deploys, crashes, Remote Configuration redeliveries). Set to 0 to skip
+# catch-up.
+CRON_STARTUP_LOOKBACK_SECONDS = 300
+
+Mode = Literal["cron", "interval"]
+
+
+@dataclass(frozen=True)
+class DueQuery:
+    query: Query
+    scheduled_time: float
+    mode: Mode
+
 
 class PostgresDataObservability(DBMAsyncJob):
     def __init__(self, check: PostgreSql, config: InstanceConfig):
         self._check = check
         self._config = config
-        self._last_execution: dict[int, float] = {}
+        self._last_execution: dict[int, float] = {}  # interval mode: last fire timestamp
         collection_interval = config.data_observability.collection_interval or 10
         super(PostgresDataObservability, self).__init__(
             check,
@@ -37,6 +54,8 @@ def __init__(self, check: PostgreSql, config: InstanceConfig):
             expected_db_exceptions=(psycopg.errors.DatabaseError,),
             job_name="data-observability",
         )
+        # Filter bad queries on check construction.
+        self._queries, self._schedulers = self._filter_valid_queries(self._do_config.queries or ())
 
     def _shutdown(self):
         self._check = None
@@ -45,14 +64,51 @@ def _shutdown(self):
     def _do_config(self):
         return self._config.data_observability
 
-    def _get_due_queries(self) -> list[Query]:
-        queries = self._do_config.queries or ()
-        now = time.time()
-        due = []
+    def _filter_valid_queries(self, queries: Iterable[Query]) -> tuple[tuple[Query, ...], dict[int, CronScheduler]]:
+        valid: list[Query] = []
+        schedulers: dict[int, CronScheduler] = {}
         for q in queries:
-            last_run = self._last_execution.get(q.monitor_id, 0.0)
-            if now - last_run >= q.interval_seconds:
-                due.append(q)
+            if q.schedule:
+                try:
+                    schedulers[q.monitor_id] = CronScheduler(q.schedule, startup_lookback=CRON_STARTUP_LOOKBACK_SECONDS)
+                except (ValueError, TypeError) as e:
+                    self._log.warning(
+                        "Skipping DO query monitor_id=%d: invalid cron schedule %r (%s). "
+                        "Check the schedule of Data Observability monitor %d.",
+                        q.monitor_id,
+                        q.schedule,
+                        e,
+                        q.monitor_id,
+                    )
+                    continue
+            elif not (q.interval_seconds and q.interval_seconds > 0):
+                self._log.warning(
+                    "Skipping DO query monitor_id=%d: neither schedule nor positive interval_seconds set",
+                    q.monitor_id,
+                )
+                continue
+            valid.append(q)
+        return tuple(valid), schedulers
+
+    def _get_due_queries(self) -> list[DueQuery]:
+        now = time.time()
+        due: list[DueQuery] = []
+        for q in self._queries:
+            if q.schedule:
+                # +0.001 so a poll landing exactly on a tick boundary is treated
+                # as due (CronScheduler.previous_tick uses strict less-than).
+                ticks = self._schedulers[q.monitor_id].due_ticks(now + 0.001)
+                if ticks:
+                    # Take the latest elapsed tick; earlier ones are already in the past
+                    # and do not need separate execution.
+                    due.append(DueQuery(q, ticks[-1], "cron"))
+            else:
+                last = self._last_execution.get(q.monitor_id)
+                if last is None or now - last >= q.interval_seconds:
+                    # Seed: treat first sight as if the previous interval just completed,
+                    # so the scheduled_time for DueQuery is now and lateness is 0.
+                    scheduled = (last + q.interval_seconds) if last is not None else now
+                    due.append(DueQuery(q, scheduled, "interval"))
         return due
 
     def _build_base_tags(self) -> list[str]:
@@ -147,16 +203,20 @@ def run_job(self):
 
         base_tags = self._build_base_tags()
 
-        for q in due_queries:
+        for due in due_queries:
+            q = due.query
             tags = base_tags + [f'monitor_id:{q.monitor_id}']
 
+            now_at_fire_start = time.time()
             with self._check.db_pool.get_connection(q.dbname) as conn:
                 result = self._execute_single_query(conn, q)
 
-            # Update scheduling timestamp immediately after execution, before
-            # metric/event emission, so a serialization failure in the event
-            # path cannot cause infinite re-execution of the same query.
-            self._last_execution[q.monitor_id] = time.time()
+            # Advance scheduling state before emission so an emit-side error cannot
+            # leave the query stuck re-firing the same tick.
+            # For cron mode, due_ticks() already advanced the scheduler's internal state.
+            now_at_fire_end = time.time()
+            if due.mode == "interval":
+                self._last_execution[q.monitor_id] = now_at_fire_end
 
             try:
                 self._check.gauge(
@@ -174,6 +234,17 @@ def run_job(self):
                     raw=True,
                 )
 
+                # Lateness measures scheduling delay only (time from tick to query start),
+                # not end-to-end result latency — query execution time is reported separately.
+                lateness = max(0.0, now_at_fire_start - due.scheduled_time)
+                self._check.gauge(
+                    'dd.postgres.data_observability.query_fire_lateness_seconds',
+                    lateness,
+                    tags=tags + [f'mode:{due.mode}'],
+                    hostname=self._check.reported_hostname,
+                    raw=True,
+                )
+
                 payload = self._build_event_payload(q, result)
                 raw_event = json.dumps(payload, default=default_json_event_encoding)
                 self._log.debug(
@@ -183,8 +254,18 @@ def run_job(self):
                     result['row_count'],
                 )
                 self._check.event_platform_event(raw_event, EVENT_TRACK_TYPE)
-            except Exception:
+            except Exception as e:
                 self._log.exception(
                     "Failed to emit metrics/event for monitor_id=%d",
                     q.monitor_id,
                 )
+                try:
+                    self._check.count(
+                        'dd.postgres.data_observability.emit_failures',
+                        1,
+                        tags=tags + [f'exc_class:{type(e).__name__}'],
+                        hostname=self._check.reported_hostname,
+                        raw=True,
+                    )
+                except Exception:
+                    pass

postgres/tests/test_config.py

@@ -610,3 +610,99 @@ def test_autodiscovery_exclude_none_does_not_error(mock_check):
     config, result = build_config(check=mock_check)
     assert result.valid
     assert config.dbname == 'main'
+
+
+# ---------------------------------------------------------------------------
+# Data Observability — schedule field config tests
+# ---------------------------------------------------------------------------
+
+
+def test_do_query_schedule_field_defaults_to_none(mock_check, minimal_instance):
+    """A DO query without schedule field has schedule=None by default."""
+    minimal_instance['data_observability'] = {
+        'enabled': True,
+        'run_sync': True,
+        'queries': [
+            {
+                'monitor_id': 1,
+                'dbname': 'mydb',
+                'query': 'SELECT 1',
+                'interval_seconds': 60,
+                'entity': {
+                    'platform': 'aws',
+                    'account': '123',
+                    'database': 'mydb',
+                    'schema': 'public',
+                    'table': 'foo',
+                },
+            }
+        ],
+    }
+    mock_check.instance = minimal_instance
+    mock_check.init_config = {}
+    config, result = build_config(check=mock_check)
+    assert result.valid
+    query = config.data_observability.queries[0]
+    assert query.schedule is None
+    assert query.interval_seconds == 60
+
+
+def test_do_query_schedule_field_parsed(mock_check, minimal_instance):
+    """A DO query with schedule field is parsed correctly; interval_seconds may be absent."""
+    minimal_instance['data_observability'] = {
+        'enabled': True,
+        'run_sync': True,
+        'queries': [
+            {
+                'monitor_id': 2,
+                'dbname': 'mydb',
+                'query': 'SELECT 1',
+                'schedule': '20 * * * *',
+                'entity': {
+                    'platform': 'aws',
+                    'account': '123',
+                    'database': 'mydb',
+                    'schema': 'public',
+                    'table': 'bar',
+                },
+            }
+        ],
+    }
+    mock_check.instance = minimal_instance
+    mock_check.init_config = {}
+    config, result = build_config(check=mock_check)
+    assert result.valid
+    query = config.data_observability.queries[0]
+    assert query.schedule == '20 * * * *'
+    assert query.interval_seconds is None
+
+
+def test_do_query_both_schedule_and_interval_parsed(mock_check, minimal_instance):
+    """A DO query with both schedule and interval_seconds is accepted; both fields present."""
+    minimal_instance['data_observability'] = {
+        'enabled': True,
+        'run_sync': True,
+        'queries': [
+            {
+                'monitor_id': 3,
+                'dbname': 'mydb',
+                'query': 'SELECT 1',
+                'schedule': '0 * * * *',
+                'interval_seconds': 3600,
+                'entity': {
+                    'platform': 'aws',
+                    'account': '123',
+                    'database': 'mydb',
+                    'schema': 'public',
+                    'table': 'baz',
+                },
+            }
+        ],
+    }
+    mock_check.instance = minimal_instance
+    mock_check.init_config = {}
+    config, result = build_config(check=mock_check)
+    assert result.valid
+    query = config.data_observability.queries[0]
+    assert query.schedule == '0 * * * *'
+    assert query.interval_seconds == 3600