feat: Launcher now used job decoder (and variables)

Alan Christie · Alan Christie · commit f15526bf7567 · 2024-11-05T12:56:54.000+01:00
diff --git a/poetry.lock b/poetry.lock
diff --git a/pyproject.toml b/pyproject.toml
@@ -14,6 +14,7 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.12"
 im-protobuf = "^7.1.0"
+im-data-manager-job-decoder = "^2.1.0"
 jsonschema = "^4.21.1"
 pyyaml = ">= 5.3.1, < 7.0"
 
diff --git a/tests/api_adapter.py b/tests/api_adapter.py
@@ -276,6 +276,7 @@ def get_job(
     ) -> Optional[Dict[str, Any]]:
         assert collection == _JOB_DEFINITIONS["collection"]
         assert job in _JOB_DEFINITIONS["jobs"]
+        assert version
 
         jd = _JOB_DEFINITIONS["jobs"][job]
         response = {"command": jd["command"]}
diff --git a/tests/instance_launcher.py b/tests/instance_launcher.py
@@ -1,9 +1,12 @@
+import json
 import os
 import subprocess
 from datetime import datetime, timezone
 from subprocess import CompletedProcess
 from typing import Any, Dict, List
 
+from decoder import decoder as job_decoder
+from decoder.decoder import TextEncoding
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 
 from tests.api_adapter import UnitTestAPIAdapter
@@ -44,14 +47,16 @@ def launch(
         self,
         *,
         project_id: str,
-        workflow_id: str,
+        running_workflow_id: str,
         running_workflow_step_id: str,
-        workflow_definition: Dict[str, Any],
-        step_specification: Dict[str, Any],
+        step_specification: str,
+        variables: Dict[str, Any],
     ) -> LaunchResult:
         assert project_id
-        assert workflow_id
+        assert running_workflow_id
         assert step_specification
+        assert isinstance(step_specification, str)
+        assert isinstance(variables, dict)
 
         assert project_id == TEST_PROJECT_ID
 
