[CoreEngine] In order to make the inference logs work, we save the container inference logs to the single dir.

Author: fedml-alex

python/fedml/computing/scheduler/comm_utils/job_monitor.py

@@ -48,6 +48,7 @@
 class JobMonitor(Singleton):
     ENDPOINT_CONTAINER_LOG_PREFIX = "endpoint"
     TIME_INTERVAL_FOR_INFERENCE_ON_GATEWAY = 60 * 10
+    ENDPOINT_CONTAINER_LOG_SUBDIR = "monitor_endpoint_logs"
 
     def __init__(self):
         if not hasattr(self, "endpoint_unavailable_counter"):
@@ -1055,8 +1056,11 @@ def monitor_endpoint_logs(self):
                 model_version = model_config.get("model_version", None)
                 endpoint_name = endpoint_json.get("end_point_name", None)
 
+                log_file_dir = os.path.join(
+                    device_client_constants.ClientConstants.get_log_file_dir(),
+                    JobMonitor.ENDPOINT_CONTAINER_LOG_SUBDIR)
                 log_file_path, program_prefix = MLOpsLoggingUtils.build_log_file_path_with_run_params(
-                    job.job_id, int(job.edge_id), device_server_constants.ServerConstants.get_log_file_dir(), is_server=True,
+                    job.job_id, int(job.edge_id), log_file_dir, is_server=False,
                     log_file_prefix=JobMonitor.ENDPOINT_CONTAINER_LOG_PREFIX,
                 )
 
@@ -1130,8 +1134,9 @@ def monitor_endpoint_logs(self):
                                 nano_second_str = container_time.split(".")[1][:9]
                                 t_datetime_obj = isoparse(container_time)
 
-                                if t_sec_offset is not None:
-                                    t_datetime_obj = t_datetime_obj + datetime.timedelta(seconds=t_sec_offset)
+                                # ISSUE: this will cause the timestamp is not correct.
+                                #if t_sec_offset is not None:
+                                #    t_datetime_obj = t_datetime_obj + datetime.timedelta(seconds=t_sec_offset)
                             except Exception as e:
                                 logging.error(f"Exception when parsing the container log time {e}")
                                 t_datetime_obj = datetime.datetime.now()

python/fedml/computing/scheduler/scheduler_core/scheduler_base_job_runner.py

@@ -12,7 +12,7 @@
 from ..comm_utils.constants import SchedulerConstants
 from ..comm_utils.job_utils import JobRunnerUtils, DockerArgs
 from ..scheduler_entry.constants import Constants
-from ....core.mlops import MLOpsMetrics
+from ....core.mlops import MLOpsMetrics, MLOpsRuntimeLogDaemon
 from ....core.mlops.mlops_device_perfs import MLOpsDevicePerfStats
 from ..comm_utils.yaml_utils import load_yaml_config
 from .general_constants import GeneralConstants
@@ -449,10 +449,16 @@ def trigger_stop_event(self):
         if self.run_process_event is not None:
             self.run_process_event.set()
 
+        time.sleep(1)
+        MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(self.run_id, self.edge_id)
+
     def trigger_completed_event(self):
         if self.run_process_completed_event is not None:
             self.run_process_completed_event.set()
 
+        time.sleep(1)
+        MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(self.run_id, self.edge_id)
+
     def execute_job_task(self, unzip_package_path, entry_file_full_path, conf_file_full_path, dynamic_args_config,
                          fedml_config_object):
         run_config = self.request_json["run_config"]

python/fedml/core/mlops/mlops_utils.py

@@ -150,10 +150,11 @@ def get_edge_id_from_args(args):
                     else:
                         edge_id = 0
             else:
-                if getattr(args, "client_id", None) is not None:
+                if getattr(args, "edge_id", None) is not None:
                     edge_id = args.edge_id
                 else:
                     edge_id = 0
+
         return edge_id
 
     @staticmethod