MemeMeow/runtime_telemetry.py

# 公共运行观测模块，为 API 与 executor 提供独立的 OTLP 导出和内容保护。
from __future__ import annotations

import json
import os
import re
import time
from pathlib import Path
from contextlib import contextmanager
from dataclasses import dataclass
from threading import Lock
from typing import Any, Iterator, Mapping
from urllib.parse import urlsplit
from uuid import uuid4

from loguru import logger
from httpx import TimeoutException
from requests import Session
from requests.exceptions import Timeout as RequestsTimeout
from opentelemetry import metrics, trace
from opentelemetry.context import Context
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, LogRecordExportResult
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor, HTTPX2ClientInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.threading import ThreadingInstrumentor
from opentelemetry.instrumentation.urllib import URLLibInstrumentor
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import MetricExportResult, PeriodicExportingMetricReader
from opentelemetry.sdk.metrics.view import ExplicitBucketHistogramAggregation, View
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import Event, ReadableSpan, SpanLimits, SpanProcessor, TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExportResult, SpanExporter
from opentelemetry.sdk.trace.sampling import ALWAYS_ON
from opentelemetry.trace import Status, StatusCode

_CREDENTIAL_KEY = re.compile(r"(?i)(password|passwd|secret|(?:access|refresh|session|auth|bearer|id)[_ .-]?token(?:$|[_ .-])|(?:^|[_ .-])token(?:$|[_ .-])|authorization|cookie|api[_-]?key|credential|signature|(?:^|[_ .-])sig(?:$|[_ .-]))")
_CREDENTIAL_TEXT = re.compile(r"(?i)(?:bearer\s+[^\s\"\x27,;&]+|(?:api[_-]?key|password|token|secret|authorization|signature|sig)[\"\x27]?\s*[=:]\s*[\"\x27]?[^\s\"\x27,;&]+)")
_BINARY_KEY = re.compile(r"(?i)(base64|image_url|image_data|file_content|binary)")
_PROPAGATOR = TraceContextTextMapPropagator()
_LOCK = Lock()
_RUNTIME: TelemetryRuntime | None = None
SECONDS_HISTOGRAM_BOUNDARIES = (0.005, 0.01, 0.025, 0.05, 0.1, 0.2, 0.25, 0.5, 1, 2, 3, 5, 10, 30, 60, 120, 300)


def duration_views() -> list[View]:
    """为秒数 histogram 提供包含数据库与 HTTP 告警阈值的官方 View。"""
    return [View(instrument_type=metrics.Histogram, instrument_unit="s", aggregation=ExplicitBucketHistogramAggregation(boundaries=SECONDS_HISTOGRAM_BOUNDARIES))]


def otlp_session() -> Session:
    """独立导出客户端直接连接宿主配置的 Collector，不使用业务出口代理。"""
    session = Session()
    session.trust_env = False
    return session


@dataclass(frozen=True)
class TelemetryConfig:
    """进程观测配置；endpoint 只由宿主环境提供。"""

    enabled: bool = False
    endpoint: str = ""
    service: str = "mememeow-api"
    environment: str = "development"
    authorization: str = ""

    @classmethod
    def from_env(cls, service: str = "mememeow-api") -> TelemetryConfig:
        """读取并校验宿主配置，配置错误在启动位置报告。"""
        enabled = os.getenv("MEMEMEOW_TELEMETRY_ENABLED", "false").lower()
        if enabled not in {"true", "false"}:
            raise ValueError("MEMEMEOW_TELEMETRY_ENABLED 必须为 true 或 false")
        endpoint = os.getenv("MEMEMEOW_TELEMETRY_ENDPOINT", "")
        if enabled == "true":
            parsed = urlsplit(endpoint)
            if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
                raise ValueError("MEMEMEOW_TELEMETRY_ENDPOINT 必须为不包含凭据或查询参数的 HTTP(S) 地址")
        authorization_file = os.getenv("MEMEMEOW_TELEMETRY_AUTH_FILE", "")
        authorization = Path(authorization_file).read_text().strip() if enabled == "true" and authorization_file else ""
        if authorization_file and enabled == "true" and (not authorization.startswith("Basic ") or "\n" in authorization):
            raise ValueError("MEMEMEOW_TELEMETRY_AUTH_FILE 必须包含单行 Basic Authorization")
        return cls(enabled == "true", endpoint.rstrip("/"), service, os.getenv("MEMEMEOW_TELEMETRY_ENVIRONMENT", "development"), authorization)


