监控

指标采集与导出集成。

使用示例

from symphra_cache import CacheManager, CacheMonitor
from symphra_cache.monitoring.prometheus import PrometheusExporter
from symphra_cache.monitoring.statsd import StatsDExporter

# 创建缓存与监控器
cache = CacheManager.from_config({"backend": "memory"})
monitor = CacheMonitor(cache)

# 执行一些操作
cache.set("user:1", {"name": "张三"})
cache.get("user:1")

# 统一指标接口
metrics = monitor.metrics
print(metrics.get_latency_stats("get"))  # {"min": ..., "max": ..., "avg": ...}

# Prometheus 导出（文本格式）
prom = PrometheusExporter(monitor, namespace="myapp", subsystem="cache")
print(prom.generate_metrics())

# StatsD 导出（发送到服务器）
# 注意：调用 send_metrics() 需要可达的 StatsD 服务器
statsd = StatsDExporter(monitor, prefix="myapp.cache")
# await statsd.send_metrics()  # 在异步上下文中调用

指标接口说明

• CacheMonitor.is_enabled() 控制监控开关（关闭时几乎零开销）。
• CacheMonitor.metrics 提供导出器期望的字段：get_count、set_count、delete_count、hit_count、miss_count。
• get_hit_rate() 与 get_total_operations() 返回命中率与总操作数。
• get_average_latency(operation) 返回 get/set 的平均延迟（毫秒）。
• get_latency_stats(operation) 返回 {min, max, avg} 的延迟统计（毫秒）。

缓存监控器

提供缓存统计、监控和健康检查功能。 线程安全。

使用示例： >>> cache = CacheManager.from_config({"backend": "memory"}) >>> monitor = CacheMonitor(cache) >>> >>> # 执行缓存操作 >>> cache.set("key", "value") >>> cache.get("key") >>> >>> # 查看统计 >>> stats = monitor.get_stats() >>> print(f"命中率: {stats.hit_rate:.2%}")

Source code in src/symphra_cache/monitor.py

class CacheMonitor:
    """
    缓存监控器

    提供缓存统计、监控和健康检查功能。
    线程安全。

    使用示例：
        >>> cache = CacheManager.from_config({"backend": "memory"})
        >>> monitor = CacheMonitor(cache)
        >>>
        >>> # 执行缓存操作
        >>> cache.set("key", "value")
        >>> cache.get("key")
        >>>
        >>> # 查看统计
        >>> stats = monitor.get_stats()
        >>> print(f"命中率: {stats.hit_rate:.2%}")
    """

    def __init__(self, cache_manager: CacheManager, *, enabled: bool = True) -> None:
        """
        初始化监控器

        Args:
            cache_manager: 缓存管理器实例
            enabled: 是否启用监控（禁用时性能开销为零）
        """
        self._cache = cache_manager
        # 为兼容导出器，公开 cache 属性
        self.cache = cache_manager

        self._enabled = enabled
        self._stats = CacheStats()
        self._lock = threading.RLock()

        # 记录延迟的 min/max（毫秒）以供导出器使用
        self._latency_min: dict[str, float] = {}
        self._latency_max: dict[str, float] = {}

        # 如果启用，替换缓存管理器的方法
        if self._enabled:
            self._wrap_cache_methods()

    def is_enabled(self) -> bool:
        """是否启用监控（为导出器兼容提供）"""
        return self._enabled

    @property
    def metrics(self) -> CacheMetricsAdapter:
        """提供与导出器兼容的指标接口"""
        return CacheMetricsAdapter(self._stats, self)

    def _wrap_cache_methods(self) -> None:
        """包装缓存管理器方法以收集统计信息"""
        original_get = self._cache.get
        original_set = self._cache.set
        original_delete = self._cache.delete

        def monitored_get(key):
            start = time.perf_counter()
            try:
                result = original_get(key)
                elapsed_s = time.perf_counter() - start
                latency_ms = elapsed_s * 1000.0
                with self._lock:
                    self._stats.gets += 1
                    self._stats.total_get_time += elapsed_s
                    # 更新 min/max（毫秒）
                    prev_min = self._latency_min.get("get")
                    prev_max = self._latency_max.get("get")
                    if prev_min is None or latency_ms < prev_min:
                        self._latency_min["get"] = latency_ms
                    if prev_max is None or latency_ms > prev_max:
                        self._latency_max["get"] = latency_ms
                    if result is not None:
                        self._stats.hits += 1
                    else:
                        self._stats.misses += 1
                return result
            except Exception as e:
                with self._lock:
                    self._stats.errors += 1
                raise e

        def monitored_set(key, value, ttl=None, ex=False, nx=False):
            start = time.perf_counter()
            try:
                result = original_set(key, value, ttl, ex, nx)
                elapsed_s = time.perf_counter() - start
                latency_ms = elapsed_s * 1000.0
                with self._lock:
                    self._stats.sets += 1
                    self._stats.total_set_time += elapsed_s
                    # 更新 min/max（毫秒）
                    prev_min = self._latency_min.get("set")
                    prev_max = self._latency_max.get("set")
                    if prev_min is None or latency_ms < prev_min:
                        self._latency_min["set"] = latency_ms
                    if prev_max is None or latency_ms > prev_max:
                        self._latency_max["set"] = latency_ms
                return result
            except Exception as e:
                with self._lock:
                    self._stats.errors += 1
                raise e

        def monitored_delete(key):
            try:
                result = original_delete(key)
                with self._lock:
                    self._stats.deletes += 1
                return result
            except Exception as e:
                with self._lock:
                    self._stats.errors += 1
                raise e

        # 替换方法
        self._cache.get = monitored_get
        self._cache.set = monitored_set
        self._cache.delete = monitored_delete

    def get_stats(self) -> CacheStats:
        """
        获取统计信息

        Returns:
            CacheStats 对象（副本）

        示例：
            >>> stats = monitor.get_stats()
            >>> print(f"命中率: {stats.hit_rate:.2%}")
            >>> print(f"平均响应时间: {stats.avg_get_time:.2f}ms")
        """
        with self._lock:
            # 返回副本
            return CacheStats(
                hits=self._stats.hits,
                misses=self._stats.misses,
                gets=self._stats.gets,
                sets=self._stats.sets,
                deletes=self._stats.deletes,
                errors=self._stats.errors,
                total_get_time=self._stats.total_get_time,
                total_set_time=self._stats.total_set_time,
                start_time=self._stats.start_time,
                last_reset=self._stats.last_reset,
            )

    def reset_stats(self) -> None:
        """
        重置统计信息

        保留 start_time，更新 last_reset。

        示例:
            >>> monitor.reset_stats()  # 重新开始统计
        """
        with self._lock:
            start_time = self._stats.start_time
            self._stats = CacheStats(start_time=start_time)
            # 清理延迟统计
            self._latency_min.clear()
            self._latency_max.clear()

    def check_health(self) -> dict:
        """
        执行健康检查

        Returns:
            健康检查结果字典

        示例：
            >>> health = monitor.check_health()
            >>> if health["healthy"]:
            ...     print("缓存健康")
        """
        try:
            # 测试基本操作
            test_key = "__health_check__"
            test_value = f"health_check_{time.time()}"

            # 测试写入
            self._cache.set(test_key, test_value, ttl=1)

            # 测试读取
            result = self._cache.get(test_key)
            read_ok = result == test_value

            # 清理
            self._cache.delete(test_key)

            # 获取后端健康状态
            backend_healthy = self._cache.backend.check_health()

            return {
                "healthy": read_ok and backend_healthy,
                "backend_healthy": backend_healthy,
                "test_passed": read_ok,
                "timestamp": time.time(),
            }
        except Exception as e:
            return {
                "healthy": False,
                "backend_healthy": False,
                "test_passed": False,
                "error": str(e),
                "timestamp": time.time(),
            }

    def get_summary(self) -> dict:
        """
        获取监控摘要

        返回统计信息和健康状态的汇总。

        Returns:
            摘要字典

        示例：
            >>> summary = monitor.get_summary()
            >>> print(summary)
        """
        stats = self.get_stats()
        health = self.check_health()

        return {
            "stats": stats.to_dict(),
            "health": health,
        }

    def __repr__(self) -> str:
        """字符串表示"""
        stats = self.get_stats()
        return (
            f"CacheMonitor(enabled={self._enabled}, "
            f"hit_rate={stats.hit_rate:.2%}, "
            f"operations={stats.gets + stats.sets + stats.deletes}, "
            f"errors={stats.errors})"
        )

