Merge pull request #229 from hubmapconsortium/devel

Merging devel to master on the way to a new release branch

.gitmodules

@@ -28,3 +28,6 @@
 [submodule "src/ingest-pipeline/airflow/dags/cwl/ome-tiff-pyramid-ims"]
 	path = src/ingest-pipeline/airflow/dags/cwl/ome-tiff-pyramid-ims
 	url = https://github.com/hubmapconsortium/ome-tiff-pyramid
+[submodule "src/ingest-pipeline/submodules/ingest-validation-tests"]
+	path = src/ingest-pipeline/submodules/ingest-validation-tests
+	url = [email protected]:hubmapconsortium/ingest-validation-tests.git

build_number

@@ -1 +1 @@
-477
+591

docker/ingest-pipeline/config/airflow.cfg

@@ -352,7 +352,9 @@ expose_stacktrace = True
 
 # Set to true to turn on authentication:
 # https://airflow.apache.org/security.html#web-authentication
-authenticate = False
+authenticate = True
+
+#auth_backend =
 
 # Filter the list of dags by owner name (requires authentication to be enabled)
 filter_by_owner = False
@@ -1052,4 +1054,4 @@ fs_group =
 # The Key-value pairs to be given to worker pods.
 # The worker pods will be given these static labels, as well as some additional dynamic labels
 # to identify the task.
-# Should be supplied in the format: ``key = value``
+# Should be supplied in the format: ``key = value``

docker/ingest-pipeline/entrypoint.sh

@@ -9,13 +9,15 @@ export PATH=/home/airflow/.local/bin:$PATH
 
 # Install custom python package if requirements.txt is present
 pip install --upgrade pip
-pip install --user 'apache-airflow[celery,crypto,postgres,redis,ssh]'
+pip install --user flask-admin
+pip install --user 'apache-airflow[celery,crypto,postgres,redis,ssh]<2.0.0'
 if [[ -e "/requirements.txt" ]]; then
     $(command -v pip) install --user -r /requirements.txt
 fi
 
 # Global defaults and back-compat
 export AIRFLOW__CORE__FERNET_KEY=`python -c 'from cryptography.fernet import Fernet; FERNET_KEY = Fernet.generate_key().decode(); print(FERNET_KEY)'`
+export AIRFLOW__WEBSERVER__SECRET_KEY=`openssl rand -hex 30`
 
 # Load DAGs examples (default: Yes)
 if [[ -z "$AIRFLOW__CORE__LOAD_EXAMPLES" && "${LOAD_EX:=n}" == n ]]; then

src/ingest-pipeline/airflow/dags/bulk_atacseq.py

@@ -52,7 +52,7 @@
 ) as dag:
     pipeline_name = 'bulk-atac-seq'
     cwl_workflows = get_absolute_workflows(
-        Path(pipeline_name, 'bulk-atac-seq-pipeline.cwl'),
+        Path('sc-atac-seq-pipeline', 'bulk-atac-seq-pipeline.cwl'),
     )
 
     def build_dataset_name(**kwargs):

src/ingest-pipeline/airflow/dags/codex_cytokit.py

@@ -55,6 +55,7 @@
         Path(pipeline_name, 'pipeline.cwl'),
         Path('portal-containers', 'ome-tiff-offsets.cwl'),
         Path('portal-containers', 'sprm-to-json.cwl'),
+        Path('portal-containers', 'sprm-to-anndata.cwl'),
     )
 
     def build_dataset_name(**kwargs):
@@ -214,10 +215,60 @@ def build_cwltool_cmd3(**kwargs):
         task_id='maybe_keep_cwl3',
         python_callable=utils.pythonop_maybe_keep,
         provide_context=True,
-        op_kwargs = {'next_op' : 'move_data',
+        op_kwargs = {'next_op' : 'prepare_cwl4',
                      'bail_op' : 'set_dataset_error',
                      'test_op' : 'pipeline_exec_cwl3'}
         )
+
+    prepare_cwl4 = DummyOperator(
+        task_id='prepare_cwl4'
+        )
+
+    def build_cwltool_cmd4(**kwargs):
+        ctx = kwargs['dag_run'].conf
+        run_id = kwargs['run_id']
+        tmpdir = utils.get_tmp_dir_path(run_id)
+        print('tmpdir: ', tmpdir)
+        parent_data_dir = ctx['parent_lz_path']
+        print('parent_data_dir: ', parent_data_dir)
+        data_dir = tmpdir / 'cwl_out'  # This stage reads input from stage 1
+        print('data_dir: ', data_dir)
+
+        command = [
+            *get_cwltool_base_cmd(tmpdir),
+            cwl_workflows[3],
+            '--input_dir',
+            data_dir / 'sprm_outputs',
+        ]
+
+        return join_quote_command_str(command)
+
+
+    t_build_cmd4 = PythonOperator(
+        task_id='build_cmd4',
+        python_callable=build_cwltool_cmd4,
+        provide_context=True,
+        )
+
+
+    t_pipeline_exec_cwl4 = BashOperator(
+        task_id='pipeline_exec_cwl4',
+        bash_command=""" \
+        tmp_dir={{tmp_dir_path(run_id)}} ; \
+        cd ${tmp_dir}/cwl_out ; \
+        {{ti.xcom_pull(task_ids='build_cmd4')}} >> ${tmp_dir}/session.log 2>&1 ; \
+        echo $?
+        """
+    )
+
+    t_maybe_keep_cwl4 = BranchPythonOperator(
+        task_id='maybe_keep_cwl4',
+        python_callable=utils.pythonop_maybe_keep,
+        provide_context=True,
+        op_kwargs = {'next_op' : 'move_data',
+                     'bail_op' : 'set_dataset_error',
+                     'test_op' : 'pipeline_exec_cwl4'}
+        )
 
 
     t_send_create_dataset = PythonOperator(
@@ -265,6 +316,7 @@ def build_cwltool_cmd3(**kwargs):
             'pipeline_exec_cwl1',
             'pipeline_exec_cwl2',
             'pipeline_exec_cwl3',
+            'pipeline_exec_cwl4',
             'move_data',
         ],
         cwl_workflows=cwl_workflows,
@@ -287,10 +339,12 @@ def build_cwltool_cmd3(**kwargs):
      >> prepare_cwl1 >> t_build_cmd1 >> t_pipeline_exec_cwl1 >> t_maybe_keep_cwl1
      >> prepare_cwl2 >> t_build_cmd2 >> t_pipeline_exec_cwl2 >> t_maybe_keep_cwl2
      >> prepare_cwl3 >> t_build_cmd3 >> t_pipeline_exec_cwl3 >> t_maybe_keep_cwl3
+     >> prepare_cwl4 >> t_build_cmd4 >> t_pipeline_exec_cwl4 >> t_maybe_keep_cwl4
      >> t_move_data >> t_expand_symlinks >> t_send_status >> t_join)
     t_maybe_keep_cwl1 >> t_set_dataset_error
     t_maybe_keep_cwl2 >> t_set_dataset_error
     t_maybe_keep_cwl3 >> t_set_dataset_error
+    t_maybe_keep_cwl4 >> t_set_dataset_error
     t_set_dataset_error >> t_join
     t_join >> t_cleanup_tmpdir