def protected_content(value: Any, secrets: tuple[str, ...] = ()) -> Any:
    """过滤结构化凭据与已知凭据值，保留 debug 文本和错误类别。"""
    if isinstance(value, Mapping):
        return {
            str(key): "[credential filtered]" if _CREDENTIAL_KEY.search(str(key)) else
            "[binary content omitted; use business file reference]" if _BINARY_KEY.search(str(key)) else
            protected_content(item, secrets)
            for key, item in value.items()
        }
    if isinstance(value, (list, tuple)):
        return [protected_content(item, secrets) for item in value]
    if isinstance(value, bytes):
        return {"content_status": "binary_reference_required", "size_bytes": len(value)}
    if isinstance(value, str):
        if value.lower().startswith("data:"):
            return {"content_status": "binary_reference_required", "encoded_size_bytes": len(value.encode("utf-8"))}
        result = value
        for secret in secrets:
            if secret:
                result = result.replace(secret, "[credential filtered]")
        return _CREDENTIAL_TEXT.sub("[credential filtered]", result)
    return value


def environment_credentials() -> tuple[str, ...]:
    """取得进程中已配置的凭据值，供专用内容过滤使用。"""
    return tuple(value for key, value in os.environ.items() if _CREDENTIAL_KEY.search(key) and value)


def record_content(span: trace.Span, name: str, value: Any, *, secrets: tuple[str, ...] = ()) -> None:
    """向当前调查 span 写入已取得的结构化内容，不生成二进制副本。"""
    if span.is_recording():
        span.set_attribute(name, json.dumps(protected_content(value, secrets + environment_credentials()), ensure_ascii=False, default=str))
        span.set_attribute("content.source", "python_parsed")
        span.set_attribute("content.completeness", "available_fields")


class _ProtectedExporter(SpanExporter):
    """导出前保护自动 instrumentation 属性并记录实际发送状态。"""

    def __init__(self, delegate: SpanExporter):
        self.delegate = delegate
        self.last_success = 0.0
        self.last_failure = 0.0

    def export(self, spans: Any) -> SpanExportResult:
        """处理 SDK 已完成的 span；发送失败和业务结果分别记录。"""
        from opentelemetry.sdk.trace import ReadableSpan

        secrets = environment_credentials()
        cleaned = [ReadableSpan(
            name=span.name, context=span.context, parent=span.parent,
            resource=span.resource, attributes={**protected_content(dict(span.attributes or {}), secrets), "content.dropped_attributes": span.dropped_attributes, "content.dropped_events": span.dropped_events},
            events=[Event(event.name, protected_content(dict(event.attributes or {}), secrets), event.timestamp) for event in span.events],
            links=span.links, kind=span.kind, status=Status(span.status.status_code, protected_content(span.status.description, secrets)),
            start_time=span.start_time, end_time=span.end_time, instrumentation_scope=span.instrumentation_scope,
        ) for span in spans]
        try:
            result = self.delegate.export(cleaned)
        except Exception as error:
            self.last_failure = time.time()
            logger.error("telemetry_trace_export_error: {}: {}", type(error).__name__, protected_content(str(error), secrets))
            raise
        if result == SpanExportResult.SUCCESS:
            self.last_success = time.time()
        else:
            self.last_failure = time.time()
            logger.error("telemetry_trace_export_failed: OTLP exporter 返回 {}", result.name)
        return result

    def shutdown(self) -> None:
        """关闭 SDK 管理的 HTTP exporter。"""
        self.delegate.shutdown()