metrics property

提供与导出器兼容的指标接口

__init__(cache_manager, *, enabled=True)

初始化监控器

Parameters:

Name	Type	Description	Default
cache_manager	CacheManager	缓存管理器实例	required
enabled	bool	是否启用监控（禁用时性能开销为零）	True

Source code in src/symphra_cache/monitor.py

def __init__(self, cache_manager: CacheManager, *, enabled: bool = True) -> None:
    """
    初始化监控器

    Args:
        cache_manager: 缓存管理器实例
        enabled: 是否启用监控（禁用时性能开销为零）
    """
    self._cache = cache_manager
    # 为兼容导出器，公开 cache 属性
    self.cache = cache_manager

    self._enabled = enabled
    self._stats = CacheStats()
    self._lock = threading.RLock()

    # 记录延迟的 min/max（毫秒）以供导出器使用
    self._latency_min: dict[str, float] = {}
    self._latency_max: dict[str, float] = {}

    # 如果启用，替换缓存管理器的方法
    if self._enabled:
        self._wrap_cache_methods()

__repr__()

字符串表示

Source code in src/symphra_cache/monitor.py

def __repr__(self) -> str:
    """字符串表示"""
    stats = self.get_stats()
    return (
        f"CacheMonitor(enabled={self._enabled}, "
        f"hit_rate={stats.hit_rate:.2%}, "
        f"operations={stats.gets + stats.sets + stats.deletes}, "
        f"errors={stats.errors})"
    )

check_health()

执行健康检查

Returns:

Type	Description
dict	健康检查结果字典

示例： >>> health = monitor.check_health() >>> if health["healthy"]: ... print("缓存健康")

Source code in src/symphra_cache/monitor.py

def check_health(self) -> dict:
    """
    执行健康检查

    Returns:
        健康检查结果字典

    示例：
        >>> health = monitor.check_health()
        >>> if health["healthy"]:
        ...     print("缓存健康")
    """
    try:
        # 测试基本操作
        test_key = "__health_check__"
        test_value = f"health_check_{time.time()}"

        # 测试写入
        self._cache.set(test_key, test_value, ttl=1)

        # 测试读取
        result = self._cache.get(test_key)
        read_ok = result == test_value

        # 清理
        self._cache.delete(test_key)

        # 获取后端健康状态
        backend_healthy = self._cache.backend.check_health()

        return {
            "healthy": read_ok and backend_healthy,
            "backend_healthy": backend_healthy,
            "test_passed": read_ok,
            "timestamp": time.time(),
        }
    except Exception as e:
        return {
            "healthy": False,
            "backend_healthy": False,
            "test_passed": False,
            "error": str(e),
            "timestamp": time.time(),
        }

get_stats()

获取统计信息

Returns:

Type	Description
CacheStats	CacheStats 对象（副本）

示例： >>> stats = monitor.get_stats() >>> print(f"命中率: {stats.hit_rate:.2%}") >>> print(f"平均响应时间: {stats.avg_get_time:.2f}ms")

Source code in src/symphra_cache/monitor.py

