postgres data-observability - address review feedback on cron schedule support

- Replace inaccurate epoch literal/banner in test_data_observability.py with a
  one-line "why" comment.
- Drop unused fake_time parameter on _make_cron_check.
- Remove narrative debugging comments and dead fake_clock/_next_run mutations
  in test_schedule_advances_after_run, test_lateness_metric_emitted_for_cron,
  and test_lateness_clamped_at_zero.
- Generalize the stale "Update scheduling timestamp" comment in run_job to
  cover both last_execution and next_run; drop the redundant fall-through
  comment in _get_due_queries.
- Tighten spec.yaml descriptions for interval_seconds/schedule to match
  runtime behavior (skipped with warning if neither is set) instead of
  over-promising "at least one must be set".
- Log a warning when cron_startup_lookback_seconds is negative instead of
  silently clamping to 0.
- Add three regression tests: default-is-300, negative-clamped-with-warning,
  no-double-fire after a startup recovery.

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

Author: mobuchowski

postgres/assets/configuration/spec.yaml

@@ -653,15 +653,16 @@ files:
                 type: string
               - name: interval_seconds
                 description: |
-                  How often (in seconds) to run this query. At least one of interval_seconds or
-                  schedule must be set. When both are present, schedule takes precedence and
-                  interval_seconds is ignored.
+                  How often (in seconds) to run this query. When both schedule and
+                  interval_seconds are present, schedule takes precedence and interval_seconds
+                  is ignored. If neither is set, the query is skipped at runtime with a warning.
                 type: integer
               - name: schedule
                 description: |
                   A standard 5-field cron expression (minute hour dom month dow) specifying
                   when to run this query. When set, takes precedence over interval_seconds.
-                  At least one of schedule or interval_seconds must be set.
+                  If neither schedule nor interval_seconds is set, the query is skipped at
+                  runtime with a warning.
                 type: string
               - name: entity
                 type: object

postgres/datadog_checks/postgres/data_observability.py

@@ -73,7 +73,13 @@ def _cron_startup_lookback_seconds(self) -> int:
         configured = self._do_config.cron_startup_lookback_seconds
         if configured is None:
             return DEFAULT_CRON_STARTUP_LOOKBACK_SECONDS
-        return max(0, configured)
+        if configured < 0:
+            self._log.warning(
+                "Invalid cron_startup_lookback_seconds=%d: must be >= 0. Cron startup recovery is disabled.",
+                configured,
+            )
+            return 0
+        return configured
 
     def _get_due_queries(self) -> list[tuple[Query, float]]:
         queries = self._do_config.queries or ()
@@ -99,7 +105,6 @@ def _get_due_queries(self) -> list[tuple[Query, float]]:
                     # First fire (no prior last_run): treat scheduled time as `now` → lateness = 0.
                     scheduled = (last_run + q.interval_seconds) if last_run > 0 else now
                     due.append((q, scheduled))
-            # mode is None → invalid query, already warned, skip
         return due
 
     def _build_base_tags(self) -> list[str]:
@@ -201,9 +206,9 @@ def run_job(self):
             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.
+            # Update scheduling state (last_execution and, for cron, next_run) before
+            # metric/event emission, so a serialization failure in the emit path
+            # cannot cause infinite re-execution of the same query.
             now_at_fire_end = time.time()
             self._last_execution[q.monitor_id] = now_at_fire_end
             if q.schedule:

postgres/tests/test_data_observability.py

@@ -16,17 +16,8 @@
 from datadog_checks.postgres import PostgreSql
 from datadog_checks.postgres.data_observability import EVENT_TRACK_TYPE
 
