fix: basic engine implementation

Author: Alan Christie

tests/api_adapter.py

@@ -68,15 +68,15 @@ def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
 
         return {"id": workflow_definition_id}
 
-    def get_workflow(self, *, workflow_definition_id: str) -> Dict[str, Any]:
+    def get_workflow(self, *, workflow_id: str) -> Dict[str, Any]:
         UnitTestAPIAdapter.mp_lock.acquire()
         with open(f"tests/{_WORKFLOW_PICKLE_FILE}", "rb") as pickle_file:
             workflow = Unpickler(pickle_file).load()
         UnitTestAPIAdapter.mp_lock.release()
 
-        if workflow_definition_id not in workflow:
+        if workflow_id not in workflow:
             return {}
-        return {"workflow": workflow[workflow_definition_id]}
+        return {"workflow": workflow[workflow_id]}
 
     def get_workflow_by_name(self, *, name: str, version: str) -> Dict[str, Any]:
         UnitTestAPIAdapter.mp_lock.acquire()
@@ -106,6 +106,21 @@ def create_running_workflow(self, *, workflow_definition_id: str) -> str:
 
         return {"id": running_workflow_id}
 
+    def set_running_workflow_done(
+        self, *, running_workflow_id: str, success: bool
+    ) -> None:
+        UnitTestAPIAdapter.mp_lock.acquire()
+        with open(f"tests/{_RUNNING_WORKFLOW_PICKLE_FILE}", "rb") as pickle_file:
+            running_workflow = Unpickler(pickle_file).load()
+
+        assert running_workflow_id in running_workflow
+        running_workflow[running_workflow_id]["done"] = True
+        running_workflow[running_workflow_id]["success"] = success
+
+        with open(f"tests/{_RUNNING_WORKFLOW_PICKLE_FILE}", "wb") as pickle_file:
+            Pickler(pickle_file).dump(running_workflow)
+        UnitTestAPIAdapter.mp_lock.release()
+
     def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
         UnitTestAPIAdapter.mp_lock.acquire()
         with open(f"tests/{_RUNNING_WORKFLOW_PICKLE_FILE}", "rb") as pickle_file:
@@ -155,6 +170,21 @@ def get_running_workflow_step(
             "running_workflow_step": running_workflow_step[running_workflow_step_id]
         }
 
+    def set_running_workflow_step_done(
+        self, *, running_workflow_step_id: str, success: bool
+    ) -> None:
+        UnitTestAPIAdapter.mp_lock.acquire()
+        with open(f"tests/{_RUNNING_WORKFLOW_STEP_PICKLE_FILE}", "rb") as pickle_file:
+            running_workflow_step = Unpickler(pickle_file).load()
+
+        assert running_workflow_step_id in running_workflow_step
+        running_workflow_step[running_workflow_step_id]["done"] = True
+        running_workflow_step[running_workflow_step_id]["success"] = success
+
+        with open(f"tests/{_RUNNING_WORKFLOW_STEP_PICKLE_FILE}", "wb") as pickle_file:
+            Pickler(pickle_file).dump(running_workflow_step)
+        UnitTestAPIAdapter.mp_lock.release()
+
     def get_running_workflow_steps(
         self, *, running_workflow_id: str
     ) -> List[Dict[str, Any]]:

tests/test_test_api_adapter.py

@@ -25,9 +25,7 @@ def test_get_unknown_workflow():
     utaa = UnitTestAPIAdapter()
 
     # Act
-    wfd = utaa.get_workflow(
-        workflow_definition_id="workflow-00000000-0000-0000-0000-000000000001"
-    )
+    wfd = utaa.get_workflow(workflow_id="workflow-00000000-0000-0000-0000-000000000001")
 
     # Assert
     assert wfd == {}
@@ -51,7 +49,7 @@ def test_get_workflow():
     wfid = response["id"]
 
     # Act
-    wf = utaa.get_workflow(workflow_definition_id=wfid)
+    wf = utaa.get_workflow(workflow_id=wfid)
 
     # Assert
     assert wf["workflow"]["name"] == "blah"

tests/test_workflow_engine_examples.py

@@ -97,6 +97,7 @@ def test_workflow_engine_with_two_step_nop(basic_engine):
     mq.join()
     print("Stopped")
     assert r_wf
