Added controller ip/port at the worker's config.

Added a mount on the data_path for the worker pod too.

Author: KFilippopolitis

.github/workflows/standalone_tests.yml

@@ -15,7 +15,7 @@ jobs:
       - name: Check out repository
         uses: actions/checkout@v3
 
-      - name: Set up python
+      - name: Set up Python
         uses: actions/setup-python@v4
         with:
           python-version: 3.8
@@ -96,6 +96,7 @@ jobs:
         id: non_smpc_tests
         run: poetry run pytest -s -m "not smpc" --cov=exareme2 --cov-report=xml:non_smpc_cov.xml tests/standalone_tests --verbosity=4
         env:
+          PYTHONPATH: ${{ github.workspace }}/exareme2:${{ github.workspace }}
           PULL_DOCKER_IMAGES: false
 
       - name: Run SMPC specific standalone tests after releasing previous test resources

README.md

@@ -58,6 +58,9 @@
    celery_cleanup_task_timeout=2
    celery_run_udf_task_timeout = 120
 
+   [controller]
+   port = 5000
+
    [privacy]
    minimum_row_count = 10
    protect_local_data = false

exareme2/algorithms/flower/flower_data_processing.py

@@ -12,6 +12,12 @@
 from sklearn.impute import SimpleImputer
 
 # Constants for project directories and environment configurations
+CONTROLLER_IP = os.getenv("CONTROLLER_IP", "127.0.0.1")
+CONTROLLER_PORT = os.getenv("CONTROLLER_PORT", 5000)
+RESULT_URL = f"http://{CONTROLLER_IP}:{CONTROLLER_PORT}/flower/result"
+INPUT_URL = f"http://{CONTROLLER_IP}:{CONTROLLER_PORT}/flower/input"
+CDES_URL = f"http://{CONTROLLER_IP}:{CONTROLLER_PORT}/cdes_metadata"
+HEADERS = {"Content-type": "application/json", "Accept": "text/plain"}
 PROJECT_ROOT = Path(__file__).resolve().parents[3]
 
 

exareme2/algorithms/flower/process_manager.py

@@ -70,6 +70,31 @@ def terminate_process(proc, logger, max_attempts=3, wait_time=10):
     logger.error(f"Failed to terminate PID {proc.pid} after {max_attempts} attempts.")
 
 
+def should_terminate_process(cmdline):
+    """Check if the process should be terminated based on its command line."""
+    return cmdline and (
+        cmdline[-1].endswith("client.py") or cmdline[-1].endswith("server.py")
+    )
+
+
+def process_garbage_collect(proc, logger):
+    """Terminate a process and handle errors."""
+    try:
+        pid = proc.pid
+        cmdline = proc.cmdline()
+        if should_terminate_process(cmdline):
+            logger.info(f"PID: {pid} - Name: {proc.name()} - Cmdline: {cmdline}")
+            terminate_process(proc, logger)
+    except psutil.NoSuchProcess:
+        logger.warn(
+            f"No process found with PID {proc.pid}. It may have already exited."
+        )
+    except psutil.AccessDenied:
+        logger.error(f"Access denied when attempting to terminate PID {proc.pid}.")
+    except Exception as e:
+        logger.error(f"An error occurred while managing PID {proc.pid}: {e}")
+
+
 class FlowerProcess:
     def __init__(self, file, parameters=None, env_vars=None, stdout=None, stderr=None):
         self.file = file
@@ -107,3 +132,9 @@ def kill_process(cls, pid, algorithm_name, logger):
             logger.error(f"Access denied when attempting to terminate PID {pid}.")
         except Exception as e:
             logger.error(f"An error occurred while managing PID {pid}: {e}")
+
+    @classmethod
+    def garbage_collect(cls, logger):
+        """Garbage collect processes matching specific criteria."""
+        for proc in psutil.process_iter(["pid", "name", "cmdline"]):
+            process_garbage_collect(proc, logger)

exareme2/controller/celery/tasks_handler.py

