OpenTelemetry 监控与可观测性

# 启用 OpenTelemetry
export OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4317
export OTEL_SERVICE_NAME=claude-code-team
export OTEL_RESOURCE_ATTRIBUTES="team=backend,env=production"
 
# 启动 Claude Code
claude

from claude_agent_sdk import Agent, ObservabilityConfig
 
agent = Agent(
    model="claude-sonnet-4-6",
    observability=ObservabilityConfig(
        exporter="otlp",
        endpoint="http://otel-collector:4317",
        service_name="code-review-agent",
        attributes={
            "team": "platform",
            "project": "auth-service"
        }
    )
)

{
  "trace_id": "abc123",
  "span_id": "def456",
  "name": "claude.task",
  "start_time": "2025-06-18T10:00:00Z",
  "end_time": "2025-06-18T10:00:05Z",
  "attributes": {
    "claude.model": "claude-sonnet-4-6",
    "claude.input_tokens": 4520,
    "claude.output_tokens": 890,
    "claude.tool_calls": 3,
    "claude.files_read": 5,
    "claude.files_written": 2
  },
  "events": [
    {"time": "10:00:01", "name": "thinking_start"},
    {"time": "10:00:03", "name": "tool_call", "attributes": {"tool": "file_read"}},
    {"time": "10:00:05", "name": "completion"}
  ]
}

{
  "timestamp": "2025-06-18T10:00:05Z",
  "level": "INFO",
  "message": "Task completed",
  "attributes": {
    "session_id": "sess_abc",
    "task": "refactor auth",
    "duration_ms": 5200,
    "success": true
  }
}

# docker-compose.yaml
version: '3'
services:
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    volumes:
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"

# alerting-rules.yaml
groups:
  - name: claude-alerts
    rules:
      - alert: HighCostSpike
        expr: |
          (
            sum(claude_cost_usd) 
            / sum(claude_cost_usd offset 1d)
          ) > 2
        for: 1h
        labels:
          severity: warning
        annotations:
          summary: "Claude Code 成本激增"
          
      - alert: HighErrorRate
        expr: |
          (
            sum(rate(claude_errors_total[5m]))
            / sum(rate(claude_requests_total[5m]))
          ) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "错误率超过 5%"
          
      - alert: SlowResponse
        expr: |
          histogram_quantile(0.95, 
            rate(claude_request_duration_seconds_bucket[5m])
          ) > 30
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "P95 响应时间超过 30 秒"

# 导出当前会话的 trace
claude export-trace --session sess_abc123 --output trace.json
 
# 分析
jq '.spans[] | {name, duration_ms}' trace.json

# 找出最耗时的操作
claude profile --last-hour
 
# 输出示例:
# Operation          Count  Avg(ms)  Total(ms)
# file_read          45     120      5400
# bash_exec          12     850      10200
# model_call         8      3200     25600

{
  "dashboard": {
    "title": "Claude Code 企业监控",
    "tags": ["claude", "ai", "monitoring"],
    "timezone": "browser",
    "panels": [
      {
        "title": "实时用量概览",
        "type": "stat",
        "targets": [
          {
            "expr": "sum(claude_tokens_input_total) + sum(claude_tokens_output_total)",
            "legendFormat": "总 Token 数"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "unit": "short",
            "color": {
              "mode": "thresholds",
              "thresholds": {
                "steps": [
                  {"color": "green", "value": null},
                  {"color": "yellow", "value": 1000000},
                  {"color": "red", "value": 5000000}
                ]
              }
            }
          }
        }
      },
      {
        "title": "Top 10 活跃用户",
        "type": "table",
        "targets": [
          {
            "expr": "topk(10, sum by (user) (claude_cost_usd_total))",
            "format": "table"
          }
        ]
      },
      {
        "title": "模型使用分布",
        "type": "piechart",
        "targets": [
          {
            "expr": "sum by (model) (claude_requests_total)",
            "legendFormat": "{{model}}"
          }
        ]
      },
      {
        "title": "错误率趋势",
        "type": "timeseries",
        "targets": [
          {
            "expr": "rate(claude_errors_total[5m]) / rate(claude_requests_total[5m])",
            "legendFormat": "错误率"
          }
        ]
      },
      {
        "title": "Token 使用热力图",
        "type": "heatmap",
        "targets": [
          {
            "expr": "sum by (le) (rate(claude_request_duration_seconds_bucket[5m]))",
            "format": "heatmap"
          }
        ]
      }
    ]
  }
}

