Data are loaded on the globalworker too.

exareme2/algorithms/flower/inputdata_preprocessing.py

@@ -39,8 +39,7 @@ def apply_inputdata(df: pd.DataFrame, inputdata: Inputdata) -> pd.DataFrame:
     return df
 
 
-def fetch_client_data(inputdata) -> pd.DataFrame:
-    FLOWER_LOGGER.error(f"BROOO {os.getenv('CSV_PATHS')}")
+def fetch_data(inputdata) -> pd.DataFrame:
     dataframes = [
         pd.read_csv(f"{os.getenv('DATA_PATH')}{csv_path}")
         for csv_path in os.getenv("CSV_PATHS").split(",")
@@ -49,20 +48,6 @@ def fetch_client_data(inputdata) -> pd.DataFrame:
     return apply_inputdata(df, inputdata)
 
 
-def fetch_server_data(inputdata) -> pd.DataFrame:
-    data_folder = Path(
-        f"{os.getenv('DATA_PATH')}/{inputdata.data_model.split(':')[0]}_v_0_1"
-    )
-    print(f"Loading data from folder: {data_folder}")
-    dataframes = [
-        pd.read_csv(data_folder / f"{dataset}.csv")
-        for dataset in inputdata.datasets
-        if (data_folder / f"{dataset}.csv").exists()
-    ]
-    df = pd.concat(dataframes, ignore_index=True)
-    return apply_inputdata(df, inputdata)
-
-
 def preprocess_data(inputdata, full_data):
     # Ensure x and y are specified and correct
     if not inputdata.x or not inputdata.y:

exareme2/algorithms/flower/logistic_regression/client.py

@@ -11,7 +11,7 @@
 from utils import set_initial_params
 from utils import set_model_params
 
-from exareme2.algorithms.flower.inputdata_preprocessing import fetch_client_data
+from exareme2.algorithms.flower.inputdata_preprocessing import fetch_data
 from exareme2.algorithms.flower.inputdata_preprocessing import get_input
 from exareme2.algorithms.flower.inputdata_preprocessing import preprocess_data
 
@@ -42,7 +42,7 @@ def evaluate(self, parameters, config):
 if __name__ == "__main__":
     model = LogisticRegression(penalty="l2", max_iter=1, warm_start=True)
     inputdata = get_input()
-    full_data = fetch_client_data(inputdata)
+    full_data = fetch_data(inputdata)
     X_train, y_train = preprocess_data(inputdata, full_data)
     set_initial_params(model, X_train, full_data, inputdata)
 

exareme2/algorithms/flower/logistic_regression/server.py

@@ -6,7 +6,7 @@
 from utils import set_initial_params
 from utils import set_model_params
 
-from exareme2.algorithms.flower.inputdata_preprocessing import fetch_server_data
+from exareme2.algorithms.flower.inputdata_preprocessing import fetch_data
 from exareme2.algorithms.flower.inputdata_preprocessing import get_input
 from exareme2.algorithms.flower.inputdata_preprocessing import post_result
 from exareme2.algorithms.flower.inputdata_preprocessing import preprocess_data
@@ -35,7 +35,7 @@ def evaluate(server_round, parameters, config):
 if __name__ == "__main__":
     model = LogisticRegression()
     inputdata = get_input()
-    full_data = fetch_server_data(inputdata)
+    full_data = fetch_data(inputdata)
     X_train, y_train = preprocess_data(inputdata, full_data)
     set_initial_params(model, X_train, full_data, inputdata)
     strategy = fl.server.strategy.FedAvg(

exareme2/controller/celery/tasks_handler.py

@@ -310,14 +310,15 @@ def start_flower_client(
         )
 
     def start_flower_server(
-        self, request_id, algorithm_name, number_of_clients, server_address
+        self, request_id, algorithm_name, number_of_clients, server_address, csv_paths
     ) -> 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,
             server_address=server_address,
+            csv_paths=csv_paths,
         )
 
     def stop_flower_server(

exareme2/controller/services/flower/controller.py

@@ -77,7 +77,7 @@ async def exec_algorithm(self, algorithm_name, algorithm_request_dto):
                     request_id, global_worker
                 )
                 server_ip = global_worker.ip
-
+                server_id = global_worker.id
             # Garbage Collect
             server_task_handler.garbage_collect()
             for handler in task_handlers:
@@ -92,7 +92,10 @@ async def exec_algorithm(self, algorithm_name, algorithm_request_dto):
 
             try:
                 server_pid = server_task_handler.start_flower_server(
-                    algorithm_name, len(task_handlers), str(server_address)
+                    algorithm_name,
+                    len(task_handlers),
+                    str(server_address),
+                    csv_paths_per_worker_id[server_id],
                 )
                 clients_pids = {
                     handler.start_flower_client(

exareme2/controller/services/flower/tasks_handler.py

@@ -41,10 +41,14 @@ def start_flower_client(
         ).get(timeout=self._tasks_timeout)
 
     def start_flower_server(
-        self, algorithm_name: str, number_of_clients: int, server_address
+        self, algorithm_name: str, number_of_clients: int, server_address, csv_paths
     ) -> int:
         return self._worker_tasks_handler.start_flower_server(
-            self._request_id, algorithm_name, number_of_clients, server_address
+            self._request_id,
+            algorithm_name,
+            number_of_clients,
+            server_address,
+            csv_paths,
         ).get(timeout=self._tasks_timeout)
 
     def stop_flower_server(self, pid: int, algorithm_name: str):

exareme2/controller/services/worker_landscape_aggregator/worker_landscape_aggregator.py

@@ -557,13 +557,8 @@ def _fetch_workers_metadata(
             .socket_addresses
         )
         workers_info = self._get_workers_info(workers_addresses)
-        local_workers = [
-            worker_info
-            for worker_info in workers_info
-            if worker_info.role == WorkerRole.LOCALWORKER
-        ]
         data_models_metadata_per_worker = self._get_data_models_metadata_per_worker(
-            local_workers
+            workers_info
         )
         return workers_info, data_models_metadata_per_worker
 

exareme2/worker/flower/starter/starter_api.py

@@ -14,8 +14,12 @@ def start_flower_client(
 
 @shared_task
 def start_flower_server(
-    request_id: str, algorithm_name: str, number_of_clients: int, server_address
+    request_id: str,
+    algorithm_name: str,
+    number_of_clients: int,
+    server_address,
+    csv_paths,
 ) -> int:
     return starter_service.start_flower_server(
-        request_id, algorithm_name, number_of_clients, server_address
+        request_id, algorithm_name, number_of_clients, server_address, csv_paths
     )

exareme2/worker/flower/starter/starter_service.py

@@ -36,7 +36,11 @@ def start_flower_client(
 
 @initialise_logger
 def start_flower_server(
-    request_id: str, algorithm_name: str, number_of_clients: int, server_address
+    request_id: str,
+    algorithm_name: str,
+    number_of_clients: int,
+    server_address,
+    csv_paths,
 ) -> int:
     env_vars = {
         "REQUEST_ID": request_id,
@@ -47,6 +51,7 @@ def start_flower_server(
         "CONTROLLER_IP": worker_config.controller.ip,
         "CONTROLLER_PORT": worker_config.controller.port,
         "DATA_PATH": worker_config.data_path,
+        "CSV_PATHS": ",".join(csv_paths),
     }
     process = FlowerProcess(f"{algorithm_name}/server.py", env_vars=env_vars)
     logger = get_logger()

exareme2/worker/worker_info/worker_info_db.py

@@ -1,5 +1,4 @@
 import json
-import warnings
 from typing import Dict
 from typing import List
 
@@ -38,7 +37,6 @@ def get_data_models() -> List[str]:
 
 
 def convert_absolute_dataset_path_to_relative(dataset_path: str) -> str:
-    warnings.warn(str(dataset_path))
     return dataset_path.split(str(worker_config.data_path))[-1]
 
 

kubernetes/templates/exareme2-globalnode.yaml

@@ -67,6 +67,25 @@ spec:
             - "select 1;"
           periodSeconds: 30
 
+      - name: db-importer
+        image: {{ .Values.exareme2_images.repository }}/exareme2_mipdb:{{ .Values.exareme2_images.version }}
+        env:
+        - name: DB_IP
+          valueFrom:
+            fieldRef:
+              fieldPath: status.podIP
+        - name: SQLITE_DB_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: spec.nodeName
+        - name: DB_PORT
+          value: "50000"
+        volumeMounts:
+        - mountPath: /opt/data
+          name: csv-data
+        - mountPath: /opt/credentials
+          name: credentials
+
       - name: rabbitmq
         image: {{ .Values.exareme2_images.repository }}/exareme2_rabbitmq:{{ .Values.exareme2_images.version }}
         env: