Logs refactoring to fit ELASTIC stack

FEDERATION env variable is now required in both controller and worker services. NODE_IDENTIFIER env variable is needed in the controller service. All loggers now have the same formatter. Added node identifier to all loggers (controller included) Refactored flower loggers to use the framework log level of the service. Minor log removals.
madgik · Oct 29, 2024 · fece1a8 · fece1a8
1 parent dbf1078
commit fece1a8
Show file tree

Hide file tree

Showing 37 changed files with 90 additions and 37 deletions.
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}] - [{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}] - [{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(