Merge branch 'aarthy/fixes' into 'enterprise'

fix(monitors): handle errored monitors

See merge request dkinternal/testgen/dataops-testgen!407

Author: ci bot

testgen/__main__.py

@@ -32,7 +32,7 @@
 from testgen.commands.run_test_execution import run_test_execution
 from testgen.commands.run_test_metadata_exporter import run_test_metadata_exporter
 from testgen.commands.run_upgrade_db_config import get_schema_revision, is_db_revision_up_to_date, run_upgrade_db_config
-from testgen.commands.test_generation import run_test_generation
+from testgen.commands.test_generation import run_monitor_generation, run_test_generation
 from testgen.common import (
     configure_logging,
     display_service,
@@ -176,6 +176,20 @@ def run_generation(test_suite_id: str | None = None, table_group_id: str | None
     click.echo("\n" + message)
 
 
+@cli.command("run-monitor-generation", help="Generates or refreshes the monitors for a table group.")
+@click.option(
+    "-t",
+    "--test-suite-id",
+    required=True,
+    type=click.STRING,
+    help="ID of the monitor suite to generate",
+)
+@with_database_session
+def generate_monitors(test_suite_id: str):
+    click.echo(f"run-monitor-generation for suite: {test_suite_id}")
+    run_monitor_generation(test_suite_id, ["Freshness_Trend", "Volume_Trend", "Schema_Drift"])
+
+
 @register_scheduler_job
 @cli.command("run-tests", help="Performs tests defined for a test suite.")
 @click.option(

testgen/commands/queries/execute_tests_query.py

@@ -180,6 +180,10 @@ def has_schema_changes(self) -> tuple[dict]:
         # Runs on App database
         return self._get_query("has_schema_changes.sql")
 
+    def get_errored_autogen_monitors(self) -> tuple[str, dict]:
+        # Runs on App database
+        return self._get_query("get_errored_autogen_monitors.sql")
+
     def get_active_test_definitions(self) -> tuple[dict]:
         # Runs on App database
         return self._get_query("get_active_test_definitions.sql")

testgen/commands/run_test_execution.py

@@ -1,6 +1,7 @@
 import logging
 import subprocess
 import threading
+from collections import defaultdict
 from datetime import UTC, datetime, timedelta
 from functools import partial
 from typing import Literal
@@ -86,12 +87,7 @@ def run_test_execution(test_suite_id: str | UUID, username: str | None = None, r
         sql_generator = TestExecutionSQL(connection, table_group, test_run)
 
         if test_suite.is_monitor:
-            schema_changes = fetch_dict_from_db(*sql_generator.has_schema_changes())[0]
-            if schema_changes["has_table_drops"]:
-                run_monitor_generation(test_suite_id, ["Freshness_Trend", "Volume_Trend", "Metric_Trend"], mode="delete")
-            if schema_changes["has_table_adds"]:
-                # Freshness monitors will be inserted after profiling
-                run_monitor_generation(test_suite_id, ["Volume_Trend"], mode="insert")
+            _sync_monitor_definitions(sql_generator)
 
         # Update the thresholds before retrieving the test definitions in the next steps
         LOG.info("Updating test thresholds based on history calculations")
@@ -190,6 +186,26 @@ def run_test_execution(test_suite_id: str | UUID, username: str | None = None, r
     """
 
 
+def _sync_monitor_definitions(sql_generator: TestExecutionSQL) -> None:
+    test_suite_id = sql_generator.test_run.test_suite_id
+
+    schema_changes = fetch_dict_from_db(*sql_generator.has_schema_changes())[0]
+    if schema_changes["has_table_drops"]:
+        run_monitor_generation(test_suite_id, ["Freshness_Trend", "Volume_Trend", "Metric_Trend"], mode="delete")
+    if schema_changes["has_table_adds"]:
+        # Freshness monitors will be inserted after profiling
+        run_monitor_generation(test_suite_id, ["Volume_Trend"], mode="insert")
+
+    # Regenerate monitors that errored in previous run
+    errored_monitors = fetch_dict_from_db(*sql_generator.get_errored_autogen_monitors())
+    if errored_monitors:
+        errored_by_type: dict[str, list[str]] = defaultdict(list)
+        for row in errored_monitors:
+            errored_by_type[row["test_type"]].append(row["table_name"])
+        for test_type, table_names in errored_by_type.items():
+            run_monitor_generation(test_suite_id, [test_type], mode="upsert", table_names=table_names)
+
+
 def _run_tests(
     sql_generator: TestExecutionSQL,
     run_type: Literal["QUERY", "METADATA"],

testgen/commands/test_generation.py

@@ -72,6 +72,7 @@ def run_monitor_generation(
     monitor_suite_id: str | UUID,
     monitors: list[MonitorTestType],
     mode: MonitorGenerationMode = "upsert",
+    table_names: list[str] | None = None,
 ) -> None:
     """
     Modes:
@@ -90,7 +91,7 @@ def run_monitor_generation(
     table_group = TableGroup.get(monitor_suite.table_groups_id)
     connection = Connection.get(table_group.connection_id)
 
-    TestGeneration(connection, table_group, monitor_suite, "Monitor", monitors).monitor_run(mode)
+    TestGeneration(connection, table_group, monitor_suite, "Monitor", monitors).monitor_run(mode, table_names=table_names)
 
 
 class TestGeneration:
@@ -123,16 +124,19 @@ def run(self) -> None:
             self._get_query("delete_stale_autogen_tests.sql"),
         ])
 
-    def monitor_run(self, mode: MonitorGenerationMode) -> None:
+    def monitor_run(self, mode: MonitorGenerationMode, table_names: list[str] | None = None) -> None:
         if mode == "delete":
             execute_db_queries([self._get_query("delete_stale_monitors.sql")])
             return
 
+        extra_params = {"INSERT_ONLY": mode == "insert"}
+        if table_names:
+            table_list = ", ".join(f"'{table}'" for table in table_names)
+            extra_params["TABLE_FILTER"] = f"AND table_name IN ({table_list})"
+
         LOG.info("Running monitor generation queries")
         execute_db_queries(
-            self._get_generation_queries(
-                extra_params={"INSERT_ONLY": mode == "insert"},
-            ),
+            self._get_generation_queries(extra_params=extra_params),
         )
 
     def _get_generation_queries(self, extra_params: dict | None = None) -> list[tuple[str, dict]]:
@@ -188,6 +192,7 @@ def _get_params(self, test_type: TestTypeParams | None = None) -> dict:
             "SQL_FLAVOR": self.flavor,
             "QUOTE": self.flavor_service.quote_character,
             "INSERT_ONLY": False,
+            "TABLE_FILTER": "",
         })
         return params
 

testgen/common/models/table_group.py

@@ -71,6 +71,10 @@ class TableGroupSummary(EntityMinimal):
     monitor_schema_anomalies: int | None
     monitor_volume_anomalies: int | None
     monitor_metric_anomalies: int | None
+    monitor_freshness_has_errors: bool | None
+    monitor_volume_has_errors: bool | None
+    monitor_schema_has_errors: bool | None
+    monitor_metric_has_errors: bool | None
     monitor_freshness_is_training: bool | None
     monitor_volume_is_training: bool | None
     monitor_metric_is_training: bool | None
@@ -266,6 +270,10 @@ def select_summary(cls, project_code: str, for_dashboard: bool = False) -> Itera
                 SUM(CASE WHEN results.test_type = 'Schema_Drift' AND results.result_code = 0 THEN 1 ELSE 0 END) AS schema_anomalies,
                 SUM(CASE WHEN results.test_type = 'Volume_Trend' AND results.result_code = 0 THEN 1 ELSE 0 END) AS volume_anomalies,
                 SUM(CASE WHEN results.test_type = 'Metric_Trend' AND results.result_code = 0 THEN 1 ELSE 0 END) AS metric_anomalies,
+                BOOL_OR(results.result_status = 'Error') FILTER (WHERE results.test_type = 'Freshness_Trend' AND ranked_test_runs.position = 1) AS freshness_has_errors,
+                BOOL_OR(results.result_status = 'Error') FILTER (WHERE results.test_type = 'Volume_Trend' AND ranked_test_runs.position = 1) AS volume_has_errors,
+                BOOL_OR(results.result_status = 'Error') FILTER (WHERE results.test_type = 'Schema_Drift' AND ranked_test_runs.position = 1) AS schema_has_errors,
+                BOOL_OR(results.result_status = 'Error') FILTER (WHERE results.test_type = 'Metric_Trend' AND ranked_test_runs.position = 1) AS metric_has_errors,
                 BOOL_AND(results.result_code = -1) FILTER (WHERE results.test_type = 'Freshness_Trend' AND ranked_test_runs.position = 1) AS freshness_is_training,
                 BOOL_AND(results.result_code = -1) FILTER (WHERE results.test_type = 'Volume_Trend' AND ranked_test_runs.position = 1) AS volume_is_training,
                 BOOL_AND(results.result_code = -1) FILTER (WHERE results.test_type = 'Metric_Trend' AND ranked_test_runs.position = 1) AS metric_is_training,
@@ -319,6 +327,10 @@ def select_summary(cls, project_code: str, for_dashboard: bool = False) -> Itera
             monitor_tables.schema_anomalies AS monitor_schema_anomalies,
             monitor_tables.volume_anomalies AS monitor_volume_anomalies,
             monitor_tables.metric_anomalies AS monitor_metric_anomalies,
+            monitor_tables.freshness_has_errors AS monitor_freshness_has_errors,
+            monitor_tables.volume_has_errors AS monitor_volume_has_errors,
+            monitor_tables.schema_has_errors AS monitor_schema_has_errors,
+            monitor_tables.metric_has_errors AS monitor_metric_has_errors,
             monitor_tables.freshness_is_training AS monitor_freshness_is_training,
             monitor_tables.volume_is_training AS monitor_volume_is_training,
             monitor_tables.metric_is_training AS monitor_metric_is_training,

testgen/common/notifications/monitor_run.py

@@ -230,6 +230,7 @@ def send_monitor_notifications(test_run: TestRun, result_list_ct=20):
                 str(table_group.id),
                 "&table_name_filter=" if table_name else "",
                 table_name if table_name else "",
+                "&source=email",
             )
         )
         try:

