Adding more test examples for unmanaged control

Author: renan-souza

.github/workflows/run_examples.sh

@@ -82,7 +82,7 @@ echo "Using examples directory: $EXAMPLES_DIR"
 echo "With Mongo? ${WITH_MONGO}"
 
 # Define the test cases
-default_tests=("instrumented_simple_example.py" "instrumented_loop_example.py" "distributed_consumer_example.py" "dask_example.py" "mlflow_example.py" "tensorboard_example.py" "single_layer_perceptron_example.py" "llm_complex/llm_main_example.py")
+default_tests=("instrumented_simple_example.py" "instrumented_loop_example.py" "distributed_consumer_example.py" "dask_example.py" "mlflow_example.py" "tensorboard_example.py" "single_layer_perceptron_example.py" "llm_complex/llm_main_example.py" "unmanaged/main.py")
 
 # Use the third argument if provided, otherwise use default tests
 if [[ -n "$3" ]]; then

examples/unmanaged/main.py

@@ -0,0 +1,55 @@
+import uuid
+import subprocess
+import os
+from flowcept.flowcept_api.flowcept_controller import Flowcept
+from pathlib import Path
+
+def start_workflow():
+    """Initialize and start the Flowcept workflow."""
+    workflow_id = str(uuid.uuid4())
+    flowcept_instance = Flowcept(
+        workflow_id=workflow_id,
+        workflow_name="Test-Workflow",
+        bundle_exec_id=workflow_id
+    )
+    flowcept_instance.start()
+    return workflow_id, flowcept_instance
+
+
+def run_task(workflow_id, script="simple_task.py"):
+    """Run an external Python script with the given workflow ID."""
+    process = subprocess.Popen(
+        ["python", script, workflow_id],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True
+    )
+
+    for line in iter(process.stdout.readline, ""):
+        print(line, end="")
+
+    process.stdout.close()
+    process.wait()
+
+
+def main():
+    """Main function to manage workflow execution."""
+
+    parent_dir = Path(__file__).resolve().parent
+    script = os.path.join(parent_dir, "simple_task.py")
+
+    workflow_id, f = start_workflow()
+
+    try:
+        run_task(workflow_id, script=script)
+    finally:
+        f.stop()
+
+    return workflow_id
+
+
+if __name__ == "__main__":
+    workflow_id = main()
+    tasks = Flowcept.db.query({"workflow_id": workflow_id})
+    assert len(tasks) == 1
+    print(f"There is one task for the workflow {workflow_id}.")

examples/unmanaged/simple_task.py

@@ -0,0 +1,29 @@
+import argparse
+
+
+def run_task(workflow_id):
+    print(f"Task ran successfully for the workflow_id {workflow_id}.")
+
+
+def parse_args():
+    # Set up argument parser
+    parser = argparse.ArgumentParser(description="Process a workflow by ID.")
+    parser.add_argument("workflow_id", type=str, help="The ID of the workflow to process")
+
+    # Parse arguments
+    args = parser.parse_args()
+    return args.workflow_id
+
+
+if __name__ == "__main__":
+    workflow_id = parse_args()
+    from flowcept import Flowcept, FlowceptTask
+    f = Flowcept(workflow_id=workflow_id,
+                 bundle_exec_id=workflow_id,
+                 start_persistence=False, save_workflow=False)
+    f.start()
+    t = FlowceptTask(workflow_id=workflow_id, activity_id="test_task")
+    run_task(workflow_id)
+    t.end()
+    f.stop()
+

resources/redis_consumer_test.py

@@ -0,0 +1,24 @@
+import redis
+import msgpack
+from flowcept.configs import MQ_HOST, MQ_PORT, MQ_CHANNEL
+
+# Connect to Redis
+redis_client = redis.Redis(host=MQ_HOST, port=MQ_PORT, db=0)
+
+# Subscribe to a channel
+pubsub = redis_client.pubsub()
+pubsub.subscribe(MQ_CHANNEL)
+
+print("Listening for messages...")
+
+
+for message in pubsub.listen():
+    print("Received a message!")
+    if message["type"] in {"psubscribe"}:
+        continue
+
+    if isinstance(message["data"], int):
+        message["data"] = str(message["data"]).encode()  # Convert to string and encode to bytes
+
+    msg_obj = msgpack.loads(message["data"], strict_map_key=False)
+    print(msg_obj)

src/flowcept/instrumentation/task_capture.py

@@ -57,6 +57,7 @@ def __init__(
         task_id: str = None,
         workflow_id: str = None,
         campaign_id: str = None,
+        activity_id: str = None,
         used: Dict = None,
         custom_metadata: Dict = None,
     ):
@@ -65,6 +66,7 @@ def __init__(
             return
         self._task = TaskObject()
         self._task.telemetry_at_start = FlowceptTask._interceptor.telemetry_capture.capture()
+        self._task.activity_id = activity_id
         self._task.started_at = time()
         self._task.task_id = task_id or str(self._task.started_at)
         self._task.workflow_id = workflow_id or Flowcept.current_workflow_id