class _MetricExporter(OTLPMetricExporter):
    """在标准 metric exporter 上保留实际导出成功与失败时间。"""

    last_success = 0.0
    last_failure = 0.0

    def export(self, metrics_data: Any, timeout_millis: float = 10000, **kwargs: Any) -> MetricExportResult:
        """通过官方 exporter 发送 metrics 并报告发送状态。"""
        try:
            result = super().export(metrics_data, timeout_millis=timeout_millis, **kwargs)
        except Exception as error:
            self.last_failure = time.time()
            logger.error("telemetry_metric_export_error: {}: {}", type(error).__name__, protected_content(str(error), environment_credentials()))
            raise
        if result == MetricExportResult.SUCCESS:
            self.last_success = time.time()
        else:
            self.last_failure = time.time()
            logger.error("telemetry_metric_export_failed: OTLP exporter 返回 {}", result.name)
        return result


class _LogExporter(OTLPLogExporter):
    """保存标准日志导出的成功与错误时间，供来源健康指标查询。"""

    last_success = 0.0
    last_failure = 0.0

    def export(self, batch: Any) -> LogRecordExportResult:
        """导出已保护的结构化事件，明确保留网络错误原因。"""
        try:
            result = super().export(batch)
        except Exception as error:
            self.last_failure = time.time()
            logger.error("telemetry_log_export_error: {}: {}", type(error).__name__, protected_content(str(error), environment_credentials()))
            raise
        if result == LogRecordExportResult.SUCCESS:
            self.last_success = time.time()
        else:
            self.last_failure = time.time()
            logger.error("telemetry_log_export_failed: OTLP exporter 返回 {}", result.name)
        return result


class ExceptionRecordProcessor(SpanProcessor):
    """将已保存的 SDK 异常事件按类型统计，日志保留单条记录的调查关联。"""

    def __init__(self, meter_provider: MeterProvider, logger_provider: LoggerProvider, secrets: tuple[str, ...] = ()):
        """在 SDK 初始化时取得内存测量与日志入口，不申请业务连接。"""
        meter = meter_provider.get_meter("mememeow.exceptions")
        self.counter = meter.create_counter("runtime_exceptions", unit="{record}")
        self.last_seen: dict[tuple[tuple[str, str], ...], float] = {}
        self.lock = Lock()
        self.logger = logger_provider.get_logger("mememeow.exceptions")
        self.secrets = secrets
        meter.create_observable_gauge("runtime_exception_last_seen", unit="s", callbacks=[self.observations])

    def observations(self, options) -> list[metrics.Observation]:
        """后台导出读取事件时间快照，标签只包含异常类别和有限操作身份。"""
        with self.lock:
            return [metrics.Observation(timestamp, dict(labels)) for labels, timestamp in self.last_seen.items()]

    def on_end(self, span: ReadableSpan) -> None:
        """span 完成时计数每条 exception 事件，并发送经凭据过滤的调查日志。"""
        attributes = span.attributes or {}
        for event in span.events:
            if event.name != "exception":
                continue
            exception_type = (event.attributes or {}).get("exception.type")
            if not isinstance(exception_type, str) or not exception_type:
                raise ValueError("telemetry_exception_type_missing: exception 事件缺少有效 exception.type")
            labels = {
                "exception_type": exception_type,
                "operation": span.name,
                "span_kind": span.kind.name,
                "http_route": attributes.get("http.route", "unavailable"),
                "http_request_method": attributes.get("http.request.method", attributes.get("http.method", "unavailable")),
            }
            self.counter.add(1, labels)
            key = tuple(sorted(labels.items()))
            observed_at = event.timestamp / 1_000_000_000
            with self.lock:
                self.last_seen[key] = max(observed_at, self.last_seen.get(key, observed_at))
            context = span.context
            content = {**labels, "trace_id": format(context.trace_id, "032x"), "span_id": format(context.span_id, "016x"), "observed_at": observed_at, "message": (event.attributes or {}).get("exception.message", "")}
            self.logger.emit(timestamp=event.timestamp, context=trace.set_span_in_context(trace.NonRecordingSpan(context)), body=json.dumps(protected_content(content, self.secrets + environment_credentials()), ensure_ascii=False), attributes={"source": "exception_events"})


