Claude API のテレメトリを ClickHouse に流して分析する — 本番環境のコスト・遅延・エラーを可視化する設計ガイド

// src/lib/telemetry.ts
import Anthropic from "@anthropic-ai/sdk";
 
interface Env {
  ANTHROPIC_API_KEY: string;
  CLICKHOUSE_URL: string;
  CLICKHOUSE_USER: string;
  CLICKHOUSE_PASSWORD: string;
  TELEMETRY_FALLBACK: KVNamespace;
}
 
interface TelemetryEvent {
  event_id: string;
  request_id: string;
  user_id: string;
  workspace_id: string;
  ts: string; // ISO8601
  model: string;
  api_version: string;
  input_tokens: number;
  output_tokens: number;
  cache_read_tokens: number;
  cache_write_5m_tokens: number;
  cache_write_1h_tokens: number;
  stop_reason: string;
  latency_ms: number;
  ttft_ms: number;
  streaming: number;
  success: number;
  error_type: string;
  http_status: number;
  feature_id: string;
  prompt_template_id: string;
  prompt_version: number;
  cost_usd: number;
  cost_source: "priced" | "unpriced";
  inference_geo: "global" | "us";
}
 
// 100 万トークンあたりの USD。公式の Pricing ページを唯一の出典にします。
// キャッシュの係数は入力単価に対して、5 分書き込み x1.25 / 1 時間書き込み x2 / 読み出し x0.1。
type Price = {
  input: number; output: number;
  cacheRead: number; cacheWrite5m: number; cacheWrite1h: number;
};
 
const PRICES_PER_MTOK: Record<string, Price> = {
  "claude-opus-5":     { input: 5, output: 25, cacheRead: 0.5, cacheWrite5m: 6.25, cacheWrite1h: 10 },
  "claude-sonnet-5":   { input: 2, output: 10, cacheRead: 0.2, cacheWrite5m: 2.5,  cacheWrite1h: 4 },
  "claude-opus-4-6":   { input: 5, output: 25, cacheRead: 0.5, cacheWrite5m: 6.25, cacheWrite1h: 10 },
  "claude-sonnet-4-6": { input: 3, output: 15, cacheRead: 0.3, cacheWrite5m: 3.75, cacheWrite1h: 6 },
  "claude-haiku-4-5":  { input: 1, output: 5,  cacheRead: 0.1, cacheWrite5m: 1.25, cacheWrite1h: 2 },
};
 
const GEO_MULTIPLIER: Record<string, number> = { global: 1, us: 1.1 };
 
function calcCost(
  model: string,
  ev: TelemetryEvent
): { cost_usd: number; cost_source: "priced" | "unpriced" } {
  const p = PRICES_PER_MTOK[model];
  // 未知のモデル ID を 0 円で埋めない。値付けできなかった事実ごと残します
  if (!p) return { cost_usd: 0, cost_source: "unpriced" };
 
  const m = 1_000_000;
  const base =
    (ev.input_tokens * p.input +
     ev.output_tokens * p.output +
     ev.cache_read_tokens * p.cacheRead +
     ev.cache_write_5m_tokens * p.cacheWrite5m +
     ev.cache_write_1h_tokens * p.cacheWrite1h) / m;
 
  return {
    cost_usd: base * (GEO_MULTIPLIER[ev.inference_geo] ?? 1),
    cost_source: "priced",
  };
}
 
export async function sendTelemetry(env: Env, event: TelemetryEvent): Promise<void> {
  // ClickHouse は JSONEachRow が最も効率がよい
  const body = JSON.stringify(event);
  const url = `${env.CLICKHOUSE_URL}/?query=INSERT+INTO+claude_api_events+FORMAT+JSONEachRow`;
 
  try {
    const res = await fetch(url, {
      method: "POST",
      headers: {
        "Authorization": `Basic ${btoa(`${env.CLICKHOUSE_USER}:${env.CLICKHOUSE_PASSWORD}`)}`,
        "Content-Type": "application/x-ndjson",
      },
      body,
    });
    if (!res.ok) {
      const text = await res.text();
      throw new Error(`ClickHouse insert failed: ${res.status} ${text}`);
    }
  } catch (err) {
    // 失敗時のフォールバック: KV にキューイングして cron で再送
    console.error("Telemetry insert failed, queuing to KV:", err);
    await env.TELEMETRY_FALLBACK.put(
      `event:${event.event_id}`,
      body,
      { expirationTtl: 7 * 24 * 60 * 60 }
    );
  }
}

// src/handler.ts
export async function handleClaudeRequest(
  req: Request,
  env: Env,
  ctx: ExecutionContext
): Promise<Response> {
  const t0 = Date.now();
  const eventId = crypto.randomUUID();
  const userId = req.headers.get("x-user-id") ?? "anonymous";
  const workspaceId = req.headers.get("x-workspace-id") ?? "default";
  const featureId = "summarize";
  const promptTemplateId = "summary-v3";
  let ttftMs = 0;
  let firstChunkSeen = false;
 
  const anthropic = new Anthropic({ apiKey: env.ANTHROPIC_API_KEY });
 
  try {
    const stream = anthropic.messages.stream({
      model: "claude-sonnet-4-6",
      max_tokens: 1024,
      messages: [{ role: "user", content: "Summarize the latest deployment notes." }],
    });
 
    for await (const _chunk of stream) {
      if (!firstChunkSeen) {
        ttftMs = Date.now() - t0;
        firstChunkSeen = true;
      }
    }
    const final = await stream.finalMessage();
 
    const event: TelemetryEvent = {
      event_id: eventId,
      request_id: (stream as { request_id?: string }).request_id ?? "",
      user_id: userId,
      workspace_id: workspaceId,
      ts: new Date().toISOString(),
      model: "claude-sonnet-4-6",
      api_version: "2023-06-01",
      input_tokens: final.usage.input_tokens,
      output_tokens: final.usage.output_tokens,
      cache_read_tokens: final.usage.cache_read_input_tokens ?? 0,
      // 合算値の cache_creation_input_tokens ではなく TTL 別の内訳を取る
      cache_write_5m_tokens: final.usage.cache_creation?.ephemeral_5m_input_tokens ?? 0,
      cache_write_1h_tokens: final.usage.cache_creation?.ephemeral_1h_input_tokens ?? 0,
      stop_reason: final.stop_reason ?? "",
      latency_ms: Date.now() - t0,
      ttft_ms: ttftMs,
      streaming: 1,
      success: 1,
      error_type: "",
      http_status: 200,
      feature_id: featureId,
      prompt_template_id: promptTemplateId,
      prompt_version: 3,
      cost_usd: 0,
      cost_source: "priced",
      inference_geo: "global",
    };
    const priced = calcCost(event.model, event);
    event.cost_usd = priced.cost_usd;
    event.cost_source = priced.cost_source;
 
    // ⚠️ 重要: バックグラウンドで送信。レスポンスを待たせない
    ctx.waitUntil(sendTelemetry(env, event));
 
    return new Response("ok");
  } catch (err) {
    const e = err as { status?: number; constructor: { name: string } };
    const event: TelemetryEvent = {
      event_id: eventId,
      request_id: "",
      user_id: userId,
      workspace_id: workspaceId,
      ts: new Date().toISOString(),
      model: "claude-sonnet-4-6",
      api_version: "2023-06-01",
      input_tokens: 0,
      output_tokens: 0,
      cache_read_tokens: 0,
      cache_write_5m_tokens: 0,
      cache_write_1h_tokens: 0,
      stop_reason: "",
      latency_ms: Date.now() - t0,
      ttft_ms: ttftMs,
      streaming: 1,
      success: 0,
      error_type: e?.constructor?.name ?? "UnknownError",
      http_status: e?.status ?? 500,
      feature_id: featureId,
      prompt_template_id: promptTemplateId,
      prompt_version: 3,
      cost_usd: 0,
      cost_source: "priced",
      inference_geo: "global",
    };
    ctx.waitUntil(sendTelemetry(env, event));
    return new Response("error", { status: 500 });
  }
}