# observability-stack.yaml
version: '3.8'
 
services:
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
 
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"
      - "14268:14268"
 
  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    volumes:
      - ./loki-config.yml:/etc/loki/local-config.yaml
 
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    volumes:
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources.yml:/etc/grafana/provisioning/datasources/datasources.yml
      - grafana-data:/var/lib/grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
 
  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"
      - "4318:4318"
 
volumes:
  prometheus-data:
  grafana-data:

export OTEL_EXPORTER_OTLP_ENDPOINT="http://otel-collector.corp.local:4317"
export OTEL_EXPORTER_OTLP_PROTOCOL="grpc"
export OTEL_RESOURCE_ATTRIBUTES="service.name=claude-code,deployment.environment=prod,team=$TEAM"
export OTEL_LOG_LEVEL=warn
export CLAUDE_TELEMETRY="on"
export CLAUDE_TELEMETRY_INCLUDE_TOOL_IO="redacted"   # 含敏感字段自动脱敏

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc: { endpoint: 0.0.0.0:4317 }
      http: { endpoint: 0.0.0.0:4318 }
 
processors:
  memory_limiter:
    check_interval: 1s
    limit_percentage: 75
    spike_limit_percentage: 25
 
  batch:
    timeout: 5s
    send_batch_size: 1024
 
  # 尾部采样：错误/慢/高成本 100%，其余 10%
  tail_sampling:
    decision_wait: 10s
    policies:
      - name: errors
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: slow
        type: latency
        latency: { threshold_ms: 5000 }
      - name: high-cost
        type: numeric_attribute
        numeric_attribute: { key: claude.tokens.output, min_value: 8000 }
      - name: random
        type: probabilistic
        probabilistic: { sampling_percentage: 10 }
 
  # 脱敏敏感字段
  attributes/redact:
    actions:
      - { key: claude.tool_args_raw, action: delete }
      - { key: prompt.text, action: hash }
      - { key: response.text, action: hash }
 
  # 资源属性补全
  resource:
    attributes:
      - { key: dc, value: ${env:DEPLOY_DC}, action: insert }
      - { key: cluster, value: ${env:CLUSTER_NAME}, action: insert }
 
  # 按权限决策分流到不同后端
  routing:
    from_attribute: claude.permission_decision
    table:
      - value: deny
        exporters: [otlp/security, otlphttp/loki]
      - value: allow
        exporters: [otlp/jaeger, prometheus]
    default_exporters: [otlp/jaeger, prometheus]
 
exporters:
  prometheus:
    endpoint: 0.0.0.0:8889
    metric_expiration: 30m
    resource_to_telemetry_conversion: { enabled: true }
 
  otlp/jaeger:
    endpoint: jaeger:4317
    tls: { insecure: true }
    sending_queue: { enabled: true, queue_size: 5000 }
    retry_on_failure: { enabled: true, initial_interval: 5s, max_elapsed_time: 5m }
 
  otlp/security:
    endpoint: security-collector.corp.local:4317
    tls: { insecure_skip_verify: false, ca_file: /etc/ssl/corp-ca.pem }
 
  otlphttp/loki:
    endpoint: http://loki:3100/otlp
 
extensions:
  health_check: { endpoint: 0.0.0.0:13133 }
  pprof:        { endpoint: 0.0.0.0:1777 }
  zpages:       { endpoint: 0.0.0.0:55679 }
 
