debug

src/engine/src/core/Server.py

@@ -374,7 +374,7 @@ def _resolve_idempotency_key(self, request):
             return idempotency_key
 
         except (AttributeError, TypeError) as e:
-            logger.info(f"{lbuf('[SERVER]')} ERROR: Failed to resolve idempotency key from provided constraints. {str(e)}. Defaulted to pipeline id '{default_idempotency_key}'")
+            logger.info(f"{lbuf('[SERVER]')} WARNING: Failed to resolve idempotency key from provided constraints. {str(e)}. Defaulted to pipeline id '{default_idempotency_key}'")
             return default_idempotency_key
 
 

src/engine/src/core/mappers/EnvMapper.py

@@ -6,6 +6,9 @@ def __init__(self, dao: WorkflowExecutorStateDAO):
         self._dao = dao
 
     def get_value_by_key(self, key):
+        print("ENVMAPPER")
+        print("KEY", key)
+        print("KEY ON ENV", self._dao.get_state().ctx.env.get(key))
         env = self._dao.get_state().ctx.env.get(key).value
         if env == None: return None
         return env.value

src/engine/src/core/tasks/executors/Function.py

@@ -147,7 +147,7 @@ def _setup_container(self) -> ContainerDetails:
         else:
             raise Exception(f"Invalid runtime: {self.task.runtime}")
 
-         # Set up env vars for the container
+        # Set up env vars for the container
         env = [
             client.V1EnvVar(
                 name="_OWE_TASK_ID",

src/engine/src/core/workflows/WorkflowExecutor.py

@@ -626,7 +626,8 @@ def _prepare_pipeline_fs(self):
         # Persist each arg to files in the pipeline file system
         env_var_value_file_repo = self.container.load("EnvVarValueFileRepository")
         env_repo = self.container.load("EnvRepository")
-        for key in self.state.ctx.pipeline.env:
+        for key in self.state.ctx.env:
+            print("KEY", key)
             env_var_value_file_repo.save(
                 self.state.ctx.pipeline.env_dir + key,
                 env_repo.get_value_by_key(key)