testgen/common/notifications/profiling_run.py

@@ -258,7 +258,7 @@ def send_profiling_run_notifications(profiling_run: ProfilingRun, result_list_ct
         return
 
     profiling_run_issues_url = "".join(
-        (PersistedSetting.get("BASE_URL", ""), "/profiling-runs:hygiene?run_id=", str(profiling_run.id))
+        (PersistedSetting.get("BASE_URL", ""), "/profiling-runs:hygiene?run_id=", str(profiling_run.id), "&source=email")
     )
 
     hygiene_issues_summary = []
@@ -304,7 +304,7 @@ def send_profiling_run_notifications(profiling_run: ProfilingRun, result_list_ct
             "id": str(profiling_run.id),
             "issues_url": profiling_run_issues_url,
             "results_url": "".join(
-                (PersistedSetting.get("BASE_URL", ""), "/profiling-runs:results?run_id=", str(profiling_run.id))
+                (PersistedSetting.get("BASE_URL", ""), "/profiling-runs:results?run_id=", str(profiling_run.id), "&source=email")
             ),
             "start_time": profiling_run.profiling_starttime,
             "end_time": profiling_run.profiling_endtime,

testgen/common/notifications/score_drop.py

@@ -180,6 +180,7 @@ def send_score_drop_notifications(notification_data: list[tuple[ScoreDefinition,
                         PersistedSetting.get("BASE_URL", ""),
                         "/quality-dashboard:score-details?definition_id=",
                         str(definition.id),
+                        "&source=email",
                     )
                 ),
                 "diff": context_diff,

testgen/common/notifications/test_run.py

@@ -328,6 +328,7 @@ def send_test_run_notifications(test_run: TestRun, result_list_ct=20, result_sta
             PersistedSetting.get("BASE_URL", ""),
             "/test-runs:results?run_id=",
             str(test_run.id),
+            "&source=email",
         )
     )
 

testgen/template/execution/get_errored_autogen_monitors.sql

@@ -0,0 +1,15 @@
+WITH prev_run AS (
+    SELECT id
+    FROM test_runs
+    WHERE test_suite_id = :TEST_SUITE_ID ::UUID
+        AND id <> :TEST_RUN_ID ::UUID
+        AND status = 'Complete'
+    ORDER BY test_starttime DESC
+    LIMIT 1
+)
+SELECT DISTINCT tr.test_type, tr.table_name
+FROM test_results tr
+INNER JOIN prev_run ON tr.test_run_id = prev_run.id
+WHERE tr.result_status = 'Error'
+    AND tr.auto_gen IS TRUE
+    AND tr.test_type IN ('Freshness_Trend', 'Volume_Trend')