class TelemetryRuntime:
    """管理 SDK 生命周期，提供内存测量与后台导出状态。"""

    def __init__(self, config: TelemetryConfig):
        self.config = config
        self.closed = False
        headers = {"Authorization": config.authorization} if config.authorization else {}
        self.instance_id = str(uuid4())
        resource = Resource.create({"service.name": config.service, "service.instance.id": self.instance_id, "deployment.environment.name": config.environment})
        self.span_exporter = _ProtectedExporter(OTLPSpanExporter(endpoint=f"{config.endpoint}/v1/traces", timeout=5, headers=headers, session=otlp_session()))
        self.tracer_provider = TracerProvider(resource=resource, sampler=ALWAYS_ON, span_limits=SpanLimits(max_attributes=1024, max_events=1024, max_attribute_length=None))
        self.tracer_provider.add_span_processor(BatchSpanProcessor(self.span_exporter, max_queue_size=2048, max_export_batch_size=128, schedule_delay_millis=2000, export_timeout_millis=5000))
        self.metric_exporter = _MetricExporter(endpoint=f"{config.endpoint}/v1/metrics", timeout=5, headers=headers, session=otlp_session())
        reader = PeriodicExportingMetricReader(self.metric_exporter, export_interval_millis=10000, export_timeout_millis=5000)
        self.meter_provider = MeterProvider(resource=resource, metric_readers=[reader], views=duration_views())
        self.log_exporter = _LogExporter(endpoint=f"{config.endpoint}/v1/logs", timeout=5, headers=headers, session=otlp_session())
        self.logger_provider = LoggerProvider(resource=resource, meter_provider=self.meter_provider)
        self.logger_provider.add_log_record_processor(BatchLogRecordProcessor(self.log_exporter, max_queue_size=2048, max_export_batch_size=128, schedule_delay_millis=2000, export_timeout_millis=5000))
        self.tracer_provider.add_span_processor(ExceptionRecordProcessor(self.meter_provider, self.logger_provider, (config.authorization,)))
        meter = self.meter_provider.get_meter("mememeow.runtime")
        signals = {"traces": self.span_exporter, "metrics": self.metric_exporter, "logs": self.log_exporter}
        meter.create_observable_gauge("telemetry_export_last_success", unit="s", callbacks=[lambda options: [metrics.Observation(exporter.last_success, {"signal": signal}) for signal, exporter in signals.items()]])
        meter.create_observable_gauge("telemetry_export_last_failure", unit="s", callbacks=[lambda options: [metrics.Observation(exporter.last_failure, {"signal": signal}) for signal, exporter in signals.items()]])

    def close(self) -> None:
        """等待有界导出并关闭 SDK，关闭错误由调用位置报告。"""
        if self.closed:
            return
        self.closed = True
        self.tracer_provider.shutdown()
        self.logger_provider.shutdown()
        self.meter_provider.shutdown()


def record_cache_decision(*, provider: str, result: str) -> None:
    """在缓存选择完成时记录一次结果，Provider 执行保留独立计数。"""
    if result not in {"hit", "miss", "refresh"}:
        raise ValueError("cache_decision_result_invalid")
    metrics.get_meter("mememeow.runtime").create_counter("runtime_cache_decisions", unit="{decision}").add(1, {"cache": "reverse_image", "provider": provider, "result": result})


def start_telemetry(service: str = "mememeow-api") -> TelemetryRuntime | None:
    """应用启动时创建一次进程 SDK；关闭配置具有明确状态。"""
    global _RUNTIME
    config = TelemetryConfig.from_env(service)
    if not config.enabled:
        return None
    with _LOCK:
        if _RUNTIME is None:
            _RUNTIME = TelemetryRuntime(config)
            trace.set_tracer_provider(_RUNTIME.tracer_provider)
            metrics.set_meter_provider(_RUNTIME.meter_provider)
            HTTPXClientInstrumentor().instrument()
            HTTPX2ClientInstrumentor().instrument()
            RequestsInstrumentor().instrument()
            URLLibInstrumentor().instrument()
            ThreadingInstrumentor().instrument()
        elif _RUNTIME.closed:
            raise RuntimeError("telemetry_runtime_closed: SDK 已随进程生命周期关闭")
        elif _RUNTIME.config != config:
            raise RuntimeError("telemetry_runtime_configuration_changed: 同一进程必须使用相同 SDK 配置")
        return _RUNTIME


