Add RequestMetricsExporter utility to export request-level metrics (#10973)

python/sglang/srt/managers/request_metrics_exporter.py

@@ -0,0 +1,209 @@
+import asyncio
+import dataclasses
+import json
+import logging
+import os
+from abc import ABC, abstractmethod
+from datetime import datetime
+from typing import List, Optional, Union
+
+from sglang.srt.managers.io_struct import EmbeddingReqInput, GenerateReqInput
+from sglang.srt.server_args import ServerArgs
+
+logger = logging.getLogger(__name__)
+
+
+class RequestMetricsExporter(ABC):
+    """Abstract base class for exporting request-level performance metrics to a data destination."""
+
+    def __init__(
+        self,
+        server_args: ServerArgs,
+        obj_skip_names: Optional[set[str]],
+        out_skip_names: Optional[set[str]],
+    ):
+        self.server_args = server_args
+        self.obj_skip_names = obj_skip_names or set()
+        self.out_skip_names = out_skip_names or set()
+
+    def _format_output_data(
+        self, obj: Union[GenerateReqInput, EmbeddingReqInput], out_dict: dict
+    ) -> dict:
+        """Format request-level output data containing performance metrics. This method
+        should be called prior to writing the data record with `self.write_record()`."""
+
+        request_params = {}
+        for field in dataclasses.fields(obj):
+            field_name = field.name
+            if field_name not in self.obj_skip_names:
+                value = getattr(obj, field_name)
+                # Convert to serializable format
+                if value is not None:
+                    request_params[field_name] = value
+
+        meta_info = out_dict.get("meta_info", {})
+        filtered_out_meta_info = {
+            k: v for k, v in meta_info.items() if k not in self.out_skip_names
+        }
+
+        request_output_data = {
+            "request_parameters": json.dumps(request_params),
+            **filtered_out_meta_info,
+        }
+        return request_output_data
+
+    @abstractmethod
+    async def write_record(
+        self, obj: Union[GenerateReqInput, EmbeddingReqInput], out_dict: dict
+    ):
+        """Write a data record corresponding to a single request, containing performance metric data."""
+        pass
+
+
+class FileRequestMetricsExporter(RequestMetricsExporter):
+    """Lightweight `RequestMetricsExporter` implementation that writes records to files on disk.
+
+    Records are written to files in the directory specified by `--export-metrics-to-file-dir`
+    server launch flag. File names are of the form `"sglang-request-metrics-{hour_suffix}.log"`.
+    """
+
+    def __init__(
+        self,
+        server_args: ServerArgs,
+        obj_skip_names: Optional[set[str]],
+        out_skip_names: Optional[set[str]],
+    ):
+        super().__init__(server_args, obj_skip_names, out_skip_names)
+        self.export_dir = getattr(server_args, "export_metrics_to_file_dir")
+        os.makedirs(self.export_dir, exist_ok=True)
+
+        # File handler state management
+        self._current_file_handler = None
+        self._current_hour_suffix = None
+
+    def _ensure_file_handler(self, hour_suffix: str):
+        """Ensure the file handler is open for the current hour suffix."""
+        if self._current_hour_suffix != hour_suffix:
+            # Close previous file handler if it exists
+            if self._current_file_handler is not None:
+                try:
+                    self._current_file_handler.close()
+                except Exception as e:
+                    logger.warning(f"Failed to close previous file handler: {e}")
+
+            # Open new file handler
+            log_filename = f"sglang-request-metrics-{hour_suffix}.log"
+            log_filepath = os.path.join(self.export_dir, log_filename)
+
+            try:
+                self._current_file_handler = open(log_filepath, "a", encoding="utf-8")
+                self._current_hour_suffix = hour_suffix
+            except Exception as e:
+                logger.error(f"Failed to open log file {log_filepath}: {e}")
+                self._current_file_handler = None
+                self._current_hour_suffix = None
+                raise
+
+    def close(self):
+        """Close the current file handler."""
+        if self._current_file_handler is not None:
+            try:
+                self._current_file_handler.close()
+            except Exception as e:
+                logger.warning(f"Failed to close file handler: {e}")
+            finally:
+                self._current_file_handler = None
+                self._current_hour_suffix = None
+
+    async def write_record(
+        self, obj: Union[GenerateReqInput, EmbeddingReqInput], out_dict: dict
+    ):
+        # Do not log health check requests, since they don't represent real user requests.
+        if isinstance(obj.rid, str) and "HEALTH_CHECK" in obj.rid:
+            return
+
+        try:
+            # Get the log file path for the current time.
+            current_time = datetime.now()
+            hour_suffix = current_time.strftime("%Y%m%d_%H")
+
+            # Ensure correct file handler is open for current hour
+            self._ensure_file_handler(hour_suffix)
+
+            if self._current_file_handler is None:
+                return
+
+            metrics_data = self._format_output_data(obj, out_dict)
+
+            def write_file():
+                json.dump(metrics_data, self._current_file_handler)
+                self._current_file_handler.write("\n")
+                self._current_file_handler.flush()
+
+            await asyncio.to_thread(write_file)
+        except Exception as e:
+            logger.exception(f"Failed to write perf metrics to file: {e}")
+
+
+class RequestMetricsExporterManager:
+    """Manager class for creating and managing RequestMetricsExporter instances."""
+
+    def __init__(
+        self,
+        server_args: ServerArgs,
+        obj_skip_names: Optional[set[str]] = None,
+        out_skip_names: Optional[set[str]] = None,
+    ):
+        self.server_args = server_args
+        self.obj_skip_names = obj_skip_names or set()
+        self.out_skip_names = out_skip_names or set()
+        self._exporters: List[RequestMetricsExporter] = []
+        self._create_exporters()
+
+    def _create_exporters(self) -> None:
+        """Create and configure RequestMetricsExporter instances based on server args."""
+        # Create standard exporters
+        self._exporters.extend(
+            create_request_metrics_exporters(
+                self.server_args, self.obj_skip_names, self.out_skip_names
+            )
+        )
+
+        # Import additional RequestMetricsExporter from private fork if available; skip otherwise.
+        try:
+            from sglang.private.managers.request_metrics_exporter_factory import (
+                create_private_request_metrics_exporters,
+            )
+
+            self._exporters.extend(
+                create_private_request_metrics_exporters(
+                    self.server_args, self.obj_skip_names, self.out_skip_names
+                )
+            )
+        except ImportError:
+            pass
+
+    def exporter_enabled(self) -> bool:
+        """Return true if at least one RequestMetricsExporter is enabled."""
+        return len(self._exporters) > 0
+
+    async def write_record(self, obj, out_dict: dict) -> None:
+        """Write a record using all configured exporters."""
+        for exporter in self._exporters:
+            await exporter.write_record(obj, out_dict)
+
+
+def create_request_metrics_exporters(
+    server_args: ServerArgs,
+    obj_skip_names: Optional[set[str]] = None,
+    out_skip_names: Optional[set[str]] = None,
+) -> List[RequestMetricsExporter]:
+    """Create and configure `RequestMetricsExporter`s based on server args."""
+    metrics_exporters = []
+
+    if server_args.export_metrics_to_file:
+        metrics_exporters.append(
+            FileRequestMetricsExporter(server_args, obj_skip_names, out_skip_names)
+        )
+
+    return metrics_exporters

python/sglang/srt/managers/tokenizer_manager.py

@@ -69,6 +69,7 @@ from sglang.srt.managers.io_struct import (
 )
 from sglang.srt.managers.mm_utils import TensorTransportMode
 from sglang.srt.managers.multimodal_processor import get_mm_processor, import_processors
+from sglang.srt.managers.request_metrics_exporter import RequestMetricsExporterManager
 from sglang.srt.managers.schedule_batch import RequestStage
 from sglang.srt.managers.scheduler import is_health_check_generate_req
 from sglang.srt.managers.scheduler_input_blocker import input_blocker_guard_region
@@ -323,6 +324,12 @@ class TokenizerManager(TokenizerCommunicatorMixin):
         self.crash_dump_request_list: deque[Tuple] = deque()
         self.crash_dump_performed = False  # Flag to ensure dump is only called once
 
+        # Initialize performance metrics loggers with proper skip names
+        _, obj_skip_names, out_skip_names = self.log_request_metadata
+        self.request_metrics_exporter_manager = RequestMetricsExporterManager(
+            self.server_args, obj_skip_names, out_skip_names
+        )
+
         # Session
         self.session_futures = {}  # session_id -> asyncio event
 
@@ -1000,6 +1007,12 @@ class TokenizerManager(TokenizerCommunicatorMixin):
                         msg = f"Finish: obj={dataclass_to_string_truncated(obj, max_length, skip_names=skip_names)}, out={dataclass_to_string_truncated(out, max_length, skip_names=out_skip_names)}"
                     logger.info(msg)
 
+                if self.request_metrics_exporter_manager.exporter_enabled():
+                    # Asynchronously write metrics for this request using the exporter manager.
+                    asyncio.create_task(
+                        self.request_metrics_exporter_manager.write_record(obj, out)
+                    )
+
                 # Check if this was an abort/error created by scheduler
                 if isinstance(out["meta_info"].get("finish_reason"), dict):
                     finish_reason = out["meta_info"]["finish_reason"]

python/sglang/srt/server_args.py

@@ -321,6 +321,10 @@ class ServerArgs:
     enable_trace: bool = False
     otlp_traces_endpoint: str = "localhost:4317"
 
+    # RequestMetricsExporter configuration
+    export_metrics_to_file: bool = False
+    export_metrics_to_file_dir: Optional[str] = None
+
     # API related
     api_key: Optional[str] = None
     served_model_name: Optional[str] = None
@@ -640,6 +644,9 @@ class ServerArgs:
         # Handle deterministic inference.
         self._handle_deterministic_inference()
 
+        # Handle exporting request-level metrics.
+        self._handle_request_metrics_exporters()
+
         # Handle any other necessary validations.
         self._handle_other_validations()
 
@@ -1849,6 +1856,13 @@ class ServerArgs:
                     "NCCL_ALGO is set to 'allreduce:tree' and custom all reduce is disabled for deterministic inference when TP size > 1."
                 )
 
+    def _handle_request_metrics_exporters(self):
+        """Handle arguments for configuring `RequestMetricsExporter` usage."""
+        if self.export_metrics_to_file and self.export_metrics_to_file_dir is None:
+            raise ValueError(
+                "--export-metrics-to-file-dir is required when --export-metrics-to-file is enabled"
+            )
+
     def _handle_other_validations(self):
         # Handle model inference tensor dump.
         if self.debug_tensor_dump_output_folder is not None:
@@ -2443,6 +2457,19 @@ class ServerArgs:
             help="Config opentelemetry collector endpoint if --enable-trace is set. format: <ip>:<port>",
         )
 
+        # RequestMetricsExporter configuration
+        parser.add_argument(
+            "--export-metrics-to-file",
+            action="store_true",
+            help="Export performance metrics for each request to local file (e.g. for forwarding to external systems).",
+        )
+        parser.add_argument(
+            "--export-metrics-to-file-dir",
+            type=str,
+            default=ServerArgs.export_metrics_to_file_dir,
+            help="Directory path for writing performance metrics files (required when --export-metrics-to-file is enabled).",
+        )
+
         # API related
         parser.add_argument(
             "--api-key",