-# ---------------------------------------------------------------------------
-# Cron schedule / lateness helpers
-# ---------------------------------------------------------------------------
-# A "frozen" epoch that sits between two hourly boundaries:
-#   xx:49:00  (previous hour + 49 min)
-#   xx:50:00  (next cron tick for "50 * * * *")
-#   xx:51:00  (tick for "51 * * * *")
-#
-# We use an absolute epoch so croniter's calendar arithmetic stays stable.
-# 2026-01-01 00:49:00 UTC  →  1751331940  (approx; exact value computed below)
-_BASE_EPOCH = calendar.timegm(datetime.datetime(2026, 1, 1, 0, 49, 0).timetuple())  # 00:49:00 UTC
+# Fixed epoch (2026-01-01 00:49:00 UTC) chosen to sit 1 minute before the ":50" cron tick used in tests.
+_BASE_EPOCH = calendar.timegm(datetime.datetime(2026, 1, 1, 0, 49, 0).timetuple())
 
 pytestmark = pytest.mark.unit
 
@@ -531,12 +522,11 @@ def test_multi_query_different_dbnames(aggregator, pg_instance):
 }
 
 
-def _make_cron_check(pg_instance, queries=None, fake_time=None):
-    """Create a check with a cron-scheduled query, optionally monkeypatching time.time."""
+def _make_cron_check(pg_instance, queries=None):
+    """Create a check with a cron-scheduled query."""
     if queries is None:
         queries = [deepcopy(CRON_QUERY)]
-    check = _create_check(pg_instance, queries=queries)
-    return check
+    return _create_check(pg_instance, queries=queries)
 
 
 def test_schedule_query_does_not_fire_before_tick(pg_instance, monkeypatch):
@@ -582,19 +572,11 @@ def test_schedule_advances_after_run(pg_instance, monkeypatch):
     check = _make_cron_check(pg_instance)
     check.db_pool = _mock_db_pool(mock_conn)
 
-    # First run: registers first future tick (00:50:00 is already past, so next is 01:50:00)
+    # First run at 00:50:05: registers next_run = 01:50:00 (croniter.get_next is strictly after now); no fire.
     check.data_observability.run_job()
-    # At this point next_run should be set (but nothing fired yet — it's first sight)
     mid = CRON_QUERY['monitor_id']
     first_next_run = check.data_observability._next_run[mid]
 
-    # Second run: now we're past the registered next_run (which was set to 00:50:00)
-    # Actually at 00:50:05 the first sight records 00:50:00 as next and skips. Let me re-check.
-    # At 00:50:05 > 00:49:00 = _BASE_EPOCH, so first sight computes next_run = 00:50:00 (already past).
-    # Wait — croniter(schedule, now).get_next() returns the first tick AFTER `now`.
-    # At now=00:50:05, get_next("50 * * * *") = 01:50:00 (the next tick after 00:50:05).
-    # So first call: next_run = 01:50:00 → skip (no fire).
-    # Advance to 01:50:05:
     current_time[0] = _BASE_EPOCH + 3665  # ~01:50:05
     check.data_observability.run_job()
 
@@ -698,16 +680,6 @@ def test_lateness_metric_emitted_for_cron(pg_instance, aggregator, monkeypatch):
     """Cron query fired late emits a positive lateness gauge with mode:cron tag."""
     # next_run will be at 00:50:00; fire at 00:52:00 → 120s lateness
     fire_time = float(_BASE_EPOCH + 180)  # 00:52:00
-
-    call_count = [0]
-
-    def fake_clock():
-        call_count[0] += 1
-        # First call: _get_due_queries registration at tick_time
-        # The sequence is complex; just use a simpler two-phase approach:
-        # Use monkeypatch to control time directly via a mutable list.
-        return current_time[0]
-
     current_time = [float(_BASE_EPOCH)]  # 00:49:00 — before the tick
     monkeypatch.setattr('datadog_checks.postgres.data_observability.time.time', lambda: current_time[0])
 
@@ -780,26 +752,9 @@ def test_lateness_clamped_at_zero(pg_instance, aggregator, monkeypatch):
     check = _make_cron_check(pg_instance)
     check.db_pool = _mock_db_pool(mock_conn)
 