+    assert r_wf["done"]
     assert r_wf["success"]
     # Now check there are the right number of RunningWorkflowStep Records
     # (and they're all set to success/done)

tests/workflow-definitions/example-two-step-nop.yaml

@@ -4,6 +4,16 @@ kind-version: "2024.1"
 name: two-step-nop
 steps:
 - name: step-1
-  specification: "{'collection': 'workflow-engine-unit-test-jobs', 'job': 'nop', 'version: '1.0.0'}"
+  specification: >-
+    {
+      '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'}"
+  specification: >-
+    {
+      'collection': 'workflow-engine-unit-test-jobs',
+      'job': 'nop',
+      'version': '1.0.0'
+    }

workflow/workflow_abc.py

@@ -69,7 +69,7 @@ def create_workflow(
     def get_workflow(
         self,
         *,
-        workflow_definition_id: str,
+        workflow_id: str,
     ) -> Dict[str, Any]:
         """Get a Workflow Record by ID."""
         # If present this should return:
@@ -103,6 +103,12 @@ def create_running_workflow(
         #    "id": "r-workflow-00000000-0000-0000-0000-000000000001",
         # }
 
+    @abstractmethod
+    def set_running_workflow_done(
+        self, *, running_workflow_id: str, success: bool
+    ) -> None:
+        """Set the success value for a RunningWorkflow Record"""
+
     @abstractmethod
     def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
         """Get a RunningWorkflow Record"""
@@ -120,6 +126,7 @@ def create_running_workflow_step(
         self,
         *,
         running_workflow_id: str,
+        step: str,
     ) -> Dict[str, Any]:
         """Create a RunningWorkflowStep Record (from a RunningWorkflow)"""
         # Should return:
@@ -142,6 +149,12 @@ def get_running_workflow_step(
         #    },
         # }
 
+    @abstractmethod
+    def set_running_workflow_step_done(
+        self, *, running_workflow_step_id: str, success: bool
+    ) -> None:
+        """Set the success value for a RunningWorkflowStep Record"""
+
     @abstractmethod
     def get_running_workflow_steps(
         self, *, running_workflow_id: str

workflow/workflow_engine.py

@@ -1,8 +1,10 @@
 """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
@@ -54,22 +56,151 @@ def handle_message(self, msg: Message) -> None:
             self._handle_workflow_message(msg)
 
     def _handle_workflow_message(self, msg: WorkflowMessage) -> None:
+        """Handles a WorkflowMessage. This is a message that signals a START or STOP
+        of a workflow. On START we will load the workflow definition and run (launch)
+        the first step."""
         assert msg
 
+        # ALL THIS CODE ADDED SIMPLY TO DEMONSTRATE THE USE OF THE API ADAPTER
+        # AND THE INSTANCE LAUNCHER FOR THE SIMPLEST OF WORKFLOWS: -
+        # THE "TWO-STEP NOP".
+        # THERE IS NWO SPECIFICATION MANIPULATION NEEDED FOR THIS EXAMPLE
+        # THE STEPS HAVE NO INPUTS OR OUTPUTS.
+        # THIS FUNCTION PROBABLY NEEDS TO BE A LOT MORE SOPHISTICATED!
+
         _LOGGER.info("WE> WorkflowMessage:\n%s", str(msg))
         if msg.action == "START":
-            _LOGGER.info("action=%s", msg.action)
-            # Load the workflow definition and run the first step...
-            rf = self._api_adapter.get_running_workflow(
+            # Using the running workflow get the workflow definition
+            response = self._api_adapter.get_running_workflow(
                 running_workflow_id=msg.running_workflow
             )
-            assert rf
-            _LOGGER.info("RunningWorkflow: %s", rf)
+            assert "running_workflow" in response
+            running_workflow = response["running_workflow"]
+            _LOGGER.info("RunningWorkflow: %s", running_workflow)
+            workflow_id = running_workflow["workflow"]["id"]
+            response = self._api_adapter.get_workflow(workflow_id=workflow_id)
+            assert "workflow" in response
+            workflow = response["workflow"]
+            # Now find the first step
+            # and create a RunningWorkflowStep record prior to launching the instance
+            response = self._api_adapter.create_running_workflow_step(
+                running_workflow_id=msg.running_workflow,
+                step=workflow["steps"][0]["name"],
+            )
+            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.
+            step = workflow["steps"][0]
+            step_specification: Dict[str, Any] = ast.literal_eval(step["specification"])
+            self._instance_launcher.launch(
+                project_id="project-000",
+                workflow_id=workflow_id,
+                running_workflow_step_id=running_workflow_step_id,
+                workflow_definition=workflow,
+                step_specification=step_specification,
+            )
 
         else:
             _LOGGER.info("action=%s", msg.action)
 
     def _handle_pod_message(self, msg: PodMessage) -> None:
+        """Handles a PodMessage. This is a message that signals the completion of a
+        step within a workflow. Steps run as "instances" and the Pod message
+        identifies the Instance. Using the Instance record we can get the
+        "running workflow step" and then identify the "running workflow" and the
+        "workflow".
+
+        First thing is to adjust the workflow step with the step's success state and
+        optional error code. If the step was successful we can find the next step
+        and launch that, or consider the last step to have run and modify the
+        running workflow record and set's it's success status."""
         assert msg
 
+        # The PodMessage has a 'instance', 'has_exit_code', and 'exit_code' values.
         _LOGGER.info("WE> PodMessage:\n%s", str(msg))
+
+        # ALL THIS CODE ADDED SIMPLY TO DEMONSTRATE THE USE OF THE API ADAPTER
+        # AND THE INSTANCE LAUNCHER FOR THE SIMPLEST OF WORKFLOWS: -
+        # THE "TWO-STEP NOP".
+        # THERE IS NWO SPECIFICATION MANIPULATION NEEDED FOR THIS EXAMPLE
+        # THE STEPS HAVE NO INPUTS OR OUTPUTS.
+        # THIS FUNCTION PROBABLY NEEDS TO BE A LOT MORE SOPHISTICATED!
+
+        # Ignore anything without an exit code.
+        if not msg.has_exit_code:
+            _LOGGER.warning("WE> PodMessage: No exit code")
+            return
+
+        instance_id: str = msg.instance
+        exit_code: int = msg.exit_code
+        _LOGGER.info(
+            "WE> PodMessage: instance=%s, exit_code=%d", instance_id, exit_code
+        )
+
+        # Ignore instances without a running workflow step
+        response = self._api_adapter.get_instance(instance_id=instance_id)
+        if "running_workflow_step" not in response:
+            _LOGGER.warning("WE> PodMessage: Without running_workflow_step")
+            return
+        running_workflow_step_id: str = response["running_workflow_step"]
+        response = self._api_adapter.get_running_workflow_step(
+            running_workflow_step_id=running_workflow_step_id
+        )
+        step_name: str = response["running_workflow_step"]["step"]
+
+        # Set the step as completed (successful or otherwise)
+        success: bool = exit_code == 0
+        self._api_adapter.set_running_workflow_step_done(
+            running_workflow_step_id=running_workflow_step_id,
+            success=success,
+        )
+
+        # Get the step's running workflow and workflow IDs and records.
+        running_workflow_id = response["running_workflow_step"]["running_workflow"]
+        assert running_workflow_id
+        response = self._api_adapter.get_running_workflow(
+            running_workflow_id=running_workflow_id
+        )
+        workflow_id = response["running_workflow"]["workflow"]["id"]
+        assert workflow_id
+        response = self._api_adapter.get_workflow(workflow_id=workflow_id)
+        workflow = response["workflow"]
+
+        end_of_workflow: bool = False
+        if success:
+            # Given the step for the instance just finished,
+            # find the next step in the workflow and launch it.
+            # If there are no more steps then the workflow is done
+            # so we need to set the running workflow as done
+            # and set it's success status too.
+            for step in workflow["steps"]:
+                if step["name"] == step_name:
+                    step_index = workflow["steps"].index(step)
+                    if step_index + 1 < len(workflow["steps"]):
+                        next_step = workflow["steps"][step_index + 1]
+                        response = self._api_adapter.create_running_workflow_step(
+                            running_workflow_id=running_workflow_id,
+                            step=next_step["name"],
+                        )
+                        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-000",
+                            workflow_id=workflow_id,
+                            running_workflow_step_id=running_workflow_step_id,
+                            workflow_definition=workflow,
+                            step_specification=step_specification,
+                        )
+                        break
+                    else:
+                        end_of_workflow = True
+
+        if end_of_workflow:
+            self._api_adapter.set_running_workflow_done(
+                running_workflow_id=running_workflow_id,
+                success=success,
+            )