feat(argowf_pipeline): Execute pipeline via ArgoWF

skit_pipelines/api/argowf.py

@@ -0,0 +1,30 @@
+from loguru import logger
+from constants import SKIT_PIPELINES_ARGO_MANIFEST
+
+
+def generate_args_list(pipeline_name: str, payload: dict):
+    cli_args_list = []
+    logger.debug(f"Incoming Payload - {payload}")
+    pipeline_name_cli_arg = f"--pipeline_name={pipeline_name}"
+    cli_args_list.append(pipeline_name_cli_arg)
+    for key, value in payload.items():
+        cli_arg_string = f"--{key}={value}"
+        cli_args_list.append(cli_arg_string)
+    logger.debug(f"Final args list - {cli_args_list}")
+    return cli_args_list
+
+
+def prepare_argo_payload(final_args_list):
+    logger.debug(f"SKIT PIPELINES MANIFEST - {SKIT_PIPELINES_ARGO_MANIFEST}")
+    workflow_spec = SKIT_PIPELINES_ARGO_MANIFEST.get("workflow", {}).get("spec", {})
+    skit_pipeline_template = workflow_spec.get("templates", [{}])[0]
+    args_list = skit_pipeline_template.get("container", {}).get("args", [])
+    args_list.extend(final_args_list)
+    logger.debug(f"Final modified SKIT_PIPELINES_ARGO_MANIFEST - {SKIT_PIPELINES_ARGO_MANIFEST}")
+    return SKIT_PIPELINES_ARGO_MANIFEST
+
+
+def initiate_workflow_via_argo(pipeline_name, payload):
+    cli_args_list = generate_args_list(pipeline_name, payload)
+    final_argo_manifest = prepare_argo_payload(cli_args_list)
+    return final_argo_manifest

skit_pipelines/api/constants.py

@@ -0,0 +1,18 @@
+import os
+import json
+from loguru import logger
+
+TRIGGER_WORKFLOW_VIA_ARGO = os.environ.get("TRIGGER_WORKFLOW_VIA_ARGO", "false").lower()
+if TRIGGER_WORKFLOW_VIA_ARGO == "true":
+    TRIGGER_WORKFLOW_VIA_ARGO = True
+else:
+    TRIGGER_WORKFLOW_VIA_ARGO = False
+
+
+try:
+    with open("./resources/skit_pipelines_argo_manifest.json", "r+") as f:
+        SKIT_PIPELINES_ARGO_MANIFEST = json.load(f)
+except Exception as e:
+    logger.error(f"Some exception occurred when attempting to open Argo Manifest - {e}")
+    SKIT_PIPELINES_ARGO_MANIFEST = {}
+

skit_pipelines/api/slack_bot.py

@@ -8,6 +8,8 @@
 import requests
 from jsoncomment import JsonComment
 from loguru import logger
+from constants import TRIGGER_WORKFLOW_VIA_ARGO
+from argowf import initiate_workflow_via_argo
 
 import skit_pipelines.constants as const
 from skit_pipelines.api.validate_input import ValidateInput
@@ -103,9 +105,14 @@ def run_pipeline(
     payload["notify"] = (
         user if "notify" not in payload else f"{payload['notify']} ,{user}"
     )
-    url_path = f"http://localhost:{server_port}/skit/pipelines/run/{pipeline_name}/"
-    access_token = read_access_token()
-    res = make_run_requests(url_path, payload, access_token)
+
+    if TRIGGER_WORKFLOW_VIA_ARGO:
+        response = initiate_workflow_via_argo(pipeline_name, payload)
+        return response
+    else:
+        url_path = f"http://localhost:{server_port}/skit/pipelines/run/{pipeline_name}/"
+        access_token = read_access_token()
+        res = make_run_requests(url_path, payload, access_token)
 
     if res.status_code == 401:
         new_access_token = read_access_token(token_fetch=True)