-    # First call: registers next_run = 01:50:00 (future)
-    check.data_observability.run_job()
-    mid = CRON_QUERY['monitor_id']
-
-    # Manually set next_run to a value slightly in the future to simulate skew
-    # (i.e., scheduled_fire_time > now_at_fire_start)
-    check.data_observability._next_run[mid] = current_time[0] + 1000.0
-
-    # Force it to look "due" by setting now past that point but making now_at_fire_start < scheduled
-    # We achieve this by having _get_due_queries return a past-tick but then fire_start < scheduled_tick.
-    # Simpler approach: directly manipulate state so the query is seen as due with scheduled > fire_start.
-    # Set next_run to a past value so it's due:
-    past_tick = current_time[0] - 5.0
-    check.data_observability._next_run[mid] = past_tick
-    # But shift time so now_at_fire_start < past_tick (synthetic skew):
-    # Actually, just verify the clamp: if scheduled_fire_time > now_at_fire_start, value = 0.
-    # We construct this by running at a time just after next_run but having time.time()
-    # return a value LESS than next_run for now_at_fire_start — achieved by decrementing after due check.
-    # The cleanest way: patch _get_due_queries directly.
-    skewed_scheduled = current_time[0] + 100.0  # scheduled in the future
+    # Patch _get_due_queries to return a scheduled_fire_time in the future of fire_start
+    # (synthetic clock-skew scenario). Lateness must clamp to 0 rather than go negative.
+    skewed_scheduled = current_time[0] + 100.0
     with patch.object(
         check.data_observability,
         '_get_due_queries',
@@ -951,3 +906,41 @@ def test_cron_startup_lookback_advances_next_run_past_recovered_tick(pg_instance
     next_run = check.data_observability._next_run[mid]
     # Recovered fire at 00:50:10 should advance _next_run to 01:50:00, not to 00:50:00.
     assert next_run >= _BASE_EPOCH + 3600  # at least one hour past _BASE_EPOCH (00:49:00)
+
+
+def test_cron_startup_lookback_default_is_300_seconds(pg_instance):
+    """The default lookback window is 5 minutes; pin the constant so a regression is loud."""
+    from datadog_checks.postgres.data_observability import DEFAULT_CRON_STARTUP_LOOKBACK_SECONDS
+
+    assert DEFAULT_CRON_STARTUP_LOOKBACK_SECONDS == 300
+    check = _make_cron_lookback_check(pg_instance, lookback_seconds=None)
+    assert check.data_observability._cron_startup_lookback_seconds() == 300
+
+
+def test_cron_startup_lookback_negative_clamped_to_zero_with_warning(pg_instance, caplog):
+    """A negative lookback value clamps to 0 and emits a warning so it isn't silently dropped."""
+    import logging
+
+    check = _make_cron_lookback_check(pg_instance, lookback_seconds=-100)
+    with caplog.at_level(logging.WARNING, logger=check.data_observability._log.name):
+        result = check.data_observability._cron_startup_lookback_seconds()
+    assert result == 0
+    assert any('Invalid cron_startup_lookback_seconds' in r.message for r in caplog.records)
+
+
+def test_cron_startup_lookback_does_not_double_fire(pg_instance, aggregator, monkeypatch):
+    """A startup-recovery fire must not re-fire the same tick on the next run_job() call."""
+    current_time = [float(_BASE_EPOCH + 70)]  # 00:50:10 — inside the default 300s lookback
+    monkeypatch.setattr('datadog_checks.postgres.data_observability.time.time', lambda: current_time[0])
+
+    mock_conn, _ = _make_mock_conn()
+    check = _make_cron_lookback_check(pg_instance, lookback_seconds=None)
+    check.db_pool = _mock_db_pool(mock_conn)
+    check.data_observability.run_job()
+    assert len(aggregator.metrics('dd.postgres.data_observability.query_executions')) == 1
+
+    # Advance a few seconds (still well before the next 01:50 tick); a second run must not re-fire.
+    current_time[0] = _BASE_EPOCH + 80
+    aggregator.reset()
+    check.data_observability.run_job()
+    assert len(aggregator.metrics('dd.postgres.data_observability.query_executions')) == 0