Added SLA parameter to the relavant task building functions

stellar · May 9, 2024 · 8982c76 · 8982c76
1 parent 9911c3b
commit 8982c76
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 1 deletion.
diff --git a/dags/stellar_etl_airflow/build_dbt_task.py b/dags/stellar_etl_airflow/build_dbt_task.py
@@ -136,6 +136,9 @@ def dbt_task(
         on_failure_callback=alert_after_max_retries,
         image_pull_policy="IfNotPresent",
         image_pull_secrets=[k8s.V1LocalObjectReference("private-docker-auth")],
+        sla=timedelta(
+            seconds=Variable.get("task_sla", deserialize_json=True)[task_name]
+        ),
     )
 
 

diff --git a/dags/stellar_etl_airflow/build_elementary_slack_alert_task.py b/dags/stellar_etl_airflow/build_elementary_slack_alert_task.py
@@ -1,5 +1,6 @@
 import base64
 import logging
+from datetime import timedelta
 
 from airflow.configuration import conf
 from airflow.models import Variable
@@ -89,4 +90,9 @@ def elementary_task(
         on_failure_callback=alert_after_max_retries,
         image_pull_policy="IfNotPresent",
         image_pull_secrets=[k8s.V1LocalObjectReference("private-docker-auth")],
+        sla=timedelta(
+            seconds=Variable.get("task_sla", deserialize_json=True)[
+                f"elementary_{task_name}"
+            ]
+        ),
     )
diff --git a/dags/stellar_etl_airflow/build_export_task.py b/dags/stellar_etl_airflow/build_export_task.py
@@ -239,5 +239,7 @@ def build_export_task(
         affinity=affinity,
         on_failure_callback=alert_after_max_retries,
         image_pull_policy=Variable.get("image_pull_policy"),
-        sla=timedelta(minutes=4),
+        sla=timedelta(
+            seconds=Variable.get("task_sla", deserialize_json=True)["export_task"]
+        ),
     )
diff --git a/dags/stellar_etl_airflow/build_time_task.py b/dags/stellar_etl_airflow/build_time_task.py
@@ -88,4 +88,9 @@ def build_time_task(
         container_resources=resources_requests,
         on_failure_callback=alert_after_max_retries,
         image_pull_policy=Variable.get("image_pull_policy"),
+        sla=timedelta(
+            seconds=Variable.get("task_sla", deserialize_json=True)[
+                "get_ledger_range_from_times"
+            ]
+        ),
     )