service:
  extensions: [health_check, pprof, zpages]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, attributes/redact, resource, tail_sampling, batch]
      exporters: [otlp/jaeger]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, resource, batch]
      exporters: [prometheus]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, attributes/redact, batch, routing]
      exporters: [otlphttp/loki, otlp/security]
  telemetry:
    metrics: { address: 0.0.0.0:8888 }
    logs:    { level: info }

# observability/instrumented_agent.py
import asyncio
import hashlib
import os
import time
from contextlib import asynccontextmanager
 
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from opentelemetry import metrics, trace
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
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.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
 
 
def _setup_otel(service: str, env: str) -> None:
    res = Resource.create({"service.name": service, "deployment.environment": env})
    tp = TracerProvider(resource=res)
    tp.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    trace.set_tracer_provider(tp)
    mp = MeterProvider(
        resource=res,
        metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter(), 30_000)],
    )
    metrics.set_meter_provider(mp)
 
 
_setup_otel(os.environ.get("OTEL_SERVICE_NAME", "claude-agent"),
            os.environ.get("DEPLOY_ENV", "dev"))
 
tracer = trace.get_tracer("claude.agent")
meter = metrics.get_meter("claude.agent")
 
turn_counter = meter.create_counter(
    "claude_turns_total", description="Number of agent turns")
token_input = meter.create_counter(
    "claude_tokens_input_total", unit="token")
token_output = meter.create_counter(
    "claude_tokens_output_total", unit="token")
tool_latency = meter.create_histogram(
    "claude_tool_latency_seconds", unit="s",
    description="Latency per tool invocation")
tool_errors = meter.create_counter(
    "claude_tool_errors_total", description="Tool failures")
 
 
@asynccontextmanager
async def traced_session(project: str, user: str):
    """围绕一次会话开启 root span，并把业务标签注入。"""
    with tracer.start_as_current_span("claude.session") as span:
        span.set_attribute("project", project)
        span.set_attribute("claude.user.email", user)
        span.set_attribute("deploy.env", os.environ.get("DEPLOY_ENV", "dev"))
        yield span
 
 
async def run_task(project: str, user: str, prompt: str) -> None:
    options = ClaudeAgentOptions(
        system_prompt="You are an expert Python engineer.",
        allowed_tools=["Read", "Grep", "Edit", "Bash"],
        max_turns=12,
    )
    labels = {"project": project, "team": os.environ.get("TEAM", "unknown")}
 
    async with traced_session(project, user):
        async with ClaudeSDKClient(options=options) as client:
            await client.query(prompt)
            async for evt in client.receive_response():
                if evt.type == "message_start":
                    turn_counter.add(1, labels)
                if evt.type == "tool_use":
                    args_hash = hashlib.sha256(
                        str(evt.tool_input).encode()).hexdigest()[:16]
                    with tracer.start_as_current_span(f"tool.{evt.tool_name}") as s:
                        s.set_attribute("claude.tool", evt.tool_name)
                        s.set_attribute("claude.tool_args_hash", args_hash)
                        start = time.perf_counter()
                        try:
                            await evt.execute()
                        except Exception as exc:
                            tool_errors.add(1, {**labels, "tool": evt.tool_name,
                                                "kind": type(exc).__name__})
                            s.record_exception(exc)
                            s.set_status(trace.Status(trace.StatusCode.ERROR))
                            raise
                        finally:
                            tool_latency.record(time.perf_counter() - start,
                                                {**labels, "tool": evt.tool_name})
                if evt.type == "message_delta" and getattr(evt, "usage", None):
                    token_input.add(evt.usage.input_tokens, labels)
                    token_output.add(evt.usage.output_tokens, labels)
 
 
