GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-04 23:17:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: propagate request correlation context

Browse Source
This commit is contained in:
jxxghp
2026-08-21 22:08:33 +08:00
parent c755c074b9
commit 47f1ff9cb4
13 changed files with 408 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/adapters/network/http.py
+11 -3
View File
@@ -17,6 +17,8 @@ from requests import Response, Session
from urllib3.exceptions import InsecureRequestWarning
from urllib.parse import unquote, quote
from app.runtime.correlation import with_correlation_header
urllib3.disable_warnings(InsecureRequestWarning)
@@ -418,7 +420,9 @@ class RequestUtils:
req_method = requests.request
else:
req_method = self._session.request
kwargs.setdefault("headers", self._headers)
kwargs["headers"] = with_correlation_header(
kwargs.get("headers", self._headers)
)
kwargs.setdefault("cookies", self._cookies)
kwargs.setdefault("proxies", self._proxies)
kwargs.setdefault("timeout", self._timeout)
@@ -1195,7 +1199,9 @@ class AsyncRequestUtils:
"""
执行实际的异步请求
"""
kwargs.setdefault("headers", self._headers)
kwargs["headers"] = with_correlation_header(
kwargs.get("headers", self._headers)
)
# 共享池下 client 自带默认 timeout，这里用每请求 timeout 覆盖以尊重实例配置
kwargs.setdefault("timeout", self._timeout)
# Cookie 在 request() 入口已按 path 处理：
@@ -1329,7 +1335,9 @@ class AsyncRequestUtils:
:return: 上下文管理器，进入后 yield httpx.Response（出错时 yield None）
"""
cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None
kwargs.setdefault("headers", self._headers)
kwargs["headers"] = with_correlation_header(
kwargs.get("headers", self._headers)
)
# 与 _make_request 保持一致：复用 keep-alive 时偶遇对端 FIN 的连接，
# 流式 GET 是幂等的，单次重试即可
app/adapters/web/correlation.py
+44
View File
@@ -0,0 +1,44 @@
"""HTTP 请求关联 ID 的 ASGI 适配器。"""
from __future__ import annotations
from typing import Any
from app.runtime.correlation import (
CORRELATION_ID_HEADER,
correlation_scope,
normalize_correlation_id,
)
class CorrelationIdMiddleware:
"""验证入口 ID、绑定请求上下文并把同一 ID 写回响应。"""
def __init__(self, app: Any) -> None:
"""保存下游 ASGI 应用。"""
self._app = app
async def __call__(self, scope: dict, receive: Any, send: Any) -> None:
"""只治理 HTTP scope，并让绑定覆盖完整流式响应生命周期。"""
if scope.get("type") != "http":
await self._app(scope, receive, send)
return
raw_headers = dict(scope.get("headers") or [])
candidate = raw_headers.get(CORRELATION_ID_HEADER.lower().encode("ascii"))
correlation_id = normalize_correlation_id(
candidate.decode("ascii", errors="ignore") if candidate else None
)
scope.setdefault("state", {})["request_id"] = correlation_id
async def send_with_correlation(message: dict) -> None:
"""在响应开始帧中覆盖为当前请求的安全关联 ID。"""
if message.get("type") == "http.response.start":
headers = list(message.get("headers") or [])
header_name = CORRELATION_ID_HEADER.lower().encode("ascii")
headers = [item for item in headers if item[0].lower() != header_name]
headers.append((header_name, correlation_id.encode("ascii")))
message["headers"] = headers
await send(message)
with correlation_scope(correlation_id):
await self._app(scope, receive, send_with_correlation)
app/factory.py
+5 -1
View File
@@ -8,6 +8,7 @@ from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException
from app.api.response import ResponseAPIRoute
from app.adapters.web.correlation import CorrelationIdMiddleware
from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry
from app.adapters.web.health import install_health_routes
from app.application.plugin.routes import configure_plugin_routes
@@ -19,8 +20,9 @@ from app.adapters.web.security.access import (
from app.application.security.token import create_access_token, decode_access_token
from app.runtime.extensions.plugin_manager import PluginManager
from app.runtime.config import settings
from app.runtime.correlation import get_correlation_id
from app.runtime.localization import LocaleHelper
from app.runtime.log import logger
from app.runtime.log import configure_correlation_id_provider, logger
from app.schemas.openai import (
AnthropicErrorDetail,
AnthropicErrorResponse,
@@ -291,6 +293,7 @@ def create_app() -> FastAPI:
"""
创建并配置 FastAPI 应用实例。
"""
configure_correlation_id_provider(get_correlation_id)
_app = FastAPI(
title=settings.PROJECT_NAME,
version=APP_VERSION,
@@ -317,6 +320,7 @@ def create_app() -> FastAPI:
allow_methods=["*"],
allow_headers=["*"],
)
_app.add_middleware(CorrelationIdMiddleware)
@_app.middleware("http")
async def locale_context_middleware(
app/runtime/correlation.py
+74
View File
@@ -0,0 +1,74 @@
"""请求与后台工作共用的关联 ID 上下文。"""
from __future__ import annotations
import re
import uuid
from contextlib import contextmanager
from contextvars import ContextVar, Token
from typing import Any, Callable, Iterator, Mapping
CORRELATION_ID_HEADER = "X-Request-ID"
MAX_CORRELATION_ID_LENGTH = 64
_VALID_CORRELATION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,63}$")
_CURRENT_CORRELATION_ID: ContextVar[str | None] = ContextVar(
"moviepilot_correlation_id",
default=None,
)
def normalize_correlation_id(candidate: str | None) -> str:
"""接受安全的调用方 ID；非法、超长或缺失值均替换为随机 ID。"""
if candidate and _VALID_CORRELATION_ID.fullmatch(candidate):
return candidate
return uuid.uuid4().hex
def get_correlation_id(default: str | None = None) -> str | None:
"""返回当前执行上下文中的关联 ID。"""
return _CURRENT_CORRELATION_ID.get() or default
def set_correlation_id(correlation_id: str) -> Token[str | None]:
"""设置当前关联 ID，并返回供调用方精确恢复的 token。"""
return _CURRENT_CORRELATION_ID.set(correlation_id)
def reset_correlation_id(token: Token[str | None]) -> None:
"""恢复设置关联 ID 之前的上下文。"""
_CURRENT_CORRELATION_ID.reset(token)
@contextmanager
def correlation_scope(correlation_id: str | None) -> Iterator[str | None]:
"""在当前同步或异步任务作用域内绑定并自动恢复关联 ID。"""
if correlation_id is None:
yield None
return
token = set_correlation_id(correlation_id)
try:
yield correlation_id
finally:
reset_correlation_id(token)
def with_correlation_header(headers: Mapping[str, str] | None) -> dict[str, str]:
"""复制请求头并在调用方未显式指定时加入当前关联 ID。"""
result = dict(headers or {})
if any(key.lower() == CORRELATION_ID_HEADER.lower() for key in result):
return result
correlation_id = get_correlation_id()
if correlation_id:
result[CORRELATION_ID_HEADER] = correlation_id
return result
def call_with_correlation(
correlation_id: str | None,
func: Callable[..., Any],
args: tuple[Any, ...],
kwargs: dict[str, Any],
) -> Any:
"""从可序列化参数恢复关联 ID 后调用函数，供子进程入口使用。"""
with correlation_scope(correlation_id):
return func(*args, **kwargs)
app/runtime/event/dispatch.py
+29 -25
View File
@@ -12,6 +12,7 @@ from app.runtime.event.binding import EventBindingResolver
from app.runtime.event.registry import EventRegistry
from app.runtime.execution import run_in_threadpool
from app.runtime.log import logger
from app.runtime.correlation import correlation_scope
from app.schemas.types import EventType
@@ -114,6 +115,7 @@ class EventDispatcher:
event_type=event.event_type,
event_data=event_data,
priority=event.priority,
correlation_id=event.correlation_id,
)
if inspect.iscoroutinefunction(handler):
asyncio.run_coroutine_threadsafe(
@@ -143,16 +145,17 @@ class EventDispatcher:
if not resolved:
return
method, binding, class_name, method_name = resolved
try:
method(event)
except Exception as err:
self._error_handler(
event=event,
module_name=binding.owner_name,
class_name=class_name,
method_name=method_name,
e=err,
)
with correlation_scope(event.correlation_id):
try:
method(event)
except Exception as err:
self._error_handler(
event=event,
module_name=binding.owner_name,
class_name=class_name,
method_name=method_name,
e=err,
)
async def invoke_async(self, handler: Callable, event: Any) -> None:
"""解析实例绑定，并按处理器类型选择协程、线程池或同步调用。"""
@@ -160,21 +163,22 @@ class EventDispatcher:
if not resolved:
return
method, binding, class_name, method_name = resolved
try:
if inspect.iscoroutinefunction(method):
await method(event)
elif binding.run_sync_in_threadpool or not class_name:
await run_in_threadpool(method, event)
else:
method(event)
except Exception as err:
self._error_handler(
event=event,
module_name=binding.owner_name,
class_name=class_name,
method_name=method_name,
e=err,
)
with correlation_scope(event.correlation_id):
try:
if inspect.iscoroutinefunction(method):
await method(event)
elif binding.run_sync_in_threadpool or not class_name:
await run_in_threadpool(method, event)
else:
method(event)
except Exception as err:
self._error_handler(
event=event,
module_name=binding.owner_name,
class_name=class_name,
method_name=method_name,
e=err,
)
@staticmethod
def should_dispatch_to_target_plugin(
app/runtime/events.py
+5 -1
View File
@@ -21,6 +21,7 @@ from app.runtime.event.dispatch import EventDispatcher
from app.runtime.event.errors import EventErrorNotifier, EventErrorPolicy
from app.runtime.event.registry import EventRegistry
from app.runtime.event.contracts import validate_event_payload
from app.runtime.correlation import get_correlation_id
DEFAULT_EVENT_PRIORITY = 10 # 事件的默认优先级
MIN_EVENT_CONSUMER_THREADS = 1 # 最小事件消费者线程数
@@ -35,11 +36,13 @@ class Event:
def __init__(self, event_type: Union[EventType, ChainEventType],
event_data: Optional[Union[Dict, ChainEventData]] = None,
priority: Optional[int] = DEFAULT_EVENT_PRIORITY):
priority: Optional[int] = DEFAULT_EVENT_PRIORITY,
correlation_id: Optional[str] = None):
"""
:param event_type: 事件的类型，支持 EventType 或 ChainEventType
:param event_data: 可选，事件携带的数据，默认为空字典
:param priority: 可选，事件的优先级，默认为 10
:param correlation_id: 生产事件时固化的请求关联 ID
"""
payload_problems = validate_event_payload(event_type, event_data)
if payload_problems:
@@ -52,6 +55,7 @@ class Event:
self.event_type = event_type # 事件类型
self.event_data = event_data or {} # 事件数据
self.priority = priority # 事件优先级
self.correlation_id = correlation_id or get_correlation_id()
def __repr__(self) -> str:
"""
app/runtime/execution.py
+3 -1
View File
@@ -1,6 +1,7 @@
import asyncio
import inspect
import time
from contextvars import copy_context
from functools import partial, wraps
from typing import Any, Callable
@@ -16,7 +17,8 @@ async def run_in_threadpool(
"""在线程中执行同步函数，保持 FastAPI 旧帮助函数的参数语义。"""
if kwargs:
func = partial(func, **kwargs)
return await run_sync(func, *args)
context = copy_context()
return await run_sync(context.run, func, *args)
def retry(ExceptionToCheck: Any,
app/runtime/log.py
+24 -5
View File
@@ -12,12 +12,11 @@ from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Any, Dict, Optional, Protocol
from typing import Any, Callable, Dict, Optional, Protocol
import click
from pydantic import BaseModel, ConfigDict
class LogConfigModel(BaseModel):
"""描述日志级别、格式和文件写入策略。"""
@@ -28,8 +27,12 @@ class LogConfigModel(BaseModel):
LOG_LEVEL: str = "INFO"
LOG_MAX_FILE_SIZE: int = 5
LOG_BACKUP_COUNT: int = 10
LOG_CONSOLE_FORMAT: str = "%(leveltext)s[%(name)s] %(asctime)s %(message)s"
LOG_FILE_FORMAT: str = "【%(levelname)s】%(asctime)s - %(message)s"
LOG_CONSOLE_FORMAT: str = (
"%(leveltext)s[%(name)s] %(asctime)s [%(correlation_id)s] %(message)s"
)
LOG_FILE_FORMAT: str = (
"【%(levelname)s】%(asctime)s [%(correlation_id)s] - %(message)s"
)
ASYNC_FILE_QUEUE_SIZE: int = 1000
ASYNC_FILE_WORKERS: int = 2
BATCH_WRITE_SIZE: int = 50
@@ -60,6 +63,7 @@ class LogEntry:
self.message = message
self.file_path = file_path
self.timestamp = timestamp or datetime.now()
self.correlation_id = _get_log_correlation_id()
class LogWriter(Protocol):
@@ -73,6 +77,18 @@ class LogWriter(Protocol):
log_settings = LogSettings()
_correlation_id_provider: Callable[[], str | None] = lambda: None
def configure_correlation_id_provider(provider: Callable[[], str | None]) -> None:
"""由组合根注入日志关联 ID 读取端口，保持日志模块为依赖叶节点。"""
global _correlation_id_provider
_correlation_id_provider = provider
def _get_log_correlation_id() -> str:
"""读取当前关联 ID；未装配或无请求上下文时返回稳定占位符。"""
return _correlation_id_provider() or "-"
class NonBlockingFileHandler:
@@ -169,7 +185,7 @@ class NonBlockingFileHandler:
@staticmethod
def _to_record(entry: LogEntry) -> logging.LogRecord:
"""把日志条目转换为标准库日志记录。"""
return logging.LogRecord(
record = logging.LogRecord(
name="",
level=getattr(logging, entry.level.upper(), logging.INFO),
pathname="",
@@ -179,6 +195,8 @@ class NonBlockingFileHandler:
exc_info=None,
created=entry.timestamp.timestamp(),
)
record.correlation_id = entry.correlation_id
return record
def _batch_writer(self) -> None:
"""持续收集队列日志，并在停止哨兵后排空已有批次。"""
@@ -258,6 +276,7 @@ class CustomFormatter(logging.Formatter):
separator = " " * max(8 - len(record.levelname), 0)
colorizer = _LEVEL_NAME_COLORS.get(record.levelno, str)
record.leveltext = colorizer(record.levelname + ":") + separator
record.correlation_id = _get_log_correlation_id()
return super().format(record)
app/runtime/thread.py
+3 -1
View File
@@ -1,4 +1,5 @@
from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context
from app.runtime.config import settings
from app.foundation.singleton import Singleton
@@ -20,7 +21,8 @@ class ThreadHelper(metaclass=Singleton):
:param kwargs: 参数
:return: future
"""
return self.pool.submit(func, *args, **kwargs)
context = copy_context()
return self.pool.submit(context.run, func, *args, **kwargs)
def shutdown(self):
"""
app/scheduler.py
+5 -1
View File
@@ -45,6 +45,7 @@ from app.runtime.gc import get_memory_usage
from app.runtime.reload import ConfigReloadMixin
from app.foundation.singleton import SingletonClass
from app.runtime.scheduling import TimerUtils
from app.runtime.correlation import call_with_correlation, get_correlation_id
lock = threading.Lock()
SCHEDULER_PROGRESS_PREFIX = "scheduler"
@@ -856,7 +857,10 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
deferred_finish = __start_coro(func(*args, **kwargs))
elif run_in_process:
# 多进程运行
p = multiprocessing.Process(target=func, args=args, kwargs=kwargs)
p = multiprocessing.Process(
target=call_with_correlation,
args=(get_correlation_id(), func, args, kwargs),
)
p.start()
p.join()
else: