mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
252 lines
9.0 KiB
Python
252 lines
9.0 KiB
Python
"""
|
||
低基数运行指标端口与进程级 Facade。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import functools
|
||
import inspect
|
||
import time
|
||
from contextlib import contextmanager
|
||
from dataclasses import dataclass
|
||
from enum import StrEnum
|
||
from typing import Any, Callable, Iterator, Mapping, Protocol, TypeVar, cast
|
||
|
||
|
||
class MetricKind(StrEnum):
|
||
"""
|
||
声明指标采用 counter、histogram 或 gauge 语义。
|
||
"""
|
||
|
||
# 累计计数，记录本次增加量
|
||
COUNTER = "counter"
|
||
# 数值分布，记录本次观测样本
|
||
HISTOGRAM = "histogram"
|
||
# 瞬时数值，记录当前状态
|
||
GAUGE = "gauge"
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class MetricSpec:
|
||
"""
|
||
声明稳定指标名称、类型与允许的低基数标签。
|
||
|
||
:param name: 稳定且唯一的指标名称
|
||
:param kind: 指标的累计、分布或瞬时值类型
|
||
:param labels: 允许写入的低基数标签名集合
|
||
"""
|
||
|
||
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("db.worker.wait", MetricKind.HISTOGRAM, frozenset()),
|
||
MetricSpec("db.worker.duration", MetricKind.HISTOGRAM, frozenset({"outcome"})),
|
||
MetricSpec("db.worker.queue.depth", MetricKind.GAUGE, frozenset()),
|
||
MetricSpec("db.worker.active", MetricKind.GAUGE, frozenset()),
|
||
MetricSpec("db.worker.rejected", MetricKind.COUNTER, frozenset()),
|
||
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(
|
||
"module.contract.legacy_hit",
|
||
MetricKind.COUNTER,
|
||
frozenset({"method", "caller_type", "abi_source"}),
|
||
),
|
||
MetricSpec(
|
||
"module.contract.result_mismatch",
|
||
MetricKind.COUNTER,
|
||
frozenset({"method", "provider_type", "problem"}),
|
||
),
|
||
MetricSpec(
|
||
"compat.facade.hit",
|
||
MetricKind.COUNTER,
|
||
frozenset({"facade", "operation", "visibility", "abi_source"}),
|
||
),
|
||
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.chat.persistence.pending", MetricKind.GAUGE, frozenset()),
|
||
MetricSpec("agent.chat.persistence.rejected", MetricKind.COUNTER, frozenset()),
|
||
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:
|
||
"""
|
||
记录一个已经通过标签合同校验的指标值。
|
||
|
||
:param spec: 已登记的指标名称、类型和标签约束
|
||
:param value: 本次指标观测值，按指标类型解释为增量或样本
|
||
:param labels: 已通过契约校验的低基数标签值映射
|
||
"""
|
||
|
||
|
||
class NoopObservationPort:
|
||
"""
|
||
未安装或未启用 exporter 时完全无副作用的默认实现。
|
||
"""
|
||
|
||
def record(self, spec: MetricSpec, value: float, labels: Mapping[str, str]) -> None:
|
||
"""
|
||
接受合法指标但不分配 exporter 资源。
|
||
"""
|
||
|
||
|
||
_observation_port: ObservationPort = NoopObservationPort()
|
||
|
||
_FacadeClass = TypeVar("_FacadeClass", bound=type)
|
||
|
||
|
||
def observe_compat_facade(facade: str) -> Callable[[_FacadeClass], _FacadeClass]:
|
||
"""
|
||
为旧 ABI Facade 的公开和私有方法记录低基数命中，不改变方法合同。
|
||
"""
|
||
|
||
def decorate(cls: _FacadeClass) -> _FacadeClass:
|
||
"""
|
||
包装兼容门面的普通、类和静态方法，保留各自的绑定方式。
|
||
"""
|
||
for name, descriptor in tuple(vars(cls).items()):
|
||
if name.startswith("__") and name.endswith("__"):
|
||
continue
|
||
visibility = "private" if name.startswith("_") else "public"
|
||
if isinstance(descriptor, classmethod):
|
||
wrapped = _wrap_compat_method(
|
||
descriptor.__func__, facade, name, visibility
|
||
)
|
||
setattr(cls, name, classmethod(wrapped))
|
||
elif isinstance(descriptor, staticmethod):
|
||
wrapped = _wrap_compat_method(
|
||
descriptor.__func__, facade, name, visibility
|
||
)
|
||
setattr(cls, name, staticmethod(wrapped))
|
||
elif callable(descriptor):
|
||
setattr(
|
||
cls,
|
||
name,
|
||
_wrap_compat_method(descriptor, facade, name, visibility),
|
||
)
|
||
return cls
|
||
|
||
return decorate
|
||
|
||
|
||
def _enforce_deprecation(facade: str, operation: str) -> None:
|
||
"""
|
||
对一次旧 Facade 命中执行其当前的废弃阶段处置。
|
||
|
||
观测回答「谁还在用」，废弃阶段回答「什么时候不再让用」，两者共用同一组标签。
|
||
未登记废弃通告的 Facade 在此完全不受影响。
|
||
|
||
:param facade: Facade 标识，与 compat.facade.hit 指标的 facade 标签一致
|
||
:param operation: 被调用的方法名
|
||
"""
|
||
from app.runtime.deprecation.policy import enforce_facade
|
||
|
||
enforce_facade(facade, operation)
|
||
|
||
|
||
def _wrap_compat_method(
|
||
method: Callable[..., Any],
|
||
facade: str,
|
||
operation: str,
|
||
visibility: str,
|
||
) -> Callable[..., Any]:
|
||
"""
|
||
包装一个兼容方法并保持同步/异步调用形态及反射元数据。
|
||
"""
|
||
if inspect.iscoroutinefunction(method):
|
||
|
||
@functools.wraps(method)
|
||
async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
|
||
"""
|
||
记录兼容调用并执行废弃策略，再等待原异步方法。
|
||
"""
|
||
record_metric(
|
||
"compat.facade.hit",
|
||
facade=facade,
|
||
operation=operation,
|
||
visibility=visibility,
|
||
abi_source="legacy_facade",
|
||
)
|
||
_enforce_deprecation(facade, operation)
|
||
return await method(*args, **kwargs)
|
||
|
||
return cast(Callable[..., Any], async_wrapper)
|
||
|
||
@functools.wraps(method)
|
||
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||
"""
|
||
记录兼容调用并执行废弃策略，再同步调用原方法。
|
||
"""
|
||
record_metric(
|
||
"compat.facade.hit",
|
||
facade=facade,
|
||
operation=operation,
|
||
visibility=visibility,
|
||
abi_source="legacy_facade",
|
||
)
|
||
_enforce_deprecation(facade, operation)
|
||
return method(*args, **kwargs)
|
||
|
||
return cast(Callable[..., Any], wrapper)
|
||
|
||
|
||
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)
|