Pipeline parameterize restructure

.env

@@ -9,9 +9,9 @@ DATA_DIR=./local_storage
 
 DUG_LOG_LEVEL=INFO
 
-ELASTIC_PASSWORD=12345
-ELASTIC_API_HOST=elasticsearch
-ELASTIC_USERNAME=elastic
+ELASTICSEARCH_PASSWORD=12345
+ELASTICSEARCH_HOST=elasticsearch
+ELASTICSEARCH_USERNAME=elastic
 
 NBOOST_API_HOST=nboost
 

.gitignore

@@ -1,4 +1,6 @@
 # Git ignore bioler plate from https://github.com/github/gitignore/blob/master/Python.gitignore
+.secret-env
+.vscode/
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

Dockerfile

@@ -1,11 +1,9 @@
-FROM apache/airflow:2.5.0-python3.10
+FROM apache/airflow:2.7.2-python3.11
+
 USER root
 RUN apt-get update && \
-    apt-get install -y git gcc python3-dev nano vim
+    apt-get install -y git nano vim 
 COPY requirements.txt requirements.txt
 USER airflow
-# dependency resolution taking hours eventually failing,
-# @TODO fix click lib dependency
-RUN pip install -r requirements.txt && \
-    pip uninstall -y elasticsearch-dsl
+RUN pip install -r requirements.txt
 RUN rm -f requirements.txt

Jenkinsfile

@@ -70,31 +70,15 @@ spec:
             steps {
                 script {
                     container(name: 'kaniko', shell: '/busybox/sh') {
-                        kaniko.buildAndPush("./Dockerfile", ["$IMAGE_NAME:$TAG1", "$IMAGE_NAME:$TAG2", "$IMAGE_NAME:$TAG3", "$IMAGE_NAME:$TAG4"])
+                        if (env.BRANCH_NAME == "main") {
+                          // Tag with latest and version iff when pushed to master
+                          kaniko.buildAndPush("./Dockerfile", ["$IMAGE_NAME:$TAG1", "$IMAGE_NAME:$TAG2", "$IMAGE_NAME:$TAG3", "$IMAGE_NAME:$TAG4"])
+                        } else {
+                          kaniko.buildAndPush("./Dockerfile", ["$IMAGE_NAME:$TAG1", "$IMAGE_NAME:$TAG2"])
+                        }
                     }
                 }
-            }
-            // post {
-            //     always {
-            //         archiveArtifacts artifacts: 'image.tar', onlyIfSuccessful: true
-            //     }
-            // }
-        }
-        // stage('Publish') {
-        //     steps {
-        //         script {
-        //             container(name: 'crane', shell: '/busybox/sh') {
-        //                 def imageTagsPushAlways = ["$IMAGE_NAME:$TAG1", "$IMAGE_NAME:$TAG2"]
-        //                 def imageTagsPushForDevelopBranch = ["$IMAGE_NAME:$TAG3"]
-        //                 def imageTagsPushForMasterBranch = ["$IMAGE_NAME:$TAG3", "$IMAGE_NAME:$TAG4"]
-        //                 image.publish(
-        //                     imageTagsPushAlways,
-        //                     imageTagsPushForDevelopBranch,
-        //                     imageTagsPushForMasterBranch
-        //                 )
-        //             }
-        //         }
-        //     }
-        // }
+            }          
+        }        
     }
 }

dags/annotate_and_index.py

@@ -0,0 +1,44 @@
+"""DAG which performs Dug annotate and index operations
+
+This DAG differes slightly from prior versions of the same functionality in
+Roger not only in that the annotation and indexing happen in the same DAG, but
+also those tasks are broken out into sub-DAGs organized by dataset. Each dataset
+has a subdag for all tasks.
+"""
+
+import os
+
+from airflow.models import DAG
+from airflow.operators.empty import EmptyOperator
+from roger.tasks import default_args, create_pipeline_taskgroup
+
+env_enabled_datasets = os.getenv(
+    "ROGER_DUG__INPUTS_DATA__SETS", "topmed,anvil").split(",")
+
+with DAG(
+        dag_id='annotate_and_index',
+        default_args=default_args,
+        schedule_interval=None
+) as dag:
+    init = EmptyOperator(task_id="init", dag=dag)
+    finish = EmptyOperator(task_id="finish", dag=dag)
+
+    from roger import pipelines
+    from roger.config import config
+    envspec = os.getenv("ROGER_DUG__INPUTS_DATA__SETS","topmed:v2.0")
+    data_sets = envspec.split(",")
+    pipeline_names = {x.split(':')[0]: x.split(':')[1] for x in data_sets}
+    for pipeline_class in pipelines.get_pipeline_classes(pipeline_names):
+        # Only use pipeline classes that are in the enabled datasets list and
+        # that have a properly defined pipeline_name attribute
+
+        # TODO
+        # Overriding environment variable just to see if this is working.
+        # name = getattr(pipeline_class, 'pipeline_name', '*not defined*')
+        # if not name in env_enabled_datasets:
+        #     continue
+
+        # Do the thing to add the pipeline's subdag to the dag in the right way
+        # . . .
+
+        init >> create_pipeline_taskgroup(dag, pipeline_class, config) >> finish