@@ -297,24 +297,24 @@ def queue_healthcheck_task(
         )
 
     def start_flower_client(
-        self, request_id, algorithm_name, worker_id
+        self, request_id, algorithm_name, server_address
     ) -> WorkerTaskResult:
         return self._queue_task(
             task_signature=TASK_SIGNATURES["start_flower_client"],
             request_id=request_id,
             algorithm_name=algorithm_name,
-            worker_id=worker_id,
+            server_address=server_address,
         )
 
     def start_flower_server(
-        self, request_id, algorithm_name, number_of_clients, worker_id
+        self, request_id, algorithm_name, number_of_clients, server_address
     ) -> WorkerTaskResult:
         return self._queue_task(
             task_signature=TASK_SIGNATURES["start_flower_server"],
             request_id=request_id,
             algorithm_name=algorithm_name,
             number_of_clients=number_of_clients,
-            worker_id=worker_id,
+            server_address=server_address,
         )
 
     def stop_flower_server(

exareme2/controller/services/flower/controller.py

@@ -7,8 +7,9 @@
 from exareme2.controller.uid_generator import UIDGenerator
 from exareme2.worker_communication import WorkerInfo
 
+FLOWER_SERVER_PORT = "8080"
+
 
-# Base Exception class for Worker-related exceptions
 class WorkerException(Exception):
     pass
 
@@ -59,23 +60,30 @@ async def exec_algorithm(self, algorithm_name, algorithm_request_dto):
                 self._create_worker_tasks_handler(request_id, worker)
                 for worker in workers_info
             ]
-            server_task_handler = (
-                task_handlers[0]
-                if len(task_handlers) == 1
-                else self._create_global_handler(request_id)
-            )
+
+            server_task_handler, server_ip = task_handlers[0], workers_info[0].ip
+            if len(task_handlers) > 1:
+                global_worker = self.worker_landscape_aggregator.get_global_worker()
+                server_task_handler = self._create_worker_tasks_handler(
+                    request_id, global_worker
+                )
+                server_ip = global_worker.ip
+
             self.flower_execution_info.set_inputdata(
                 inputdata=algorithm_request_dto.inputdata
             )
             server_pid = None
             clients_pids = {}
+            server_address = f"{server_ip}:{FLOWER_SERVER_PORT}"
 
             try:
                 server_pid = server_task_handler.start_flower_server(
-                    algorithm_name, len(task_handlers)
+                    algorithm_name, len(task_handlers), str(server_address)
                 )
                 clients_pids = {
-                    handler.start_flower_client(algorithm_name): handler
+                    handler.start_flower_client(
+                        algorithm_name, str(server_address)
+                    ): handler
                     for handler in task_handlers
                 }
 
@@ -102,7 +110,10 @@ async def exec_algorithm(self, algorithm_name, algorithm_request_dto):
 
     def _create_global_handler(self, request_id):
         global_worker = self.worker_landscape_aggregator.get_global_worker()