@@ -73,17 +78,36 @@ def launch(
         execution_directory = f"project-root/{project_id}"
         os.makedirs(execution_directory, exist_ok=True)
 
-        # Just run the Python module that matched the 'job' in the step specification.
-        # Don't care about 'version' or 'collection'. It will be relative to the
-        # execution directory.
-        job: str = step_specification["job"]
-        job_module = f"{_JOB_DIRECTORY}/{job}.py"
-        assert os.path.isfile(job_module)
-
-        job_cmd: List[str] = ["python", job_module]
-        print(f"Running job command: {job_module}")
+        # Apply variables to the step's Job command.
+        step_specification_map = json.loads(step_specification)
+        job = self._api_adapter.get_job(
+            collection=step_specification_map["collection"],
+            job=step_specification_map["job"],
+            version="do-not-care",
+        )
+        assert job
+
+        # Now apply the variables to the command
+        decoded_command, status = job_decoder.decode(
+            job["command"],
+            variables,
+            running_workflow_step_id,
+            TextEncoding.JINJA2_3_0,
+        )
+        print(f"Decoded command: {decoded_command}")
+        print(f"Status: {status}")
+        assert status
+
+        # Now run the decoded command, which will be in the _JOB_DIRECTORY
+        command = f"{_JOB_DIRECTORY}/{decoded_command}"
+        command_list = command.split()
+        module = command_list[0]
+        print(f"Module: {module}")
+        assert os.path.isfile(module)
+        subprocess_cmd: List[str] = ["python"] + command_list
+        print(f"Subprocess command: {subprocess_cmd}")
         completed_process: CompletedProcess = subprocess.run(
-            job_cmd, check=False, cwd=execution_directory
+            subprocess_cmd, check=False, cwd=execution_directory
         )
 
         # Simulate a PodMessage (that will contain the instance ID),
@@ -103,5 +127,5 @@ def launch(
             error_msg=None,
             instance_id=instance_id,
             task_id=task_id,
-            command=" ".join(job_cmd),
+            command=" ".join(subprocess_cmd),
         )
diff --git a/tests/job-definitions/job-definitions.yaml b/tests/job-definitions/job-definitions.yaml
@@ -31,7 +31,7 @@ jobs:
 
   smiles-to-file:
     command: >-
-      string-to-file.py --smiles {{ smiles }} --output {{ outputFile }}
+      smiles-to-file.py --smiles {{ smiles }} --output {{ outputFile }}
     variables:
       outputs:
         type: object
diff --git a/tests/test_test_instance_launcher.py b/tests/test_test_instance_launcher.py
@@ -1,4 +1,5 @@
 # Tests for the decoder package.
+import json
 
 import pytest
 
@@ -40,10 +41,12 @@ def test_launch_nop(basic_launcher):
     # Act
     result = launcher.launch(
         project_id=TEST_PROJECT_ID,
-        workflow_id="workflow-00000000-0000-0000-0000-000000000001",
+        running_workflow_id="r-workflow-00000000-0000-0000-0000-000000000001",
         running_workflow_step_id=rwfsid,
-        workflow_definition={},
-        step_specification={"job": "nop"},
+        step_specification=json.dumps(
+            {"collection": "workflow-engine-unit-test-jobs", "job": "nop"}
+        ),
+        variables={},
     )
 
     # Assert
@@ -70,10 +73,12 @@ def test_launch_nop_fail(basic_launcher):
     # Act
     result = launcher.launch(
         project_id=TEST_PROJECT_ID,
-        workflow_id="workflow-00000000-0000-0000-0000-000000000001",
+        running_workflow_id="r-workflow-00000000-0000-0000-0000-000000000001",
         running_workflow_step_id=rwfsid,
-        workflow_definition={},
-        step_specification={"job": "nop-fail"},
+        step_specification=json.dumps(
+            {"collection": "workflow-engine-unit-test-jobs", "job": "nop-fail"}
+        ),
+        variables={},
     )
 
     # Assert
@@ -100,16 +105,20 @@ def test_launch_smiles_to_file(basic_launcher):
     # Act
     result = launcher.launch(
         project_id=TEST_PROJECT_ID,
-        workflow_id="workflow-00000000-0000-0000-0000-000000000001",
+        running_workflow_id="r-workflow-00000000-0000-0000-0000-000000000001",
         running_workflow_step_id=rwfsid,
-        workflow_definition={},
-        step_specification={
-            "job": "smiles-to-file",
-            "variables": {"smiles": "C1=CC=CC=C1", "outputFile": "output.smi"},
-        },
+        step_specification=json.dumps(
+            {
+                "collection": "workflow-engine-unit-test-jobs",
+                "job": "smiles-to-file",
+            }
+        ),
+        variables={"smiles": "C1=CC=CC=C1", "outputFile": "output.smi"},
     )
 
     # Assert
     assert result.error == 0
     assert result.command.startswith("python ")
-    assert result.command.endswith("tests/jobs/smiles-to-file.py")
+    assert result.command.endswith(
+        "tests/jobs/smiles-to-file.py --smiles C1=CC=CC=C1 --output output.smi"
+    )
diff --git a/tests/workflow-definitions/example-nop-fail.yaml b/tests/workflow-definitions/example-nop-fail.yaml
@@ -8,7 +8,7 @@ steps:
 - name: step-1
   specification: >-
     {
-      'collection': 'workflow-engine-unit-test-jobs',
-      'job': 'nop-fail',
-      'version': '1.0.0'
+      "collection": "workflow-engine-unit-test-jobs",
+      "job": "nop-fail",
+      "version": "1.0.0"
     }
diff --git a/tests/workflow-definitions/example-two-step-nop.yaml b/tests/workflow-definitions/example-two-step-nop.yaml
@@ -9,14 +9,14 @@ steps:
 - name: step-1
   specification: >-
     {
-      'collection': 'workflow-engine-unit-test-jobs',
-      'job': 'nop',
-      'version': '1.0.0'
+      "collection": "workflow-engine-unit-test-jobs",
+      "job": "nop",
+      "version": "1.0.0"
     }
 - name: step-2
   specification: >-
     {
-      'collection': 'workflow-engine-unit-test-jobs',
-      'job': 'nop',
-      'version': '1.0.0'
+      "collection": "workflow-engine-unit-test-jobs",
+      "job": "nop",
+      "version": "1.0.0"
     }
diff --git a/tests/workflow-definitions/minimal.yaml b/tests/workflow-definitions/minimal.yaml
@@ -4,4 +4,5 @@ kind-version: "2024.1"
 name: workflow-minimal
 steps:
 - name: step-1
-  specification: "{}"
+  specification: >-
+    {}
diff --git a/workflow/workflow_abc.py b/workflow/workflow_abc.py
@@ -28,10 +28,10 @@ def launch(
         self,
         *,
         project_id: str,
-        workflow_id: str,
+        running_workflow_id: str,
         running_workflow_step_id: str,
-        workflow_definition: Dict[str, Any],
-        step_specification: Dict[str, Any],
+        step_specification: str,
+        variables: Dict[str, Any],
     ) -> LaunchResult:
         """Launch a (Job) Instance"""
 
diff --git a/workflow/workflow_engine.py b/workflow/workflow_engine.py
@@ -1,10 +1,8 @@
 """The WorkflowEngine execution logic.
 """
 
-import ast
 import logging
 import sys
-from typing import Any, Dict
 
 from google.protobuf.message import Message
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
@@ -79,6 +77,7 @@ def _handle_workflow_message(self, msg: WorkflowMessage) -> None:
             _LOGGER.info("RunningWorkflow: %s", running_workflow)
             project_id = running_workflow["project_id"]
             workflow_id = running_workflow["workflow"]["id"]
+            variables = running_workflow["variables"]
             response = self._api_adapter.get_workflow(workflow_id=workflow_id)
             assert "workflow" in response
             workflow = response["workflow"]
@@ -90,16 +89,15 @@ def _handle_workflow_message(self, msg: WorkflowMessage) -> None:
             )
             assert "id" in response
             running_workflow_step_id = response["id"]
