madgik · ThanKarab · Nov 1, 2024 · Oct 29, 2024
diff --git a/README.md b/README.md
@@ -42,6 +42,7 @@
 
    ```
    ip = "172.17.0.1"
+   federation = "dementia"
    log_level = "DEBUG"
    framework_log_level ="INFO"
    monetdb_image = "madgik/exareme2_db:dev"

diff --git a/exareme2/algorithms/exareme2/kmeans.py b/exareme2/algorithms/exareme2/kmeans.py
@@ -70,12 +70,11 @@ def run(self, data, metadata):
         curr_iter = 0
         centers_to_compute = global_result2
         centers_to_compute_global = global_result
-        print(centers_to_compute)
 
-        init_centers = get_transfer_data(global_result)["centers"]
-
-        init_centers_array = numpy.array(init_centers)
-        init_centers_list = init_centers_array.tolist()
+        # init_centers = get_transfer_data(global_result)["centers"]
+        #
+        # init_centers_array = numpy.array(init_centers)
+        # init_centers_list = init_centers_array.tolist()
         while True:
             metrics_local = local_run(
                 func=compute_metrics2,
@@ -103,7 +102,6 @@ def run(self, data, metadata):
                     title="K-Means Centers",
                     centers=new_centers_array.tolist(),
                 )
-                print("finished after " + str(curr_iter))
                 return ret_obj
 
             else:

diff --git a/exareme2/algorithms/exareme2/udfgen/udfio.py b/exareme2/algorithms/exareme2/udfgen/udfio.py
@@ -23,7 +23,7 @@ def get_logger(udf_name: str, request_id: str):
 
     log_level = os.getenv(LOG_LEVEL_ENV_VARIABLE, LOG_LEVEL_DEFAULT_VALUE)
     formatter = logging.Formatter(
-        f"%(asctime)s - %(levelname)s - MONETDB - PYTHONUDF - {udf_name}(%(lineno)d) - {request_id} - %(message)s"
+        f"%(asctime)s - %(levelname)s - {udf_name}(%(lineno)d) - [UNKNOWN] - [exareme2-monetdb] - [UNKNOWN] - [{request_id}] - %(message)s"
     )
     # StreamHandler
     sh = logging.StreamHandler()

diff --git a/exareme2/algorithms/flower/__init__.py b/exareme2/algorithms/flower/__init__.py
@@ -3,21 +3,22 @@
 
 from flwr.common.logger import FLOWER_LOGGER
 
-for handler in FLOWER_LOGGER.handlers:
-    FLOWER_LOGGER.removeHandler(handler)
-
-FLOWER_LOGGER.setLevel(logging.DEBUG)
-
-request_id = os.getenv("REQUEST_ID", "NO-REQUEST_ID")
+node_identifier = os.getenv("WORKER_IDENTIFIER", "NO-IDENTIFIER")
+federation = os.getenv("FEDERATION", "NO-FEDERATION")
 worker_role = os.getenv("WORKER_ROLE", "NO-ROLE")
-worker_identifier = os.getenv("WORKER_IDENTIFIER", "NO-IDENTIFIER")
+framework_log_level = os.getenv("FRAMEWORK_LOG_LEVEL", "INFO")
+request_id = os.getenv("REQUEST_ID", "NO-REQUEST_ID")
 
 flower_formatter = logging.Formatter(
-    f"%(asctime)s - %(levelname)s - FLOWER - {worker_role} - {worker_identifier} - %(module)s - %(funcName)s(%(lineno)d) - {request_id} - %(message)s"
+    f"%(asctime)s - %(levelname)s - %(module)s.%(funcName)s(%(lineno)d) - [{federation}] - [exareme2-flower-{worker_role.lower()}] - [{node_identifier}] - [{request_id}] - %(message)s"
 )
 
 # Configure console logger
 console_handler = logging.StreamHandler()
-console_handler.setLevel(logging.DEBUG)
+console_handler.setLevel(framework_log_level)
 console_handler.setFormatter(flower_formatter)
+
+for handler in FLOWER_LOGGER.handlers:
+    FLOWER_LOGGER.removeHandler(handler)
+FLOWER_LOGGER.setLevel(framework_log_level)
 FLOWER_LOGGER.addHandler(console_handler)
diff --git a/exareme2/controller/config.toml b/exareme2/controller/config.toml
@@ -1,3 +1,6 @@
+node_identifier = "$NODE_IDENTIFIER"
+federation = "$FEDERATION"
+
 log_level = "$LOG_LEVEL"
 framework_log_level = "$FRAMEWORK_LOG_LEVEL"
 

diff --git a/exareme2/controller/logger.py b/exareme2/controller/logger.py
@@ -19,7 +19,7 @@ def get_request_logger(request_id):
 def init_logger(request_id, log_level=None):
     logger = logging.getLogger(request_id)
     formatter = logging.Formatter(
-        f"%(asctime)s - %(levelname)s - CONTROLLER - %(module)s - %(funcName)s(%(lineno)d) - {request_id} - %(message)s"
+        f"%(asctime)s - %(levelname)s - %(module)s.%(funcName)s(%(lineno)d) - [{ctrl_config.federation}] - [exareme2-controller] - [{ctrl_config.node_identifier}] - [{request_id}] - %(message)s"
     )
 
     # StreamHandler

diff --git a/exareme2/controller/quart/loggers.py b/exareme2/controller/quart/loggers.py
@@ -5,10 +5,10 @@
     "version": 1,
     "formatters": {
         "controller_background_service_frm": {
-            "format": "%(asctime)s - %(levelname)s - CONTROLLER - BACKGROUND - %(module)s - %(funcName)s(%(lineno)d) - %(message)s",
+            "format": f"%(asctime)s - %(levelname)s - %(module)s.%(funcName)s(%(lineno)d) - [{ctrl_config.federation}] - [exareme2-controller] - [{ctrl_config.node_identifier}] - [BACKGROUND] - %(message)s"
         },
         "framework": {
-            "format": "%(asctime)s - %(levelname)s - CONTROLLER - WEBAPI - %(message)s"
+            "format": f"%(asctime)s - %(levelname)s - WEBAPI FRAMEWORK - [{ctrl_config.federation}] - [exareme2-controller] - [{ctrl_config.node_identifier}] - [FRAMEWORK] - %(message)s"
         },
     },
     "handlers": {

diff --git a/exareme2/worker/config.toml b/exareme2/worker/config.toml
@@ -1,5 +1,7 @@
 identifier = "$WORKER_IDENTIFIER"
 role = "$WORKER_ROLE"
+federation = "$FEDERATION"
+
 data_path = "$DATA_PATH"
 
 log_level = "$LOG_LEVEL"

diff --git a/exareme2/worker/flower/starter/starter_service.py b/exareme2/worker/flower/starter/starter_service.py
@@ -15,8 +15,11 @@ def start_flower_client(
         "MONETDB_PASSWORD": worker_config.monetdb.local_password,
         "MONETDB_DB": worker_config.monetdb.database,
         "REQUEST_ID": request_id,
-        "WORKER_ROLE": worker_config.role,
+        "FEDERATION": worker_config.federation,
         "WORKER_IDENTIFIER": worker_config.identifier,
+        "WORKER_ROLE": worker_config.role,
+        "LOG_LEVEL": worker_config.log_level,
+        "FRAMEWORK_LOG_LEVEL": worker_config.framework_log_level,
         "SERVER_ADDRESS": server_address,
         "NUMBER_OF_CLIENTS": worker_config.monetdb.database,
         "CONTROLLER_IP": worker_config.controller.ip,
@@ -28,9 +31,9 @@ def start_flower_client(
     process = FlowerProcess(f"{algorithm_folder_path}/client.py", env_vars=env_vars)
     logger = get_logger()
 
-    logger.info("Starting client.py")
+    logger.info("Starting flower client...")
     pid = process.start(logger)
-    logger.info(f"Started client.py process id: {pid}")
+    logger.info(f"Started flower client, with process id: {pid}")
     return pid
 
 
@@ -43,9 +46,12 @@ def start_flower_server(
     csv_paths,
 ) -> int:
     env_vars = {
-        "REQUEST_ID": request_id,
+        "FEDERATION": worker_config.federation,
         "WORKER_ROLE": worker_config.role,
         "WORKER_IDENTIFIER": worker_config.identifier,
+        "LOG_LEVEL": worker_config.log_level,
+        "FRAMEWORK_LOG_LEVEL": worker_config.framework_log_level,
+        "REQUEST_ID": request_id,
         "SERVER_ADDRESS": server_address,
         "NUMBER_OF_CLIENTS": number_of_clients,
         "CONTROLLER_IP": worker_config.controller.ip,
@@ -55,7 +61,7 @@ def start_flower_server(
     }
     process = FlowerProcess(f"{algorithm_folder_path}/server.py", env_vars=env_vars)
     logger = get_logger()
-    logger.info("Starting server.py")
+    logger.info("Starting flower server...")
     pid = process.start(logger)
-    logger.info(f"Started server.py process id: {pid}")
+    logger.info(f"Started flower server, with process id: {pid}")
     return pid
diff --git a/exareme2/worker/utils/celery_app.py b/exareme2/worker/utils/celery_app.py
@@ -40,7 +40,7 @@
 def setup_celery_logging(*args, **kwargs):
     logger = logging.getLogger()
     formatter = logging.Formatter(
-        f"%(asctime)s - %(levelname)s - WORKER - {worker_config.role} - {worker_config.identifier} - CELERY - FRAMEWORK - %(message)s"
+        f"%(asctime)s - %(levelname)s - CELERY FRAMEWORK - [{worker_config.federation}] - [exareme2-controller] - [{worker_config.identifier}] - [FRAMEWORK] - %(message)s"
     )
 
     # StreamHandler

diff --git a/exareme2/worker/utils/logger.py b/exareme2/worker/utils/logger.py
@@ -17,7 +17,7 @@ def init_logger(request_id):
     logger = logging.getLogger(request_id)
 
     formatter = logging.Formatter(
-        f"%(asctime)s - %(levelname)s - WORKER - {worker_config.role} - {worker_config.identifier} - %(module)s - %(funcName)s(%(lineno)d) - {request_id} - %(message)s"
+        f"%(asctime)s - %(levelname)s - %(module)s.%(funcName)s(%(lineno)d) - [{worker_config.federation}] - [exareme2-{worker_config.role.lower()}] - [{worker_config.identifier}] - [{request_id}] - %(message)s"
     )
 
     # StreamHandler

diff --git a/kubernetes/templates/exareme2-controller.yaml b/kubernetes/templates/exareme2-controller.yaml
@@ -29,6 +29,12 @@ spec:
         - mountPath: /opt/cleanup
           name: cleanup-file
         env:
+        - name: NODE_IDENTIFIER
+          valueFrom:
+            fieldRef:
+              fieldPath: spec.nodeName
+        - name: FEDERATION
+          value: {{ .Values.federation }}
         - name: LOG_LEVEL
           value: {{ .Values.log_level }}
         - name: FRAMEWORK_LOG_LEVEL

diff --git a/kubernetes/templates/exareme2-globalnode.yaml b/kubernetes/templates/exareme2-globalnode.yaml
@@ -133,6 +133,8 @@ spec:
           value: {{ .Values.globalworker_identifier }}
         - name: WORKER_ROLE
           value: "GLOBALWORKER"
+        - name: FEDERATION
+          value: {{ .Values.federation }}
         - name: LOG_LEVEL
           value: {{ .Values.log_level }}
         - name: FRAMEWORK_LOG_LEVEL

diff --git a/kubernetes/templates/exareme2-localnode.yaml b/kubernetes/templates/exareme2-localnode.yaml
@@ -147,6 +147,8 @@ spec:
               fieldPath: spec.nodeName
         - name: WORKER_ROLE
           value: "LOCALWORKER"
+        - name: FEDERATION
+          value: {{ .Values.federation }}
         - name: LOG_LEVEL
           value: {{ .Values.log_level }}
         - name: FRAMEWORK_LOG_LEVEL

diff --git a/kubernetes/values.yaml b/kubernetes/values.yaml
@@ -4,13 +4,13 @@ exareme2_images:
   repository: madgik
   version: latest
 
+federation: dementia
+
 log_level: INFO
 framework_log_level: ERROR
 
 max_concurrent_experiments: 32
 
-globalworker_identifier: globalworker
-
 db:
   credentials_location: /opt/exareme2/credentials
   storage_location: /opt/exareme2/db
@@ -26,6 +26,8 @@ controller:
   workers_cleanup_interval: 60
   cleanup_file_folder: /opt/cleanup
 
+globalworker_identifier: globalworker
+
 smpc:
   enabled: false
   optional: false

diff --git a/tasks.py b/tasks.py
@@ -140,6 +140,7 @@ def create_configs(c):
 
         worker_config["identifier"] = worker["id"]
         worker_config["role"] = worker["role"]
+        worker_config["federation"] = deployment_config["federation"]
         worker_config["log_level"] = deployment_config["log_level"]
         worker_config["framework_log_level"] = deployment_config["framework_log_level"]
         worker_config["controller"]["ip"] = deployment_config["ip"]
@@ -206,6 +207,8 @@ def create_configs(c):
     with open(CONTROLLER_CONFIG_TEMPLATE_FILE) as fp:
         template_controller_config = toml.load(fp)
     controller_config = copy.deepcopy(template_controller_config)
+    controller_config["node_identifier"] = "controller"
+    controller_config["federation"] = deployment_config["federation"]
     controller_config["log_level"] = deployment_config["log_level"]
     controller_config["framework_log_level"] = deployment_config["framework_log_level"]
 

diff --git a/tests/algorithm_validation_tests/five_node_deployment_template.toml b/tests/algorithm_validation_tests/five_node_deployment_template.toml
@@ -1,4 +1,5 @@
 ip = "172.17.0.1"
+federation = "algorithm_tests"
 log_level = "INFO"
 framework_log_level ="INFO"
 monetdb_image = "madgik/exareme2_db:testing"

diff --git a/tests/algorithm_validation_tests/one_node_deployment_template.toml b/tests/algorithm_validation_tests/one_node_deployment_template.toml
@@ -1,4 +1,5 @@
 ip = "172.17.0.1"
+federation = "algorithm_tests"
 log_level = "INFO"
 framework_log_level ="INFO"
 monetdb_image = "madgik/exareme2_db:testing"

diff --git a/tests/standalone_tests/algorithms/exareme2/test_longitudinal_transformer.py b/tests/standalone_tests/algorithms/exareme2/test_longitudinal_transformer.py
@@ -199,6 +199,7 @@ def db(globalworker_db_cursor):
 
 @pytest.mark.slow
 @pytest.mark.database
+@pytest.mark.usefixtures("monetdb_globalworker")
 class TestLongitudinalTransformerUdf_WithDb:
     test_table = "test_longitudinal_table"
 

diff --git a/tests/standalone_tests/conftest.py b/tests/standalone_tests/conftest.py
@@ -939,7 +939,7 @@ def _create_worker_service(algo_folders_env_variable_val, worker_config_filepath
     )
 
     # Check that celery started
-    _search_for_string_in_logfile("CELERY - FRAMEWORK - celery@.* ready.", logpath)
+    _search_for_string_in_logfile("celery@.* ready.", logpath)
 
     print(f"Created worker service with id '{worker_id}' and process id '{proc.pid}'.")
     return proc
@@ -1272,9 +1272,7 @@ def _create_controller_service(
     _search_for_string_in_logfile("Running on", logpath)
 
     # Check that workers were loaded
-    _search_for_string_in_logfile(
-        "INFO - CONTROLLER - BACKGROUND - federation_info_logs", logpath
-    )
+    _search_for_string_in_logfile("Workers:", logpath)
     print(f"\nCreated controller service on port '{service_port}'.")
 
     return proc

diff --git a/...agreggator/test_federation_info_script.py → ...lone_tests/test_federation_info_script.py b/...agreggator/test_federation_info_script.py → ...lone_tests/test_federation_info_script.py
@@ -31,6 +31,8 @@
 def controller_config_dict_mock():
     controller_config = {
         "log_level": "INFO",
+        "node_identifier": "controller",
+        "federation": "standalone_tests",
     }
     yield controller_config
 

diff --git a/tests/standalone_tests/testing_env_configs/test_controller.toml b/tests/standalone_tests/testing_env_configs/test_controller.toml
@@ -1,3 +1,6 @@
+node_identifier = "controller"
+federation = "standalone_tests"
+
 log_level = "DEBUG"
 framework_log_level = "INFO"
 deployment_type = "LOCAL"

diff --git a/tests/standalone_tests/testing_env_configs/test_external_smpc_controller.toml b/tests/standalone_tests/testing_env_configs/test_external_smpc_controller.toml
@@ -1,3 +1,6 @@
+node_identifier = "controller"
+federation = "standalone_tests"
+
 log_level = "DEBUG"
 framework_log_level = "INFO"
 deployment_type = "LOCAL"

diff --git a/tests/standalone_tests/testing_env_configs/test_external_smpc_dp_controller.toml b/tests/standalone_tests/testing_env_configs/test_external_smpc_dp_controller.toml
@@ -1,3 +1,6 @@
+node_identifier = "controller"
+federation = "standalone_tests"
+
 log_level = "DEBUG"
 framework_log_level = "INFO"
 deployment_type = "LOCAL"

diff --git a/tests/standalone_tests/testing_env_configs/test_external_smpc_globalworker.toml b/tests/standalone_tests/testing_env_configs/test_external_smpc_globalworker.toml
@@ -1,4 +1,5 @@
 identifier = "testsmpcglobalworker"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "GLOBALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_external_smpc_localworker1.toml b/tests/standalone_tests/testing_env_configs/test_external_smpc_localworker1.toml
@@ -1,4 +1,5 @@
 identifier = "testsmpclocalworker1"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "LOCALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_external_smpc_localworker2.toml b/tests/standalone_tests/testing_env_configs/test_external_smpc_localworker2.toml
@@ -1,4 +1,5 @@
 identifier = "testsmpclocalworker2"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "LOCALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_globalworker.toml b/tests/standalone_tests/testing_env_configs/test_globalworker.toml
@@ -1,4 +1,5 @@
 identifier = "testglobalworker"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "GLOBALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_localworker1.toml b/tests/standalone_tests/testing_env_configs/test_localworker1.toml
@@ -1,4 +1,5 @@
 identifier = "testlocalworker1"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "LOCALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_localworker2.toml b/tests/standalone_tests/testing_env_configs/test_localworker2.toml
@@ -1,4 +1,5 @@
 identifier = "testlocalworker2"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "LOCALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_localworkertmp.toml b/tests/standalone_tests/testing_env_configs/test_localworkertmp.toml
@@ -1,4 +1,5 @@
 identifier = "testlocalworkertmp"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "LOCALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_smpc_controller.toml b/tests/standalone_tests/testing_env_configs/test_smpc_controller.toml
@@ -1,3 +1,6 @@
+node_identifier = "controller"
+federation = "standalone_tests"
+
 log_level = "DEBUG"
 framework_log_level = "INFO"
 deployment_type = "LOCAL"

diff --git a/tests/standalone_tests/testing_env_configs/test_smpc_globalworker.toml b/tests/standalone_tests/testing_env_configs/test_smpc_globalworker.toml
@@ -1,4 +1,5 @@
 identifier = "testsmpcglobalworker"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "GLOBALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_smpc_localworker1.toml b/tests/standalone_tests/testing_env_configs/test_smpc_localworker1.toml
@@ -1,4 +1,5 @@
 identifier = "testsmpclocalworker1"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "LOCALWORKER"

diff --git a/tests/standalone_tests/testing_env_configs/test_smpc_localworker2.toml b/tests/standalone_tests/testing_env_configs/test_smpc_localworker2.toml
@@ -1,4 +1,5 @@
 identifier = "testsmpclocalworker2"
+federation = "standalone_tests"
 log_level = "DEBUG"
 framework_log_level = "INFO"
 role = "LOCALWORKER"

diff --git a/tests/standalone_tests/worker/exareme2/test_monet_db_facade.py b/tests/standalone_tests/worker/exareme2/test_monet_db_facade.py
@@ -31,6 +31,7 @@ def patch_worker_logger():
                 "log_level": "DEBUG",
                 "role": "localworker",
                 "identifier": "localworkertmp",
+                "federation": "standalone_tests",
             },
         ),
     ), patch(