-- 値付けできなかったトラフィックを毎朝拾う（0 件であることを確認する）
SELECT
    model,
    count() AS requests,
    sum(input_tokens + output_tokens) AS tokens,
    min(ts) AS first_seen,
    max(ts) AS last_seen
FROM claude_api_events
WHERE ts >= now() - INTERVAL 1 DAY
  AND cost_source = 'unpriced'
GROUP BY model
ORDER BY tokens DESC;

-- 分単位の事前集計テーブル
CREATE TABLE claude_minute_metrics (
    minute DateTime,
    workspace_id LowCardinality(String),
    feature_id LowCardinality(String),
    model LowCardinality(String),
 
    request_count UInt32,
    success_count UInt32,
    error_count UInt32,
    total_input_tokens UInt64,
    total_output_tokens UInt64,
    total_cache_read_tokens UInt64,
    total_cost_usd Decimal64(6),
 
    p50_latency_ms AggregateFunction(quantile(0.5), UInt32),
    p95_latency_ms AggregateFunction(quantile(0.95), UInt32),
    p99_latency_ms AggregateFunction(quantile(0.99), UInt32)
)
ENGINE = SummingMergeTree(
    -- 合算する列を明示する。省略すると「ソートキー以外の数値列すべて」が対象になります
    (request_count, success_count, error_count,
     total_input_tokens, total_output_tokens, total_cache_read_tokens,
     total_cost_usd)
)
PARTITION BY toYYYYMM(minute)
ORDER BY (workspace_id, feature_id, model, minute);
 
-- 元テーブルへの INSERT を自動で集計テーブルに反映する MV
CREATE MATERIALIZED VIEW claude_minute_metrics_mv
TO claude_minute_metrics
AS SELECT
    toStartOfMinute(ts) AS minute,
    workspace_id,
    feature_id,
    model,
    count() AS request_count,
    sumIf(1, success = 1) AS success_count,
    sumIf(1, success = 0) AS error_count,
    sum(input_tokens) AS total_input_tokens,
    sum(output_tokens) AS total_output_tokens,
    sum(cache_read_tokens) AS total_cache_read_tokens,
    sum(cost_usd) AS total_cost_usd,
    quantileState(0.5)(latency_ms) AS p50_latency_ms,
    quantileState(0.95)(latency_ms) AS p95_latency_ms,
    quantileState(0.99)(latency_ms) AS p99_latency_ms
FROM claude_api_events
GROUP BY minute, workspace_id, feature_id, model;

-- ダッシュボード用クエリ（数十ミリ秒で返る）
SELECT
    minute,
    sum(total_cost_usd) AS cost_usd,
    sum(request_count) AS requests,
    quantileMerge(0.95)(p95_latency_ms) AS p95_ms
FROM claude_minute_metrics
WHERE workspace_id = 'tenant_abc'
  AND minute >= now() - INTERVAL 24 HOUR
GROUP BY minute
ORDER BY minute;

SELECT
    toStartOfDay(ts) AS day,
    model,
    sum(cost_usd) AS cost_usd,
    sum(input_tokens + output_tokens) AS total_tokens
FROM claude_api_events
WHERE ts >= now() - INTERVAL 30 DAY
GROUP BY day, model
ORDER BY day, cost_usd DESC;

SELECT
    prompt_template_id,
    count() AS requests,
    sum(cost_usd) AS total_cost,
    avg(cost_usd) AS avg_cost,
    avg(output_tokens) AS avg_output_tokens,
    quantile(0.95)(latency_ms) AS p95_latency_ms
FROM claude_api_events
WHERE ts >= now() - INTERVAL 7 DAY
  AND success = 1
GROUP BY prompt_template_id
ORDER BY total_cost DESC
LIMIT 20;

SELECT
    feature_id,
    sum(cache_read_tokens) AS cache_read,
    sum(input_tokens) AS total_input,
    round(sum(cache_read_tokens) / nullif(sum(input_tokens), 0) * 100, 2) AS cache_hit_pct
