[BugFix] Seperate prometheus multiproc dir for single-server multi-dp services

Author: liyonghua0910

fastdeploy/engine/common_engine.py

@@ -67,6 +67,7 @@
 )
 from fastdeploy.inter_communicator.fmq import FMQ
 from fastdeploy.metrics.metrics import main_process_metrics
+from fastdeploy.metrics.prometheus_multiprocess_setup import setup_dp_prometheus_dir
 from fastdeploy.model_executor.guided_decoding import schema_checker
 from fastdeploy.plugins.token_processor import load_token_processor_plugins
 from fastdeploy.spec_decode import SpecMethod
@@ -2570,6 +2571,7 @@ def launch_components(self):
                 self.launched_expert_service_signal.value[0] = 1
                 self.dp_processed = []
                 self.dp_engine_worker_queue_server = []
+                base_prom_dir = os.environ.get("PROMETHEUS_MULTIPROC_DIR")
                 for i in range(
                     1,
                     self.cfg.parallel_config.data_parallel_size // self.cfg.nnode,
@@ -2608,10 +2610,13 @@ def launch_components(self):
                         f"Engine is initialized successfully with {self.cfg.parallel_config.tensor_parallel_size}"
                         + f" data parallel id {i}"
                     )
+                    setup_dp_prometheus_dir(i, base_prom_dir)
                     self.dp_processed[-1].start()
                     while self.launched_expert_service_signal.value[i] == 0:
                         time.sleep(1)
 
+                setup_dp_prometheus_dir(0, base_prom_dir)
+
     def check_worker_initialize_status(self):
         """
         Check the initlialize status of workers by stdout logging

fastdeploy/engine/engine.py

@@ -45,6 +45,7 @@
 from fastdeploy.engine.request import Request
 from fastdeploy.inter_communicator import EngineWorkerQueue, IPCSignal
 from fastdeploy.metrics.metrics import main_process_metrics
+from fastdeploy.metrics.prometheus_multiprocess_setup import setup_dp_prometheus_dir
 from fastdeploy.platforms import current_platform
 from fastdeploy.utils import EngineError, console_logger, envs, llm_logger
 
@@ -805,6 +806,7 @@ def launch_components(self):
                 self.launched_expert_service_signal.value[0] = 1
                 self.dp_processed = []
                 self.dp_engine_worker_queue_server = []
+                base_prom_dir = os.environ.get("PROMETHEUS_MULTIPROC_DIR")
                 for i in range(
                     1,
                     self.cfg.parallel_config.data_parallel_size // self.cfg.nnode,
@@ -842,8 +844,11 @@ def launch_components(self):
                         f"Engine is initialized successfully with {self.cfg.parallel_config.tensor_parallel_size}"
                         + f" data parallel id {i}"
                     )
+                    setup_dp_prometheus_dir(i, base_prom_dir)
                     self.dp_processed[-1].start()
 
+                setup_dp_prometheus_dir(0, base_prom_dir)
+
                 for i in range(
                     1,
                     self.cfg.parallel_config.data_parallel_size // self.cfg.nnode,

fastdeploy/entrypoints/openai/multi_api_server.py

@@ -20,6 +20,7 @@
 import sys
 import time
 
+from fastdeploy.metrics.prometheus_multiprocess_setup import setup_dp_prometheus_dir
 from fastdeploy.platforms import current_platform
 from fastdeploy.utils import find_free_ports, get_logger, is_port_available
 
@@ -108,13 +109,7 @@ def start_servers(
         env["FD_ENABLE_MULTI_API_SERVER"] = "1"
         env["FD_LOG_DIR"] = env.get("FD_LOG_DIR", "log") + f"/log_{i}"
         if "PROMETHEUS_MULTIPROC_DIR" in env:
-            prom_dir = env.get("PROMETHEUS_MULTIPROC_DIR")
-            prom_dir_i = os.path.join(os.path.dirname(prom_dir), os.path.basename(prom_dir) + f"_dp{i}")
-            # Create the directory if it doesn't exist
-            if not os.path.exists(prom_dir_i):
-                os.makedirs(prom_dir_i, exist_ok=True)
-            env["PROMETHEUS_MULTIPROC_DIR"] = prom_dir_i
-            logger.info(f"Set PROMETHEUS_MULTIPROC_DIR for DP {i}: {prom_dir_i}")
+            setup_dp_prometheus_dir(i, env["PROMETHEUS_MULTIPROC_DIR"], env)
 
         cmd = [
             sys.executable,

fastdeploy/metrics/prometheus_multiprocess_setup.py

@@ -44,3 +44,18 @@ def setup_multiprocess_prometheus():
             "will properly handle cleanup."
         )
         return os.environ["PROMETHEUS_MULTIPROC_DIR"]
+
+
+def setup_dp_prometheus_dir(dp_id, base_dir, env_dict=None):
+    """Set up an isolated PROMETHEUS_MULTIPROC_DIR for a given data parallel rank.
+
+    Args:
+        dp_id: The data parallel rank id.
+        base_dir: The base PROMETHEUS_MULTIPROC_DIR to derive from.
+        env_dict: If provided, write to this dict instead of os.environ.
+    """
+    prom_dir_dp = os.path.join(base_dir, f"dp{dp_id}")
+    os.makedirs(prom_dir_dp, exist_ok=True)
+    target = env_dict if env_dict is not None else os.environ
+    target["PROMETHEUS_MULTIPROC_DIR"] = prom_dir_dp
+    llm_logger.info(f"Set PROMETHEUS_MULTIPROC_DIR for DP {dp_id}: {prom_dir_dp}")