diff --git a/python/sglang/srt/managers/request_metrics_exporter.py b/python/sglang/srt/managers/request_metrics_exporter.py new file mode 100644 index 000000000..95f1fda8f --- /dev/null +++ b/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 diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 9c5c6fa12..4dbf078c8 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/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"] diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index 2bca47fc7..036748f7c 100644 --- a/python/sglang/srt/server_args.py +++ b/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: :", ) + # 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",