diff --git a/dags/elementary_report_dag.py b/dags/elementary_report_dag.py new file mode 100644 index 00000000..d94cb3ea --- /dev/null +++ b/dags/elementary_report_dag.py @@ -0,0 +1,32 @@ +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 ( + alert_sla_miss, + get_default_dag_args, + init_sentry, +) + +init_sentry() + +with DAG( + "elementary_report", + default_args=get_default_dag_args(), + start_date=datetime(2024, 6, 25, 0, 0), + description="This DAG creates elementary report and send it to slack", + schedule="*/60 * * * *", # Runs every 60 minutes + user_defined_filters={ + "container_resources": lambda s: k8s.V1ResourceRequirements(requests=s), + }, + max_active_runs=1, + catchup=False, +) as dag: + + # Trigger elementary + elementary_alerts = elementary_task(dag, "generate_report", resource_cfg="dbt",command="send-report") + + elementary_alerts diff --git a/dags/stellar_etl_airflow/build_elementary_slack_alert_task.py b/dags/stellar_etl_airflow/build_elementary_slack_alert_task.py index 2b6b88e5..9d1839aa 100644 --- a/dags/stellar_etl_airflow/build_elementary_slack_alert_task.py +++ b/dags/stellar_etl_airflow/build_elementary_slack_alert_task.py @@ -23,6 +23,7 @@ def elementary_task( dag, task_name, resource_cfg="default", + command="monitor" ): namespace = conf.get("kubernetes", "NAMESPACE") @@ -45,7 +46,7 @@ def elementary_task( slack_secret_name = Variable.get("dbt_elementary_secret") secret = access_secret(slack_secret_name, "default") args = [ - "monitor", + f"{command}", "--slack-token", f"{secret}", "--slack-channel-name",