可观测性数据存储成本优化：采样、聚合与冷热分层

一、你的 Prometheus 存储账单上个月 6 万，而其中 80% 的指标从来没人查过

可观测性数据的存储成本是典型的"沉默杀手"——初期每天几 GB 的监控数据往 Prometheus/Elasticsearch/Loki 里灌，一年后每天几百 GB，月账单 5-6 万。更扎心的是，这些数据中 80% 从未被任何人查询过——高精度监控数据（15 秒抓取一次）在事件发生 30 分钟后就不再有人关心了。

成本优化的三个杠杆：采样（数据精度降级）、聚合（预计算减少原始存储）、冷热分层（把低价值数据挪到廉价存储）。这三者不是互斥的——一套完整的存储优化策略通常是三者组合使用。

关键是"降精度不降可观测性"。你不需要永久保留 15 秒粒度的指标——7 天后的指标用 5 分钟粒度就够了，30 天后的指标用 1 小时粒度就能满足趋势分析需求。

二、底层机制与原理剖析

三层降成本策略：

采样（Trace 和 Log 层面）：不是所有数据都值同样精度。分布式 Trace 的采样率是最直接的杠杆——错误 Trace 100% 保留（排障必需），正常 Trace 只保留 10%（评估性能趋势足够）。应用日志按错误级别过滤——ERROR 级别日志全部保留，INFO 级别日志仅保留采样。

预聚合（Metrics 层面）：这是 ROI 最高的优化。Prometheus/VictoriaMetrics 原生支持 recording rules——预先计算如sum(rate(http_requests_total[5m]))这样的聚合结果，持久化聚合结果，然后允许删除原始高精度数据。查询常见面板（如 QPS 趋势图）时直接查聚合结果，不需要实时计算。

冷热分层（时间维度）：Thanos 和 Cortex 都支持对象存储作为长期存储。热数据（0-7 天）放在本地 SSD（VictoriaMetrics），温数据（7-30 天）放在 S3 Standard（Thanos Sidecar 上传），冷数据（30 天+）可以迁移到 S3 Glacier。

三、生产级代码实现

# prometheus-recording-rules.yaml # 预聚合规则：按频率分层聚合 --- groups: # 第一层：5 分钟聚合（7 天后从原始数据转为这个粒度） - name: aggregation_5min interval: 5m rules: # HTTP 请求速率（5 分钟） - record: job:http_requests_total:rate5m expr: rate(http_requests_total[5m]) # HTTP 请求错误率 - record: job:http_errors:rate5m expr: rate(http_requests_total{status=~"5.."}[5m]) # P95 延迟（30 秒桶的预聚合） - record: job:http_request_duration:p99_5m expr: histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m])) # 内存用量（max） - record: job:memory_usage:max_5m expr: max_over_time(process_resident_memory_bytes[5m]) # 第二层：1 小时聚合（30 天后降为这个粒度） - name: aggregation_1h interval: 1h rules: # 从 5 分钟聚合再聚合到 1 小时 - record: job:http_requests_total:rate1h expr: rate(job:http_requests_total:rate5m[1h]) - record: job:http_request_duration:p99_1h expr: max_over_time(job:http_request_duration:p99_5m[1h])