-        return self._create_worker_tasks_handler(request_id, global_worker)
+        return (
+            self._create_worker_tasks_handler(request_id, global_worker),
+            global_worker.ip,
+        )
 
     async def _cleanup(
         self, algorithm_name, server_task_handler, server_pid, clients_pids

exareme2/controller/services/flower/flower_io_registry.py

@@ -62,12 +62,13 @@ async def get_result(self) -> Dict[str, Any]:
     async def get_result_with_timeout(self) -> Dict[str, Any]:
         try:
             await asyncio.wait_for(self.get_result(), self._timeout)
+            self._logger.debug(f"Result with timeout: {self._result}")
+            return self._result.content
         except asyncio.TimeoutError:
             error = f"Failed to get result: operation timed out after {self._timeout} seconds"
             self._logger.error(error)
             self._result = Result(content={"error": error}, status=Status.FAILURE)
-        self._logger.debug(f"Result with timeout: {self._result}")
-        return self._result.content
+            raise TimeoutError(error)
 
     def get_status(self) -> Status:
         """Returns the current status of the execution."""

exareme2/controller/services/flower/tasks_handler.py

@@ -29,14 +29,16 @@ def worker_id(self) -> str:
     def worker_data_address(self) -> str:
         return self._db_address
 
-    def start_flower_client(self, algorithm_name) -> int:
+    def start_flower_client(self, algorithm_name, server_address) -> int:
         return self._worker_tasks_handler.start_flower_client(
-            self._request_id, algorithm_name, self._worker_id
+            self._request_id, algorithm_name, server_address
         ).get(timeout=self._tasks_timeout)
 
-    def start_flower_server(self, algorithm_name: str, number_of_clients: int) -> int:
+    def start_flower_server(
+        self, algorithm_name: str, number_of_clients: int, server_address
+    ) -> int:
         return self._worker_tasks_handler.start_flower_server(
-            self._request_id, algorithm_name, number_of_clients, self._worker_id
+            self._request_id, algorithm_name, number_of_clients, server_address
         ).get(timeout=self._tasks_timeout)
 
     def stop_flower_server(self, pid: int, algorithm_name: str):

exareme2/worker/Dockerfile

@@ -23,10 +23,16 @@ ENV PYTHONUNBUFFERED=1 \
     MONETDB_LOCAL_PASSWORD="executor" \
     MONETDB_PUBLIC_USERNAME="guest" \
     MONETDB_PUBLIC_PASSWORD="guest" \
+    DATA_PATH="/opt/data" \
     CODE_PATH="/opt/code"
 ENV PATH="$POETRY_HOME/bin:$CODE_PATH:$PATH"
 WORKDIR $CODE_PATH
 
+ENV PYTHONPATH=$CODE_PATH
+RUN mkdir $DATA_PATH
+VOLUME $DATA_PATH
+
+
 #######################################################
 # Installing poetry and dependencies
 #######################################################

exareme2/worker/README.md

@@ -17,6 +17,9 @@ WORKER_IDENTIFIER=globalworker
 WORKER_ROLE=GLOBALWORKER
 LOG_LEVEL=INFO
 FRAMEWORK_LOG_LEVEL=INFO
+DATA_PATH=/opt/data/
+CONTROLLER_IP=172.17.0.1
+CONTROLLER_PORT=5000
 PROTECT_LOCAL_DATA=false
 RABBITMQ_IP=172.17.0.1
 RABBITMQ_PORT=5670

exareme2/worker/__init__.py

@@ -1,14 +1,19 @@
 import os
 from importlib.resources import open_text
+from pathlib import Path
 
 import envtoml
 
 from exareme2 import AttrDict
 from exareme2 import worker
 
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+TEST_DATA_FOLDER = PROJECT_ROOT / "tests" / "test_data"
+
 if config_file := os.getenv("EXAREME2_WORKER_CONFIG_FILE"):
     with open(config_file) as fp:
         config = AttrDict(envtoml.load(fp))
+        config.data_path = TEST_DATA_FOLDER
 else:
     with open_text(worker, "config.toml") as fp:
         config = AttrDict(envtoml.load(fp))

exareme2/worker/config.toml

@@ -1,9 +1,14 @@
 identifier = "$WORKER_IDENTIFIER"
 role = "$WORKER_ROLE"
+data_path = "$DATA_PATH"
 
 log_level = "$LOG_LEVEL"
 framework_log_level = "$FRAMEWORK_LOG_LEVEL"
 
+[controller]
+ip = "$CONTROLLER_IP"
+port = "$CONTROLLER_PORT"
+
 [privacy]
 minimum_row_count = 10
 protect_local_data = "$PROTECT_LOCAL_DATA"

exareme2/worker/flower/starter/flower_api.py

@@ -4,14 +4,16 @@
 
 
 @shared_task
-def start_flower_client(request_id: str, algorithm_name, worker_id) -> int:
-    return flower_service.start_flower_client(request_id, algorithm_name, worker_id)
+def start_flower_client(request_id: str, algorithm_name, server_address) -> int:
+    return flower_service.start_flower_client(
+        request_id, algorithm_name, server_address
+    )
 
 
 @shared_task
 def start_flower_server(
-    request_id: str, algorithm_name: str, number_of_clients: int, worker_id
+    request_id: str, algorithm_name: str, number_of_clients: int, server_address
 ) -> int:
     return flower_service.start_flower_server(
-        request_id, algorithm_name, number_of_clients, worker_id
+        request_id, algorithm_name, number_of_clients, server_address
     )

exareme2/worker/flower/starter/flower_service.py

@@ -1,53 +1,46 @@
-from exareme2 import ALGORITHM_FOLDERS
 from exareme2.algorithms.flower.process_manager import FlowerProcess
 from exareme2.worker import config as worker_config
 from exareme2.worker.utils.logger import get_logger
 from exareme2.worker.utils.logger import initialise_logger
 
-# Dictionary to keep track of running processes
-running_processes = {}
-SERVER_ADDRESS = "0.0.0.0:8080"
-
 
 @initialise_logger
-def start_flower_client(request_id: str, algorithm_name, worker_id) -> int:
+def start_flower_client(request_id: str, algorithm_name, server_address) -> int:
     env_vars = {
         "MONETDB_IP": worker_config.monetdb.ip,
         "MONETDB_PORT": worker_config.monetdb.port,
         "MONETDB_USERNAME": worker_config.monetdb.local_username,
         "MONETDB_PASSWORD": worker_config.monetdb.local_password,
         "MONETDB_DB": worker_config.monetdb.database,
-        "SERVER_ADDRESS": SERVER_ADDRESS,
+        "SERVER_ADDRESS": server_address,
         "NUMBER_OF_CLIENTS": worker_config.monetdb.database,
+        "CONTROLLER_IP": worker_config.controller.ip,
+        "CONTROLLER_PORT": worker_config.controller.port,
+        "DATA_PATH": worker_config.data_path,
     }
-    with open(f"/tmp/exareme2/{worker_id}.out", "a") as f:
-        process = FlowerProcess(
-            f"{algorithm_name}/client.py", env_vars=env_vars, stderr=f, stdout=f
-        )
-        running_processes[request_id] = process
-        logger = get_logger()
+    process = FlowerProcess(f"{algorithm_name}/client.py", env_vars=env_vars)
+    logger = get_logger()
 
-        logger.info("Starting client.py")
-        pid = process.start(logger)
+    logger.info("Starting client.py")
+    pid = process.start(logger)
     logger.info(f"Started client.py process id: {pid}")
     return pid
 
 
 @initialise_logger
 def start_flower_server(
-    request_id: str, algorithm_name: str, number_of_clients: int, worker_id
+    request_id: str, algorithm_name: str, number_of_clients: int, server_address
 ) -> int:
     env_vars = {
-        "SERVER_ADDRESS": SERVER_ADDRESS,
+        "SERVER_ADDRESS": server_address,
         "NUMBER_OF_CLIENTS": number_of_clients,
+        "CONTROLLER_IP": worker_config.controller.ip,
+        "CONTROLLER_PORT": worker_config.controller.port,
+        "DATA_PATH": worker_config.data_path,
     }
-    with open(f"/tmp/exareme2/{worker_id}.out", "a") as f:
-        process = FlowerProcess(
-            f"{algorithm_name}/server.py", env_vars=env_vars, stderr=f, stdout=f
-        )
-        running_processes[request_id] = process
-        logger = get_logger()
-        logger.info("Starting server.py")
-        pid = process.start(logger)
+    process = FlowerProcess(f"{algorithm_name}/server.py", env_vars=env_vars)
+    logger = get_logger()
+    logger.info("Starting server.py")
+    pid = process.start(logger)
     logger.info(f"Started server.py process id: {pid}")
     return pid

kubernetes/templates/exareme2-globalnode.yaml

@@ -23,6 +23,9 @@ spec:
       - name: db-data
         hostPath:
           path: {{ .Values.db.storage_location }}
+      - name: csv-data
+        hostPath:
+          path: {{ .Values.db.csvs_location }}
       containers:
       - name: monetdb
         image: {{ .Values.exareme2_images.repository }}/exareme2_db:{{ .Values.exareme2_images.version }}
@@ -97,6 +100,9 @@ spec:
 
       - name: worker
         image: {{ .Values.exareme2_images.repository }}/exareme2_worker:{{ .Values.exareme2_images.version }}
+        volumeMounts:
+        - mountPath: /opt/data
+          name: csv-data
         env:
         - name: WORKER_IDENTIFIER
           value: "globalworker"
@@ -106,6 +112,13 @@ spec:
           value: {{ .Values.log_level }}
         - name: FRAMEWORK_LOG_LEVEL
           value: {{ .Values.framework_log_level }}
+        - name: CONTROLLER_IP
+          value: exareme2-controller-service
+        - name: CONTROLLER_PORT
+          valueFrom:
+            configMapKeyRef:
+              name: exareme2-controller-service
+              key: EXAREME2_CONTROLLER_SERVICE_SERVICE_PORT
         - name: PROTECT_LOCAL_DATA
           value: "false"      # The GLOBALWORKER does not need to secure its data, since they are not private.
         - name: CELERY_TASKS_TIMEOUT