def get_stats(self) -> CacheStats:
    """
    获取统计信息

    Returns:
        CacheStats 对象（副本）

    示例：
        >>> stats = monitor.get_stats()
        >>> print(f"命中率: {stats.hit_rate:.2%}")
        >>> print(f"平均响应时间: {stats.avg_get_time:.2f}ms")
    """
    with self._lock:
        # 返回副本
        return CacheStats(
            hits=self._stats.hits,
            misses=self._stats.misses,
            gets=self._stats.gets,
            sets=self._stats.sets,
            deletes=self._stats.deletes,
            errors=self._stats.errors,
            total_get_time=self._stats.total_get_time,
            total_set_time=self._stats.total_set_time,
            start_time=self._stats.start_time,
            last_reset=self._stats.last_reset,
        )

get_summary()

获取监控摘要

返回统计信息和健康状态的汇总。

Returns:

Type	Description
dict	摘要字典

示例： >>> summary = monitor.get_summary() >>> print(summary)

Source code in src/symphra_cache/monitor.py

def get_summary(self) -> dict:
    """
    获取监控摘要

    返回统计信息和健康状态的汇总。

    Returns:
        摘要字典

    示例：
        >>> summary = monitor.get_summary()
        >>> print(summary)
    """
    stats = self.get_stats()
    health = self.check_health()

    return {
        "stats": stats.to_dict(),
        "health": health,
    }

is_enabled()

是否启用监控（为导出器兼容提供）

Source code in src/symphra_cache/monitor.py

def is_enabled(self) -> bool:
    """是否启用监控（为导出器兼容提供）"""
    return self._enabled

reset_stats()

重置统计信息

保留 start_time，更新 last_reset。

示例

monitor.reset_stats() # 重新开始统计

Source code in src/symphra_cache/monitor.py

def reset_stats(self) -> None:
    """
    重置统计信息

    保留 start_time，更新 last_reset。

    示例:
        >>> monitor.reset_stats()  # 重新开始统计
    """
    with self._lock:
        start_time = self._stats.start_time
        self._stats = CacheStats(start_time=start_time)
        # 清理延迟统计
        self._latency_min.clear()
        self._latency_max.clear()

缓存统计信息

所有计数器都是累积值，可通过 reset() 重置。

Source code in src/symphra_cache/monitor.py

@dataclass
class CacheStats:
    """
    缓存统计信息

    所有计数器都是累积值，可通过 reset() 重置。
    """

    # 命中统计
    hits: int = 0  # 缓存命中次数
    misses: int = 0  # 缓存未命中次数

    # 操作统计
    gets: int = 0  # get 操作次数
    sets: int = 0  # set 操作次数
    deletes: int = 0  # delete 操作次数

    # 错误统计
    errors: int = 0  # 错误次数

    # 性能统计
    total_get_time: float = 0.0  # get 操作总耗时（秒）
    total_set_time: float = 0.0  # set 操作总耗时（秒）

    # 时间戳
    start_time: float = field(default_factory=time.time)  # 统计开始时间
    last_reset: float = field(default_factory=time.time)  # 上次重置时间

    @property
    def hit_rate(self) -> float:
        """命中率（0-1）"""
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0

    @property
    def miss_rate(self) -> float:
        """未命中率(0-1)"""
        total = self.hits + self.misses
        return self.misses / total if total > 0 else 0.0

    @property
    def avg_get_time(self) -> float:
        """平均 get 操作耗时（毫秒）"""
        return (self.total_get_time / self.gets * 1000) if self.gets > 0 else 0.0

    @property
    def avg_set_time(self) -> float:
        """平均 set 操作耗时（毫秒）"""
        return (self.total_set_time / self.sets * 1000) if self.sets > 0 else 0.0

    @property
    def uptime(self) -> float:
        """运行时间（秒）"""
        return time.time() - self.start_time

    def to_dict(self) -> dict:
        """转换为字典"""
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": self.hit_rate,
            "miss_rate": self.miss_rate,
            "gets": self.gets,
            "sets": self.sets,
            "deletes": self.deletes,
            "errors": self.errors,
            "avg_get_time_ms": self.avg_get_time,
            "avg_set_time_ms": self.avg_set_time,
            "uptime_seconds": self.uptime,
            "start_time": self.start_time,
            "last_reset": self.last_reset,
        }

avg_get_time property

平均 get 操作耗时（毫秒）

avg_set_time property

平均 set 操作耗时（毫秒）

hit_rate property

命中率（0-1）

miss_rate property

未命中率(0-1)

uptime property

运行时间（秒）

to_dict()

转换为字典

Source code in src/symphra_cache/monitor.py

def to_dict(self) -> dict:
    """转换为字典"""
    return {
        "hits": self.hits,
        "misses": self.misses,
        "hit_rate": self.hit_rate,
        "miss_rate": self.miss_rate,
        "gets": self.gets,
        "sets": self.sets,
        "deletes": self.deletes,
        "errors": self.errors,
        "avg_get_time_ms": self.avg_get_time,
        "avg_set_time_ms": self.avg_set_time,
        "uptime_seconds": self.uptime,
        "start_time": self.start_time,
        "last_reset": self.last_reset,
    }

Prometheus 指标导出器

将缓存监控指标转换为 Prometheus 格式。

支持的指标类型： - Counter: 累积计数器（操作次数） - Gauge: 瞬时值（缓存大小、命中率） - Histogram: 分布统计（延迟分布）

使用示例： >>> exporter = PrometheusExporter(monitor) >>> metrics_text = exporter.generate_metrics()

Source code in src/symphra_cache/monitoring/prometheus.py