if __name__ == "__main__":
    asyncio.run(run_task(
        project="finops-rules",
        user=os.environ.get("USER_EMAIL", "[email protected]"),
        prompt="审查 src/risk/ 下的代码并给出 3 个性能改进建议。",
    ))

# 按团队的 Token 消耗速率
sum by (team) (rate(claude_tokens_output_total[5m]))
  + sum by (team) (rate(claude_tokens_input_total[5m]))
 
# 按工具的 P95 延迟
histogram_quantile(0.95,
  sum by (tool, le) (rate(claude_tool_latency_seconds_bucket[5m]))
)
 
# 工具错误率
sum by (tool) (rate(claude_tool_errors_total[5m]))
  /
sum by (tool) (rate(claude_tool_latency_seconds_count[5m]))

# 1) Loki：把单次会话的所有日志按时间排出来
{service_name="claude-code"} | json | claude_session_id = "$session_id"
  | line_format "{{.ts}} {{.claude_tool}} {{.message}}"
 
# 2) Loki：找最近 1h 内权限被拒绝的所有调用
{service_name="claude-code"} | json
  | claude_permission_decision = "deny"
  | line_format "{{.ts}} user={{.claude_user_email}} tool={{.claude_tool}} reason={{.deny_reason}}"

-- 3) Tempo TraceQL：找耗时 > 30s 且产生 deny 的会话
{ resource.service.name = "claude-code" }
  | duration > 30s
  | events.name = "permission.deny"
  | select(claude.session_id, claude.user.email, duration)

# prometheus/rules/claude-finops.yml
groups:
  - name: claude-finops
    interval: 30s
    rules:
      - record: claude:cost_usd:rate5m
        expr: |
          (
            sum by (team, model) (rate(claude_tokens_input_total[5m])) *
              on(model) group_left() claude_model_price_input
          )
          +
          (
            sum by (team, model) (rate(claude_tokens_output_total[5m])) *
              on(model) group_left() claude_model_price_output
          )
 
      - record: claude:cost_usd:month_to_date
        expr: |
          sum by (team) (
            sum_over_time(
              (claude:cost_usd:rate5m * 300)[31d:5m]
            )
          )
 
      - alert: ClaudeTeamBudgetWarning
        expr: claude:cost_usd:month_to_date / on(team) claude_team_budget_usd > 0.8
        for: 10m
        labels: { severity: warning }
        annotations:
          summary: "团队 {{ $labels.team }} 已用 80% 月预算"
 
      - alert: ClaudeTeamBudgetExceeded
        expr: claude:cost_usd:month_to_date / on(team) claude_team_budget_usd > 1
        for: 5m
        labels: { severity: critical }
        annotations:
          summary: "团队 {{ $labels.team }} 已超月预算"

# budget_guard.py —— 收到告警后写降级标记到 Redis
import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
 
import redis
 
r = redis.Redis.from_url(os.environ["REDIS_URL"])
TTL = 6 * 3600  # 6 小时窗口
 
 
class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        for alert in body.get("alerts", []):
            team = alert["labels"].get("team")
            sev = alert["labels"].get("severity")
            status = alert.get("status")
            if not team:
                continue
            key = f"team:{team}:degraded"
            if status == "firing" and sev == "critical":
                r.setex(key, TTL, "haiku")
            elif status == "firing" and sev == "warning":
                r.setex(key, TTL, "sonnet")
            elif status == "resolved":
                r.delete(key)
        self.send_response(204); self.end_headers()
 
 
if __name__ == "__main__":
    HTTPServer(("0.0.0.0", 8080), Handler).serve_forever()

# pre_run_check.py
import os, redis
from claude_agent_sdk import ClaudeAgentOptions
 
r = redis.Redis.from_url(os.environ["REDIS_URL"])
 