FROM claude_api_events
WHERE ts >= now() - INTERVAL 7 DAY
GROUP BY feature_id
ORDER BY total_input DESC;

SELECT
    user_id,
    feature_id,
    count() AS requests_in_window,
    sum(cost_usd) AS cost_usd,
    sum(input_tokens) AS input_tokens
FROM claude_api_events
WHERE ts >= now() - INTERVAL 5 MINUTE
GROUP BY user_id, feature_id
HAVING requests_in_window > 50
ORDER BY requests_in_window DESC;

SELECT
    feature_id,
    prompt_template_id,
    count() AS suspicious_count,
    avg(output_tokens) AS avg_output,
    quantile(0.99)(output_tokens) AS p99_output
FROM claude_api_events
WHERE ts >= now() - INTERVAL 1 HOUR
  AND success = 1
  AND (output_tokens < 10 OR output_tokens > 3000)
GROUP BY feature_id, prompt_template_id
HAVING suspicious_count > 5;

WITH baseline AS (
    SELECT
        feature_id,
        avg(latency_ms) AS baseline_avg,
        stddevPop(latency_ms) AS baseline_std
    FROM claude_api_events
    WHERE ts BETWEEN now() - INTERVAL 1 DAY AND now() - INTERVAL 1 HOUR
      AND success = 1
    GROUP BY feature_id
)
SELECT
    e.feature_id,
    avg(e.latency_ms) AS recent_avg,
    b.baseline_avg,
    (avg(e.latency_ms) - b.baseline_avg) / nullif(b.baseline_std, 0) AS z_score
FROM claude_api_events e
JOIN baseline b USING (feature_id)
WHERE e.ts >= now() - INTERVAL 5 MINUTE
GROUP BY e.feature_id, b.baseline_avg, b.baseline_std
HAVING z_score > 3;

-- テナント単位の月次コストランキング
SELECT
    workspace_id,
    sum(cost_usd) AS api_cost_usd,
    sum(input_tokens + output_tokens) AS total_tokens,
    countDistinct(user_id) AS active_users,
    sum(cost_usd) / nullif(countDistinct(user_id), 0) AS cost_per_user
FROM claude_api_events
WHERE ts >= toStartOfMonth(now())
GROUP BY workspace_id
ORDER BY api_cost_usd DESC
LIMIT 50;

-- 月単位・モデル別の集計（Anthropic コンソールと比較する）
SELECT
    toStartOfMonth(ts) AS month,
    model,
    sum(input_tokens) AS input_tokens,
    sum(output_tokens) AS output_tokens,
    sum(cache_read_tokens) AS cache_read_tokens,
    sum(cache_write_5m_tokens) AS cache_write_5m_tokens,
    sum(cache_write_1h_tokens) AS cache_write_1h_tokens,
    countIf(cost_source = 'unpriced') AS unpriced_requests,
    round(sum(cost_usd), 2) AS estimated_cost_usd
FROM claude_api_events
WHERE ts >= now() - INTERVAL 3 MONTH
GROUP BY month, model
ORDER BY month, model;

-- 先週と今週の比較（feature_id 別）
WITH last_week AS (
    SELECT feature_id, sum(cost_usd) AS cost
    FROM claude_api_events
    WHERE ts BETWEEN now() - INTERVAL 14 DAY AND now() - INTERVAL 7 DAY
    GROUP BY feature_id
),
this_week AS (
    SELECT feature_id, sum(cost_usd) AS cost
    FROM claude_api_events
    WHERE ts >= now() - INTERVAL 7 DAY
    GROUP BY feature_id
)
SELECT
    coalesce(t.feature_id, l.feature_id) AS feature_id,
    l.cost AS last_week_cost,
    t.cost AS this_week_cost,
    (t.cost - l.cost) / nullif(l.cost, 0) * 100 AS pct_change
FROM this_week t
FULL OUTER JOIN last_week l USING (feature_id)
ORDER BY this_week_cost DESC;