class PrometheusExporter:
    """
    Prometheus 指标导出器

    将缓存监控指标转换为 Prometheus 格式。

    支持的指标类型：
    - Counter: 累积计数器（操作次数）
    - Gauge: 瞬时值（缓存大小、命中率）
    - Histogram: 分布统计（延迟分布）

    使用示例：
        >>> exporter = PrometheusExporter(monitor)
        >>> metrics_text = exporter.generate_metrics()
    """

    def __init__(
        self,
        monitor: CacheMonitor,
        namespace: str = "symphra_cache",
        subsystem: str = "cache",
        labels: dict[str, str] | None = None,
    ) -> None:
        """
        初始化 Prometheus 导出器

        Args:
            monitor: 缓存监控器
            namespace: 指标命名空间
            subsystem: 子系统名称
            labels: 全局标签
        """
        self.monitor = monitor
        self.namespace = namespace
        self.subsystem = subsystem
        self.labels = labels or {}
        self._start_time = time.time()

    def _format_labels(self, extra_labels: dict[str, str] | None = None) -> str:
        """
        格式化标签

        Args:
            extra_labels: 额外标签

        Returns:
            格式化的标签字符串
        """
        all_labels = self.labels.copy()
        if extra_labels:
            all_labels.update(extra_labels)

        if not all_labels:
            return ""

        label_strs = []
        for key, value in all_labels.items():
            # 转义特殊字符
            escaped_value = str(value).replace('"', '\\"').replace("\n", "\\n")
            label_strs.append(f'{key}="{escaped_value}"')

        return "{" + ",".join(label_strs) + "}"

    def _generate_counter_metrics(self) -> str:
        """
        生成 Counter 指标

        Returns:
            Counter 指标文本
        """
        metrics = self.monitor.metrics
        lines = []

        # 操作计数器
        lines.append(f"# HELP {self._metric_name('operations_total')} Total cache operations")
        lines.append(f"# TYPE {self._metric_name('operations_total')} counter")

        operations = [
            ("get", metrics.get_count),
            ("set", metrics.set_count),
            ("delete", metrics.delete_count),
            ("hit", metrics.hit_count),
            ("miss", metrics.miss_count),
        ]

        for operation, count in operations:
            if count > 0:
                labels = self._format_labels({"operation": operation})
                lines.append(f"{self._metric_name('operations_total')}{labels} {count}")

        return "\n".join(lines)

    def _generate_gauge_metrics(self) -> str:
        """
        生成 Gauge 指标

        Returns:
            Gauge 指标文本
        """
        metrics = self.monitor.metrics
        lines = []

        # 缓存大小
        lines.append(f"# HELP {self._metric_name('size')} Current cache size")
        lines.append(f"# TYPE {self._metric_name('size')} gauge")
        try:
            cache_size = len(self.monitor.cache)
            lines.append(f"{self._metric_name('size')}{self._format_labels()} {cache_size}")
        except Exception:
            lines.append(f"{self._metric_name('size')}{self._format_labels()} 0")

        # 命中率
        lines.append(f"# HELP {self._metric_name('hit_rate')} Cache hit rate")
        lines.append(f"# TYPE {self._metric_name('hit_rate')} gauge")
        hit_rate = metrics.get_hit_rate()
        lines.append(f"{self._metric_name('hit_rate')}{self._format_labels()} {hit_rate}")

        # 运行时间
        lines.append(f"# HELP {self._metric_name('uptime_seconds')} Cache uptime in seconds")
        lines.append(f"# TYPE {self._metric_name('uptime_seconds')} gauge")
        uptime = time.time() - self._start_time
        lines.append(f"{self._metric_name('uptime_seconds')}{self._format_labels()} {uptime}")

        return "\n".join(lines)

    def _generate_histogram_metrics(self) -> str:
        """
        生成 Histogram 指标

        Returns:
            Histogram 指标文本
        """
        metrics = self.monitor.metrics
        lines = []

        # GET 操作延迟分布
        lines.append(
            f"# HELP {self._metric_name('get_duration_seconds')} Time spent on GET operations"
        )
        lines.append(f"# TYPE {self._metric_name('get_duration_seconds')} histogram")

        # Prometheus histogram buckets (秒)
        buckets = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]

        get_count = metrics.get_count
        if get_count > 0:
            avg_latency = metrics.get_average_latency("get") / 1000  # 转换为秒

            # 生成 bucket 计数（简化实现）
            for bucket in buckets:
                # 假设正态分布，计算 bucket 内的请求数
                bucket_count = int(
                    get_count * self._normal_cdf(bucket, avg_latency, avg_latency * 0.5)
                )
                labels = self._format_labels({"le": str(bucket)})
                lines.append(
                    f"{self._metric_name('get_duration_seconds')}_bucket{labels} {bucket_count}"
                )

            # 总计数和总和
            lines.append(
                f"{self._metric_name('get_duration_seconds')}_count{self._format_labels()} {get_count}"
            )
            lines.append(
                f"{self._metric_name('get_duration_seconds')}_sum{self._format_labels()} {avg_latency * get_count}"
            )

        # SET 操作延迟分布
        lines.append(
            f"# HELP {self._metric_name('set_duration_seconds')} Time spent on SET operations"
        )
        lines.append(f"# TYPE {self._metric_name('set_duration_seconds')} histogram")

        set_count = metrics.set_count
        if set_count > 0:
            avg_latency = metrics.get_average_latency("set") / 1000

            for bucket in buckets:
                bucket_count = int(
                    set_count * self._normal_cdf(bucket, avg_latency, avg_latency * 0.5)
                )
                labels = self._format_labels({"le": str(bucket)})
                lines.append(
                    f"{self._metric_name('set_duration_seconds')}_bucket{labels} {bucket_count}"
                )

            lines.append(
                f"{self._metric_name('set_duration_seconds')}_count{self._format_labels()} {set_count}"
            )
            lines.append(
                f"{self._metric_name('set_duration_seconds')}_sum{self._format_labels()} {avg_latency * set_count}"
            )

        return "\n".join(lines)

    def _normal_cdf(self, x: float, mean: float, std: float) -> float:
        """
        正态分布累积分布函数（简化实现）

        Args:
            x: 输入值
            mean: 均值
            std: 标准差

        Returns:
            CDF 值
        """
        import math

        return 0.5 * (1 + math.erf((x - mean) / (std * math.sqrt(2))))

    def _metric_name(self, name: str) -> str:
        """
        生成完整的指标名称

        Args:
            name: 指标名称

        Returns:
            完整的指标名称
        """
        parts = []
        if self.namespace:
            parts.append(self.namespace)
        if self.subsystem:
            parts.append(self.subsystem)
        parts.append(name)
        return "_".join(parts)

    def generate_metrics(self) -> str:
        """
        生成 Prometheus 格式的指标文本

        Returns:
            Prometheus 指标文本
        """
        if not self.monitor.is_enabled():
            return "# Cache monitoring is disabled"

        lines = []

        # 添加元信息
        lines.append(f"# Symphra Cache Metrics - {time.strftime('%Y-%m-%d %H:%M:%S')}")
        lines.append("# Generated by PrometheusExporter")
        lines.append("")

        # 生成不同类型的指标
        lines.append(self._generate_counter_metrics())
        lines.append("")
        lines.append(self._generate_gauge_metrics())
        lines.append("")
        lines.append(self._generate_histogram_metrics())

        return "\n".join(lines)

    def get_metrics_handler(self) -> Callable[[], str]:
        """
        获取指标处理器函数

        Returns:
            返回指标文本的函数
        """
        return self.generate_metrics

    def create_pushgateway_client(
        self,
        gateway_url: str,
        job_name: str,
        instance: str = "",
    ) -> PrometheusPushgatewayClient:
        """
        创建 Pushgateway 客户端

        Args:
            gateway_url: Pushgateway URL
            job_name: 作业名称
            instance: 实例标识符

        Returns:
            Pushgateway 客户端
        """
        return PrometheusPushgatewayClient(
            exporter=self,
            gateway_url=gateway_url,
            job_name=job_name,
            instance=instance or self._get_default_instance(),
        )

    def _get_default_instance(self) -> str:
        """
        获取默认实例标识符

        Returns:
            实例标识符
        """
        import os
        import socket

        hostname = socket.gethostname()
        pid = os.getpid()
        return f"{hostname}:{pid}"

    def update_labels(self, labels: dict[str, str]) -> None:
        """
        更新全局标签

        Args:
            labels: 新的标签字典
        """
        self.labels.update(labels)

