基于OpenTelemetry与Prometheus构建Claude Code可观测性监控体系

1. 项目概述为Claude Code构建可观测性监控体系如果你正在使用Claude Code进行AI辅助编程并且对它的使用成本、效率以及内部运行状态感到好奇那么今天分享的这个项目正是为你准备的。我最近花了不少时间基于OpenTelemetry、Prometheus、Loki和Grafana这一套当前最主流的可观测性技术栈搭建了一套专门用于监控Claude Code的完整方案。简单来说它能把Claude Code这个“黑盒”变成一个“透明盒”让你能清晰地看到每一次代码生成请求消耗了多少Token、花费了多少钱、缓存命中率如何甚至是代码变更的轨迹。这个项目的核心价值在于“量化”和“洞察”。在AI工具日益普及的今天尤其是对于团队或频繁使用的开发者而言无节制的使用很容易导致成本失控。通过这套监控体系你不仅能实时掌握开支还能深入分析使用模式比如哪些时段的请求最密集、哪些类型的提示词Prompt最耗资源、缓存机制是否真的在为你省钱。这不仅仅是简单的数据展示而是通过构建一个从数据采集、处理到可视化分析的完整管道为你提供数据驱动的决策依据从而优化使用策略提升开发效率的同时有效控制成本。2. 技术栈选型与架构设计思路2.1 为什么选择OpenTelemetry Prometheus Loki Grafana在开始动手之前明确技术选型的理由至关重要。市面上监控方案很多我选择这套组合拳是基于以下几个核心考量OpenTelemetry (OTel) 统一的数据采集标准OTel是目前云原生领域事实上的可观测性数据标准。它最大的优势在于提供了与供应商无关的API、SDK和工具用于生成、收集和导出遥测数据指标、日志、链路。对于Claude Code监控我们主要利用其指标Metrics和日志Logs能力。使用OTel意味着我们的采集端是标准化的未来如果需要切换后端存储或可视化工具比如从Prometheus换成其他时序数据库采集代码几乎无需改动避免了被单一技术栈锁定的风险。Prometheus 专为监控而生的时序数据库Prometheus是监控领域的“瑞士军刀”特别擅长处理多维数据模型和强大的查询语言PromQL。Claude Code的每次调用都可以抽象为一系列指标例如claude_code_request_duration_seconds请求耗时、claude_code_tokens_used_total累计Token使用量。Prometheus会以固定的时间间隔Scrape从OTel Collector拉取这些指标并存储起来。它的拉模型Pull非常适合这种需要持续收集系统状态数据的场景并且与Grafana的集成是天作之合。Loki 轻量级的日志聚合系统与专注于指标的Prometheus不同Loki专门处理日志。Claude Code在运行过程中会产生大量文本日志例如每次请求的提示词内容、生成的代码片段、错误信息等。传统的ELK栈Elasticsearch, Logstash, Kibana虽然强大但资源消耗也大。Loki的设计哲学是“只索引元数据不索引日志内容”它通过标签Labels来索引和查询日志流这使得它非常轻量和高效特别适合与Prometheus和Grafana搭配形成指标和日志的统一观测平台。Grafana 强大的可视化与告警平台Grafana是这套体系的“大脑”和“眼睛”。它可以从Prometheus和Loki中查询数据并绘制成直观的仪表盘Dashboard。你可以创建诸如“每日成本趋势图”、“Token使用热力图”、“缓存命中率仪表”等面板。更重要的是Grafana支持灵活的告警规则设置比如当每分钟成本超过某个阈值或缓存命中率低于50%时自动通过邮件、Slack等渠道通知你实现主动监控。注意这套架构是云原生监控的经典模式其学习曲线相对平缓社区活跃有海量的现成仪表盘模板和问题解决方案可供参考这能极大降低我们后续的维护和排错成本。2.2 整体架构数据流解析理解了每个组件的作用我们来看它们是如何协同工作的。整个系统的数据流可以清晰地分为三层数据采集层Claude Code应用通过集成OpenTelemetry SDK在代码关键位置如发起API调用、收到响应、访问缓存埋点生成指标和日志。这些数据被发送到OpenTelemetry Collector。数据处理与存储层OTel Collector作为一个管道接收数据后可以进行过滤、加工然后分别将指标Metrics导出到Prometheus将日志Logs导出到Loki。两者各司其职进行存储。可视化与告警层Grafana配置好Prometheus和Loki作为数据源从中查询数据渲染成用户友好的图表和仪表盘并基于定义好的规则触发告警。这种分层架构职责清晰扩展性强。例如未来如果想增加链路追踪Tracing功能只需在OTel Collector中配置将Trace数据导出到Jaeger或Tempo即可其他部分无需大动。3. 核心组件部署与配置实操理论部分清晰后我们进入实战环节。我推荐使用Docker Compose来部署这套环境它能一键拉起所有服务并处理好网络互联是最快捷的方式。3.1 使用Docker Compose一键部署首先你需要确保本地已安装Docker和Docker Compose。然后创建一个项目目录例如claude-code-monitoring并在其中创建docker-compose.yml文件。version: 3.8 services: # OpenTelemetry Collector - 数据收集与转发枢纽 otel-collector: image: otel/opentelemetry-collector-contrib:latest command: [--config/etc/otel-collector-config.yaml] volumes: - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml ports: - 4317:4317 # OTLP gRPC 接收端口 - 4318:4318 # OTLP HTTP 接收端口 - 8889:8889 # 健康检查/指标端口 depends_on: - prometheus - loki # Prometheus - 指标存储与查询 prometheus: image: prom/prometheus:latest volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml - prometheus_data:/prometheus command: - --config.file/etc/prometheus/prometheus.yml - --storage.tsdb.path/prometheus - --web.console.libraries/etc/prometheus/console_libraries - --web.console.templates/etc/prometheus/consoles - --storage.tsdb.retention.time30d # 数据保留30天 - --web.enable-lifecycle ports: - 9090:9090 # Loki - 日志聚合 loki: image: grafana/loki:latest command: -config.file/etc/loki/local-config.yaml volumes: - loki_data:/loki ports: - 3100:3100 # Grafana - 可视化平台 grafana: image: grafana/grafana-enterprise:latest environment: - GF_SECURITY_ADMIN_PASSWORDadmin # 首次登录密码请务必修改 - GF_INSTALL_PLUGINSgrafana-clock-panel volumes: - grafana_data:/var/lib/grafana - ./grafana/provisioning:/etc/grafana/provisioning ports: - 3000:3000 depends_on: - prometheus - loki volumes: prometheus_data: loki_data: grafana_data:这个Compose文件定义了四个核心服务。接下来我们需要配置最关键的两个文件OTel Collector的配置和Prometheus的配置。3.2 OpenTelemetry Collector 配置详解OTel Collector的配置文件otel-collector-config.yaml决定了数据如何被接收、处理和导出。这是整个系统的“交通指挥中心”。receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 processors: batch: # 批处理处理器将数据打包发送以提高效率 timeout: 1s send_batch_size: 1024 memory_limiter: # 内存限制器防止内存溢出 check_interval: 1s limit_mib: 512 spike_limit_mib: 256 exporters: debug: verbosity: detailed prometheus: endpoint: 0.0.0.0:8889 namespace: claude_code const_labels: monitor: claude-code-observability loki: endpoint: http://loki:3100/loki/api/v1/push labels: attributes: # 将OTel资源属性中的service.name作为Loki日志流的标签 service.name: service_name default_labels_enabled: # 禁用 exporter 名称作为默认标签保持标签简洁 exporter: false service: pipelines: metrics: receivers: [otlp] processors: [memory_limiter, batch] exporters: [debug, prometheus] logs: receivers: [otlp] processors: [memory_limiter, batch] exporters: [debug, loki]配置关键点解析receivers:定义了数据入口。这里配置了OTLP协议OpenTelemetry Protocol的gRPC和HTTP端口Claude Code的SDK将向这里发送数据。exporters:定义了数据出口。prometheusexporter 会将指标数据以Prometheus格式暴露在8889端口lokiexporter 会将日志发送到Loki服务的API。service.pipelines:将接收器、处理器、导出器连接成管道。我们建立了两条独立的管道分别处理指标和日志。3.3 Prometheus 抓取配置Prometheus需要知道去哪里拉取指标数据。创建prometheus.yml配置文件global: scrape_interval: 15s # 每15秒抓取一次数据 evaluation_interval: 15s # 每15秒评估一次告警规则 scrape_configs: - job_name: otel-collector static_configs: - targets: [otel-collector:8889] # 抓取OTel Collector暴露的指标 scrape_interval: 10s # 对Collector可以抓取频繁一些 - job_name: prometheus static_configs: - targets: [localhost:9090] alerting: alertmanagers: - static_configs: - targets: [] # 在实际生产环境中可以配置Alertmanager地址以实现更复杂的告警路由和静默 rule_files: # - first_rules.yml # - second_rules.yml配置完成后在项目目录下执行docker-compose up -d等待所有容器启动。你可以通过docker-compose ps检查状态并通过以下地址访问服务Grafana:http://localhost:3000(用户名: admin, 密码: admin)Prometheus:http://localhost:9090Loki: 通常不直接提供UI可通过Grafana查询。4. Claude Code应用侧埋点与集成监控后端就绪后下一步就是在你的Claude Code应用或调用Claude API的脚本中集成OpenTelemetry SDK进行埋点。这里以Python为例因为Claude Code的API调用通常用Python实现。4.1 安装必要的Python库首先安装OpenTelemetry相关的Python包pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp4.2 初始化OpenTelemetry并发送数据在你的Claude Code调用代码中添加以下初始化逻辑。建议将其放在应用启动时执行一次。import logging from opentelemetry import metrics from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry._logs import set_logger_provider from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler from opentelemetry.sdk._logs.export import BatchLogRecordProcessor from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter # 1. 配置指标Metrics metric_exporter OTLPMetricExporter(endpointhttp://localhost:4317, insecureTrue) metric_reader PeriodicExportingMetricReader(metric_exporter, export_interval_millis5000) meter_provider MeterProvider(metric_readers[metric_reader]) metrics.set_meter_provider(meter_provider) # 获取一个计量器Meter用于创建指标 meter metrics.get_meter(__name__) # 创建几个核心指标 request_counter meter.create_counter( nameclaude_code.requests.total, descriptionTotal number of Claude Code API requests, unit1 ) token_counter meter.create_counter( nameclaude_code.tokens.total, descriptionTotal tokens consumed, unit1 ) request_duration meter.create_histogram( nameclaude_code.request.duration, descriptionDuration of Claude Code API requests, unitms ) cache_hit_counter meter.create_counter( nameclaude_code.cache.hits.total, descriptionTotal cache hits, unit1 ) cache_miss_counter meter.create_counter( nameclaude_code.cache.misses.total, descriptionTotal cache misses, unit1 ) # 2. 配置日志Logs - 发送到OTel Collector log_exporter OTLPLogExporter(endpointhttp://localhost:4317, insecureTrue) logger_provider LoggerProvider() set_logger_provider(logger_provider) logger_provider.add_log_record_processor(BatchLogRecordProcessor(log_exporter)) # 将OpenTelemetry的LoggingHandler添加到你的应用日志器中 handler LoggingHandler(levellogging.INFO, logger_providerlogger_provider) logging.getLogger().addHandler(handler)4.3 在业务代码中埋点现在在你实际调用Claude API的函数周围添加指标记录和日志记录。import time import logging from opentelemetry import context logger logging.getLogger(__name__) def call_claude_code(prompt: str, use_cache: bool True): 模拟调用Claude Code API的函数 start_time time.time() attributes {model: claude-3-opus, operation: code_generation} try: # 模拟缓存逻辑 if use_cache and check_cache(prompt): cache_hit_counter.add(1, attributes) logger.info(fCache hit for prompt: {prompt[:50]}...) # ... 从缓存获取结果 result [CACHED] Simulated code output else: cache_miss_counter.add(1, attributes) logger.info(fCache miss, calling API for prompt: {prompt[:50]}...) # 模拟API调用 time.sleep(0.5) # 模拟网络延迟 # ... 实际调用Claude API result Simulated code output # 模拟返回的Token数 tokens_used len(prompt) // 4 100 # 记录Token使用量 token_counter.add(tokens_used, attributes) # 记录一次请求 request_counter.add(1, attributes) # 记录请求耗时 duration (time.time() - start_time) * 1000 # 转为毫秒 request_duration.record(duration, attributes) logger.info(fRequest completed in {duration:.2f}ms) return result except Exception as e: logger.error(fClaude Code API call failed: {e}, exc_infoTrue) # 可以记录错误相关的指标 error_attributes attributes.copy() error_attributes[error] type(e).__name__ request_counter.add(1, error_attributes) raise def check_cache(prompt: str) - bool: # 简化的缓存检查逻辑 return False # 假设总是未命中这段代码做了以下几件事记录请求次数每次调用call_claude_coderequest_counter加1。记录Token用量在API调用成功后根据模拟逻辑累加Token数到token_counter。记录请求耗时使用直方图request_duration记录每次请求的耗时分布这对于分析性能瓶颈至关重要。记录缓存命中/未命中通过cache_hit_counter和cache_miss_counter来评估缓存策略的有效性。输出结构化日志通过标准的Pythonlogging模块记录信息、错误这些日志会被LoggingHandler捕获并发送到Loki。实操心得埋点时为指标添加有意义的标签Attributes是后续进行多维分析的关键。例如为request_counter添加model模型版本、operation操作类型如code_generation、code_explain等标签。这样在Grafana中你可以轻松地按模型或操作类型来聚合和对比数据。但也要注意标签的基数Cardinality不能过高否则会给Prometheus带来巨大压力。5. Grafana仪表盘配置与核心监控视图当数据开始流入后我们就可以在Grafana中创建强大的监控仪表盘了。登录Grafana后首先需要添加数据源。5.1 添加Prometheus和Loki数据源点击左侧齿轮图标 -Data sources-Add data source。选择Prometheus。在URL一栏填写http://prometheus:9090因为它们在同一个Docker网络中。点击Save test应该显示“Data source is working”。再次Add data source选择Loki。URL填写http://loki:3100。点击Save test。5.2 构建Claude Code核心监控仪表盘创建一个新的Dashboard并添加以下核心面板面板1请求量与耗时概览Stat Time Series查询A (Stat):sum(rate(claude_code_requests_total[5m]))- 显示最近5分钟的平均每秒请求数RPS。查询B (Time Series):rate(claude_code_request_duration_ms_bucket[5m])- 使用Heatmap可视化或Histogram模式查看耗时分布。可以配合histogram_quantile(0.95, rate(...[5m]))来显示P95延迟线。作用一目了然看到系统负载和性能表现。面板2Token消耗与成本估算Time Series查询A:sum(rate(claude_code_tokens_total[1h]))- 每小时Token消耗速率。查询B (估算成本):假设每百万Token输入$10输出$30。创建一个混合查询# 假设我们通过标签区分了输入/输出Token这里用token_type标签示例 (sum(rate(claude_code_tokens_total{token_typeinput}[1h])) * 10 / 1e6) (sum(rate(claude_code_tokens_total{token_typeoutput}[1h])) * 30 / 1e6)这个查询结果就是预估的每小时成本美元。你可以将其乘以24来估算日成本。作用实时监控资源消耗和成本趋势是成本控制的核心。面板3缓存效率分析Time Series Gauge查询A (缓存命中率):rate(claude_code_cache_hits_total[5m]) / (rate(claude_code_cache_hits_total[5m]) rate(claude_code_cache_misses_total[5m])) * 100查询B (Gauge):直接使用上面的公式但显示当前瞬时值作为一个仪表盘中的仪表Gauge面板。作用直观展示缓存策略的效果。如果命中率持续偏低就需要考虑优化缓存键Cache Key设计或扩大缓存容量。面板4错误与异常监控Logs Panel添加一个Logs类型面板数据源选择Loki。查询:{service_nameyour-claude-code-service} | json | levelerror。这个查询会过滤出错误级别的日志| json会自动解析日志中的JSON字段如果日志是JSON格式的话。作用集中查看所有错误信息结合日志上下文快速定位问题。面板5用户/操作分析Table查询:topk(10, sum by (user_id, operation) (rate(claude_code_requests_total[1h])))。这里假设你为指标添加了user_id和operation标签。作用以表格形式展示过去一小时内请求量最高的10个用户及其最常执行的操作便于识别重度用户和使用模式。将这些面板合理布局你就得到了一个功能全面的Claude Code监控指挥中心。6. 高级技巧告警规则与成本优化策略监控的最终目的不仅是“看到”还要能“预警”和“优化”。6.1 在Grafana中配置关键告警进入Grafana Alerting页面创建告警规则异常高成本告警规则名称ClaudeCodeHourlyCostSpike查询使用上面面板2的成本估算查询例如sum(rate(claude_code_tokens_total[1h])) 1000000假设1小时内Token消耗超过100万。条件WHEN last() OF query(A, 1h, now) IS ABOVE 1000000通知渠道配置邮件、Slack或钉钉。服务性能下降告警规则名称ClaudeCodeHighLatency查询histogram_quantile(0.95, rate(claude_code_request_duration_ms_bucket[5m]))条件WHEN last() OF query(A, 5m, now) IS ABOVE 5000P95延迟超过5秒。作用及时发现API响应变慢可能是网络问题或Claude服务端异常。缓存失效告警规则名称ClaudeCodeLowCacheHitRate查询使用面板3的缓存命中率公式。条件WHEN last() OF query(A, 15m, now) IS BELOW 30命中率持续15分钟低于30%。作用提示缓存策略可能需要进行调整。6.2 基于监控数据的成本优化实践有了数据支撑优化就有了方向。以下是我在实践中总结的几个有效策略策略一识别并优化“Token消耗大户”在Grafana中分析claude_code_tokens_total按user_id或prompt_type需要埋点时添加此标签的分布。你可能会发现某些用于生成长篇文档或复杂算法的提示词消耗了不成比例的Token。针对这些场景可以优化提示词使其更精确、简洁。对输出结果进行分页或流式处理避免一次性生成过长的内容。考虑对某些通用、固定的代码片段如项目脚手架、常用工具函数建立本地模板库而非每次都让AI生成。策略二设计更智能的缓存策略监控面板上的缓存命中率是优化缓存的第一指南针。如果命中率低审查缓存键Cache Key确保缓存键能准确区分不同语义的请求。例如“写一个快速排序函数”和“写一个Python的快速排序函数”可能应该被缓存为不同的条目。可以考虑对提示词进行语义哈希或标准化处理如去除多余空格、统一术语后再作为缓存键。实施分层缓存对于完全相同的提示词缓存结果永久有效或设置长TTL。对于相似但不完全相同的提示词可以探索使用向量数据库进行相似度匹配返回最相似的缓存结果作为参考从而减少全新生成的需求。策略三错峰与限流通过监控面板观察请求的时序分布。如果发现工作时段如上午10-12点请求过于集中可以考虑对非紧急的、批处理性质的代码生成任务如批量生成单元测试、代码注释设置调度安排在夜间或低峰期执行。在应用层面实现简单的限流机制例如使用令牌桶算法为每个用户或每个团队设置每分钟/每小时的最大请求次数防止个别脚本异常或滥用导致成本激增。这套监控体系的价值正是在于它将模糊的“感觉”变成了清晰的“数据”。你不再需要猜测“这个月AI花了多少钱”而是可以精确地知道“周二下午的代码重构任务消耗了总成本的40%”。这种可见性是进行任何有效管理和优化的基石。