import os
import json
from typing import Any
from enum import Enum
from dataclasses import dataclass, field, asdict
from pathlib import Path
from motor.common.resources.instance import ParallelConfig, PDRole
from motor.common.resources.dispatch import (
DISPATCH_PROFILE_KEY,
DispatchPlan,
DispatchProfile,
classify_vllm_dispatch_profile,
dispatch_capabilities_for_profile,
)
from motor.config.resolver import ConfigResolver
from motor.config.tls_config import TLSConfig
from motor.common.utils.env import Env
from motor.common.utils.patch_check import safe_open
from motor.common.logger import get_logger
from motor.config.config_utils import (
ConfigKey,
format_config_summary_header,
raise_if_config_errors,
reload_dataclass_config_from_json,
save_instance_config_to_json,
_update_tls_config,
MGMT_TLS_CONFIG,
)
from motor.config.log_config import LoggingConfig
from motor.config.port_allocator_config import PortAllocatorConfig
FILE_ENCODING = "utf-8"
PP = "pp_size"
TP = "tp_size"
DP = "dp_size"
BASIC_CONFIG_KEY = "basic_config"
SNAPSHOT_CONFIG_KEY = "snapshot_config"
PARALLEL_CONFIG_KEY = "parallel_config"
MOTOR_NODE_MANAGER_CONFIG_KEY = "motor_nodemanger_config"
MOTOR_ENGINE_ENCODE_CONFIG_KEY = "motor_engine_encode_config"
MOTOR_ENGINE_PREFILL_CONFIG_KEY = "motor_engine_prefill_config"
MOTOR_ENGINE_DECODE_CONFIG_KEY = "motor_engine_decode_config"
MOTOR_ENGINE_UNION_CONFIG_KEY = "motor_engine_union_config"
MOTOR_CONTAINER_SNAPSHOT_CONFIG_KEY = "motor_container_snapshot_config"
ENGINE_CONFIG_KEY = "engine_config"
KV_TRANSFER_CONFIG_KEY = "kv_transfer_config"
KV_CONNECTOR_KEY = "kv_connector"
MULTICONNECTOR = "MultiConnector"
KV_CONNECTOR_EXTRA_CONFIG_KEY = "kv_connector_extra_config"
CONNECTORS_KEY = "connectors"
KV_PORT_KEY = "kv_port"
LOOPUP_RPC_PORT_KEY = "lookup_rpc_port"
SERVER_LIST = "server_list"
DEVICE = "device"
HARDWARE_TYPE_KEY = "hardware_type"
MODEL_NAME_KEY = "model_name"
ENGINE_TYPE_KEY = "engine_type"
DISPATCH_CAPABILITIES_KEY = "dispatch_capabilities"
ENABLE_MULTI_ENDPOINTS_KEY = "enable_multi_endpoints"
ENGINE_TYPE_VLLM = "vllm"
ENGINE_TYPE_SGLANG = "sglang"
ENABLE_SNAPSHOT_KEY = "enable_snapshot"
SNAPSHOT_METADATA_PATH_KEY = "snapshot_metadata_path"
logger = get_logger(__name__)
class HardwareType(str, Enum):
TYPE_800I_A2 = "800I-A2"
TYPE_800I_A3 = "800I-A3"
TYPE_350_ATLAS_8 = "350-Atlas-8"
TYPE_350_ATLAS_16 = "350-Atlas-16"
TYPE_350_ATLAS_4P_8 = "350-Atlas-4p-8"
TYPE_350_ATLAS_4P_16 = "350-Atlas-4p-16"
TYPE_850_ATLAS_8P_8 = "850-Atlas-8p-8"
TYPE_850_SUPERPOD_ATLAS_8 = "850-SuperPod-Atlas-8"
TYPE_950_SUPERPOD_ATLAS_8 = "950-SuperPod-Atlas-8"
def __repr__(self) -> str:
return str.__repr__(self.value)
@classmethod
def is_a5(cls, hardware_type: str) -> bool:
return hardware_type in {member.value for member in cls if member not in (cls.TYPE_800I_A2, cls.TYPE_800I_A3)}
@dataclass
class BasicConfig:
"""Basic configuration class"""
job_name: str = Env.job_name
role: PDRole = PDRole.ROLE_U
model_name: str = ""
engine_type: str | None = None
dispatch_capabilities: list[str] = field(default_factory=list)
hardware_type: HardwareType = HardwareType.TYPE_800I_A3
heartbeat_interval_seconds: int = 3
device_num: int = 0
parallel_config: ParallelConfig = field(default_factory=ParallelConfig)
enable_multi_endpoints: bool = True
nnodes: int = 1
@dataclass
class SnapshotConfig:
"""Snapshot configuration class"""
enable_snapshot: bool = False
snapshot_metadata_path: str = ""
@dataclass
class APIConfig:
"""API configuration class"""
pod_ip: str | None = field(default_factory=lambda: Env.pod_ip or "127.0.0.1")
node_manager_port: int = 1026
@dataclass
class EndpointConfig:
endpoint_num: int = 0
base_port: int = 10000
mgmt_ports: list[str] = field(default_factory=list)
service_ports: list[str] = field(default_factory=list)
@dataclass
class SingleContainerNodemanagerConfig:
single_container_flag: bool = False
node_manager_port_offset: int | None = None
base_port_offset: int | None = None
device_offset: int | None = None
device_num: int | None = None
kv_port: int | None = None
lookup_rpc_port: int | None = None
dp_rpc_port: int | None = None
@classmethod
def from_json(cls, user_config_data: dict[str, Any]) -> "SingleContainerNodemanagerConfig":
config = cls()
deploy_mode = user_config_data.get("motor_deploy_config", {}).get("deploy_mode", "")
if deploy_mode != "single_container" or not Env.role or Env.index is None:
return config
config.single_container_flag = True
p_instances_num = user_config_data['motor_deploy_config']['p_instances_num']
d_instances_num = user_config_data['motor_deploy_config']['d_instances_num']
encode_section = user_config_data.get(MOTOR_ENGINE_ENCODE_CONFIG_KEY, {})
prefill_section = user_config_data[MOTOR_ENGINE_PREFILL_CONFIG_KEY]
decode_section = user_config_data[MOTOR_ENGINE_DECODE_CONFIG_KEY]
encode_resolver = ConfigResolver(encode_section) if encode_section else None
prefill_resolver = ConfigResolver(prefill_section)
decode_resolver = ConfigResolver(decode_section)
encode_parallel_config = encode_resolver.get_parallel_config() if encode_resolver else {}
prefill_parallel_config = prefill_resolver.get_parallel_config()
decode_parallel_config = decode_resolver.get_parallel_config()
e_dp_size = encode_parallel_config.get(DP, 1)
p_dp_size = prefill_parallel_config.get(DP, 1)
d_dp_size = decode_parallel_config.get(DP, 1)
e_world_size = encode_parallel_config.get("world_size", 0)
p_world_size = prefill_parallel_config["world_size"]
d_world_size = decode_parallel_config["world_size"]
index = int(Env.index)
d_node_manager_port_offset = p_instances_num * p_dp_size + index
d_base_port_offset = (p_instances_num * p_dp_size + index * d_dp_size) * 2
d_device_offset = p_instances_num * p_world_size + index * d_world_size
e_node_manager_port_offset = d_instances_num * d_dp_size + d_node_manager_port_offset
e_base_port_offset = d_base_port_offset + e_dp_size
e_device_offset = d_device_offset + e_world_size
kv_port_offset = 0
lookup_rpc_port_offset = 0
dp_rpc_port_offset = 0
if Env.role == 'prefill':
config.node_manager_port_offset = index
config.base_port_offset = index * d_dp_size * 2
config.device_offset = index * p_world_size
config.device_num = p_world_size
kv_port_offset = config.device_offset
lookup_rpc_port_offset = index
dp_rpc_port_offset = index
elif Env.role == 'decode':
config.node_manager_port_offset = d_node_manager_port_offset
config.base_port_offset = d_base_port_offset
config.device_offset = d_device_offset
config.device_num = d_world_size
kv_port_offset = config.device_offset
lookup_rpc_port_offset = p_instances_num + index
dp_rpc_port_offset = p_instances_num + index
elif Env.role == 'encode':
config.node_manager_port_offset = e_node_manager_port_offset
config.base_port_offset = e_base_port_offset
config.device_offset = e_device_offset
config.device_num = e_world_size
kv_port_offset = config.device_offset
lookup_rpc_port_offset = index
dp_rpc_port_offset = index
kv_config = user_config_data[MOTOR_ENGINE_PREFILL_CONFIG_KEY][ENGINE_CONFIG_KEY].get(KV_TRANSFER_CONFIG_KEY, {})
if kv_config:
if kv_config[KV_CONNECTOR_KEY] == MULTICONNECTOR:
connectors = kv_config[KV_CONNECTOR_EXTRA_CONFIG_KEY][CONNECTORS_KEY]
config.kv_port = int(connectors[0][KV_PORT_KEY]) + kv_port_offset
config.lookup_rpc_port = int(connectors[1][LOOPUP_RPC_PORT_KEY]) + lookup_rpc_port_offset
else:
config.kv_port = int(kv_config[KV_PORT_KEY]) + kv_port_offset
config.dp_rpc_port = int(prefill_parallel_config["dp_rpc_port"]) + dp_rpc_port_offset
return config
@dataclass
class NodeManagerFaultToleranceConfig:
"""Fault tolerance configuration for NodeManager"""
enable_fault_tolerance: bool = False
zmq_pub_port: int = 0
@dataclass
class NodeManagerConfig:
"""Global configuration singleton for node manager"""
api_config: APIConfig = field(default_factory=APIConfig)
mgmt_tls_config: TLSConfig = field(default_factory=TLSConfig)
endpoint_config: EndpointConfig = field(default_factory=EndpointConfig)
basic_config: BasicConfig = field(default_factory=BasicConfig)
snapshot_config: SnapshotConfig = field(default_factory=SnapshotConfig)
logging_config: LoggingConfig = field(default_factory=LoggingConfig)
single_container_config: SingleContainerNodemanagerConfig = field(default_factory=SingleContainerNodemanagerConfig)
fault_tolerance_config: NodeManagerFaultToleranceConfig = field(default_factory=NodeManagerFaultToleranceConfig)
port_allocator_config: PortAllocatorConfig = field(default_factory=PortAllocatorConfig)
config_path: str | None = field(default=None, init=False)
last_modified: float | None = field(default=None, init=False)
def __post_init__(self):
"""Validate configuration after initialization"""
if not hasattr(self, "config_path") or self.config_path is None:
self.config_path = Env.user_config_path or Env.config_path
try:
if self.config_path and os.path.exists(self.config_path):
self.last_modified = os.path.getmtime(self.config_path)
except (OSError, IOError):
pass
self.validate_config()
@classmethod
def from_json(cls, config_path: str | None = None) -> "NodeManagerConfig":
"""Load configuration from user_config.json"""
if config_path is None:
config_path = Env.user_config_path or Env.config_path
config_path_obj = Path(config_path) if config_path else None
logger.info("Loading configuration files: config=%s", config_path_obj)
config = cls()
config.config_path = config_path
config_data = {}
raw = None
if config_path_obj is not None and os.path.exists(str(config_path_obj)):
with safe_open(str(config_path_obj), "r") as f:
raw = json.load(f)
logger.info("Successfully loaded config file: %s", config_path_obj)
config.single_container_config = SingleContainerNodemanagerConfig.from_json(raw)
if isinstance(raw, dict):
cls._discard_user_dispatch_capabilities(raw)
config_data = cls._load_node_manager_config_data(raw)
else:
config_data = raw
cls._update_from_config_data(config, config_data)
else:
logger.warning("Config file does not exist, using default configuration: %s", config_path_obj)
cls._set_device_count_from_config(config, raw)
config.validate_config()
if config_path_obj is not None and config_path_obj.exists():
config.last_modified = config_path_obj.stat().st_mtime
logger.info("Configuration loading completed")
return config
@classmethod
def _discard_user_dispatch_capabilities(cls, user_cfg: dict[str, Any]) -> None:
"""Remove user overrides; capabilities are derived from engine semantics."""
removed = False
root_basic = user_cfg.get(BASIC_CONFIG_KEY)
if isinstance(root_basic, dict):
removed = root_basic.pop(DISPATCH_CAPABILITIES_KEY, None) is not None or removed
engine_keys = (
MOTOR_ENGINE_ENCODE_CONFIG_KEY,
MOTOR_ENGINE_PREFILL_CONFIG_KEY,
MOTOR_ENGINE_DECODE_CONFIG_KEY,
MOTOR_ENGINE_UNION_CONFIG_KEY,
)
for engine_key in engine_keys:
engine_config = user_cfg.get(engine_key)
if not isinstance(engine_config, dict):
continue
removed = engine_config.pop(DISPATCH_CAPABILITIES_KEY, None) is not None or removed
node_manager_config = engine_config.get(MOTOR_NODE_MANAGER_CONFIG_KEY)
if not isinstance(node_manager_config, dict):
continue
basic_config = node_manager_config.get(BASIC_CONFIG_KEY)
if isinstance(basic_config, dict):
removed = basic_config.pop(DISPATCH_CAPABILITIES_KEY, None) is not None or removed
if removed:
logger.warning(
"User-configured dispatch_capabilities is no longer supported and was ignored. "
"Configure kv_transfer_config or dispatch_profile instead."
)
@classmethod
def _load_node_manager_config_data(cls, user_cfg: dict[str, Any]) -> dict[str, Any]:
"""Load node_manager_config from engine config based on role"""
engine_config_key = None
if Env.role == "encode":
engine_config_key = MOTOR_ENGINE_ENCODE_CONFIG_KEY
elif Env.role == "prefill":
engine_config_key = MOTOR_ENGINE_PREFILL_CONFIG_KEY
elif Env.role == "decode":
engine_config_key = MOTOR_ENGINE_DECODE_CONFIG_KEY
elif Env.role in ("union", "both"):
engine_config_key = MOTOR_ENGINE_UNION_CONFIG_KEY
if not engine_config_key or engine_config_key not in user_cfg:
return user_cfg
engine_config = user_cfg[engine_config_key]
if "motor_nodemanger_config" in engine_config:
config_data = engine_config.get("motor_nodemanger_config", {})
else:
config_data = {}
if BASIC_CONFIG_KEY not in config_data:
config_data[BASIC_CONFIG_KEY] = {}
resolver = ConfigResolver(engine_config)
config_data[BASIC_CONFIG_KEY][MODEL_NAME_KEY] = resolver.get_model_name("")
config_data[BASIC_CONFIG_KEY][ENGINE_TYPE_KEY] = engine_config.get(ENGINE_TYPE_KEY)
config_data[BASIC_CONFIG_KEY][DISPATCH_CAPABILITIES_KEY] = cls._infer_dispatch_capabilities(engine_config)
config_data[BASIC_CONFIG_KEY][HARDWARE_TYPE_KEY] = user_cfg["motor_deploy_config"][HARDWARE_TYPE_KEY]
engine_cfg = engine_config.get(ENGINE_CONFIG_KEY, {})
try:
config_data[BASIC_CONFIG_KEY]["nnodes"] = int(engine_cfg.get("nnodes", 1))
except (TypeError, ValueError):
config_data[BASIC_CONFIG_KEY]["nnodes"] = 1
if Env.role in ("encode", "prefill", "decode", "union", "both"):
config_data[BASIC_CONFIG_KEY]["parallel_config"] = resolver.get_parallel_config()
config_data[BASIC_CONFIG_KEY][ENABLE_MULTI_ENDPOINTS_KEY] = resolver.get_enable_multi_endpoints()
nnodes = config_data[BASIC_CONFIG_KEY].get("nnodes", 1)
pc = config_data[BASIC_CONFIG_KEY].get("parallel_config", {})
pcp_size = pc.get("pcp_size", 1)
if isinstance(nnodes, int) and nnodes > 1 and pcp_size > 1 and pcp_size % nnodes == 0:
per_node_pcp = pcp_size // nnodes
per_node_tp = pc.get("tp_size", 1)
per_node_pp = pc.get("pp_size", 1)
pc["local_world_size"] = per_node_pcp * per_node_tp * per_node_pp
logger.info(
"Cross-node PCP detected (nnodes=%d, pcp=%d): per-node local_world_size adjusted from %d to %d",
nnodes,
pcp_size,
pcp_size * per_node_tp * per_node_pp,
pc["local_world_size"],
)
_update_tls_config([MGMT_TLS_CONFIG], config_data, user_cfg)
snapshot_cfg = user_cfg.get(MOTOR_CONTAINER_SNAPSHOT_CONFIG_KEY, {})
config_data[SNAPSHOT_CONFIG_KEY] = {
ENABLE_SNAPSHOT_KEY: snapshot_cfg.get(ENABLE_SNAPSHOT_KEY, False),
SNAPSHOT_METADATA_PATH_KEY: snapshot_cfg.get(SNAPSHOT_METADATA_PATH_KEY, ""),
}
return config_data
@classmethod
def _infer_dispatch_capabilities(cls, engine_config: dict[str, Any]) -> list[str]:
"""Infer Motor dispatch capabilities from engine-native config."""
engine_type = str(engine_config.get(ENGINE_TYPE_KEY, "")).strip().lower()
native_engine_config = engine_config.get(ENGINE_CONFIG_KEY, {})
if not isinstance(native_engine_config, dict):
native_engine_config = {}
if engine_type == ENGINE_TYPE_SGLANG:
return [DispatchPlan.CONCURRENT_ENGINE_SYNC.value]
if engine_type == ENGINE_TYPE_VLLM:
profile = classify_vllm_dispatch_profile(
native_engine_config,
explicit_profile=engine_config.get(DISPATCH_PROFILE_KEY),
)
capabilities = dispatch_capabilities_for_profile(profile)
if not capabilities and profile == DispatchProfile.UNKNOWN:
logger.warning(
"Unable to infer vLLM dispatch capability from kv_transfer_config. "
"Configure a supported connector or set dispatch_profile explicitly."
)
return capabilities
return []
@classmethod
def _update_from_config_data(cls, config: "NodeManagerConfig", cfg: dict[str, Any]):
"""Update configuration from config JSON data"""
def update_config_from_dict(config_obj, config_dict):
"""Update configuration object fields from dictionary, only for existing keys"""
for key, value in config_dict.items():
if hasattr(config_obj, key):
setattr(config_obj, key, value)
if "logging_config" in cfg:
update_config_from_dict(config.logging_config, cfg["logging_config"])
if "api_config" in cfg:
update_config_from_dict(config.api_config, cfg["api_config"])
if "mgmt_tls_config" in cfg:
update_config_from_dict(config.mgmt_tls_config, cfg["mgmt_tls_config"])
if "fault_tolerance_config" in cfg:
update_config_from_dict(config.fault_tolerance_config, cfg["fault_tolerance_config"])
if "port_allocator_config" in cfg:
update_config_from_dict(config.port_allocator_config, cfg["port_allocator_config"])
if "endpoint_config" in cfg:
update_config_from_dict(config.endpoint_config, cfg["endpoint_config"])
if SNAPSHOT_CONFIG_KEY in cfg:
update_config_from_dict(config.snapshot_config, cfg[SNAPSHOT_CONFIG_KEY])
if BASIC_CONFIG_KEY in cfg:
basic_cfg = cfg[BASIC_CONFIG_KEY]
update_config_from_dict(config.basic_config, basic_cfg)
if "parallel_config" in basic_cfg:
pc = basic_cfg["parallel_config"]
if isinstance(pc, dict):
config.basic_config.parallel_config = ParallelConfig(**pc)
if config.single_container_config.single_container_flag:
config.api_config.node_manager_port += config.single_container_config.node_manager_port_offset
config.endpoint_config.base_port += config.single_container_config.base_port_offset
try:
role = Env.role
config.basic_config.role = PDRole(role)
logger.info("Role from environment: %s", role)
logger.info("Role from config: %s", config.basic_config.role)
except ValueError as e:
raise ValueError("Invalid role value from environment") from e
@classmethod
def _set_device_count_from_config(cls, config: "NodeManagerConfig", raw: dict | None):
"""
Set device count from config based on role.
For prefill role, use p_pod_npu_num.
For decode role, use d_pod_npu_num.
For single_container mode, use parallel_config.world_size instead.
"""
try:
if not isinstance(raw, dict) or "motor_deploy_config" not in raw:
logger.warning("No motor_deploy_config found, using default device configuration")
config.basic_config.device_num = 0
config.endpoint_config.endpoint_num = 0
config.endpoint_config.service_ports = []
config.endpoint_config.mgmt_ports = []
return
if config.single_container_config.single_container_flag:
cls._set_device_count_for_single_container(config)
else:
cls._set_device_count_for_normal_mode(config, raw)
cls._generate_endpoint_ports(config)
except Exception as e:
logger.error("Failed to get device count from config: %s", e)
config.basic_config.device_num = 0
config.endpoint_config.endpoint_num = 0
config.endpoint_config.service_ports = []
config.endpoint_config.mgmt_ports = []
@classmethod
def _set_device_count_for_single_container(cls, config: "NodeManagerConfig"):
"""Set device count for single container mode using parallel_config.world_size"""
device_count = config.basic_config.parallel_config.world_size
if device_count > 0:
logger.info("Single container mode: using world_size %d from parallel_config", device_count)
config.basic_config.device_num = device_count
else:
logger.warning("Single container mode: world_size is 0, falling back to single_container_config.device_num")
config.basic_config.device_num = config.single_container_config.device_num or 0
@classmethod
def _set_device_count_for_normal_mode(cls, config: "NodeManagerConfig", raw: dict):
"""Set device count for normal mode using motor_deploy_config"""
deploy_config = raw["motor_deploy_config"]
if Env.role == "encode":
device_count = deploy_config.get("e_pod_npu_num", 0)
elif Env.role == "prefill":
device_count = deploy_config.get("p_pod_npu_num", 0)
elif Env.role == "decode":
device_count = deploy_config.get("d_pod_npu_num", 0)
elif Env.role in ("union", "both"):
device_count = deploy_config.get("hybrid_pod_npu_num", 0)
else:
device_count = 0
if device_count > 0:
logger.info("Using %d devices from config for role %s", device_count, Env.role)
config.basic_config.device_num = device_count
else:
logger.warning("No device count found in config for role %s", Env.role)
config.basic_config.device_num = 0
@classmethod
def _generate_endpoint_ports(cls, config: "NodeManagerConfig"):
"""
Calculate endpoint number based on tensor parallel & pipeline parallel config.
Example: tp=2, pp=4 => 8 devices per pod
"""
dp = config.basic_config.parallel_config.dp_size
devices_per_dp = config.basic_config.parallel_config.local_world_size
if (config.basic_config.enable_multi_endpoints and config.basic_config.device_num < devices_per_dp) or dp < 1:
raise ValueError(
f"Device count ({config.basic_config.device_num}) must bigger than "
f"or equal to devices per dp ({devices_per_dp}) "
f"and dp must be bigger than 0"
)
if not config.basic_config.enable_multi_endpoints:
config.endpoint_config.endpoint_num = 1
else:
config.endpoint_config.endpoint_num = min(dp, config.basic_config.device_num // devices_per_dp)
config.endpoint_config.service_ports = [
str(config.endpoint_config.base_port + i * 2) for i in range(config.endpoint_config.endpoint_num)
]
config.endpoint_config.mgmt_ports = [
str(config.endpoint_config.base_port + i * 2 + 1) for i in range(config.endpoint_config.endpoint_num)
]
logger.info(
"Generate endpoint ports successfully: endpoint_num: %d, mgmt_ports: %s, service_ports: %s.",
config.endpoint_config.endpoint_num,
config.endpoint_config.mgmt_ports,
config.endpoint_config.service_ports,
)
def validate_config(self) -> None:
"""Validate the validity of configuration values"""
errors = []
if self.api_config.node_manager_port <= 0 or self.api_config.node_manager_port > 65535:
errors.append("node_manager_port must be in range 1-65535")
if self.endpoint_config.base_port < 0 or self.endpoint_config.base_port > 65535:
errors.append("base_port must be in range 0-65535")
if self.endpoint_config.endpoint_num < 0:
errors.append("endpoint_num cannot be negative")
if self.basic_config.heartbeat_interval_seconds <= 0:
errors.append("heartbeat_interval_seconds must be greater than 0")
valid_log_levels = ["DEBUG", "INFO", "WARNING", "ERROR"]
if self.logging_config.log_level.upper() not in valid_log_levels:
errors.append(f"log_level must be one of: {', '.join(valid_log_levels)}")
if self.logging_config.log_max_line_length <= 0:
errors.append("log_max_line_length must be greater than 0")
raise_if_config_errors(errors)
def reload(self) -> bool:
"""Reload configuration from files"""
return reload_dataclass_config_from_json(
self,
NodeManagerConfig.from_json,
skip=frozenset({"config_path", "last_modified"}),
skip_private=False,
success_message="NodeManager configuration reload successful",
error_message="Failed to reload NodeManager configuration: %s",
)
def to_dict(self) -> dict[str, Any]:
"""Convert configuration to dictionary with grouped structure"""
config_dict = asdict(self)
if hasattr(self.basic_config.parallel_config, "model_dump"):
config_dict["basic_config"]["parallel_config"] = self.basic_config.parallel_config.model_dump()
config_dict.pop("config_path", None)
config_dict.pop("last_modified", None)
config_dict["basic_config"].pop(DISPATCH_CAPABILITIES_KEY, None)
return config_dict
def save_to_json(self, config_path: str | None = None) -> bool:
"""Save configuration to JSON files"""
return save_instance_config_to_json(
self,
config_path,
config_key=ConfigKey.MOTOR_NODEMANAGER,
file_encoding=FILE_ENCODING,
component_name="node manager",
missing_path_message="Save paths not specified",
)
def get_config_summary(self) -> str:
"""Get configuration summary information"""
title = " " * 22 + "NodeManager Configuration Summary"
return (
format_config_summary_header(title) + " Logging Configuration:\n"
f" ├─ Log Level: {self.logging_config.log_level}\n"
f" ├─ Log File: {self.logging_config.host_log_dir}\n"
f" └─ Log Max Line Length: {self.logging_config.log_max_line_length}\n"
"\n"
" Network Configuration:\n"
f" ├─ Node Manager Port: {self.api_config.node_manager_port}\n"
f" ├─ Pod IP: {self.api_config.pod_ip}\n"
f" └─ TLS: {'Enabled' if self.mgmt_tls_config.enable_tls else 'Disabled'}\n"
"\n"
" Basic Configuration:\n"
f" ├─ Job Name: {self.basic_config.job_name}\n"
f" ├─ Role: {self.basic_config.role}\n"
f" ├─ Model: {self.basic_config.model_name}\n"
f" ├─ Device Count: {self.basic_config.device_num}\n"
f" ├─ Multi Endpoints: {self.basic_config.enable_multi_endpoints}\n"
f" ├─ Endpoint Count: {self.endpoint_config.endpoint_num}\n"
f" └─ Hardware Type: {self.basic_config.hardware_type}\n"
"\n"
" Snapshot Configuration:\n"
f" ├─ Snapshot Enable: "
f"{'Enabled' if self.snapshot_config.enable_snapshot else 'Disabled'}\n"
f" └─ Metadata Path: {self.snapshot_config.snapshot_metadata_path or '(default)'}\n"
"\n"
" Parallel Configuration:\n"
f" ├─ TP Size: TP={self.basic_config.parallel_config.tp_size}\n"
f" ├─ PP Size: PP={self.basic_config.parallel_config.pp_size}\n"
f" ├─ DP Size: DP={self.basic_config.parallel_config.dp_size}\n"
f" ├─ EP Size: EP={self.basic_config.parallel_config.ep_size}\n"
f" ├─ PCP Size: PCP={self.basic_config.parallel_config.pcp_size}\n"
f" └─ World Size: World Size={self.basic_config.parallel_config.world_size}\n"
f"{'=' * 80}"
)