__init__(monitor, namespace='symphra_cache', subsystem='cache', labels=None)

初始化 Prometheus 导出器

Parameters:

Name	Type	Description	Default
monitor	CacheMonitor	缓存监控器	required
namespace	str	指标命名空间	'symphra_cache'
subsystem	str	子系统名称	'cache'
labels	dict[str, str] | None	全局标签	None

Source code in src/symphra_cache/monitoring/prometheus.py

def __init__(
    self,
    monitor: CacheMonitor,
    namespace: str = "symphra_cache",
    subsystem: str = "cache",
    labels: dict[str, str] | None = None,
) -> None:
    """
    初始化 Prometheus 导出器

    Args:
        monitor: 缓存监控器
        namespace: 指标命名空间
        subsystem: 子系统名称
        labels: 全局标签
    """
    self.monitor = monitor
    self.namespace = namespace
    self.subsystem = subsystem
    self.labels = labels or {}
    self._start_time = time.time()

create_pushgateway_client(gateway_url, job_name, instance='')

创建 Pushgateway 客户端

Parameters:

Name	Type	Description	Default
gateway_url	str	Pushgateway URL	required
job_name	str	作业名称	required
instance	str	实例标识符	''

Returns:

Type	Description
PrometheusPushgatewayClient	Pushgateway 客户端

Source code in src/symphra_cache/monitoring/prometheus.py

def create_pushgateway_client(
    self,
    gateway_url: str,
    job_name: str,
    instance: str = "",
) -> PrometheusPushgatewayClient:
    """
    创建 Pushgateway 客户端

    Args:
        gateway_url: Pushgateway URL
        job_name: 作业名称
        instance: 实例标识符

    Returns:
        Pushgateway 客户端
    """
    return PrometheusPushgatewayClient(
        exporter=self,
        gateway_url=gateway_url,
        job_name=job_name,
        instance=instance or self._get_default_instance(),
    )

generate_metrics()

生成 Prometheus 格式的指标文本

Returns:

Type	Description
str	Prometheus 指标文本

Source code in src/symphra_cache/monitoring/prometheus.py

