HUBBLE 444 - Refactor Elementary monitoring to run every 30 min

dags/dbt_data_quality_alerts_dag.py

@@ -0,0 +1,37 @@
+from datetime import datetime
+
+from airflow import DAG
+from airflow.operators.empty import EmptyOperator
+from kubernetes.client import models as k8s
+from stellar_etl_airflow.build_dbt_task import dbt_task
+from stellar_etl_airflow.build_elementary_slack_alert_task import elementary_task
+from stellar_etl_airflow.default import get_default_dag_args, init_sentry
+
+init_sentry()
+
+dag = DAG(
+    "dbt_data_quality_alerts",
+    default_args=get_default_dag_args(),
+    start_date=datetime(2024, 6, 11, 0, 0),
+    description="This DAG runs dbt tests and Elementary alerts at a half-hourly cadence",
+    schedule_interval="*/30 * * * *",
+    user_defined_filters={
+        "container_resources": lambda s: k8s.V1ResourceRequirements(requests=s),
+    },
+    max_active_runs=1,
+    catchup=False,
+    tags=["dbt-data-quality", "dbt-elementary-alerts"],
+)
+
+
+# DBT tests to run
+dbt_unit_tests = dbt_task(
+    dag,
+    command_type="test",
+    tag="singular_test",
+)
+unit_tests_elementary_alerts = elementary_task(dag, "dbt_data_quality")
+start_tests = EmptyOperator(task_id="start_tests_task")
+
+# DAG task graph
+start_tests >> dbt_unit_tests >> unit_tests_elementary_alerts

dags/dbt_enriched_base_tables_dag.py

@@ -4,7 +4,6 @@
 from kubernetes.client import models as k8s
 from stellar_etl_airflow.build_cross_dependency_task import build_cross_deps
 from stellar_etl_airflow.build_dbt_task import dbt_task
-from stellar_etl_airflow.build_elementary_slack_alert_task import elementary_task
 from stellar_etl_airflow.default import get_default_dag_args, init_sentry
 
 init_sentry()
@@ -30,16 +29,12 @@
 wait_on_state_table = build_cross_deps(dag, "wait_on_state_table", "state_table_export")
 
 # DBT models to run
-enriched_history_operations_task = dbt_task(dag, tag="enriched_history_operations")
+enriched_history_operations_task = dbt_task(
+    dag, tag="enriched_history_operations", excluded="singular_test"
+)
 current_state_task = dbt_task(dag, tag="current_state")
 
-elementary = elementary_task(dag, "dbt_enriched_base_tables")
-
 # DAG task graph
 wait_on_history_table >> enriched_history_operations_task
 
 wait_on_state_table >> current_state_task
-
-enriched_history_operations_task >> elementary
-
-current_state_task >> elementary

dags/dbt_stellar_marts_dag.py

@@ -4,7 +4,6 @@
 from kubernetes.client import models as k8s
 from stellar_etl_airflow.build_cross_dependency_task import build_cross_deps
 from stellar_etl_airflow.build_dbt_task import dbt_task
-from stellar_etl_airflow.build_elementary_slack_alert_task import elementary_task
 from stellar_etl_airflow.default import get_default_dag_args, init_sentry
 
 init_sentry()
@@ -49,8 +48,6 @@
 soroban = dbt_task(dag, tag="soroban")
 snapshot_state = dbt_task(dag, tag="snapshot_state")
 
-elementary = elementary_task(dag, "dbt_stellar_marts")
-
 # DAG task graph
 wait_on_dbt_enriched_base_tables >> ohlc_task >> liquidity_pool_trade_volume_task
 
@@ -68,17 +65,3 @@
 wait_on_dbt_enriched_base_tables >> history_assets
 wait_on_dbt_enriched_base_tables >> soroban
 wait_on_dbt_enriched_base_tables >> snapshot_state
-
-mgi_task >> elementary
-liquidity_providers_task >> elementary
-liquidity_pools_values_task >> elementary
-liquidity_pools_value_history_task >> elementary
-trade_agg_task >> elementary
-fee_stats_agg_task >> elementary
-asset_stats_agg_task >> elementary
-network_stats_agg_task >> elementary
-partnership_assets_task >> elementary
-history_assets >> elementary
-soroban >> elementary
-liquidity_pool_trade_volume_task >> elementary
-snapshot_state >> elementary

dags/stellar_etl_airflow/build_dbt_task.py

@@ -64,6 +64,7 @@ def dbt_task(
     flag="select",
     operator="",
     command_type="build",
+    excluded=None,
     resource_cfg="default",
 ):
     namespace = conf.get("kubernetes", "NAMESPACE")
@@ -97,6 +98,15 @@ def dbt_task(
         args.append(",".join(models))
     else:
         args.append(models[0])
+    # --exclude selector added for necessary use cases
+    # Argument should be string or list of strings
+    if excluded:
+        task_name = f"{task_name}_with_exclude"
+        args.append("--exclude")
+        if isinstance(excluded, list):
+            args.append(",".join(excluded))
+        else:
+            args.append(excluded)
 
     if Variable.get("dbt_full_refresh_models", deserialize_json=True).get(task_name):
         args.append("--full-refresh")

dags/stellar_etl_airflow/build_elementary_slack_alert_task.py

@@ -3,9 +3,7 @@
 
 from airflow.configuration import conf
 from airflow.models import Variable
-from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import (
-    KubernetesPodOperator,
-)
+from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
 from kubernetes import client, config
 from kubernetes.client import models as k8s
 from stellar_etl_airflow.default import alert_after_max_retries
@@ -87,4 +85,5 @@ def elementary_task(
         on_failure_callback=alert_after_max_retries,
         image_pull_policy="IfNotPresent",
         image_pull_secrets=[k8s.V1LocalObjectReference("private-docker-auth")],
+        trigger_rule="all_done",
     )