120 lines
4.3 KiB
Python
120 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from app.core.config import settings
|
|
from app.core.files import read_runtime_json, write_runtime_json
|
|
from app.core.redis_client import get_redis
|
|
|
|
|
|
DEFAULT_RUNTIME_SETTINGS = {
|
|
"worker_mode": settings.worker_mode,
|
|
"worker_service_name": settings.worker_service_name,
|
|
"api_service_name": settings.api_service_name,
|
|
"sync_agent_service_name": settings.sync_agent_service_name,
|
|
"worker_log_sync_enabled": False,
|
|
"worker_log_sync_mode": "key",
|
|
"control_node_autoresume_enabled": False,
|
|
"claim_recent_jobs_first": False,
|
|
"claim_recent_jobs_limit": 0,
|
|
"claim_recent_jobs_window_hours": 0,
|
|
"claim_batch_floor": 0,
|
|
"claim_batch_ceil": 0,
|
|
"submit_backlog_floor": 0,
|
|
"submit_backlog_ceil": 0,
|
|
"dispatch_cap_multiplier": 1,
|
|
"pending_buffer_cap_multiplier": 1,
|
|
}
|
|
|
|
RUNTIME_SETTINGS_REDIS_KEY = "domain_tool:runtime_settings"
|
|
CONFIG_UPDATE_CHANNEL = "domain_tool:config_update"
|
|
|
|
|
|
def _normalize_worker_log_sync_mode(value: object) -> str:
|
|
return "full" if str(value or "").strip().lower() == "full" else "key"
|
|
|
|
|
|
def _normalize_bool(value: object, default: bool = False) -> bool:
|
|
if value is None:
|
|
return bool(default)
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value or "").strip().lower() not in {"", "0", "false", "no", "off"}
|
|
|
|
|
|
def _normalize_non_negative_int(value: object, default: int = 0) -> int:
|
|
try:
|
|
normalized = int(value)
|
|
except (TypeError, ValueError):
|
|
normalized = int(default)
|
|
return max(0, normalized)
|
|
|
|
|
|
def _normalize_positive_int(value: object, default: int = 1) -> int:
|
|
try:
|
|
normalized = int(value)
|
|
except (TypeError, ValueError):
|
|
normalized = int(default)
|
|
return max(1, normalized)
|
|
|
|
|
|
def normalize_runtime_settings(payload: dict | None) -> dict:
|
|
merged = dict(DEFAULT_RUNTIME_SETTINGS)
|
|
if isinstance(payload, dict):
|
|
for key in DEFAULT_RUNTIME_SETTINGS:
|
|
if key in payload and payload[key] is not None:
|
|
merged[key] = payload[key]
|
|
|
|
worker_mode = str(merged.get("worker_mode") or "").strip()
|
|
if worker_mode not in {"windows-local", "linux-systemd"}:
|
|
merged["worker_mode"] = DEFAULT_RUNTIME_SETTINGS["worker_mode"]
|
|
|
|
for key in ("worker_service_name", "api_service_name", "sync_agent_service_name"):
|
|
value = str(merged.get(key) or "").strip()
|
|
merged[key] = value or DEFAULT_RUNTIME_SETTINGS[key]
|
|
|
|
merged["worker_log_sync_enabled"] = _normalize_bool(merged.get("worker_log_sync_enabled", False), default=False)
|
|
merged["worker_log_sync_mode"] = _normalize_worker_log_sync_mode(merged.get("worker_log_sync_mode"))
|
|
merged["control_node_autoresume_enabled"] = _normalize_bool(
|
|
merged.get("control_node_autoresume_enabled", False),
|
|
default=False,
|
|
)
|
|
merged["claim_recent_jobs_first"] = _normalize_bool(
|
|
merged.get("claim_recent_jobs_first", False),
|
|
default=False,
|
|
)
|
|
for key in ("claim_batch_floor", "claim_batch_ceil", "submit_backlog_floor", "submit_backlog_ceil"):
|
|
merged[key] = _normalize_non_negative_int(merged.get(key), DEFAULT_RUNTIME_SETTINGS[key])
|
|
for key in (
|
|
"claim_recent_jobs_limit",
|
|
"claim_recent_jobs_window_hours",
|
|
"dispatch_cap_multiplier",
|
|
"pending_buffer_cap_multiplier",
|
|
):
|
|
if key in {"claim_recent_jobs_limit", "claim_recent_jobs_window_hours"}:
|
|
merged[key] = _normalize_non_negative_int(merged.get(key), DEFAULT_RUNTIME_SETTINGS[key])
|
|
continue
|
|
merged[key] = _normalize_positive_int(merged.get(key), DEFAULT_RUNTIME_SETTINGS[key])
|
|
return merged
|
|
|
|
|
|
def get_runtime_settings() -> dict:
|
|
stored = read_runtime_json("runtime_settings.json", default={})
|
|
return normalize_runtime_settings(stored)
|
|
|
|
|
|
def _sync_runtime_settings_update(runtime_settings: dict) -> None:
|
|
try:
|
|
redis_client = get_redis()
|
|
redis_client.set(RUNTIME_SETTINGS_REDIS_KEY, json.dumps(runtime_settings, ensure_ascii=False))
|
|
redis_client.publish(CONFIG_UPDATE_CHANNEL, "runtime_settings")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def update_runtime_settings(payload: dict) -> dict:
|
|
merged = normalize_runtime_settings({**get_runtime_settings(), **(payload or {})})
|
|
write_runtime_json("runtime_settings.json", merged)
|
|
_sync_runtime_settings_update(merged)
|
|
return merged
|