Verify workflow success in CI on Airflow (#48)

* Adding airflow status checker

* Adding airflow status checker

* Adding airflow package to requirement

* Adding the dag run status via API route

* Adding the dag run status via API route

* Adding the dag auth as user and pass

* Adding final kill in case of long running DAG

* Added few lines and changed code as requested

.github/workflows/provider.yaml

@@ -64,6 +64,12 @@ jobs:
           astro dev run dags trigger lakeFS_workflow
           sleep 30
 
+      - name : Run DAG state check script
+        id : dag_status_id
+        run: |
+            chmod +x dag_status.py
+            python3 dag_status.py
+
       - name: Wait until Airflow makes output file available on main
         env:
           # To avoid the lack of region - see https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html

dag_status.py

@@ -0,0 +1,37 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+import sys
+import time
+import requests
+
+
+def get_latest_state():
+    url = 'http://localhost:8080/api/v1/dags/lakeFS_workflow/dagRuns'
+    username = 'admin'
+    password = 'admin'
+    response = requests.get(url, auth=(username, password))
+    dag_details = {}
+    latest = max(response.json()['dag_runs'], key=lambda k: \
+                 k['execution_date'])
+    state = latest['state']
+    return state
+
+
+def dag_state():
+    state = get_latest_state()
+    timeout = time.time() + 60 * 5  # 5 minutes from now
+    while state != 'success' and state != 'failed' and state \
+        != 'skipped':
+        time.sleep(5)
+        if time.time() > timeout:
+            return 1
+        state = get_latest_state()
+        continue
+    if state == 'success':
+        return 0
+    else:
+        return 1
+
+
+sys.exit(dag_state())

requirements.txt

@@ -1,2 +1,3 @@
 lakefs_client~=0.91.0
 setuptools~=56.0.0
+requests~=2.28.2