add destination path to remote inbox

tapis-project · Nov 10, 2023 · 5ba782c · 5ba782c
1 parent 8dee7c1
commit 5ba782c
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 5 deletions.
diff --git a/src/api/src/backend/views/ETLPipelines.py b/src/api/src/backend/views/ETLPipelines.py
@@ -174,6 +174,10 @@ def post(self, request, group_id, *_, **__):
                         "type": "string",
                         "value": body.remote_inbox.globus_endpoint_id
                     },
+                    "GLOBUS_DESTINATION_PATH": {
+                        "type": "string",
+                        "value": body.remote_inbox.globus_destination_path
+                    },
                     "GLOBUS_CLIENT_ID": {
                         "type": "string",
                         "value": body.remote_inbox.globus_client_id

diff --git a/src/api/src/backend/views/http/etl.py b/src/api/src/backend/views/http/etl.py
@@ -41,6 +41,7 @@ class GlobusRemoteInbox(BaseModel):
     globus_auth: GlobusAuth
     globus_endpoint_id: str
     globus_client_id: str
+    globus_destination_path: str
 
 class S3Auth(BaseModel):
     access_key: str

diff --git a/src/engine/src/core/workflows/executors/WorkflowExecutor.py b/src/engine/src/core/workflows/executors/WorkflowExecutor.py
@@ -246,7 +246,11 @@ def _prepare_tasks(self):
             # Fetch task templates
             if task.uses != None:
                 template_mapper = TemplateMapper(cache_dir=self.state.ctx.pipeline.git_cache_dir)
-                task = template_mapper.map(task, task.uses)
+                try:
+                    task = template_mapper.map(task, task.uses)
+                except Exception as e:
+                    # Trigger the terminal state callback.
+                    self._on_pipeline_terminal_state(event=PIPELINE_FAILED, message=str(e))
 
             # Add a key to the output for the task
             self.state.ctx.output[task.id] = None
@@ -337,15 +341,15 @@ def _on_task_terminal_state(self, task, task_result):
         return self._fetch_ready_tasks()
 
     @interceptable()
-    def _on_pipeline_terminal_state(self, event=None):
+    def _on_pipeline_terminal_state(self, event=None, message=""):
         # No event was provided. Determine if complete or failed from number
         # of failed tasks
         if event == None:
             event = PIPELINE_FAILED if len(self.state.failed) > 0 else PIPELINE_COMPLETED
 
         msg = "COMPLETED"
-        if event == PIPELINE_FAILED: msg = "FAILED"
-        elif event == PIPELINE_TERMINATED: msg = "TERMINATED"
+        if event == PIPELINE_FAILED: msg = "FAILED" + f" {message}"
+        elif event == PIPELINE_TERMINATED: msg = "TERMINATED" + f" {message}"
 
         self.state.ctx.logger.info(self.p_str(msg))
 

diff --git a/src/engine/src/helpers/TemplateMapper.py b/src/engine/src/helpers/TemplateMapper.py
@@ -35,7 +35,10 @@ def map(self, obj: Union[Pipeline, Task], uses: Uses) -> Union[Pipeline, Task]:
         """
 
         # Clone git repository specified on the pipeline.uses if exists
-        template = self.template_repo.get_by_uses(uses)
+        try:
+            template = self.template_repo.get_by_uses(uses)
+        except Exception as e:
+            raise Exception(f"Template mapping error: {e}")
 
         # Resolve which class the final object should have
         obj_class = Pipeline