def generate_metrics(self) -> str:
    """
    生成 Prometheus 格式的指标文本

    Returns:
        Prometheus 指标文本
    """
    if not self.monitor.is_enabled():
        return "# Cache monitoring is disabled"

    lines = []

    # 添加元信息
    lines.append(f"# Symphra Cache Metrics - {time.strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append("# Generated by PrometheusExporter")
    lines.append("")

    # 生成不同类型的指标
    lines.append(self._generate_counter_metrics())
    lines.append("")
    lines.append(self._generate_gauge_metrics())
    lines.append("")
    lines.append(self._generate_histogram_metrics())

    return "\n".join(lines)

get_metrics_handler()

获取指标处理器函数

Returns:

Type	Description
Callable[[], str]	返回指标文本的函数

Source code in src/symphra_cache/monitoring/prometheus.py

def get_metrics_handler(self) -> Callable[[], str]:
    """
    获取指标处理器函数

    Returns:
        返回指标文本的函数
    """
    return self.generate_metrics

update_labels(labels)

更新全局标签

Parameters:

Name	Type	Description	Default
labels	dict[str, str]	新的标签字典	required

Source code in src/symphra_cache/monitoring/prometheus.py

def update_labels(self, labels: dict[str, str]) -> None:
    """
    更新全局标签

    Args:
        labels: 新的标签字典
    """
    self.labels.update(labels)

StatsD 指标导出器

将缓存监控指标转换为 StatsD 格式并通过 UDP 发送。

支持的指标类型： - Counter: 计数器（操作次数） - Timer: 计时器（延迟） - Gauge: 瞬时值（缓存大小、命中率）

使用示例： >>> exporter = StatsDExporter(monitor, host="localhost", port=8125) >>> await exporter.send_metrics()

Source code in src/symphra_cache/monitoring/statsd.py

class StatsDExporter:
    """
    StatsD 指标导出器

    将缓存监控指标转换为 StatsD 格式并通过 UDP 发送。

    支持的指标类型：
    - Counter: 计数器（操作次数）
    - Timer: 计时器（延迟）
    - Gauge: 瞬时值（缓存大小、命中率）

    使用示例：
        >>> exporter = StatsDExporter(monitor, host="localhost", port=8125)
        >>> await exporter.send_metrics()
    """

    def __init__(
        self,
        monitor: CacheMonitor,
        host: str = "localhost",
        port: int = 8125,
        prefix: str = "symphra.cache",
        sample_rate: float = 1.0,
        protocol: str = "udp",
        batch_size: int = 10,
    ) -> None:
        """
        初始化 StatsD 导出器

        Args:
            monitor: 缓存监控器
            host: StatsD 服务器主机
            port: StatsD 服务器端口
            prefix: 指标前缀
            sample_rate: 采样率 (0.0-1.0)
            protocol: 传输协议 ("udp", "tcp")
            batch_size: 批量发送大小
        """
        self.monitor = monitor
        self.host = host
        self.port = port
        self.prefix = prefix
        self.sample_rate = max(0.0, min(1.0, sample_rate))  # 确保在有效范围内
        self.protocol = protocol.lower()
        self.batch_size = batch_size
        self._socket: socket.socket | None = None
        self._tcp_writer: asyncio.StreamWriter | None = None
        self._tcp_reader: asyncio.StreamReader | None = None
        self._is_connected = False
        self._pending_metrics: list[str] = []
        self._lock = asyncio.Lock()

    async def connect(self) -> None:
        """
        建立到 StatsD 服务器的连接
        """
        if self._is_connected:
            return

        try:
            if self.protocol == "udp":
                self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                self._socket.setblocking(False)
            elif self.protocol == "tcp":
                reader, writer = await asyncio.open_connection(self.host, self.port)
                self._tcp_reader = reader
                self._tcp_writer = writer
            else:
                raise ValueError(f"不支持的协议: {self.protocol}")

            self._is_connected = True

        except Exception as e:
            print(f"连接 StatsD 服务器失败: {e}")
            self._is_connected = False

    async def disconnect(self) -> None:
        """
        断开连接
        """
        if self.protocol == "udp" and self._socket:
            self._socket.close()
        elif self.protocol == "tcp" and self._tcp_writer:
            self._tcp_writer.close()
            await self._tcp_writer.wait_closed()

        self._is_connected = False
        self._socket = None
        self._tcp_writer = None
        self._tcp_reader = None

    def _format_metric_name(self, name: str) -> str:
        """
        格式化指标名称

        Args:
            name: 原始指标名称

        Returns:
            格式化的指标名称
        """
        return f"{self.prefix}.{name}"

    def _generate_counter_metrics(self) -> list[str]:
        """
        生成计数器指标

        Returns:
            计数器指标列表
        """
        metrics = self.monitor.metrics
        metric_lines = []

        # 操作计数器
        operations = [
            ("get", metrics.get_count),
            ("set", metrics.set_count),
            ("delete", metrics.delete_count),
            ("hit", metrics.hit_count),
            ("miss", metrics.miss_count),
        ]

        for operation, count in operations:
            if count > 0:
                metric_name = self._format_metric_name(f"operations.{operation}")
                metric_lines.append(f"{metric_name}:{count}|c")

        return metric_lines

    def _generate_timer_metrics(self) -> list[str]:
        """
        生成计时器指标

        Returns:
            计时器指标列表
        """
        metrics = self.monitor.metrics
        metric_lines = []

        # GET 操作延迟
        if metrics.get_count > 0:
            avg_latency = metrics.get_average_latency("get")
            min_latency = metrics.get_latency_stats("get")["min"]
            max_latency = metrics.get_latency_stats("get")["max"]

            metric_lines.extend(
                [
                    f"{self._format_metric_name('get.latency.avg')}:{avg_latency:.3f}|ms",
                    f"{self._format_metric_name('get.latency.min')}:{min_latency:.3f}|ms",
                    f"{self._format_metric_name('get.latency.max')}:{max_latency:.3f}|ms",
                ]
            )

        # SET 操作延迟
        if metrics.set_count > 0:
            avg_latency = metrics.get_average_latency("set")
            min_latency = metrics.get_latency_stats("set")["min"]
            max_latency = metrics.get_latency_stats("set")["max"]

            metric_lines.extend(
                [
                    f"{self._format_metric_name('set.latency.avg')}:{avg_latency:.3f}|ms",
                    f"{self._format_metric_name('set.latency.min')}:{min_latency:.3f}|ms",
                    f"{self._format_metric_name('set.latency.max')}:{max_latency:.3f}|ms",
                ]
            )

        return metric_lines

    def _generate_gauge_metrics(self) -> list[str]:
        """
        生成 Gauge 指标

        Returns:
            Gauge 指标列表
        """
        metrics = self.monitor.metrics
        metric_lines = []

        # 缓存大小
        try:
            cache_size = len(self.monitor.cache)
            metric_lines.append(f"{self._format_metric_name('size')}:{cache_size}|g")
        except Exception:
            metric_lines.append(f"{self._format_metric_name('size')}:0|g")

        # 命中率
        hit_rate = metrics.get_hit_rate()
        metric_lines.append(f"{self._format_metric_name('hit_rate')}:{hit_rate:.3f}|g")

        # 总操作数
        total_ops = metrics.get_total_operations()
        metric_lines.append(f"{self._format_metric_name('operations.total')}:{total_ops}|g")

        return metric_lines

    async def _send_udp_metrics(self, metric_lines: list[str]) -> bool:
        """
        通过 UDP 发送指标

        Args:
            metric_lines: 指标行列表

        Returns:
            发送是否成功
        """
        if not self._socket or not self._is_connected:
            return False

        try:
            # 合并指标为单个数据报（注意 UDP 数据报大小限制）
            for i in range(0, len(metric_lines), 10):  # 每10个指标一个数据报
                batch = metric_lines[i : i + 10]
                if not batch:
                    continue

                message = "\n".join(batch).encode("utf-8")

                # 检查数据报大小（通常限制为 1500 字节）
                if len(message) > 1400:  # 留一些余量
                    # 分割大数据报
                    for line in batch:
                        if len(line.encode("utf-8")) <= 1400:
                            await asyncio.get_event_loop().sock_sendto(
                                self._socket, line.encode("utf-8"), (self.host, self.port)
                            )
                else:
                    await asyncio.get_event_loop().sock_sendto(
                        self._socket, message, (self.host, self.port)
                    )

            return True

        except Exception as e:
            print(f"UDP 发送失败: {e}")
            return False

    async def _send_tcp_metrics(self, metric_lines: list[str]) -> bool:
        """
        通过 TCP 发送指标

        Args:
            metric_lines: 指标行列表

        Returns:
            发送是否成功
        """
        if not self._tcp_writer or not self._is_connected:
            return False

        try:
            # 合并指标为单个消息
            message = "\n".join(metric_lines).encode("utf-8") + b"\n"
            self._tcp_writer.write(message)
            await self._tcp_writer.drain()
            return True

        except Exception as e:
            print(f"TCP 发送失败: {e}")
            return False

    async def send_metrics(self, metric_lines: list[str] | None = None) -> bool:
        """
        发送指标到 StatsD 服务器

        Args:
            metric_lines: 要发送的指标行列表，None 表示发送所有指标

        Returns:
            发送是否成功
        """
        if not self.monitor.is_enabled():
            return True

        if metric_lines is None:
            metric_lines = self.generate_all_metrics()

        if not metric_lines:
            return True

        # 建立连接
        if not self._is_connected:
            await self.connect()
            if not self._is_connected:
                return False

        # 应用采样率
        if self.sample_rate < 1.0:
            import random

            metric_lines = [line for line in metric_lines if random.random() < self.sample_rate]

        try:
            if self.protocol == "udp":
                return await self._send_udp_metrics(metric_lines)
            else:
                return await self._send_tcp_metrics(metric_lines)

        except Exception as e:
            print(f"发送指标失败: {e}")
            await self.disconnect()
            return False

    def generate_all_metrics(self) -> list[str]:
        """
        生成所有指标

        Returns:
            指标行列表
        """
        all_metrics = []
        all_metrics.extend(self._generate_counter_metrics())
        all_metrics.extend(self._generate_timer_metrics())
        all_metrics.extend(self._generate_gauge_metrics())
        return all_metrics

    async def schedule_periodic_send(self, interval: float = 30.0) -> None:
        """
        安排周期性发送指标

        Args:
            interval: 发送间隔（秒）
        """
        while True:
            try:
                await asyncio.sleep(interval)
                await self.send_metrics()
            except asyncio.CancelledError:
                break
            except Exception as e:
                print(f"周期性发送失败: {e}")

    def add_custom_metric(self, name: str, value: float, metric_type: str = "g") -> None:
        """
        添加自定义指标

        Args:
            name: 指标名称
            value: 指标值
            metric_type: 指标类型 ("c", "g", "ms")
        """
        metric_line = f"{self._format_metric_name(name)}:{value}|{metric_type}"
        self._pending_metrics.append(metric_line)

    async def flush_pending_metrics(self) -> bool:
        """
        刷新待发送的指标

        Returns:
            刷新是否成功
        """
        if not self._pending_metrics:
            return True

        success = await self.send_metrics(self._pending_metrics)
        if success:
            self._pending_metrics.clear()

        return success

    def get_connection_status(self) -> dict[str, Any]:
        """
        获取连接状态

        Returns:
            连接状态信息
        """
        return {
            "connected": self._is_connected,
            "protocol": self.protocol,
            "host": self.host,
            "port": self.port,
            "pending_metrics": len(self._pending_metrics),
        }

    async def __aenter__(self) -> StatsDExporter:
        """异步上下文管理器入口"""
        await self.connect()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: Any | None,
    ) -> None:
        """异步上下文管理器出口"""
        await self.disconnect()

    def __del__(self) -> None:
        """析构函数"""
        if self._socket:
            self._socket.close()