def emit_event(source: str, event: Mapping[str, Any], *, attributes: Mapping[str, Any] | None = None) -> None:
    """从已取得的内存快照发送结构化日志，后台导出不占用业务连接。"""
    if _RUNTIME is None or _RUNTIME.closed:
        return
    secrets = environment_credentials() + (_RUNTIME.config.authorization,)
    _RUNTIME.logger_provider.get_logger("mememeow.events").emit(
        timestamp=time.time_ns(),
        body=json.dumps(protected_content(dict(event), secrets), ensure_ascii=False, default=str),
        attributes={"source": source, **protected_content(dict(attributes or {}), secrets)},
    )


def trace_carrier() -> dict[str, str]:
    """生成 W3C 上下文，执行协议的授权仍由现有凭据负责。"""
    carrier: dict[str, str] = {}
    _PROPAGATOR.inject(carrier)
    return carrier


def correlation_attributes() -> dict[str, str]:
    """从当前 SDK span 复制受信调查字段，供已有执行快照传递。"""
    span = trace.get_current_span()
    attributes = span.attributes if isinstance(span, ReadableSpan) else {}
    return {key: value for key, value in (attributes or {}).items() if key in {"account.id", "scope.id"} and isinstance(value, str)}


def telemetry_snapshot() -> dict[str, Any]:
    """复制当前受信追踪及调查字段，供业务已有持久记录恢复执行关联。"""
    return {"trace_context": trace_carrier(), "attributes": correlation_attributes()}


@contextmanager
def restored_operation(operation: str, *, snapshot: Mapping[str, Any] | None, attributes: Mapping[str, Any]) -> Iterator[trace.Span]:
    """从持久快照创建独立后台 span，前序请求作为 link，身份字段仅用于调查。"""
    saved = snapshot or {}
    parent = trace.get_current_span(extract_carrier(saved.get("trace_context", {}))).get_span_context()
    links = [trace.Link(parent)] if parent.is_valid else []
    correlation = {key: value for key, value in saved.get("attributes", {}).items() if key in {"account.id", "scope.id"} and isinstance(value, str)}
    with trace.get_tracer("mememeow.runtime").start_as_current_span(operation, context=Context(), links=links, attributes={**correlation, **attributes}) as span:
        yield span


def extract_carrier(carrier: Mapping[str, str]) -> Context:
    """读取仅用于关联的 W3C traceparent 与 tracestate。"""
    return _PROPAGATOR.extract(dict(carrier))


@contextmanager
def observed_operation(operation: str, *, unit: str, attributes: Mapping[str, Any] | None = None, context: Context | None = None) -> Iterator[trace.Span]:
    """测量已知执行边界，保留具体错误并继续原调用的异常行为。"""
    started = time.monotonic()
    meter = metrics.get_meter("mememeow.runtime")
    labels = {"operation": operation, "unit": unit, **{key: value for key, value in (attributes or {}).items() if key in {"provider", "model"} and isinstance(value, str)}}
    with trace.get_tracer("mememeow.runtime").start_as_current_span(operation, context=context, attributes=protected_content({**correlation_attributes(), **dict(attributes or {})}, environment_credentials())) as span:
        result = "succeeded"
        try:
            yield span
        except BaseException as error:
            result = "timeout" if isinstance(error, (TimeoutError, TimeoutException, RequestsTimeout)) else "failed"
            span.set_status(Status(StatusCode.ERROR, protected_content(str(error), environment_credentials())))
            span.set_attribute("error.type", type(error).__name__)
            raise
        finally:
            span.set_attribute("observation.unit", unit)
            span.set_attribute("operation.result", result)
            meter.create_counter("runtime_operations", unit="{operation}").add(1, {**labels, "result": result})
            meter.create_histogram("runtime_operation_duration", unit="s").record(time.monotonic() - started, labels)