# observability-cost-optimizer.py """ 可观测性数据成本优化工具 功能： 1. 分析当前存储用量和查询模式 2. 计算降精度后的成本节省 3. 生成迁移计划 """ import logging from typing import Dict, List, Tuple from dataclasses import dataclass from datetime import datetime, timedelta logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class RetentionTier: """存储分层配置""" name: str # hot / warm / cold max_age_days: int # 数据保留天数 resolution: str # 15s / 5m / 1h storage_type: str # SSD / S3 / Glacier cost_per_gb_month: float # 每 GB 每月成本（元） @dataclass class DataSource: """数据源（Prometheus / Loki / Tempo）""" name: str daily_ingest_gb: float # 每天新写入数据量（GB） current_retention_days: int # 当前保留天数 query_activity: Dict[str, float] # 查询分布：{age_days: percentage} # 示例: {1: 0.5, 7: 0.3, 30: 0.15, 90: 0.05} # 表示 50% 查询在 1 天内，30% 在 7 天内 class CostOptimizer: """ 成本优化分析器 分析逻辑： 1. 扫描当前查询模式——确定哪些时间段的数据被高频查询 2. 计算每个时间段的"数据价值"（查询频率 / 存储成本） 3. 根据价值决定保留精度和存储层 """ # 默认分层策略 DEFAULT_TIERS = { "metrics": [ RetentionTier("hot", 7, "15s", "SSD", 100), RetentionTier("warm", 30, "5m", "S3 Standard", 20), RetentionTier("cold", 365, "1h", "S3 Glacier", 5), ], "logs": [ RetentionTier("hot", 3, "full", "SSD", 100), RetentionTier("warm", 14, "sampled(10%)", "S3 Standard", 20), RetentionTier("cold", 90, "errors_only", "S3 Glacier", 5), ], "traces": [ RetentionTier("hot", 3, "100%", "SSD", 100), RetentionTier("warm", 14, "errors(100%) + normal(10%)", "S3 Standard", 20), RetentionTier("cold", 30, "errors_only", "S3 Glacier", 5), ], } def analyze(self, source: DataSource, data_type: str = "metrics") -> Dict: """分析单个数据源的成本和优化空间""" tiers = self.DEFAULT_TIERS.get(data_type, self.DEFAULT_TIERS["metrics"]) # 1. 当前成本 current_daily_cost = source.daily_ingest_gb * 100 # 全部热存储 # 2. 分层后成本 optimized_daily_cost = self._calculate_tiered_cost(source, tiers) # 3. 计算节省 monthly_current = current_daily_cost * 30 monthly_optimized = optimized_daily_cost * 30 saving = monthly_current - monthly_optimized saving_pct = (saving / monthly_current * 100) if monthly_current > 0 else 0 return { "source": source.name, "type": data_type, "current_monthly_cost": round(monthly_current, 2), "optimized_monthly_cost": round(monthly_optimized, 2), "monthly_saving": round(saving, 2), "saving_percent": round(saving_pct, 1), "annual_saving": round(saving * 12, 2), "tier_details": [ { "tier": tier.name, "age_range": f"0-{tier.max_age_days}天", "resolution": tier.resolution, "storage": tier.storage_type, "monthly_cost": round( tier.cost_per_gb_month * source.daily_ingest_gb * tier.max_age_days, 2 ), } for tier in tiers ], } def _calculate_tiered_cost(self, source: DataSource, tiers: List[RetentionTier]) -> float: """计算分层存储的日均成本""" total_cost = 0.0 cumulative_days = 0 for tier in tiers: days_in_tier = min(tier.max_age_days, source.current_retention_days - cumulative_days) if days_in_tier <= 0: break # 分层存储成本 = 日摄入量 × 该层天数 × 该层单位成本 total_cost += source.daily_ingest_gb * days_in_tier * tier.cost_per_gb_month / 30 cumulative_days += days_in_tier return total_cost def generate_report(self, sources: List[DataSource]) -> str: """生成优化报告""" lines = [ "=" * 60, "可观测性数据存储成本优化报告", "=" * 60, "", ] total_current = 0 total_optimized = 0 for source in sources: types = ["metrics", "logs", "traces"] for dtype in types: result = self.analyze(source, dtype) total_current += result["current_monthly_cost"] total_optimized += result["optimized_monthly_cost"] lines.append(f"\n--- {source.name} ({dtype}) ---") lines.append(f" 当前月成本: ¥{result['current_monthly_cost']:,.0f}") lines.append(f" 优化后月成本: ¥{result['optimized_monthly_cost']:,.0f}") lines.append(f" 月节省: ¥{result['monthly_saving']:,.0f} ({result['saving_percent']}%)") for detail in result["tier_details"]: lines.append( f" [{detail['tier']}] {detail['age_range']} " f"{detail['resolution']} @ {detail['storage']} " f"≈ ¥{detail['monthly_cost']:,.0f}/月" ) total_saving = total_current - total_optimized lines.extend([ "", "=" * 40, f"总计: ¥{total_current:,.0f} → ¥{total_optimized:,.0f}", f"月节省: ¥{total_saving:,.0f} ({total_saving/total_current*100:.1f}%)", f"年节省: ¥{total_saving * 12:,.0f}", "=" * 60, ]) return "\n".join(lines) # --------------------------------------------------------------------------- # 示例 # --------------------------------------------------------------------------- if __name__ == "__main__": optimizer = CostOptimizer() # 模拟数据源 prometheus = DataSource( name="Prometheus", daily_ingest_gb=50, # 每天 50GB current_retention_days=90, # 保留 90 天 query_activity={1: 0.6, 7: 0.3, 30: 0.1}, ) loki = DataSource( name="Loki", daily_ingest_gb=120, # 每天 120GB current_retention_days=30, query_activity={1: 0.7, 7: 0.2, 14: 0.1}, ) report = optimizer.generate_report([prometheus, loki]) print(report)

四、边界分析与架构权衡

采样率的正确设定：

• 采样率低了 → 可能漏掉重要的异常信号。正常 Trace 10% 采样率意味着 90% 的请求没有 Trace 记录
• 补救：Head-based sampling（在 Trace 开始时决定）→ Tail-based sampling（在 Trace 结束后根据"有没有错误"决定），后者可以做到"错误 Trace 100% 保留，正常 Trace 按比例"

预聚合丢失的信息：

• 5 分钟聚合的 P99 丢失了"在 5 分钟内的瞬时抖动"。如果某个服务的 P99 在第 2 分钟飙到 5 秒但第 3-5 分钟正常，5 分钟聚合会平滑掉这个异常
• 补救：预聚合时保留 min/max 值而不仅仅是 avg/P99

冷存储的查询延迟：

• S3 Glacier 取回数据需要几分钟到几小时——发生 30 天前的事故复盘时，查冷存储数据需要提前"解冻"
• 建议：温存储（S3 Standard）保留到 30 天，只有 30 天+ 的才进 Glacier

五、总结

可观测性存储成本优化的核心是"降精度不降可观测性"。采样降 log/trace 的存储量，预聚合降 metrics 的存储量，冷热分层降低价值数据的存储成本。关键是先分析查询模式——80% 的查询集中在最近 7 天的数据——然后把 80% 的成本花在这 20% 的高频数据上。Prometheus recording rules + Thanos 对象存储在工程上是成熟组合，能把月成本从 6 万降到 1 万以内。