-            # The specification is a string here.
-            # It needs to be a dictionary for the launch() method.
+            # The step specification is a string here - pass it directly to the launcher
+            # which will get the Job command and apply the provided variables to it.
             step = workflow["steps"][0]
-            step_specification: Dict[str, Any] = ast.literal_eval(step["specification"])
             self._instance_launcher.launch(
                 project_id=project_id,
-                workflow_id=workflow_id,
+                running_workflow_id=msg.running_workflow,
                 running_workflow_step_id=running_workflow_step_id,
-                workflow_definition=workflow,
-                step_specification=step_specification,
+                step_specification=step["specification"],
+                variables=variables,
             )
 
         else:
@@ -165,6 +163,7 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
         )
         project_id = response["running_workflow"]["project_id"]
         workflow_id = response["running_workflow"]["workflow"]["id"]
+        variables = response["running_workflow"]["variables"]
         assert workflow_id
         response = self._api_adapter.get_workflow(workflow_id=workflow_id)
         workflow = response["workflow"]
@@ -187,15 +186,12 @@ def _handle_pod_message(self, msg: PodMessage) -> None:
                         )
                         assert "id" in response
                         running_workflow_step_id = response["id"]
-                        step_specification: Dict[str, Any] = ast.literal_eval(
-                            next_step["specification"]
-                        )
                         self._instance_launcher.launch(
                             project_id=project_id,
-                            workflow_id=workflow_id,
+                            running_workflow_id=running_workflow_id,
                             running_workflow_step_id=running_workflow_step_id,
-                            workflow_definition=workflow,
-                            step_specification=step_specification,
+                            step_specification=next_step["specification"],
+                            variables=variables,
                         )
                         end_of_workflow = False
                         break

Original file line number	Diff line number	Diff line change
`@@ -8,7 +8,7 @@ steps:`
`8`	`8`	`- name: step-1`
`9`	`9`	`specification: >-`
`10`	`10`	`{`
`11`		`- 'collection': 'workflow-engine-unit-test-jobs',`
`12`		`- 'job': 'nop-fail',`
`13`		`- 'version': '1.0.0'`
	`11`	`+ "collection": "workflow-engine-unit-test-jobs",`
	`12`	`+ "job": "nop-fail",`
	`13`	`+ "version": "1.0.0"`
`14`	`14`	`}`
Original file line number	Diff line number	Diff line change
`@@ -9,14 +9,14 @@ steps:`
`9`	`9`	`- name: step-1`
`10`	`10`	`specification: >-`
`11`	`11`	`{`
`12`		`- 'collection': 'workflow-engine-unit-test-jobs',`
`13`		`- 'job': 'nop',`
`14`		`- 'version': '1.0.0'`
	`12`	`+ "collection": "workflow-engine-unit-test-jobs",`
	`13`	`+ "job": "nop",`
	`14`	`+ "version": "1.0.0"`
`15`	`15`	`}`
`16`	`16`	`- name: step-2`
`17`	`17`	`specification: >-`
`18`	`18`	`{`
`19`		`- 'collection': 'workflow-engine-unit-test-jobs',`
`20`		`- 'job': 'nop',`
`21`		`- 'version': '1.0.0'`
	`19`	`+ "collection": "workflow-engine-unit-test-jobs",`
	`20`	`+ "job": "nop",`
	`21`	`+ "version": "1.0.0"`
`22`	`22`	`}`