def build_options(team: str) -> ClaudeAgentOptions:
    forced = r.get(f"team:{team}:degraded")
    forced_model = forced.decode() if forced else None
    base = ClaudeAgentOptions(
        system_prompt="You are an engineering assistant.",
        allowed_tools=["Read", "Grep", "Edit", "Bash"],
        max_turns=10,
    )
    if forced_model:
        base.model = forced_model  # 强制降级到便宜模型
        base.max_turns = min(base.max_turns, 6)
    return base

- alert:
 ClaudeAnomalousSession
  expr: |
    (
      sum by (claude_session_id) (rate(claude_tool_errors_total[2m])) > 3
    )
    or
    (
      sum by (claude_session_id) (
        increase(claude_permission_denials_total[2m])
      ) > 0
    )
  for: 1m
  labels: { severity: critical, route: forensics }
  annotations:
    summary: "异常会话 {{ $labels.claude_session_id }}"

# forensics_worker.py
import hashlib
import io
import json
import os
import pathlib
import subprocess
import tarfile
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer
 
import boto3
 
S3 = boto3.client("s3")
BUCKET = os.environ["EVIDENCE_BUCKET"]
TEMPO = os.environ["TEMPO_URL"]
LOG_DIR = pathlib.Path(os.environ.get("CLAUDE_LOG_DIR", "/var/log/claude"))
 
 
def fetch_trace(session_id: str) -> bytes:
    url = f"{TEMPO}/api/search?tags=claude.session_id={session_id}&limit=1"
    with urllib.request.urlopen(url, timeout=15) as resp:
        meta = json.loads(resp.read())
    if not meta.get("traces"):
        return b"{}"
    trace_id = meta["traces"][0]["traceID"]
    with urllib.request.urlopen(f"{TEMPO}/api/traces/{trace_id}", timeout=30) as resp:
        return resp.read()
 
 
def collect(session_id: str) -> tuple[bytes, str]:
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
        # trace
        trace_bytes = fetch_trace(session_id)
        info = tarfile.TarInfo("trace.json")
        info.size = len(trace_bytes)
        tar.addfile(info, io.BytesIO(trace_bytes))
        # 本地 jsonl 日志
        log_path = LOG_DIR / f"{session_id}.jsonl"
        if log_path.exists():
            tar.add(log_path, arcname="session.jsonl")
        # 当时代码快照（用 git stash + bundle）
        rev = subprocess.run(["git", "rev-parse", "HEAD"],
                             capture_output=True, text=True).stdout.strip()
        bundle = subprocess.run(
            ["git", "bundle", "create", "/tmp/repo.bundle", "HEAD"],
            capture_output=True)
        if bundle.returncode == 0:
            tar.add("/tmp/repo.bundle", arcname=f"repo-{rev[:12]}.bundle")
        # 元数据
        meta = {
            "session_id": session_id,
            "captured_at": int(time.time()),
            "git_head": rev,
            "host": os.uname().nodename,
        }
        meta_bytes = json.dumps(meta, indent=2).encode()
        info = tarfile.TarInfo("meta.json")
        info.size = len(meta_bytes)
        tar.addfile(info, io.BytesIO(meta_bytes))
    raw = buf.getvalue()
    digest = hashlib.sha256(raw).hexdigest()
    return raw, digest
 
 
class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        payload = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        for alert in payload.get("alerts", []):
            sid = alert["labels"].get("claude_session_id")
            if not sid or alert.get("status") != "firing":
                continue
            blob, digest = collect(sid)
            key = f"forensics/{time.strftime('%Y/%m/%d')}/{sid}.tgz"
            S3.put_object(
                Bucket=BUCKET, Key=key, Body=blob,
                ServerSideEncryption="aws:kms",
                Metadata={"sha256": digest, "session_id": sid},
                ObjectLockMode="COMPLIANCE",
                ObjectLockRetainUntilDate=time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ",
                    time.gmtime(time.time() + 365 * 24 * 3600)),
            )
        self.send_response(204); self.end_headers()
 
 
if __name__ == "__main__":
    HTTPServer(("0.0.0.0", 9090), Handler).serve_forever()