Merge pull request #19 from mmacata/actinia-worker-http-launcher

mmacata · web-flow · commit b070f015ab9b · 2026-01-06T10:25:24.000+01:00
Add actinia-worker-http-launcher
diff --git a/.flake8 b/.flake8
@@ -7,6 +7,7 @@
 exclude = .git,.pycache,build,.eggs
 
 per-file-ignores =
+    ./tests/actinia_worker_http_launcher.py: E501
     ./tests/cloudevent_receiver_server.py: E501
     # remove when implemented:
     ./src/actinia_cloudevent_plugin/api/cloudevent.py: E501
diff --git a/.vscode/tasks.json b/.vscode/tasks.json
@@ -17,12 +17,19 @@
 			"command": "docker compose -f docker/docker-compose-vscode.yml up -d --build --force-recreate event-receiver-server",
 			"isBackground": true
 		},
+		{
+			"label": "start-actinia-worker-http-launcher",
+			"type": "shell",
+			"command": "docker compose -f docker/docker-compose-vscode.yml up -d --build --force-recreate actinia-worker-http-launcher",
+			"isBackground": true
+		},
 		{
 			"type": "docker-run",
 			"label": "docker-run: debug",
 			"dependsOn": [
 				"docker-build",
-				"start-cloudevent-receiver-dummy-server"
+				"start-cloudevent-receiver-dummy-server",
+				"start-actinia-worker-http-launcher"
 			],
 			"python": {
 				"module": "flask",
diff --git a/Makefile b/Makefile
@@ -19,13 +19,13 @@ dist:
 	python3 setup.py dist
 
 test:
-	./tests_with_cloudevent_receiver.sh
+	pytest
 
 unittest:
-	./tests_with_cloudevent_receiver.sh unittest
+	pytest -m "unittest"
 
 devtest:
-	./tests_with_cloudevent_receiver.sh dev
+	pytest -m "dev"
 
 integrationtest:
-	./tests_with_cloudevent_receiver.sh integrationtest
+	pytest -m "integrationtest"
diff --git a/config/mount/sample.ini b/config/mount/sample.ini
@@ -1,8 +1,9 @@
 [ACTINIA]
-processing_base_url = http://actinia-core:8088/api/v3
+enqueue_job_base_url = http://actinia-core:8088/api/v3
 use_actinia_modules = True
 user = actinia-gdi
 password = actinia-gdi
+worker_http_launcher_url = http://actinia-worker-http-launcher:8000/launch
 
 [EVENTRECEIVER]
 url = http://event-receiver-server:3000/
diff --git a/docker/Dockerfile b/docker/Dockerfile
@@ -18,9 +18,7 @@ RUN pip3 install --no-cache-dir gunicorn && \
     pip3 install --no-cache-dir setuptools pwgen==0.8.2.post0 pytest==8.3.5 pytest-cov==6.0.0 python-dotenv
 
 COPY . /src/actinia-cloudevent-plugin/
-RUN pip3 install --no-cache-dir -e /src/actinia-cloudevent-plugin/ && \
-    # For tests:
-    chmod a+x /src/actinia-cloudevent-plugin/tests_with_cloudevent_receiver.sh
+RUN pip3 install --no-cache-dir -e /src/actinia-cloudevent-plugin/
 
 RUN echo "RUNNING_SINCE=`date`" > /src/.env
 
diff --git a/docker/Dockerfile_actinia-worker-http-launcher b/docker/Dockerfile_actinia-worker-http-launcher
@@ -0,0 +1,14 @@
+FROM alpine:3.23
+
+RUN apk update; \
+    apk add --no-cache python3 python3-dev make
+ENV PATH="/opt/venv/bin:$PATH"
+RUN /usr/bin/python -m venv --system-site-packages --without-pip /opt/venv
+RUN python -m ensurepip && pip3 install --no-cache-dir --upgrade pip
+
+RUN pip install --no-cache-dir --upgrade fastapi[standard] docker
+
+COPY tests/actinia_worker_http_launcher.py /src/app.py
+
+WORKDIR /src
+CMD ["fastapi", "run", "/src/app.py", "--port", "8000"]
diff --git a/docker/docker-compose-vscode.yml b/docker/docker-compose-vscode.yml
@@ -9,6 +9,18 @@ services:
     networks:
       actinia-dev:
 
+  actinia-worker-http-launcher:
+    container_name: actinia-worker-http-launcher
+    build:
+      context: ..
+      dockerfile: docker/Dockerfile_actinia-worker-http-launcher
+    ports:
+      - "8000:8000"
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+    networks:
+      actinia-dev:
+
 networks:
   actinia-dev:
     name: actinia-docker_actinia-dev
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
@@ -23,7 +23,7 @@ services:
       dockerfile: docker/Dockerfile_eventreceiver
 
   actinia-core:
-    image: mundialis/actinia:2.12.2
+    image: mundialis/actinia:2.13.0
     depends_on:
       - valkey
 
diff --git a/ruff.toml b/ruff.toml
@@ -7,6 +7,7 @@
 # N802 Function name should be lowercase
 # N816 Variable in global scope should not be mixedCas
 # PLC0415 import-outside-top-level
+# PLR0912 Too many branches (13 > 12)
 # PLR0913 too-many-arguments (of function definition)
 # PLR0917 too-many-positional-arguments (of function definition)
 # PLR2004 magic-value-comparison (unnamed numerical constants ("magic") values)
@@ -23,7 +24,7 @@ lint.ignore = ["ANN001", "ANN201", "ANN202", "E501", "FA102", "N802", "N816", "P
 [lint.per-file-ignores]
 "tests/testsuite.py" = [ "PLC0415",]
 "src/actinia_cloudevent_plugin/resources/logging.py" = ["A005",]
-"src/actinia_cloudevent_plugin/resources/config.py" = ["SIM102","S105"]
+"src/actinia_cloudevent_plugin/resources/config.py" = ["PLR0912", "SIM102","S105"]
 "tests/integrationtests/test_cloudevent.py" = ["PLR2004",]
 "tests/cloudevent_receiver_server.py" = ["S104"]
 "tests/integrationtests/test_hook.py" = ["PLR2004",]
diff --git a/src/actinia_cloudevent_plugin/api/cloudevents.py b/src/actinia_cloudevent_plugin/api/cloudevents.py
@@ -25,7 +25,7 @@
 from actinia_cloudevent_plugin.model.response_models import (
     SimpleStatusCodeResponseModel,
 )
-from actinia_cloudevent_plugin.resources.config import EVENTRECEIVER
+from actinia_cloudevent_plugin.resources.config import ACTINIA, EVENTRECEIVER
 from actinia_cloudevent_plugin.resources.logging import log
 
 
@@ -62,6 +62,7 @@ def post(self) -> SimpleStatusCodeResponseModel:
         # With received process chain start actinia process + return cloudevent
         actinia_resp = start_actinia_job(event_received)
         queue_name = actinia_resp["queue"]
+        resource_id = actinia_resp["resource_id"]
 
         try:
             if queue_name == "local":
@@ -70,15 +71,12 @@ def post(self) -> SimpleStatusCodeResponseModel:
                 # job is processed directly.
                 log.info("No need to start actinia-worker")
             else:
-                pass
-                # TODO: Send event to JobSink
-                # TODO: Configure JobSink URL
-                # url = TODO
-                # new_event = send_binary_cloud_event(
-                #     event_received,
-                #     queue_name,
-                #     url,
-                # )
+                url = f"{ACTINIA.worker_http_launcher_url}/{queue_name}"
+                new_event = send_binary_cloud_event(
+                    event_received,
+                    queue_name,
+                    url,
+                )
 
             # Send event to configured broker
             url = EVENTRECEIVER.url
@@ -88,11 +86,15 @@ def post(self) -> SimpleStatusCodeResponseModel:
                 actinia_resp["queue"],
                 url,
             )
-            return SimpleStatusCodeResponseModel(
-                status=204,
-                message=self.msg.replace("<EVENT1>", event_received["id"])
+            response = {
+                "status": 201,
+                "message": self.msg.replace("<EVENT1>", event_received["id"])
                 .replace("<EVENT2>", new_event["id"])
                 .replace("<ACTINIA_JOB>", queue_name),
-            )
+                "actinia_queue_name": queue_name,
+                "actinia_job": resource_id,
+            }
+            return make_response(jsonify(response), 201)
+
         except ConnectionError as e:
             return f"Connection ERROR when returning cloudevent: {e}"
diff --git a/src/actinia_cloudevent_plugin/core/cloudevents.py b/src/actinia_cloudevent_plugin/core/cloudevents.py
@@ -50,7 +50,7 @@ def start_actinia_job(event) -> str:
         project = project.split(".")[0]
         mapset = project.split(".")[1]
 
-    url = f"{ACTINIA.processing_base_url}/projects/{project}/"
+    url = f"{ACTINIA.enqueue_job_base_url}/projects/{project}/"
     if not mapset:
         # emphemeral processing
         if ACTINIA.use_actinia_modules:
diff --git a/src/actinia_cloudevent_plugin/resources/config.py b/src/actinia_cloudevent_plugin/resources/config.py
@@ -26,10 +26,11 @@
 class ACTINIA:
     """Default config for actinia processing."""
 
-    processing_base_url = "http://localhost:8088/"
+    enqueue_job_base_url = "http://localhost:8088/"
     use_actinia_modules = True
     user = "actinia-gdi"
     password = "actinia-gdi"
+    worker_http_launcher_url = "http://localhost:8000/launch"
 
 
 class EVENTRECEIVER:
@@ -73,10 +74,10 @@ def __init__(self) -> None:
 
         # ACTINIA
         if config.has_section("ACTINIA"):
-            if config.has_option("ACTINIA", "processing_base_url"):
-                ACTINIA.processing_base_url = config.get(
+            if config.has_option("ACTINIA", "enqueue_job_base_url"):
+                ACTINIA.enqueue_job_base_url = config.get(
                     "ACTINIA",
-                    "processing_base_url",
+                    "enqueue_job_base_url",
                 )
             if config.has_option("ACTINIA", "use_actinia_modules"):
                 ACTINIA.use_actinia_modules = config.getboolean(
@@ -87,7 +88,11 @@ def __init__(self) -> None:
                 ACTINIA.user = config.get("ACTINIA", "user")
             if config.has_option("ACTINIA", "password"):
                 ACTINIA.password = config.get("ACTINIA", "password")
-
+            if config.has_option("ACTINIA", "worker_http_launcher_url"):
+                ACTINIA.worker_http_launcher_url = config.get(
+                    "ACTINIA",
+                    "worker_http_launcher_url",
+                )
         # LOGGING
         if config.has_section("LOGCONFIG"):
             if config.has_option("LOGCONFIG", "logfile"):
diff --git a/tests/actinia_worker_http_launcher.py b/tests/actinia_worker_http_launcher.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+"""SPDX-FileCopyrightText: (c) 2025 by mundialis GmbH & Co. KG.
+
+SPDX-License-Identifier: GPL-3.0-or-later
+
+Endpoint to start an actinia-worker container and pass a queue name
+"""
+
+__license__ = "GPL-3.0-or-later"
+__author__ = "Carmen Tawalika"
+__copyright__ = "Copyright 2025 mundialis GmbH & Co. KG"
+__maintainer__ = "mundialis GmbH & Co. KG"
+
+
+from fastapi import FastAPI
+
+from docker import from_env
+
+app = FastAPI()
+client = from_env()
+
+"""Hints for development
+
+This http server serves as a mockup for a knative JobSink.
+Example to start a job from JobSink is by posting a CloudEvent.
+
+kubectl run curl --image=curlimages/curl --rm=true -ti -- -X POST -v \
+   -H "content-type: application/json"  \
+   -H "ce-specversion: 1.0" \
+   -H "ce-source: my/curl/command" \
+   -H "ce-type: org.actinia.queuename" \
+   -H "ce-id: 150" \
+   -H "ce-actiniaqueuename: job_queue_resource_id-4263d323-829e-4a09-b3f3-6fbbfbd72a67" \
+   -d "{"pc":"dummyprocesschain"}" \
+   http://job-sink.knative-eventing.svc.cluster.local/processing/actinia-processing-job
+
+The started container has the CloudEvent written to an event file.
+
+{
+  "specversion": "1.0",
+  "id": "150",
+  "source": "my/curl/command",
+  "type": "org.actinia.queuename",
+  "datacontenttype": "application/json",
+  "data": {
+    "pc": "dummyprocesschain"
+  },
+  "actiniaqueuename": "job_queue_resource_id-4263d323-829e-4a09-b3f3-6fbbfbd72a67"
+}
+
+The job will parse the queue name and start the worker with a helper script.
+This workaround is not needed here, the queue name is not mounted but passed
+via HTTP GET parameter.
+"""
+
+
+@app.post("/launch/{queue_name}")
+def start_container(queue_name: str):
+    """Start actinia-worker and hand over queue name.
+
+    Receive an actinia queue_name and launch an actinia-worker via
+    docker to start the worker watching for the queue and process it.
+
+    Args:
+        queue_name (str): actinia queue name to pass to the worker
+
+    Returns:
+        dict: response including status
+
+    """
+    client.containers.run(
+        "actiniacore",
+        command=f"-q {queue_name}",
+        entrypoint="actinia-worker",
+        network="actinia-docker_actinia-dev",
+        name="actinia-worker",
+        auto_remove=True,
+        detach=True,
+    )
+
+    return {"status": "started actinia-worker container"}
diff --git a/tests/integrationtests/test_cloudevent.py b/tests/integrationtests/test_cloudevent.py
@@ -75,7 +75,7 @@ def test_post_cloudevent(self) -> None:
             resp,
             Response,
         ), "The response is not of type Response"
-        assert resp.status_code == 200, "The status code is not 200"
+        assert resp.status_code == 201, "The status code is not 201"
         assert hasattr(resp, "json"), "The response has no attribute 'json'"
         assert (
             "message" in resp.json
diff --git a/tests/integrationtests/test_hook.py b/tests/integrationtests/test_hook.py
@@ -77,7 +77,7 @@ def test_post_cloudevent(self) -> None:
             resp,
             Response,
         ), "The response is not of type Response"
-        assert resp.status_code == 200, "The status code is not 200"
+        assert resp.status_code == 201, "The status code is not 201"
         assert hasattr(resp, "json"), "The response has no attribute 'json'"
         assert (
             "message" in resp.json
diff --git a/tests_with_cloudevent_receiver.sh b/tests_with_cloudevent_receiver.sh
