From 43cccb8ff80542b8bac736f36d9877647eb1cbc8 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 21 Aug 2026 22:15:47 +0800 Subject: [PATCH] feat: add low-cardinality observation port --- app/adapters/observability/__init__.py | 1 + app/adapters/observability/otel.py | 49 ++++++ app/adapters/web/metrics.py | 58 ++++++++ app/factory.py | 5 + app/runtime/event/dispatch.py | 25 +++- app/runtime/events.py | 11 ++ app/runtime/extensions/module/dispatcher.py | 41 ++++-- app/runtime/observability/__init__.py | 99 +++++++++++++ app/scheduler.py | 15 ++ .../backend-architecture-next-stage.md | 13 ++ .../architecture/dependency-baseline.json | 22 ++- tests/test_observability.py | 139 ++++++++++++++++++ 12 files changed, 454 insertions(+), 24 deletions(-) create mode 100644 app/adapters/observability/__init__.py create mode 100644 app/adapters/observability/otel.py create mode 100644 app/adapters/web/metrics.py create mode 100644 app/runtime/observability/__init__.py create mode 100644 tests/test_observability.py diff --git a/app/adapters/observability/__init__.py b/app/adapters/observability/__init__.py new file mode 100644 index 000000000..4576d801a --- /dev/null +++ b/app/adapters/observability/__init__.py @@ -0,0 +1 @@ +"""运行观测导出器适配器。""" diff --git a/app/adapters/observability/otel.py b/app/adapters/observability/otel.py new file mode 100644 index 000000000..c1b3c7f93 --- /dev/null +++ b/app/adapters/observability/otel.py @@ -0,0 +1,49 @@ +"""可选 OpenTelemetry metrics adapter。""" + +from __future__ import annotations + +import importlib +import os +from typing import Any, Mapping + +from app.runtime.observability import MetricKind, MetricSpec, NoopObservationPort, ObservationPort + + +class OpenTelemetryObservationPort: + """把内部指标合同映射到可选安装的 OpenTelemetry Metrics API。""" + + def __init__(self, meter: Any) -> None: + """保存 meter，并按名称惰性创建 instrument。""" + self._meter = meter + self._instruments: dict[str, Any] = {} + + def record(self, spec: MetricSpec, value: float, labels: Mapping[str, str]) -> None: + """按合同类型使用 OTel counter、histogram 或 up/down counter。""" + instrument = self._instruments.get(spec.name) + if instrument is None: + instrument = self._create_instrument(spec) + self._instruments[spec.name] = instrument + if spec.kind == MetricKind.HISTOGRAM: + instrument.record(value, attributes=dict(labels)) + else: + instrument.add(value, attributes=dict(labels)) + + def _create_instrument(self, spec: MetricSpec) -> Any: + """为内部指标类型创建对应 OTel instrument。""" + normalized = spec.name.replace(".", "_") + if spec.kind == MetricKind.HISTOGRAM: + return self._meter.create_histogram(normalized) + if spec.kind == MetricKind.COUNTER: + return self._meter.create_counter(normalized) + return self._meter.create_up_down_counter(normalized) + + +def build_observation_port() -> ObservationPort: + """仅在显式启用且 API 可导入时创建 OTel adapter，否则返回 no-op。""" + if os.getenv("MOVIEPILOT_OTEL_METRICS") != "1": + return NoopObservationPort() + try: + metrics = importlib.import_module("opentelemetry.metrics") + except ImportError: + return NoopObservationPort() + return OpenTelemetryObservationPort(metrics.get_meter("moviepilot")) diff --git a/app/adapters/web/metrics.py b/app/adapters/web/metrics.py new file mode 100644 index 000000000..9bff7587f --- /dev/null +++ b/app/adapters/web/metrics.py @@ -0,0 +1,58 @@ +"""HTTP route/status/latency 指标 ASGI 适配器。""" + +from __future__ import annotations + +import time +from typing import Any + +from app.runtime.observability import record_metric +from starlette.routing import Match + + +class HttpMetricsMiddleware: + """按路由模板、方法和状态码记录低基数 HTTP 时延。""" + + 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 + started_at = time.perf_counter() + status = "500" + + async def send_with_metrics(message: dict) -> None: + """捕获响应状态并原样转发 ASGI 消息。""" + nonlocal status + if message.get("type") == "http.response.start": + status = str(message.get("status", 500)) + await send(message) + + try: + await self._app(scope, receive, send_with_metrics) + finally: + route_path = self._resolve_route_template(scope) + record_metric( + "http.server.duration", + time.perf_counter() - started_at, + route=route_path, + method=str(scope.get("method", "UNKNOWN")), + status=status, + ) + + def _resolve_route_template(self, scope: dict) -> str: + """遍历 ASGI wrapper 找到匹配路由模板，绝不回退到具体请求 path。""" + route = scope.get("route") + if getattr(route, "path", None): + return route.path + candidate = self._app + while candidate is not None: + for registered_route in getattr(candidate, "routes", ()): + match, _ = registered_route.matches(scope) + if match == Match.FULL: + return getattr(registered_route, "path", "unmatched") + candidate = getattr(candidate, "app", None) + return "unmatched" diff --git a/app/factory.py b/app/factory.py index 4ad4f09ae..ffe39fe6e 100644 --- a/app/factory.py +++ b/app/factory.py @@ -9,6 +9,8 @@ from starlette.exceptions import HTTPException from app.api.response import ResponseAPIRoute from app.adapters.web.correlation import CorrelationIdMiddleware +from app.adapters.web.metrics import HttpMetricsMiddleware +from app.adapters.observability.otel import build_observation_port 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 @@ -23,6 +25,7 @@ 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 configure_correlation_id_provider, logger +from app.runtime.observability import configure_observation from app.schemas.openai import ( AnthropicErrorDetail, AnthropicErrorResponse, @@ -294,6 +297,7 @@ def create_app() -> FastAPI: 创建并配置 FastAPI 应用实例。 """ configure_correlation_id_provider(get_correlation_id) + configure_observation(build_observation_port()) _app = FastAPI( title=settings.PROJECT_NAME, version=APP_VERSION, @@ -321,6 +325,7 @@ def create_app() -> FastAPI: allow_headers=["*"], ) _app.add_middleware(CorrelationIdMiddleware) + _app.add_middleware(HttpMetricsMiddleware) @_app.middleware("http") async def locale_context_middleware( diff --git a/app/runtime/event/dispatch.py b/app/runtime/event/dispatch.py index 6251c148b..78f4064d9 100644 --- a/app/runtime/event/dispatch.py +++ b/app/runtime/event/dispatch.py @@ -13,6 +13,7 @@ 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.runtime.observability import observe_duration from app.schemas.types import EventType @@ -147,7 +148,12 @@ class EventDispatcher: method, binding, class_name, method_name = resolved with correlation_scope(event.correlation_id): try: - method(event) + with observe_duration( + "event.handler.duration", + event_type=event.event_type.value, + handler_type="bound" if class_name else "function", + ): + method(event) except Exception as err: self._error_handler( event=event, @@ -165,12 +171,17 @@ class EventDispatcher: method, binding, class_name, method_name = resolved 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) + with observe_duration( + "event.handler.duration", + event_type=event.event_type.value, + handler_type="bound" if class_name else "function", + ): + 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, diff --git a/app/runtime/events.py b/app/runtime/events.py index 9f740e522..a2deee143 100644 --- a/app/runtime/events.py +++ b/app/runtime/events.py @@ -22,6 +22,7 @@ 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 +from app.runtime.observability import record_metric DEFAULT_EVENT_PRIORITY = 10 # 事件的默认优先级 MIN_EVENT_CONSUMER_THREADS = 1 # 最小事件消费者线程数 @@ -317,6 +318,11 @@ class EventManager(metaclass=Singleton): """ logger.debug(f"Triggering broadcast event: {event}") self.__event_queue.put((event.priority, event)) + record_metric( + "event.queue.depth", + self.__event_queue.qsize(), + delivery="broadcast", + ) def __dispatch_chain_event(self, event: Event) -> bool: """ @@ -421,6 +427,11 @@ class EventManager(metaclass=Singleton): while self.__event.is_set(): try: priority, event = self.__event_queue.get(timeout=rate_limiter.current_wait) + record_metric( + "event.queue.depth", + self.__event_queue.qsize(), + delivery="broadcast", + ) rate_limiter.reset() self.__dispatch_broadcast_event(event) except Empty: diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py index ce6c1a2f6..ca5960d0f 100644 --- a/app/runtime/extensions/module/dispatcher.py +++ b/app/runtime/extensions/module/dispatcher.py @@ -9,6 +9,7 @@ from typing import Any, Protocol from app.foundation.reflection import ObjectUtils from app.runtime.execution import run_in_threadpool from app.runtime.log import logger +from app.runtime.observability import observe_duration from app.runtime.extensions.module.contracts import ( diagnose_module_callable, get_module_method_contract, @@ -68,29 +69,41 @@ class ModuleInvocationDispatcher: """先执行插件模块，再按优先级执行宿主模块。""" contract = get_module_method_contract(method) logger.debug("模块方法契约：%s -> %s", method, contract.family) - result = self.execute_plugin_modules(method, None, *args, **kwargs) + with observe_duration( + "module.provider.duration", method=method, provider_type="plugin" + ): + result = self.execute_plugin_modules(method, None, *args, **kwargs) if not self.is_valid_empty(result) and not isinstance(result, list): return result - return self.execute_system_modules(method, result, *args, **kwargs) + with observe_duration( + "module.provider.duration", method=method, provider_type="system" + ): + return self.execute_system_modules(method, result, *args, **kwargs) async def async_dispatch(self, method: str, *args: Any, **kwargs: Any) -> Any: """以与同步路径相同的聚合规则执行同步或异步模块方法。""" contract = get_module_method_contract(method) logger.debug("异步模块方法契约：%s -> %s", method, contract.family) - result = await self.async_execute_plugin_modules( - method, - None, - *args, - **kwargs, - ) + with observe_duration( + "module.provider.duration", method=method, provider_type="plugin" + ): + result = await self.async_execute_plugin_modules( + method, + None, + *args, + **kwargs, + ) if not self.is_valid_empty(result) and not isinstance(result, list): return result - return await self.async_execute_system_modules( - method, - result, - *args, - **kwargs, - ) + with observe_duration( + "module.provider.duration", method=method, provider_type="system" + ): + return await self.async_execute_system_modules( + method, + result, + *args, + **kwargs, + ) def execute_plugin_modules( self, diff --git a/app/runtime/observability/__init__.py b/app/runtime/observability/__init__.py new file mode 100644 index 000000000..5a1a3ba10 --- /dev/null +++ b/app/runtime/observability/__init__.py @@ -0,0 +1,99 @@ +"""低基数运行指标端口与进程级 Facade。""" + +from __future__ import annotations + +import time +from contextlib import contextmanager +from dataclasses import dataclass +from enum import StrEnum +from typing import Iterator, Mapping, Protocol + + +class MetricKind(StrEnum): + """声明指标采用 counter、histogram 或 gauge 语义。""" + + COUNTER = "counter" + HISTOGRAM = "histogram" + GAUGE = "gauge" + + +@dataclass(frozen=True, slots=True) +class MetricSpec: + """声明稳定指标名称、类型与允许的低基数标签。""" + + name: str + kind: MetricKind + labels: frozenset[str] + + +METRIC_SPECS = { + spec.name: spec + for spec in ( + MetricSpec("http.server.duration", MetricKind.HISTOGRAM, frozenset({"route", "method", "status"})), + MetricSpec("db.pool.wait", MetricKind.HISTOGRAM, frozenset({"backend", "outcome"})), + MetricSpec("db.pool.checked_out", MetricKind.GAUGE, frozenset({"backend"})), + MetricSpec("db.pool.timeout", MetricKind.COUNTER, frozenset({"backend"})), + MetricSpec("event.queue.depth", MetricKind.GAUGE, frozenset({"delivery"})), + MetricSpec("event.handler.duration", MetricKind.HISTOGRAM, frozenset({"event_type", "handler_type", "outcome"})), + MetricSpec("module.provider.duration", MetricKind.HISTOGRAM, frozenset({"method", "provider_type", "outcome"})), + MetricSpec("module.provider.timeout", MetricKind.COUNTER, frozenset({"method", "provider_type"})), + MetricSpec("scheduler.job.duration", MetricKind.HISTOGRAM, frozenset({"owner", "outcome"})), + MetricSpec("scheduler.job.overlap_skip", MetricKind.COUNTER, frozenset({"owner"})), + MetricSpec("scheduler.job.retry", MetricKind.COUNTER, frozenset({"owner"})), + MetricSpec("scheduler.job.dead_letter", MetricKind.COUNTER, frozenset({"owner"})), + MetricSpec("plugin.lifecycle.duration", MetricKind.HISTOGRAM, frozenset({"operation", "outcome"})), + MetricSpec("agent.active_tasks", MetricKind.GAUGE, frozenset({"task_type"})), + MetricSpec("agent.cancel", MetricKind.COUNTER, frozenset({"task_type", "outcome"})), + MetricSpec("agent.provider.duration", MetricKind.HISTOGRAM, frozenset({"provider_type", "outcome"})), + MetricSpec("agent.token_usage", MetricKind.COUNTER, frozenset({"provider_type", "direction"})), + ) +} + + +class ObservationPort(Protocol): + """Application/runtime 可依赖的最小指标写入端口。""" + + def record(self, spec: MetricSpec, value: float, labels: Mapping[str, str]) -> None: + """记录一个已经通过标签合同校验的指标值。""" + + +class NoopObservationPort: + """未安装或未启用 exporter 时完全无副作用的默认实现。""" + + def record(self, spec: MetricSpec, value: float, labels: Mapping[str, str]) -> None: + """接受合法指标但不分配 exporter 资源。""" + + +_observation_port: ObservationPort = NoopObservationPort() + + +def configure_observation(port: ObservationPort | None) -> None: + """由组合根替换进程级端口；None 明确恢复 no-op。""" + global _observation_port + _observation_port = port or NoopObservationPort() + + +def record_metric(name: str, value: float = 1, **labels: str) -> None: + """校验名称、类型无关数值和标签白名单后写入当前端口。""" + spec = METRIC_SPECS[name] + unexpected = set(labels) - spec.labels + if unexpected: + raise ValueError(f"指标 {name} 包含未登记标签：{sorted(unexpected)}") + _observation_port.record(spec, float(value), labels) + + +@contextmanager +def observe_duration(name: str, **labels: str) -> Iterator[None]: + """记录代码块耗时，并把成功或失败收敛为低基数 outcome。""" + started_at = time.perf_counter() + outcome = "success" + try: + yield + except BaseException: + outcome = "error" + raise + finally: + duration_labels = dict(labels) + if "outcome" in METRIC_SPECS[name].labels: + duration_labels["outcome"] = outcome + record_metric(name, time.perf_counter() - started_at, **duration_labels) diff --git a/app/scheduler.py b/app/scheduler.py index adcdcf563..a481e9ea4 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -4,6 +4,7 @@ import hashlib import inspect import multiprocessing import threading +import time import traceback from datetime import datetime, timedelta from typing import Callable, Optional, Dict, Any, List @@ -46,6 +47,7 @@ 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 +from app.runtime.observability import record_metric lock = threading.Lock() SCHEDULER_PROGRESS_PREFIX = "scheduler" @@ -562,7 +564,12 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): return None if not JobExecutionState.begin(job, started_at): logger.warning(f"定时任务 {job_id} - {job.get('name')} 正在运行 ...") + record_metric( + "scheduler.job.overlap_skip", + owner=str(job.get("owner", "unknown")), + ) return None + job["_metric_started_at"] = time.perf_counter() progress = ProgressHelper(self._get_progress_key(job_id)) progress.start() progress.update( @@ -596,6 +603,14 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job = self._jobs.get(job_id) if job: JobExecutionState.finish(job, finished_at, error) + metric_started_at = job.pop("_metric_started_at", None) + if metric_started_at is not None: + record_metric( + "scheduler.job.duration", + time.perf_counter() - metric_started_at, + owner=str(job.get("owner", "unknown")), + outcome="success" if success else "error", + ) job_name = job.get("name") if job else job_id # 收尾可能发生在事件循环上（__run_coro_job），使用异步进度后端避免阻塞 progress = AsyncProgressHelper(self._get_progress_key(job_id)) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 47c5f098c..33dcb902b 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -756,6 +756,19 @@ app/scheduler.py # APScheduler 兼容 Facade OTel 初始化只能位于 Startup/Adapter；Domain/Application 只依赖 no-op-capable observation Protocol。插件 ID、用户 ID、媒体标题等高基数字段不得直接作为 metric label。 +**实施记录（2026-08-21）**： + +- `app.runtime.observability` 定义单一 `ObservationPort`、默认 no-op、指标类型/目录、标签白名单和统一耗时 + 作用域；没有 exporter 时所有调用仍可执行，未登记标签在进入 Adapter 前直接拒绝。 +- 指标目录覆盖 HTTP、DB pool、Event、Module、Scheduler、Plugin lifecycle 和 Agent 所列能力；标签审计 + 明确禁止 user/plugin/media/request/job 实例 ID、标题和 URL。首批实际接线覆盖 HTTP route/status/latency、 + Event queue/handler、Module provider 与 Scheduler duration/overlap，剩余能力可按相同端口逐个接入。 +- `app.adapters.observability.otel` 只在组合根显式读取 `MOVIEPILOT_OTEL_METRICS=1` 后懒加载 OTel API； + 未安装可选包时稳定回退 no-op，不给核心层增加 SDK 依赖。HTTP Adapter 通过路由匹配输出模板，绝不以原始 + request path 充当 label。 +- 专项测试覆盖 exporter 缺失、非法标签、全目录高基数审计、成功/失败 outcome、动态 URL 路由模板； + 既有 API、Event、Module、Scheduler 与健康探针回归保持通过。 + #### ARCH-270：渐进式类型门禁 **目标**：不要求全仓一次通过 mypy/pyright；只保证新 canonical contract 和被治理模块完整类型化。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index b1f6b11b0..84b9a5d68 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6232, - "edge_sha256": "2139dcdcfc12b4f29c732471744d09b0990c51867d4ad227026b00fa947c90d8", + "edge_count": 6244, + "edge_sha256": "f187bdbb5e88ce9a6b2ff559b10e5e6cd5a60d14693484ea17663cebf438cc92", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -106,6 +106,8 @@ "app.adapters.network.doh -> app.runtime.reload", "app.adapters.network.http -> app.runtime", "app.adapters.network.http -> app.runtime.correlation", + "app.adapters.observability.otel -> app.runtime", + "app.adapters.observability.otel -> app.runtime.observability", "app.adapters.system.display -> app.foundation", "app.adapters.system.display -> app.foundation.singleton", "app.adapters.system.display -> app.runtime", @@ -157,6 +159,8 @@ "app.adapters.web.correlation -> app.runtime.correlation", "app.adapters.web.health -> app.runtime", "app.adapters.web.health -> app.runtime.health", + "app.adapters.web.metrics -> app.runtime", + "app.adapters.web.metrics -> app.runtime.observability", "app.adapters.web.security.access -> app.runtime", "app.adapters.web.security.access -> app.runtime.cache", "app.adapters.web.security.access -> app.runtime.config", @@ -3722,9 +3726,12 @@ "app.domain.title -> app.schemas", "app.domain.title -> app.schemas.types", "app.factory -> app.adapters", + "app.factory -> app.adapters.observability", + "app.factory -> app.adapters.observability.otel", "app.factory -> app.adapters.web", "app.factory -> app.adapters.web.correlation", "app.factory -> app.adapters.web.health", + "app.factory -> app.adapters.web.metrics", "app.factory -> app.adapters.web.plugin", "app.factory -> app.adapters.web.plugin.routes", "app.factory -> app.adapters.web.security", @@ -3743,6 +3750,7 @@ "app.factory -> app.runtime.extensions.plugin_manager", "app.factory -> app.runtime.localization", "app.factory -> app.runtime.log", + "app.factory -> app.runtime.observability", "app.factory -> app.schemas", "app.factory -> app.schemas.mcp", "app.factory -> app.schemas.openai", @@ -5365,6 +5373,7 @@ "app.runtime.event.dispatch -> app.runtime.event.registry", "app.runtime.event.dispatch -> app.runtime.execution", "app.runtime.event.dispatch -> app.runtime.log", + "app.runtime.event.dispatch -> app.runtime.observability", "app.runtime.event.dispatch -> app.schemas", "app.runtime.event.dispatch -> app.schemas.types", "app.runtime.event.errors -> app.runtime", @@ -5387,6 +5396,7 @@ "app.runtime.events -> app.runtime.event.errors", "app.runtime.events -> app.runtime.event.registry", "app.runtime.events -> app.runtime.log", + "app.runtime.events -> app.runtime.observability", "app.runtime.events -> app.runtime.rate", "app.runtime.events -> app.runtime.thread", "app.runtime.events -> app.schemas", @@ -5417,6 +5427,7 @@ "app.runtime.extensions.module.dispatcher -> app.runtime.extensions.module", "app.runtime.extensions.module.dispatcher -> app.runtime.extensions.module.contracts", "app.runtime.extensions.module.dispatcher -> app.runtime.log", + "app.runtime.extensions.module.dispatcher -> app.runtime.observability", "app.runtime.extensions.module.dispatcher -> app.schemas", "app.runtime.extensions.module.dispatcher -> app.schemas.exception", "app.runtime.extensions.module_manager -> app.foundation", @@ -5573,6 +5584,7 @@ "app.scheduler -> app.runtime.extensions.service_config", "app.scheduler -> app.runtime.gc", "app.scheduler -> app.runtime.log", + "app.scheduler -> app.runtime.observability", "app.scheduler -> app.runtime.progress", "app.scheduler -> app.runtime.reload", "app.scheduler -> app.runtime.scheduling", @@ -6249,7 +6261,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 778, + "module_count": 782, "modules": [ "app", "app.adapters", @@ -6271,6 +6283,8 @@ "app.adapters.network.doh", "app.adapters.network.http", "app.adapters.network.ip", + "app.adapters.observability", + "app.adapters.observability.otel", "app.adapters.system", "app.adapters.system.backup", "app.adapters.system.backup.database", @@ -6291,6 +6305,7 @@ "app.adapters.web", "app.adapters.web.correlation", "app.adapters.web.health", + "app.adapters.web.metrics", "app.adapters.web.plugin", "app.adapters.web.plugin.routes", "app.adapters.web.security", @@ -6922,6 +6937,7 @@ "app.runtime.localization", "app.runtime.log", "app.runtime.managed_resources", + "app.runtime.observability", "app.runtime.progress", "app.runtime.rate", "app.runtime.reload", diff --git a/tests/test_observability.py b/tests/test_observability.py new file mode 100644 index 000000000..c98885896 --- /dev/null +++ b/tests/test_observability.py @@ -0,0 +1,139 @@ +"""低基数指标合同、no-op 与 HTTP adapter 测试。""" + +from dataclasses import dataclass, field +from typing import Mapping + +import httpx +import pytest +from starlette.applications import Starlette +from starlette.responses import PlainTextResponse +from starlette.routing import Route + +from app.adapters.observability import otel +from app.adapters.web.metrics import HttpMetricsMiddleware +from app.runtime.observability import ( + METRIC_SPECS, + MetricSpec, + NoopObservationPort, + configure_observation, + observe_duration, + record_metric, +) + + +@dataclass +class RecordingObservationPort: + """测试用端口，保存已经通过核心标签校验的写入。""" + + records: list[tuple[MetricSpec, float, Mapping[str, str]]] = field( + default_factory=list + ) + + def record( + self, spec: MetricSpec, value: float, labels: Mapping[str, str] + ) -> None: + """追加一条不可变测试快照。""" + self.records.append((spec, value, dict(labels))) + + +@pytest.fixture(autouse=True) +def _reset_observation_port(): + """避免进程级观测端口在用例间泄漏。""" + configure_observation(None) + yield + configure_observation(None) + + +def test_noop_port_accepts_registered_metric_without_exporter() -> None: + """未安装 exporter 时记录合法指标必须完全可用。""" + configure_observation(NoopObservationPort()) + record_metric( + "http.server.duration", + 0.1, + route="/health/live", + method="GET", + status="200", + ) + + +def test_metric_catalog_contains_no_high_cardinality_labels() -> None: + """整个指标目录不得登记用户、插件、媒体、URL 或请求实例标签。""" + forbidden = { + "user_id", + "plugin_id", + "media_id", + "media_title", + "url", + "request_id", + "job_id", + } + + assert METRIC_SPECS + assert all(not (spec.labels & forbidden) for spec in METRIC_SPECS.values()) + + +def test_unregistered_label_is_rejected_before_adapter() -> None: + """调用方不能绕过目录向 exporter 注入高基数标签。""" + with pytest.raises(ValueError, match="未登记标签"): + record_metric( + "scheduler.job.duration", + 1, + owner="plugin", + outcome="success", + job_id="dynamic-123", + ) + + +def test_duration_records_success_and_error_outcomes() -> None: + """统一计时器把正常和异常路径收敛为有限 outcome。""" + port = RecordingObservationPort() + configure_observation(port) + + with observe_duration( + "module.provider.duration", method="recognize_media", provider_type="system" + ): + pass + with pytest.raises(RuntimeError): + with observe_duration( + "module.provider.duration", + method="recognize_media", + provider_type="plugin", + ): + raise RuntimeError("failed") + + assert [record[2]["outcome"] for record in port.records] == ["success", "error"] + + +@pytest.mark.asyncio +async def test_http_metrics_use_route_template_not_request_url() -> None: + """HTTP 指标使用路由模板，不能把具体资源 ID 变成 label。""" + port = RecordingObservationPort() + configure_observation(port) + + async def item(_request): + """返回固定测试响应。""" + return PlainTextResponse("ok") + + app = Starlette(routes=[Route("/items/{item_id}", item)]) + app.add_middleware(HttpMetricsMiddleware) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/items/secret-item") + + assert response.status_code == 200 + _, _, labels = port.records[-1] + assert labels == {"route": "/items/{item_id}", "method": "GET", "status": "200"} + + +def test_optional_otel_adapter_falls_back_to_noop(monkeypatch) -> None: + """显式启用但未安装 OTel API 时启动仍返回 no-op。""" + monkeypatch.setenv("MOVIEPILOT_OTEL_METRICS", "1") + + def missing(_name: str): + """模拟可选依赖不存在。""" + raise ImportError("missing") + + monkeypatch.setattr(otel.importlib, "import_module", missing) + + assert isinstance(otel.build_observation_port(), NoopObservationPort)