__aenter__() async

异步上下文管理器入口

Source code in src/symphra_cache/monitoring/statsd.py

async def __aenter__(self) -> StatsDExporter:
    """异步上下文管理器入口"""
    await self.connect()
    return self

__aexit__(exc_type, exc_val, exc_tb) async

异步上下文管理器出口

Source code in src/symphra_cache/monitoring/statsd.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any | None,
) -> None:
    """异步上下文管理器出口"""
    await self.disconnect()

__del__()

析构函数

Source code in src/symphra_cache/monitoring/statsd.py

def __del__(self) -> None:
    """析构函数"""
    if self._socket:
        self._socket.close()

__init__(monitor, host='localhost', port=8125, prefix='symphra.cache', sample_rate=1.0, protocol='udp', batch_size=10)

初始化 StatsD 导出器

Parameters:

Name	Type	Description	Default
monitor	CacheMonitor	缓存监控器	required
host	str	StatsD 服务器主机	'localhost'
port	int	StatsD 服务器端口	8125
prefix	str	指标前缀	'symphra.cache'
sample_rate	float	采样率 (0.0-1.0)	1.0
protocol	str	传输协议 ("udp", "tcp")	'udp'
batch_size	int	批量发送大小	10

Source code in src/symphra_cache/monitoring/statsd.py

def __init__(
    self,
    monitor: CacheMonitor,
    host: str = "localhost",
    port: int = 8125,
    prefix: str = "symphra.cache",
    sample_rate: float = 1.0,
    protocol: str = "udp",
    batch_size: int = 10,
) -> None:
    """
    初始化 StatsD 导出器

    Args:
        monitor: 缓存监控器
        host: StatsD 服务器主机
        port: StatsD 服务器端口
        prefix: 指标前缀
        sample_rate: 采样率 (0.0-1.0)
        protocol: 传输协议 ("udp", "tcp")
        batch_size: 批量发送大小
    """
    self.monitor = monitor
    self.host = host
    self.port = port
    self.prefix = prefix
    self.sample_rate = max(0.0, min(1.0, sample_rate))  # 确保在有效范围内
    self.protocol = protocol.lower()
    self.batch_size = batch_size
    self._socket: socket.socket | None = None
    self._tcp_writer: asyncio.StreamWriter | None = None
    self._tcp_reader: asyncio.StreamReader | None = None
    self._is_connected = False
    self._pending_metrics: list[str] = []
    self._lock = asyncio.Lock()

add_custom_metric(name, value, metric_type='g')

添加自定义指标

Parameters:

Name	Type	Description	Default
name	str	指标名称	required
value	float	指标值	required
metric_type	str	指标类型 ("c", "g", "ms")	'g'

Source code in src/symphra_cache/monitoring/statsd.py

def add_custom_metric(self, name: str, value: float, metric_type: str = "g") -> None:
    """
    添加自定义指标

    Args:
        name: 指标名称
        value: 指标值
        metric_type: 指标类型 ("c", "g", "ms")
    """
    metric_line = f"{self._format_metric_name(name)}:{value}|{metric_type}"
    self._pending_metrics.append(metric_line)

connect() async

建立到 StatsD 服务器的连接

Source code in src/symphra_cache/monitoring/statsd.py

async def connect(self) -> None:
    """
    建立到 StatsD 服务器的连接
    """
    if self._is_connected:
        return

    try:
        if self.protocol == "udp":
            self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self._socket.setblocking(False)
        elif self.protocol == "tcp":
            reader, writer = await asyncio.open_connection(self.host, self.port)
            self._tcp_reader = reader
            self._tcp_writer = writer
        else:
            raise ValueError(f"不支持的协议: {self.protocol}")

        self._is_connected = True

    except Exception as e:
        print(f"连接 StatsD 服务器失败: {e}")
        self._is_connected = False

disconnect() async

断开连接

Source code in src/symphra_cache/monitoring/statsd.py

async def disconnect(self) -> None:
    """
    断开连接
    """
    if self.protocol == "udp" and self._socket:
        self._socket.close()
    elif self.protocol == "tcp" and self._tcp_writer:
        self._tcp_writer.close()
        await self._tcp_writer.wait_closed()

    self._is_connected = False
    self._socket = None
    self._tcp_writer = None
    self._tcp_reader = None

flush_pending_metrics() async

刷新待发送的指标

Returns:

Type	Description
bool	刷新是否成功

Source code in src/symphra_cache/monitoring/statsd.py

async def flush_pending_metrics(self) -> bool:
    """
    刷新待发送的指标

    Returns:
        刷新是否成功
    """
    if not self._pending_metrics:
        return True

    success = await self.send_metrics(self._pending_metrics)
    if success:
        self._pending_metrics.clear()

    return success

generate_all_metrics()

生成所有指标

Returns:

Type	Description
list[str]	指标行列表

Source code in src/symphra_cache/monitoring/statsd.py

def generate_all_metrics(self) -> list[str]:
    """
    生成所有指标

    Returns:
        指标行列表
    """
    all_metrics = []
    all_metrics.extend(self._generate_counter_metrics())
    all_metrics.extend(self._generate_timer_metrics())
    all_metrics.extend(self._generate_gauge_metrics())
    return all_metrics

get_connection_status()

获取连接状态

Returns:

Type	Description
dict[str, Any]	连接状态信息

Source code in src/symphra_cache/monitoring/statsd.py

def get_connection_status(self) -> dict[str, Any]:
    """
    获取连接状态

    Returns:
        连接状态信息
    """
    return {
        "connected": self._is_connected,
        "protocol": self.protocol,
        "host": self.host,
        "port": self.port,
        "pending_metrics": len(self._pending_metrics),
    }

schedule_periodic_send(interval=30.0) async

安排周期性发送指标

Parameters:

Name	Type	Description	Default
interval	float	发送间隔（秒）	30.0

Source code in src/symphra_cache/monitoring/statsd.py

async def schedule_periodic_send(self, interval: float = 30.0) -> None:
    """
    安排周期性发送指标

    Args:
        interval: 发送间隔（秒）
    """
    while True:
        try:
            await asyncio.sleep(interval)
            await self.send_metrics()
        except asyncio.CancelledError:
            break
        except Exception as e:
            print(f"周期性发送失败: {e}")

send_metrics(metric_lines=None) async

发送指标到 StatsD 服务器

Parameters:

Name	Type	Description	Default
metric_lines	list[str] | None	要发送的指标行列表，None 表示发送所有指标	None

Returns:

Type	Description
bool	发送是否成功

Source code in src/symphra_cache/monitoring/statsd.py

async def send_metrics(self, metric_lines: list[str] | None = None) -> bool:
    """
    发送指标到 StatsD 服务器

    Args:
        metric_lines: 要发送的指标行列表，None 表示发送所有指标

    Returns:
        发送是否成功
    """
    if not self.monitor.is_enabled():
        return True

    if metric_lines is None:
        metric_lines = self.generate_all_metrics()

    if not metric_lines:
        return True

    # 建立连接
    if not self._is_connected:
        await self.connect()
        if not self._is_connected:
            return False

    # 应用采样率
    if self.sample_rate < 1.0:
        import random

        metric_lines = [line for line in metric_lines if random.random() < self.sample_rate]

    try:
        if self.protocol == "udp":
            return await self._send_udp_metrics(metric_lines)
        else:
            return await self._send_tcp_metrics(metric_lines)

    except Exception as e:
        print(f"发送指标失败: {e}")
        await self.disconnect()
        return False