障害情報の分析を自律化してみた(環境詳細)

# アラート起点の自律調査(レベル1〜3)の実験スタック。
# 監視対象は自作の計算ジョブ投入システム(job-api / slot-manager)。メトリクス・ログ・トレースを
# OTLP で Collector に送り、Prometheus / Loki / Tempo に振り分ける。
# 受け口(investigate.py)は claude CLI が必要なので docker ホストで動かす。
#   Grafana      : http://localhost:3082 (admin / admin)
#   Prometheus   : http://localhost:9092
#   Alertmanager : http://localhost:9094
#   mcp-grafana  : http://localhost:8097/mcp
#   Tempo MCP    : http://localhost:3200/api/mcp
#   Loki         : http://localhost:3101
#   job-api      : http://localhost:8100
#   slot-manager : http://localhost:8101
# 既存の Grafana / Prometheus / Loki と衝突しないようポートをずらしてある。
name: grafana-alert

x-app: &app
  build: ./app
  restart: unless-stopped
  environment:
    OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector:4317
    OTEL_EXPORTER_OTLP_INSECURE: "true"
    OTEL_SEMCONV_STABILITY_OPT_IN: http
    PYTHONUNBUFFERED: "1"
  depends_on:
    - otel-collector

services:
  job-api:
    <<: *app
    command: uvicorn job_api:app --host 0.0.0.0 --port 8000 --no-access-log
    ports:
      - "8100:8000"
    environment:
      OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector:4317
      OTEL_EXPORTER_OTLP_INSECURE: "true"
      OTEL_SEMCONV_STABILITY_OPT_IN: http
      PYTHONUNBUFFERED: "1"
      SLOT_MANAGER_URL: http://slot-manager:8000
    depends_on:
      - otel-collector
      - slot-manager

  slot-manager:
    <<: *app
    command: uvicorn slot_manager:app --host 0.0.0.0 --port 8000 --no-access-log
    ports:
      - "8101:8000"

  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.160.0
    volumes:
      - ./otel-collector/config.yaml:/etc/otelcol-contrib/config.yaml:ro
    ports:
      - "4317:4317"
      - "4318:4318"
      - "8889:8889"
    depends_on:
      - loki
      - tempo
    restart: unless-stopped

  loki:
    image: grafana/loki:3.7.7
    ports:
      - "3101:3100"
    volumes:
      - ./loki/config.yaml:/etc/loki/config.yaml:ro
      - ./loki/rules:/loki/rules:ro
      - loki-data:/loki
    command:
      - -config.file=/etc/loki/config.yaml
    restart: unless-stopped

  tempo:
    image: grafana/tempo:3.0.0
    user: "0"
    ports:
      - "3200:3200"
    volumes:
      - ./tempo/config.yaml:/etc/tempo/config.yaml:ro
      - tempo-data:/var/tempo
    command:
      - -config.file=/etc/tempo/config.yaml
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:v3.14.0
    ports:
      - "9092:9090"
    volumes:
      - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - ./prometheus/alert_rules.yml:/etc/prometheus/alert_rules.yml:ro
      - prometheus-data:/prometheus
    command:
      - --config.file=/etc/prometheus/prometheus.yml
      - --storage.tsdb.retention.time=15d
    restart: unless-stopped

  alertmanager:
    image: prom/alertmanager:v0.28.1
    ports:
      - "9094:9093"
    volumes:
      - ./alertmanager/alertmanager.yml:/etc/alertmanager/alertmanager.yml:ro
    command:
      - --config.file=/etc/alertmanager/alertmanager.yml
    extra_hosts:
      - "host.docker.internal:host-gateway"
    restart: unless-stopped

  grafana:
    image: grafana/grafana:13.2.1
    ports:
      - "3082:3000"
    environment:
      GF_SECURITY_ADMIN_USER: admin
      GF_SECURITY_ADMIN_PASSWORD: admin
      GF_USERS_ALLOW_SIGN_UP: "false"
      # スクリーンショット用にログインなしの閲覧を許可(自宅 PC 内に閉じている前提)
      GF_AUTH_ANONYMOUS_ENABLED: "true"
      GF_AUTH_ANONYMOUS_ORG_ROLE: Viewer
    volumes:
      - ./grafana/provisioning:/etc/grafana/provisioning:ro
      - ./grafana/dashboards:/var/lib/grafana/dashboards:ro
      - grafana-data:/var/lib/grafana
    depends_on:
      - prometheus
      - loki
      - tempo
    restart: unless-stopped

  mcp-grafana:
    image: grafana/mcp-grafana:1.4.1
    ports:
      - "8097:8000"
    environment:
      GRAFANA_URL: http://grafana:3000
      # 実験なので admin の基本認証。本番はサービスアカウントのトークンにする
      GRAFANA_USERNAME: admin
      GRAFANA_PASSWORD: admin
    command:
      - -t
      - streamable-http
      - --address
      - 0.0.0.0:8000
      # エージェントは読み取り専用。annotation の書き戻しは受け口が REST API で行う
      - --disable-write
      # 調査に使わないツール群を落として、ツール定義のトークン量を減らす(64 個 → 20 個弱)。
      # 残すのは alerting / annotations / datasource / prometheus / loki / proxied(Tempo の MCP)
      - --disable-admin
      - --disable-agento11y
      - --disable-api
      - --disable-asserts
      - --disable-assistant
      - --disable-cloudwatch
      - --disable-config
      - --disable-dashboard
      - --disable-docs
      - --disable-elasticsearch
      - --disable-examples
      - --disable-folder
      - --disable-graphite
      - --disable-incident
      - --disable-influxdb
      - --disable-navigation
      - --disable-oncall
      - --disable-plugin
      - --disable-provisioning
      - --disable-pyroscope
      - --disable-quickwit
      - --disable-rendering
      - --disable-runpanelquery
      - --disable-search
      - --disable-sift
      - --disable-snapshot
      - --disable-snowflake
      - --disable-sql
      - --disable-user
      - --allowed-hosts
      - ${MCP_ALLOWED_HOSTS:-*}
      - --allowed-origins
      - ${MCP_ALLOWED_ORIGINS:-*}
    depends_on:
      - grafana
    restart: unless-stopped

  # ローカル LLM 用。Claude Code の Anthropic 形式を llama.cpp(OpenAI 互換)へ変換する
  litellm:
    image: ghcr.io/berriai/litellm:main-stable
    ports:
      - "4000:4000"
    volumes:
      - ./litellm/config.yaml:/app/config.yaml:ro
    command:
      - --config
      - /app/config.yaml
      - --port
      - "4000"
    extra_hosts:
      - "host.docker.internal:host-gateway"
    restart: unless-stopped

  # 社内チャットの代わり。Alertmanager の発火通知と受け口の調査レポートを #alerts に投稿する
  mattermost-db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: mmuser
      POSTGRES_PASSWORD: mmpass
      POSTGRES_DB: mattermost
    volumes:
      - mattermost-db:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U mmuser -d mattermost"]
      interval: 5s
      timeout: 3s
      retries: 20
    restart: unless-stopped

  mattermost:
    image: mattermost/mattermost-team-edition:11.11.0
    ports:
      - "8065:8065"
    environment:
      MM_SQLSETTINGS_DRIVERNAME: postgres
      MM_SQLSETTINGS_DATASOURCE: postgres://mmuser:mmpass@mattermost-db:5432/mattermost?sslmode=disable&connect_timeout=10
      MM_SERVICESETTINGS_SITEURL: ${MM_SITEURL:-http://localhost:8065}
      MM_SERVICESETTINGS_ENABLEINCOMINGWEBHOOKS: "true"
      MM_SERVICESETTINGS_ENABLEPOSTUSERNAMEOVERRIDE: "true"
      MM_SERVICESETTINGS_ENABLEPOSTICONOVERRIDE: "true"
      MM_EMAILSETTINGS_REQUIREEMAILVERIFICATION: "false"
      MM_PLUGINSETTINGS_ENABLEMARKETPLACE: "false"
      MM_LOGSETTINGS_CONSOLELEVEL: ERROR
    volumes:
      - mattermost-config:/mattermost/config
      - mattermost-data:/mattermost/data
      - mattermost-logs:/mattermost/logs
      - mattermost-plugins:/mattermost/plugins
      - mattermost-client-plugins:/mattermost/client/plugins
    depends_on:
      mattermost-db:
        condition: service_healthy
    restart: unless-stopped

volumes:
  mattermost-db:
  mattermost-config:
  mattermost-data:
  mattermost-logs:
  mattermost-plugins:
  mattermost-client-plugins:
  prometheus-data:
  grafana-data:
  loki-data:
  tempo-data:

# アプリからの OTLP を受けて 3 つの保存先に振り分ける。
#   トレース   -> Tempo(OTLP gRPC)
#   ログ       -> Loki(OTLP HTTP。service.name などがインデックスのラベルになる)
#   メトリクス -> :8889 で Prometheus 形式で公開し、Prometheus がスクレイプする
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 2s

exporters:
  otlp/tempo:
    endpoint: tempo:4317
    tls:
      insecure: true
  otlphttp/loki:
    endpoint: http://loki:3100/otlp
  prometheus:
    endpoint: 0.0.0.0:8889
    # service.name などのリソース属性を service_name のようなラベルにも展開する
    resource_to_telemetry_conversion:
      enabled: true

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlp/tempo]
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [prometheus]
    logs:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlphttp/loki]

global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - /etc/prometheus/alert_rules.yml

alerting:
  alertmanagers:
    - static_configs:
        - targets: ["alertmanager:9093"]

scrape_configs:
  - job_name: prometheus
    static_configs:
      - targets: ["localhost:9090"]

  # アプリのメトリクスは OTLP で Collector に届き、Collector が Prometheus 形式で公開する。
  # honor_labels: Collector が付ける job / instance(= service.name / service.instance.id)を優先する
  - job_name: otel-collector
    honor_labels: true
    static_configs:
      - targets: ["otel-collector:8889"]

# 実験 1〜3 に対応するアラートルール。
# しきい値は人が書く。ラベルに原因のヒント(scenario など)を入れない。
# 実験 1(負荷)と実験 3(下流の DB の遅延)は同じ JobApiLatencyHigh を鳴らし、原因だけが違う。
groups:
  - name: job-system
    rules:
      - alert: JobApiLatencyHigh
        expr: histogram_quantile(0.95, sum(rate(jobs_duration_seconds_bucket[2m])) by (le)) > 0.5
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "ジョブ投入 API の p95 レイテンシが 1 分以上 0.5 秒を超えている"
          description: "直近 2 分の p95 = {{ $value | humanizeDuration }}"

      - alert: JobErrorRateHigh
        expr: sum(rate(jobs_total{outcome="failed"}[2m])) / sum(rate(jobs_total[2m])) > 0.2
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "ジョブ投入の失敗率が 1 分以上 20% を超えている"
          description: "直近 2 分の失敗率 = {{ $value | humanizePercentage }}"

      - alert: ScrapeTargetDown
        expr: up == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "スクレイプ対象 {{ $labels.job }} が応答しない"
          description: "{{ $labels.instance }} の up が 1 分以上 0"

# alertmanager.yml のひな型。setup_mattermost.py が {{MATTERMOST_HOOK_ID}} を埋めて alertmanager.yml を書く。
# 発火したアラートは 2 か所へ送る:
#   1. docker ホストの受け口(investigate.py、ポート 9095)。AI 調査を起動する
#   2. Mattermost の #alerts(Slack 互換の incoming webhook)。運用者への通知
route:
  receiver: ai-investigator
  group_by: ["alertname"]
  group_wait: 10s
  group_interval: 1m
  repeat_interval: 4h

receivers:
  - name: ai-investigator
    webhook_configs:
      - url: http://host.docker.internal:9095/alert
        send_resolved: false
    slack_configs:
      - api_url: http://mattermost:8065/hooks/{{MATTERMOST_HOOK_ID}}
        channel: "#alerts"
        username: alertmanager
        send_resolved: true
        title: "{{ .CommonLabels.alertname }} [{{ .Status | toUpper }}]"
        text: '{{ range .Alerts }}{{ .Annotations.summary }}{{ if .Annotations.description }}: {{ .Annotations.description }}{{ end }}{{ "\n" }}{{ end }}'
        color: '{{ if eq .Status "firing" }}danger{{ else }}good{{ end }}'

# 単一ノードの Loki。OTLP で受けたログをローカルディスクに保存する。
# ruler を有効にして、ログ由来のアラート(rules/fake/*.yml)を Alertmanager に送る。
auth_enabled: false

server:
  http_listen_port: 3100
  grpc_listen_port: 9096

common:
  path_prefix: /loki
  storage:
    filesystem:
      chunks_directory: /loki/chunks
      rules_directory: /loki/rules
  replication_factor: 1
  ring:
    kvstore:
      store: inmemory

schema_config:
  configs:
    - from: "2024-04-01"
      store: tsdb
      object_store: filesystem
      schema: v13
      index:
        prefix: index_
        period: 24h

limits_config:
  allow_structured_metadata: true
  volume_enabled: true

ruler:
  storage:
    type: local
    local:
      directory: /loki/rules
  rule_path: /tmp/loki-rules
  alertmanager_url: http://alertmanager:9093
  ring:
    kvstore:
      store: inmemory
  enable_api: true

# ログ由来のアラート(実験 2b)。メトリクスには何も出ず、ログの行数だけで鳴る。
# "fake" は auth_enabled: false のときのテナント名。
groups:
  - name: job-system
    rules:
      - alert: SlotsRunningLow
        expr: sum(count_over_time({service_name="slot-manager"} |= "slots running low" [5m])) > 3
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "計算枠が少ないという警告ログが 5 分で 3 行を超えた"

# 単一ノードの Tempo。OTLP で受けたトレースをローカルディスクに保存する。
# query_frontend.mcp_server を有効にすると http://tempo:3200/api/mcp に
# MCP サーバ(traceql-search / get-trace など)が立つ。mcp-grafana には Tempo ツールが無いので併用する。
server:
  http_listen_port: 3200

distributor:
  receivers:
    otlp:
      protocols:
        grpc:
          endpoint: 0.0.0.0:4317
        http:
          endpoint: 0.0.0.0:4318

storage:
  trace:
    backend: local
    local:
      path: /var/tempo/blocks
    wal:
      path: /var/tempo/wal

query_frontend:
  mcp_server:
    enabled: true

apiVersion: 1
datasources:
  - name: Prometheus
    type: prometheus
    uid: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true

  - name: Loki
    type: loki
    uid: loki
    access: proxy
    url: http://loki:3100

  - name: Tempo
    type: tempo
    uid: tempo
    access: proxy
    url: http://tempo:3200
    jsonData:
      tracesToLogsV2:
        datasourceUid: loki
        spanStartTimeShift: "-5m"
        spanEndTimeShift: "5m"
        filterByTraceID: true
      serviceMap:
        datasourceUid: prometheus

apiVersion: 1
providers:
  - name: grafana-alert
    folder: ""
    type: file
    options:
      path: /var/lib/grafana/dashboards

{
  "uid": "job-system",
  "title": "Job system",
  "tags": ["grafana-alert"],
  "timezone": "browser",
  "schemaVersion": 39,
  "refresh": "10s",
  "time": { "from": "now-30m", "to": "now" },
  "annotations": {
    "list": [
      {
        "name": "AI investigation",
        "datasource": { "type": "grafana", "uid": "-- Grafana --" },
        "enable": true,
        "iconColor": "orange",
        "target": {
          "type": "tags",
          "tags": ["ai-investigation"],
          "limit": 100,
          "matchAny": true
        }
      }
    ]
  },
  "panels": [
    {
      "id": 1,
      "type": "timeseries",
      "title": "ジョブ投入 p95 レイテンシ(アラート式)",
      "gridPos": { "x": 0, "y": 0, "w": 12, "h": 8 },
      "datasource": { "type": "prometheus", "uid": "prometheus" },
      "fieldConfig": {
        "defaults": {
          "unit": "s",
          "thresholds": {
            "mode": "absolute",
            "steps": [
              { "color": "green", "value": null },
              { "color": "red", "value": 0.5 }
            ]
          },
          "custom": { "thresholdsStyle": { "mode": "line" } }
        },
        "overrides": []
      },
      "targets": [
        {
          "refId": "A",
          "expr": "histogram_quantile(0.95, sum(rate(jobs_duration_seconds_bucket[2m])) by (le))",
          "legendFormat": "p95"
        }
      ]
    },
    {
      "id": 2,
      "type": "timeseries",
      "title": "ジョブ投入レート(結末別)",
      "gridPos": { "x": 12, "y": 0, "w": 12, "h": 8 },
      "datasource": { "type": "prometheus", "uid": "prometheus" },
      "fieldConfig": { "defaults": { "unit": "reqps" }, "overrides": [] },
      "targets": [
        {
          "refId": "A",
          "expr": "sum(rate(jobs_total[1m])) by (outcome)",
          "legendFormat": "{{outcome}}"
        }
      ]
    },
    {
      "id": 3,
      "type": "timeseries",
      "title": "ジョブ投入の失敗率(アラート式)",
      "gridPos": { "x": 0, "y": 8, "w": 12, "h": 8 },
      "datasource": { "type": "prometheus", "uid": "prometheus" },
      "fieldConfig": {
        "defaults": {
          "unit": "percentunit",
          "min": 0,
          "max": 1,
          "thresholds": {
            "mode": "absolute",
            "steps": [
              { "color": "green", "value": null },
              { "color": "red", "value": 0.2 }
            ]
          },
          "custom": { "thresholdsStyle": { "mode": "line" } }
        },
        "overrides": []
      },
      "targets": [
        {
          "refId": "A",
          "expr": "sum(rate(jobs_total{outcome=\"failed\"}[2m])) / sum(rate(jobs_total[2m]))",
          "legendFormat": "failed"
        }
      ]
    },
    {
      "id": 4,
      "type": "timeseries",
      "title": "サービス別 HTTP p95(自動計装)",
      "gridPos": { "x": 12, "y": 8, "w": 12, "h": 8 },
      "datasource": { "type": "prometheus", "uid": "prometheus" },
      "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] },
      "targets": [
        {
          "refId": "A",
          "expr": "histogram_quantile(0.95, sum(rate(http_server_request_duration_seconds_bucket[2m])) by (le, service_name))",
          "legendFormat": "{{service_name}}"
        }
      ]
    },
    {
      "id": 5,
      "type": "logs",
      "title": "WARNING / ERROR ログ",
      "gridPos": { "x": 0, "y": 16, "w": 16, "h": 9 },
      "datasource": { "type": "loki", "uid": "loki" },
      "options": {
        "showTime": true,
        "sortOrder": "Descending",
        "wrapLogMessage": true
      },
      "targets": [
        {
          "refId": "A",
          "expr": "{service_name=~\"job-api|slot-manager\"} | detected_level=~\"warn|error\"",
          "queryType": "range"
        }
      ]
    },
    {
      "id": 6,
      "type": "stat",
      "title": "スクレイプ対象 up",
      "gridPos": { "x": 16, "y": 16, "w": 8, "h": 9 },
      "datasource": { "type": "prometheus", "uid": "prometheus" },
      "fieldConfig": {
        "defaults": {
          "thresholds": {
            "mode": "absolute",
            "steps": [
              { "color": "red", "value": null },
              { "color": "green", "value": 1 }
            ]
          }
        },
        "overrides": []
      },
      "options": {
        "reduceOptions": { "calcs": ["lastNotNull"] },
        "textMode": "value_and_name"
      },
      "targets": [{ "refId": "A", "expr": "up", "legendFormat": "{{job}}" }]
    }
  ]
}

# ローカル LLM を Claude Code から使うための変換プロキシ。
# Claude Code は Anthropic Messages API(/v1/messages)を話すので、LiteLLM で受けて
# OpenAI 互換の llama.cpp(既存スタックの :8099)へ流す。
#   claude 側: ANTHROPIC_BASE_URL=http://localhost:4000  ANTHROPIC_AUTH_TOKEN=sk-local  --model qwen3.6-35b-a3b
model_list:
  - model_name: qwen3.6-35b-a3b
    litellm_params:
      model: openai/qwen3.6-35b-a3b
      api_base: http://host.docker.internal:8099/v1
      api_key: none
      timeout: 600

litellm_settings:
  drop_params: true
  set_verbose: false

general_settings:
  master_key: sk-local

FROM python:3.13-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

fastapi
uvicorn
httpx
opentelemetry-sdk
opentelemetry-exporter-otlp-proto-grpc
opentelemetry-instrumentation-fastapi
opentelemetry-instrumentation-httpx

"""OTel の配線。トレース・メトリクス・ログをまとめて OTLP(gRPC)で Collector に送る。

送り先は環境変数 OTEL_EXPORTER_OTLP_ENDPOINT(compose では otel-collector:4317)。
ログは標準 logging に LoggingHandler を足すだけで、trace_id / span_id が自動で付く。
"""

from __future__ import annotations

import logging
import os
import socket

from opentelemetry import metrics, trace
from opentelemetry._logs import set_logger_provider
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
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


def setup(service_name: str) -> tuple[trace.Tracer, metrics.Meter]:
    resource = Resource.create({
        "service.name": service_name,
        "service.version": os.environ.get("APP_VERSION", "0.1.0"),
        "service.instance.id": socket.gethostname(),
    })

    tp = TracerProvider(resource=resource)
    tp.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    trace.set_tracer_provider(tp)

    mp = MeterProvider(
        resource=resource,
        metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter(), export_interval_millis=5000)],
    )
    metrics.set_meter_provider(mp)

    lp = LoggerProvider(resource=resource)
    lp.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter()))
    set_logger_provider(lp)

    # 標準出力(docker logs)にも同じ行を出す
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    logging.getLogger().addHandler(LoggingHandler(level=logging.NOTSET, logger_provider=lp))

    return trace.get_tracer(service_name), metrics.get_meter(service_name)

"""job-api サービス。計算ジョブの投入を受けて slot-manager に計算枠の確保を頼む。

- POST /jobs {"user", "job", "size"}: slot-manager の /reserve を呼び、結果を返す。
  メトリクス jobs(結末別の件数)と jobs.duration(所要秒)を記録する
- POST /admin/fault: 故障注入。{"duplicate_calls": 8} で 1 ジョブあたり /reserve を 8 回呼ぶ(冪等でない再送のバグ)。
  {"reset": true} で戻す
- GET /admin/fault、GET /health

メトリクスのラベルは outcome(ok / failed)だけ。ジョブ種別やエラー内容はログとスパン属性にだけ出る。
"""

from __future__ import annotations

import itertools
import logging
import os
import time

import httpx
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

import telemetry

tracer, meter = telemetry.setup("job-api")
log = logging.getLogger("job-api")

app = FastAPI(title="job-api")
FastAPIInstrumentor.instrument_app(app, excluded_urls="health,admin")
HTTPXClientInstrumentor().instrument()

SLOT_MANAGER_URL = os.environ.get("SLOT_MANAGER_URL", "http://slot-manager:8000").rstrip("/")
FAULT: dict[str, int] = {"duplicate_calls": 1}
# 種別ごとの見積もり時間(秒)。ログに載せるだけ
ESTIMATE_S = {"mip": 600, "lp": 60, "simulation": 300, "ml": 1800}

jobs_counter = meter.create_counter("jobs", description="ジョブ投入の結末別の件数")
jobs_duration = meter.create_histogram(
    "jobs.duration", unit="s", description="ジョブ投入 1 件の所要時間(枠の確保まで)",
    explicit_bucket_boundaries_advisory=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10],
)
_seq = itertools.count(1000)
_client = httpx.AsyncClient(timeout=30.0, limits=httpx.Limits(max_connections=200, max_keepalive_connections=50))


@app.post("/jobs")
async def submit_job(body: dict):
    t0 = time.perf_counter()
    job_id = f"J-{next(_seq)}"
    job = body.get("job", "")
    size = int(body.get("size", 1))
    user = body.get("user", "anonymous")
    attrs = {"job.id": job_id, "user.id": user, "job.type": job, "job.size": size}
    span = trace.get_current_span()
    for k, v in attrs.items():
        span.set_attribute(k, v)
    log.info("job received", extra=attrs)

    outcome, status, detail = "ok", 200, None
    try:
        for _ in range(max(1, FAULT["duplicate_calls"])):
            r = await _client.post(f"{SLOT_MANAGER_URL}/reserve", json={"job": job, "size": size})
        if r.status_code != 200:
            outcome, status = "failed", r.status_code
            try:
                detail = r.json().get("error", r.text)
            except ValueError:
                detail = r.text
    except httpx.HTTPError as e:
        outcome, status, detail = "failed", 502, f"slot-manager unreachable: {type(e).__name__}: {e}"

    elapsed = time.perf_counter() - t0
    jobs_counter.add(1, {"outcome": outcome})
    jobs_duration.record(elapsed, {"outcome": outcome})
    span.set_attribute("job.outcome", outcome)

    if outcome == "ok":
        log.info("job submitted", extra={**attrs, "job.estimate_s": ESTIMATE_S.get(job, 0)})
        return {"job_id": job_id, "job": job, "size": size, "estimate_s": ESTIMATE_S.get(job, 0)}
    log.error("job failed: %s", detail, extra={**attrs, "http.response.status_code": status})
    return JSONResponse({"job_id": job_id, "error": detail}, status_code=status)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/admin/fault")
async def get_fault():
    return {"fault": FAULT, "slot_manager_url": SLOT_MANAGER_URL}


@app.post("/admin/fault")
async def set_fault(body: dict):
    if body.get("reset"):
        FAULT["duplicate_calls"] = 1
    if "duplicate_calls" in body:
        FAULT["duplicate_calls"] = int(body["duplicate_calls"])
    return await get_fault()

"""slot-manager サービス。計算ジョブごとに計算枠(ライセンス席、GPU/CPU スロット)を確保する。job-api の下流。

- POST /reserve {"job", "size"}: 枠を確認して確保する。 DB へのアクセスは sleep で模擬し、
  接続プール(DB_POOL 本)を取り合う。プール待ちは db.pool.acquire、問い合わせは db.query のスパンになる
- POST /admin/fault: 故障注入。{"latency_ms": 300} で DB を遅くする、{"slots": {"mip": 0}} で枠数を書き換える、
  {"error_rate": 0.5} で DB エラーを混ぜる。{"reset": true} で全部戻す
- GET /admin/fault: 現在の故障設定
- GET /health

枠切れは 409 とエラーログ「failed to reserve: no free slot: <job>」で表す。
メトリクスにはジョブ種別のラベルを付けない(どの種別かはログにしか出ない)。
"""

from __future__ import annotations

import asyncio
import logging
import os
import random

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

import telemetry

tracer, meter = telemetry.setup("slot-manager")
log = logging.getLogger("slot-manager")

app = FastAPI(title="slot-manager")
FastAPIInstrumentor.instrument_app(app, excluded_urls="health,admin")

BIG = 10**9
JOB_TYPES = ("mip", "lp", "simulation", "ml")
SLOTS: dict[str, int] = {job: BIG for job in JOB_TYPES}
FAULT: dict[str, float] = {"latency_ms": 0, "error_rate": 0.0}
DB_BASE_MS = int(os.environ.get("DB_BASE_MS", "40"))
DB_POOL = int(os.environ.get("DB_POOL", "4"))
LOW_WATERMARK = 50

_pool = asyncio.Semaphore(DB_POOL)


async def db_query(operation: str) -> None:
    """DB 呼び出しの模擬。プール待ちと問い合わせ本体を別のスパンにする。"""
    with tracer.start_as_current_span("db.pool.acquire"):
        await _pool.acquire()
    try:
        with tracer.start_as_current_span("db.query") as span:
            span.set_attribute("db.system.name", "sqlite")
            span.set_attribute("db.operation.name", operation)
            await asyncio.sleep((DB_BASE_MS + FAULT["latency_ms"]) / 1000)
    finally:
        _pool.release()


@app.post("/reserve")
async def reserve(body: dict):
    job = body.get("job", "")
    size = int(body.get("size", 1))
    if job not in SLOTS:
        return JSONResponse({"error": f"unknown job type: {job}"}, status_code=400)

    await db_query("SELECT slots")
    if random.random() < FAULT["error_rate"]:
        log.error("db connection reset while reserving", extra={"job.type": job})
        return JSONResponse({"error": "db error"}, status_code=500)
    if SLOTS[job] < size:
        log.error(
            "failed to reserve: no free slot: %s", job,
            extra={"job.type": job, "job.size": size, "slots.remaining": SLOTS[job]},
        )
        return JSONResponse({"error": "no free slot"}, status_code=409)

    SLOTS[job] -= size
    await db_query("UPDATE slots")
    log.info("reserved: %s x%d", job, size, extra={"job.type": job, "job.size": size, "slots.remaining": SLOTS[job]})
    if SLOTS[job] < LOW_WATERMARK:
        log.warning(
            "slots running low: %s", job,
            extra={"job.type": job, "slots.remaining": SLOTS[job]},
        )
    return {"job": job, "remaining": SLOTS[job]}


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/admin/fault")
async def get_fault():
    return {"fault": FAULT, "slots": {k: v for k, v in SLOTS.items()}, "db_base_ms": DB_BASE_MS, "db_pool": DB_POOL}


@app.post("/admin/fault")
async def set_fault(body: dict):
    if body.get("reset"):
        FAULT.update({"latency_ms": 0, "error_rate": 0.0})
        for job in JOB_TYPES:
            SLOTS[job] = BIG
    if "latency_ms" in body:
        FAULT["latency_ms"] = float(body["latency_ms"])
    if "error_rate" in body:
        FAULT["error_rate"] = float(body["error_rate"])
    for job, n in (body.get("slots") or {}).items():
        if job in SLOTS:
            SLOTS[job] = int(n)
    return await get_fault()

"""Alertmanager の webhook を受けて Claude Code(headless)に調査させる受け口。

- POST /alert に Alertmanager の webhook ペイロードが届く
- firing のアラートごとに `claude -p` を起動し、Grafana MCP(読み取り専用。Prometheus / Loki / Tempo)で調べさせる
- 結果を reports/ に Markdown で保存し、Grafana の annotation に書き戻す

自作するのはこのファイルだけ。標準ライブラリのみで動く。
"""

from __future__ import annotations

import base64
import json
import os
import shlex
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

HERE = Path(__file__).resolve().parent
PORT = int(os.environ.get("PORT", "9095"))
CLAUDE_BIN = os.environ.get("CLAUDE_BIN", "claude")
CLAUDE_MODEL = os.environ.get("CLAUDE_MODEL", "sonnet")
MCP_CONFIG = os.environ.get("MCP_CONFIG", str(HERE / "mcp.json"))
PROMPT_FILE = os.environ.get("PROMPT_FILE", str(HERE / "PROMPT.md"))
REPORT_DIR = Path(os.environ.get("REPORT_DIR", str(HERE / "reports")))
GRAFANA_URL = os.environ.get("GRAFANA_URL", "http://localhost:3082").rstrip("/")
GRAFANA_SA_TOKEN = os.environ.get("GRAFANA_SA_TOKEN", "")
# トークンが無ければ基本認証(実験用。既定は admin:admin)
GRAFANA_BASIC_AUTH = os.environ.get("GRAFANA_BASIC_AUTH", "admin:admin")
# エージェントに許すツール。mcp-grafana の MCP だけ(Tempo のツールも mcp-grafana が tempo_* として中継する)
ALLOWED_TOOLS = os.environ.get("ALLOWED_TOOLS", "mcp__grafana__*")
# 同じアラート(fingerprint)はこの秒数の間は再調査しない
DEDUP_SECONDS = int(os.environ.get("DEDUP_SECONDS", "1800"))
# Mattermost の incoming webhook。未設定なら .env の MATTERMOST_HOOK_ID から組み立てる(setup_mattermost.py が書く)
MATTERMOST_URL = os.environ.get("MATTERMOST_URL", "http://localhost:8065").rstrip("/")
MATTERMOST_WEBHOOK_URL = os.environ.get("MATTERMOST_WEBHOOK_URL", "")
# 投稿先チャンネル。モデルごとに分けるときは受け口の起動時に変える(例: sonnet / qwen)
MATTERMOST_CHANNEL = os.environ.get("MATTERMOST_CHANNEL", "alerts")
MATTERMOST_MAX_CHARS = 16000  # Mattermost の投稿上限は 16,383 文字
# Claude Code の組み込みツール。エージェントには MCP ツールだけを残す
BUILTIN_TOOLS = "Bash,PowerShell,Edit,Write,MultiEdit,NotebookEdit,Read,Glob,Grep,LS,WebFetch,WebSearch,Agent,TodoWrite,Skill"

_recent: dict[str, float] = {}
_lock = threading.Lock()


def log(msg: str) -> None:
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)


def build_prompt(alert: dict) -> str:
    template = Path(PROMPT_FILE).read_text(encoding="utf-8")
    return template.replace("{alert_json}", json.dumps(alert, ensure_ascii=False, indent=2))


def run_claude(prompt: str) -> dict:
    """claude -p を起動して JSON 結果を返す。組み込みツールは拒否し、MCP ツールだけ許可する。"""
    cmd = [
        *shlex.split(CLAUDE_BIN, posix=(os.name != "nt")),
        "-p",
        "--model", CLAUDE_MODEL,
        "--output-format", "json",
        "--mcp-config", MCP_CONFIG,
        "--strict-mcp-config",
        "--allowedTools", ALLOWED_TOOLS,
        # 組み込みツールは名前で全部拒否する(--tools "" だと MCP ツールの許可まで消える)
        "--disallowedTools", BUILTIN_TOOLS,
        "--no-session-persistence",
    ]
    started = time.monotonic()
    proc = subprocess.run(cmd, input=prompt, capture_output=True, text=True, encoding="utf-8")
    elapsed = time.monotonic() - started
    try:
        data = json.loads(proc.stdout)
    except json.JSONDecodeError:
        data = None
    if proc.returncode != 0 and not (isinstance(data, dict) and data.get("result")):
        raise RuntimeError(f"claude exited {proc.returncode}: {proc.stderr[-1500:]} {proc.stdout[-1500:]}")
    if proc.returncode != 0:
        # API エラーなどで途中終了しても、結果(エラー文)を残す。ローカルモデルのコンテキスト超過の記録用
        log(f"claude exited {proc.returncode}(結果は保存する): {str(data.get('result'))[:200]}")
        data["is_error"] = True
    data.setdefault("duration_ms", int(elapsed * 1000))
    data["_stderr"] = proc.stderr
    return data


def grafana_auth_header() -> str:
    if GRAFANA_SA_TOKEN:
        return f"Bearer {GRAFANA_SA_TOKEN}"
    return "Basic " + base64.b64encode(GRAFANA_BASIC_AUTH.encode("utf-8")).decode("ascii")


def post_annotation(alert: dict, report_text: str) -> None:
    labels = alert.get("labels", {})
    started = alert.get("startsAt", "")
    try:
        t = int(datetime.fromisoformat(started.replace("Z", "+00:00")).timestamp() * 1000)
    except ValueError:
        t = int(time.time() * 1000)
    body = {
        "time": t,
        "tags": ["ai-investigation", labels.get("alertname", "unknown")],
        "text": f"[{labels.get('alertname')}] AI 調査\n\n" + report_text[:1500],
    }
    req = urllib.request.Request(
        f"{GRAFANA_URL}/api/annotations",
        data=json.dumps(body).encode("utf-8"),
        headers={
            "Content-Type": "application/json",
            "Authorization": grafana_auth_header(),
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            log(f"annotation 作成: HTTP {resp.status}")
    except urllib.error.URLError as e:
        log(f"annotation 作成に失敗: {e}")


def mattermost_webhook_url() -> str:
    if MATTERMOST_WEBHOOK_URL:
        return MATTERMOST_WEBHOOK_URL
    env = HERE / ".env"
    if env.exists():
        for line in env.read_text(encoding="utf-8").splitlines():
            if line.startswith("MATTERMOST_HOOK_ID="):
                return f"{MATTERMOST_URL}/hooks/{line.split('=', 1)[1].strip()}"
    return ""


def post_mattermost(alert: dict, report_text: str, result: dict) -> None:
    """調査レポートの全文を Mattermost の #alerts に投稿する(Markdown はそのまま描画される)"""
    url = mattermost_webhook_url()
    if not url:
        log("Mattermost の webhook が未設定なので投稿しない")
        return
    labels = alert.get("labels", {})
    head = (f"### [{labels.get('alertname')}] AI 調査レポート\n"
            f"発火 {alert.get('startsAt', '')} / モデル {CLAUDE_MODEL} / "
            f"所要 {result.get('duration_ms', 0) / 1000:.0f} 秒、{result.get('num_turns')} ターン\n\n")
    text = head + report_text
    if len(text) > MATTERMOST_MAX_CHARS:
        text = text[:MATTERMOST_MAX_CHARS] + "\n\n…(上限のため省略)"
    body = {"channel": MATTERMOST_CHANNEL, "username": f"ai-investigator ({CLAUDE_MODEL})", "text": text}
    req = urllib.request.Request(url, data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
                                 headers={"Content-Type": "application/json"}, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            log(f"Mattermost 投稿: HTTP {resp.status}")
    except (urllib.error.URLError, OSError) as e:
        log(f"Mattermost 投稿に失敗: {e}")


def investigate(alert: dict) -> None:
    labels = alert.get("labels", {})
    name = labels.get("alertname", "unknown")
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    log(f"調査開始: {name} {labels}")
    try:
        result = run_claude(build_prompt(alert))
    except Exception as e:  # noqa: BLE001
        log(f"調査失敗: {name}: {e}")
        return
    report = result.get("result", "")
    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    path = REPORT_DIR / f"{stamp}_{name}.md"
    header = (
        f"# {name}\n\n"
        f"- 開始: {alert.get('startsAt')}\n"
        f"- ラベル: `{json.dumps(labels, ensure_ascii=False)}`\n"
        f"- モデル: {CLAUDE_MODEL}\n"
        f"- 所要: {result.get('duration_ms', 0) / 1000:.1f} 秒、ターン数: {result.get('num_turns')}\n\n"
    )
    path.write_text(header + report + "\n", encoding="utf-8")
    if result.get("_stderr"):
        # エージェントの標準エラー(local_agent.py ならツール呼び出しの記録)を並べて残す
        path.with_suffix(".log").write_text(result["_stderr"], encoding="utf-8")
    with (REPORT_DIR / "results.jsonl").open("a", encoding="utf-8") as f:
        f.write(json.dumps({
            "time": stamp,
            "alertname": name,
            "model": CLAUDE_MODEL,
            "duration_ms": result.get("duration_ms"),
            "num_turns": result.get("num_turns"),
            "is_error": bool(result.get("is_error")),
            "usage": result.get("usage"),
            "report": str(path.name),
        }, ensure_ascii=False) + "\n")
    log(f"レポート保存: {path}")
    post_annotation(alert, report)
    post_mattermost(alert, report, result)


class Handler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:  # noqa: N802
        if self.path != "/alert":
            self.send_error(404)
            return
        length = int(self.headers.get("Content-Length", "0"))
        try:
            payload = json.loads(self.rfile.read(length) or b"{}")
        except json.JSONDecodeError:
            self.send_error(400, "invalid json")
            return
        self.send_response(200)
        self.end_headers()
        now = time.time()
        for alert in payload.get("alerts", []):
            if alert.get("status") != "firing":
                continue
            key = alert.get("fingerprint") or json.dumps(alert.get("labels", {}), sort_keys=True)
            with _lock:
                if now - _recent.get(key, 0) < DEDUP_SECONDS:
                    log(f"スキップ(直近に調査済み): {alert.get('labels', {}).get('alertname')}")
                    continue
                _recent[key] = now
            threading.Thread(target=investigate, args=(alert,), daemon=True).start()

    def log_message(self, fmt: str, *args) -> None:  # noqa: D102
        return


def main() -> None:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")  # Windows のコンソールでも日本語ログを崩さない
    server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    log(f"受け口を起動: http://0.0.0.0:{PORT}/alert (claude={CLAUDE_BIN}, model={CLAUDE_MODEL}, tools={ALLOWED_TOOLS})")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    sys.exit(main())

"""Mattermost の初期設定(標準ライブラリのみ)。

    python setup_mattermost.py [--url http://localhost:8065] [--admin-password ...]

1. 最初のユーザー(= システム管理者)admin を作る
2. チーム ops とチャンネル alerts を作る
3. alerts への incoming webhook を作る
4. webhook の id を .env(MATTERMOST_HOOK_ID)に書き、alertmanager/alertmanager.tmpl.yml から alertmanager/alertmanager.yml を生成する

再実行しても既存のものを使い回す(冪等)。終わったら `docker compose restart alertmanager` で通知先を反映する。
"""

from __future__ import annotations

import argparse
import json
import re
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

HERE = Path(__file__).resolve().parent


def api(base: str, method: str, path: str, body: dict | None = None, token: str | None = None):
    req = urllib.request.Request(f"{base}/api/v4{path}", method=method,
                                 data=json.dumps(body).encode("utf-8") if body is not None else None,
                                 headers={"Content-Type": "application/json"})
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            raw = resp.read()
            return resp.status, (json.loads(raw) if raw else None), resp.headers
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or b"{}"), e.headers


def wait_ready(base: str) -> None:
    for _ in range(60):
        try:
            with urllib.request.urlopen(f"{base}/api/v4/system/ping", timeout=5) as resp:
                if resp.status == 200:
                    return
        except (urllib.error.URLError, TimeoutError, OSError):
            pass
        time.sleep(3)
    raise SystemExit("Mattermost が起動しない")


def main() -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--url", default="http://localhost:8065")
    p.add_argument("--admin-user", default="admin")
    p.add_argument("--admin-password", default="Admin-pass-1234")
    p.add_argument("--admin-email", default="admin@example.invalid")
    p.add_argument("--team", default="ops")
    p.add_argument("--channel", default="alerts")
    a = p.parse_args()
    base = a.url.rstrip("/")
    wait_ready(base)

    # 1. 管理者(最初のユーザーがシステム管理者になる)
    status, body, _ = api(base, "POST", "/users", {"email": a.admin_email, "username": a.admin_user, "password": a.admin_password})
    print(f"user: {status} {body.get('username') if isinstance(body, dict) else body}")
    status, body, headers = api(base, "POST", "/users/login", {"login_id": a.admin_user, "password": a.admin_password})
    if status != 200:
        raise SystemExit(f"login failed: {status} {body}")
    token = headers["Token"]

    # 2. チームとチャンネル
    status, team, _ = api(base, "GET", f"/teams/name/{a.team}", token=token)
    if status != 200:
        status, team, _ = api(base, "POST", "/teams", {"name": a.team, "display_name": a.team.capitalize(), "type": "O"}, token=token)
    print(f"team: {team['name']} ({team['id']})")
    status, ch, _ = api(base, "GET", f"/teams/{team['id']}/channels/name/{a.channel}", token=token)
    if status != 200:
        status, ch, _ = api(base, "POST", "/channels", {"team_id": team["id"], "name": a.channel, "display_name": a.channel, "type": "O"}, token=token)
    print(f"channel: {ch['name']} ({ch['id']})")

    # 3. incoming webhook(display_name で既存を探す)
    status, hooks, _ = api(base, "GET", f"/hooks/incoming?team_id={team['id']}&per_page=200", token=token)
    hook = next((h for h in (hooks or []) if h.get("display_name") == "alert-investigation"), None)
    if not hook:
        status, hook, _ = api(base, "POST", "/hooks/incoming", {
            "channel_id": ch["id"], "display_name": "alert-investigation",
            "description": "Alertmanager の発火通知と AI 調査レポート",
        }, token=token)
    print(f"webhook: {hook['id']}")

    # 4. .env と alertmanager.yml
    env_path = HERE / ".env"
    lines = env_path.read_text(encoding="utf-8").splitlines() if env_path.exists() else []
    lines = [l for l in lines if not l.startswith("MATTERMOST_HOOK_ID=")]
    lines.append(f"MATTERMOST_HOOK_ID={hook['id']}")
    env_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    tmpl = (HERE / "alertmanager" / "alertmanager.tmpl.yml").read_text(encoding="utf-8")
    (HERE / "alertmanager" / "alertmanager.yml").write_text(tmpl.replace("{{MATTERMOST_HOOK_ID}}", hook["id"]), encoding="utf-8")
    print(f"wrote .env and alertmanager/alertmanager.yml. next: docker compose restart alertmanager")
    print(f"login: {base}  user={a.admin_user}  channel=~{a.channel}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

あなたは Grafana に接続された SRE の調査エージェントです。
Alertmanager から次のアラートが届きました。

```json
{alert_json}
```

監視対象は計算ジョブの投入システムで、`job-api` と `slot-manager` の 2 サービスからなります。
`job-api` はジョブ 1 件ごとに `slot-manager` の `POST /reserve` を呼び、`slot-manager` は DB で計算枠を確認して確保します。
3 つのデータソースが使えます。

- Prometheus(メトリクス): `jobs_total{outcome}`、`jobs_duration_seconds_bucket` など。`mcp__grafana__query_prometheus`
- Loki(ログ): `{service_name="job-api"}`、`{service_name="slot-manager"}`。`mcp__grafana__query_loki_logs`
- Tempo(トレース): `mcp__grafana__tempo_traceql-search` と `mcp__grafana__tempo_get-trace`

MCP ツール(`mcp__grafana__*`)だけを使って原因を調べてください。
ツールは読み取り専用です。設定変更や書き込みはできません。

クエリの書き方:

- PromQL の時間範囲は `startsAt` の前後 30 分。range クエリには `stepSeconds=60` を渡す
- LogQL は 1 本のストリームにフィルタを足す(例: `{service_name="slot-manager"} |~ "(?i)error|warn"`)。`or` で 2 本をつながない
- TraceQL の例: `{resource.service.name="job-api" && duration > 500ms}`。検索には開始・終了時刻を必ず渡す
- ツールがエラーを返したら、クエリを 1 度だけ書き直すか別の手段に切り替える

手順:

1. `query_prometheus` でアラート式そのものと、関連するメトリクスをアラートの `startsAt` の前後 30 分で確認する
2. 仮説を 2〜3 個立て、それぞれをクエリで検証する
3. メトリクスで原因が閉じないときは、`startsAt` の前後 5 分のログとトレースを見る
   - ログ: `query_loki_logs` で `startsAt` 前後の ERROR / WARNING 行を引き、原因を示す行をそのまま引用する
   - トレース: `tempo_traceql-search` で遅い、または失敗したトレースを探し、`tempo_get-trace` でどのサービスのどのスパンが時間を使っているかを見る
4. `get_annotations` で直近のデプロイや過去の調査結果があれば参照する

調べ方の原則:

- エラーの原因は、エラーを返した側(下流のサービス)のログから引く。呼び出し側に残る HTTP ステータス(409 や 502)は結果であって原因ではない
- 失敗が一部に偏っていないか(種別・対象・ユーザーなど)を、ログの属性で集計して確かめる。偏りがあれば、その値を結論に書く
- 過去の調査結果(annotation)は参考に留める。今回のアラートの時間帯のデータで確認できたことだけを根拠にし、過去の障害を今回の原因に流用しない

回答は日本語で、次の見出し構成の Markdown にしてください。

## 症状

アラートの内容と、実際に観測した値(数字と時刻を含める)。

## 仮説と検証

| #   | 仮説 | 検証に使ったクエリ | 結果 | 判定 |
| --- | ---- | ------------------ | ---- | ---- |

## 結論

最も可能性の高い原因を 1〜3 行で。確信度(高 / 中 / 低)を添える。
根拠にしたログ行やスパン名があれば、そのまま引用する。

## 次の一手

運用者が取るべき行動を箇条書きで 3 つまで。

{
  "mcpServers": {
    "grafana": {
      "type": "http",
      "url": "http://localhost:8097/mcp"
    }
  }
}

"""ローカル LLM 用の最小エージェント。受け口(investigate.py)から `claude -p` の代わりに呼ぶ。

    uv run --no-project --with mcp --with openai python local_agent.py \
        --model qwen3.6-35b-a3b --mcp-config mcp.json [--allowedTools ...]

- stdin にプロンプト、stdout に `claude --output-format json` と同じ形の JSON を返す
- MCP サーバ(mcp.json の先頭)に Streamable HTTP でつなぎ、ツール一覧を OpenAI 互換の function calling に変換する
- LLM は OpenAI 互換 API(既定 http://localhost:8099/v1、llama.cpp)。ツール呼び出しがなくなるまで回す
- コンテキストが小さいので、ツールの結果は LOCAL_MAX_TOOL_CHARS で切り詰め、
  prompt_tokens が LOCAL_COMPACT_AT を超えたら古いツール結果を短くする

Claude Code のシステムプロンプト(約 18k トークン)を持たないので、同じ 32k のモデルでも調査が最後まで回る。
受け口の CLAUDE_BIN に "uv run --no-project --with mcp --with openai python local_agent.py" を入れると差し替わる。
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import sys
import time

from mcp import ClientSession
from openai import OpenAI

try:  # mcp SDK の版で名前が違う
    from mcp.client.streamable_http import streamable_http_client as streamablehttp_client
except ImportError:  # pragma: no cover
    from mcp.client.streamable_http import streamablehttp_client

LLM_BASE = os.environ.get("LOCAL_LLM_BASE", "http://localhost:8099/v1")
MAX_TURNS = int(os.environ.get("LOCAL_MAX_TURNS", "30"))
# 既定は llama.cpp 64k 向け(ツール 20 回 / 結果 6,000 文字 / 52k で圧縮)。32k なら 14 / 3000 / 24000 にする
MAX_TOOL_CALLS = int(os.environ.get("LOCAL_MAX_TOOL_CALLS", "20"))
MAX_TOOL_CHARS = int(os.environ.get("LOCAL_MAX_TOOL_CHARS", "6000"))
COMPACT_AT = int(os.environ.get("LOCAL_COMPACT_AT", "52000"))
KEEP_RECENT = int(os.environ.get("LOCAL_KEEP_RECENT", "4"))
TOOL_ALLOW = [t for t in os.environ.get("LOCAL_TOOLS", "").split(",") if t]

# 枠組みの説明だけを書く。調査の手順やクエリの書き方は Claude Code と共通の PROMPT.md に置く
SYSTEM = (
    "ツールは 1 回に 1 つずつ呼び、結果を読んでから次を決めてください。"
    f"ツール呼び出しは合計 {MAX_TOOL_CALLS} 回までです。各結果の末尾に残り回数を示します。"
    "残りが少なくなったら、手元の事実で報告を書いてください。"
    "tempo_get-trace の結果は「service | span | duration_ms | detail」の表で返ります。"
)


def log(msg: str) -> None:
    print(f"[local_agent] {msg}", file=sys.stderr, flush=True)


def to_openai_tool(t) -> dict:
    schema = (getattr(t, "input_schema", None) or getattr(t, "inputSchema", None)
              or {"type": "object", "properties": {}})
    return {"type": "function", "function": {
        "name": t.name,
        "description": (t.description or "")[:800],
        "parameters": schema,
    }}


def shape_result(name: str, text: str) -> str:
    """大きな JSON を小さなモデル向けに要約する。Tempo の検索結果とトレースが対象。他はそのまま返す"""
    if not text.lstrip().startswith("{"):
        return text
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        return text
    if name.endswith("traceql-search") and isinstance(data.get("traces"), list):
        lines = ["traceID | root service | root span | duration_ms"]
        for t in data["traces"][:15]:
            lines.append(f"{t.get('traceID')} | {t.get('rootServiceName')} | {t.get('rootTraceName')} | {t.get('durationMs')}")
        lines.append(f"({len(data['traces'])} traces。詳細は tempo_get-trace で 1 件ずつ)")
        return "\n".join(lines)
    if name.endswith("get-trace") and isinstance(data.get("trace"), dict):
        rows = []
        for svc in data["trace"].get("services", []):
            for scope in svc.get("scopes", []):
                for sp in scope.get("spans", []):
                    a = sp.get("attributes", {}) or {}
                    detail = a.get("db.operation.name") or a.get("http.route") or a.get("url.full") or ""
                    status = sp.get("status", {}).get("code", "")
                    err = " ERROR" if "ERROR" in status else ""
                    rows.append((int(sp.get("startTimeUnixNano", 0)), svc.get("serviceName"), sp.get("name"),
                                 float(sp.get("durationMs", 0)), detail, err,
                                 a.get("http.response.status_code", "")))
        rows.sort()
        lines = [f"trace {data['trace'].get('traceId')} : service | span | duration_ms | detail"]
        for _, s, n, d, detail, err, code in rows[:40]:
            lines.append(f"{s} | {n} | {d:.1f} | {detail} {code}{err}".rstrip())
        if len(rows) > 40:
            lines.append(f"(全 {len(rows)} スパンのうち先頭 40)")
        return "\n".join(lines)
    return text


def compact(messages: list[dict]) -> int:
    """古いツール結果を短くしてコンテキストを空ける。最近の KEEP_RECENT 件は残す"""
    idx = [i for i, m in enumerate(messages) if m["role"] == "tool" and not m.get("_compacted")]
    n = 0
    for i in idx[:-KEEP_RECENT] if len(idx) > KEEP_RECENT else []:
        m = messages[i]
        if len(m["content"]) > 300:
            m["content"] = m["content"][:300] + "…(古い結果なので省略)"
            m["_compacted"] = True
            n += 1
    return n


def dedupe_paragraphs(text: str) -> str:
    """反復ループで同じ段落が並んだ出力を、初出だけ残して畳む(判定のため元の長さは注記する)"""
    paras = text.split("\n\n")
    out, seen = [], set()
    for p in paras:
        k = p.strip()
        if k and k in seen:
            continue
        seen.add(k)
        out.append(p)
    if len(out) < len(paras):
        out.append(f"\n(同じ段落の繰り返し {len(paras) - len(out)} 個を省略)")
    return "\n\n".join(out)


def strip_private(messages: list[dict]) -> list[dict]:
    return [{k: v for k, v in m.items() if not k.startswith("_")} for m in messages]


async def run(model: str, mcp_url: str, prompt: str) -> dict:
    t0 = time.monotonic()
    usage = {"input_tokens": 0, "output_tokens": 0}
    tool_log: list[str] = []
    async with streamablehttp_client(mcp_url) as streams:
        read, write = streams[0], streams[1]
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = (await session.list_tools()).tools
            if TOOL_ALLOW:
                tools = [t for t in tools if t.name in TOOL_ALLOW]
            oai_tools = [to_openai_tool(t) for t in tools]
            log(f"tools={len(oai_tools)} model={model} llm={LLM_BASE}")

            client = OpenAI(base_url=LLM_BASE, api_key=os.environ.get("LOCAL_LLM_KEY", "none"), timeout=600)
            messages: list[dict] = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt}]
            turns = 0
            result = ""
            seen: dict[str, int] = {}
            calls_left = MAX_TOOL_CALLS
            for turn in range(MAX_TURNS):
                last = turn == MAX_TURNS - 1 or calls_left <= 0
                if calls_left <= 0 and messages[-1]["role"] != "user":
                    messages.append({"role": "user", "content": "ツール呼び出しの上限に達しました。ここまでに得た事実だけで、"
                                    "指示された見出し構成(症状 / 仮説と検証 / 結論 / 次の一手)の報告を書いてください。"})
                resp = client.chat.completions.create(
                    model=model, messages=strip_private(messages),
                    tools=oai_tools, tool_choice="none" if last else "auto",
                    max_tokens=int(os.environ.get("LOCAL_MAX_TOKENS", "4000")),
                    # Qwen3 系の推奨値。温度を下げすぎると同じ文を繰り返し始める
                    temperature=float(os.environ.get("LOCAL_TEMPERATURE", "0.6")),
                    top_p=0.95,
                    presence_penalty=float(os.environ.get("LOCAL_PRESENCE_PENALTY", "1.0")),
                )
                turns += 1
                if resp.usage:
                    usage["input_tokens"] += resp.usage.prompt_tokens or 0
                    usage["output_tokens"] += resp.usage.completion_tokens or 0
                    log(f"turn {turns}: prompt_tokens={resp.usage.prompt_tokens} completion={resp.usage.completion_tokens}")
                    if (resp.usage.prompt_tokens or 0) > COMPACT_AT:
                        log(f"compacted {compact(messages)} tool results")
                msg = resp.choices[0].message
                assistant: dict = {"role": "assistant", "content": msg.content or ""}
                reasoning = getattr(msg, "reasoning_content", None)
                if reasoning:
                    assistant["reasoning_content"] = reasoning
                if msg.tool_calls:
                    assistant["tool_calls"] = [{
                        "id": tc.id, "type": "function",
                        "function": {"name": tc.function.name, "arguments": tc.function.arguments or "{}"},
                    } for tc in msg.tool_calls]
                messages.append(assistant)
                if not msg.tool_calls:
                    result = msg.content or ""
                    break
                for tc in msg.tool_calls:
                    name = tc.function.name
                    try:
                        args = json.loads(tc.function.arguments or "{}")
                    except json.JSONDecodeError:
                        args = {}
                    key = name + json.dumps(args, sort_keys=True, ensure_ascii=False)
                    tool_log.append(name)
                    if seen.get(key, 0) >= 1:
                        # 同じ呼び出しの繰り返し。結果を返さず、手元の情報で書くよう促す
                        seen[key] += 1
                        log(f"  {name} repeated x{seen[key]} -> skipped")
                        messages.append({"role": "tool", "tool_call_id": tc.id,
                                         "content": "この呼び出しは既に同じ引数で実行済みです。前の結果を使い、"
                                                    "新しい事実が要らなければ報告を書いてください。"})
                        continue
                    seen[key] = 1
                    try:
                        r = await session.call_tool(name, args)
                        text = "\n".join(getattr(c, "text", "") for c in r.content) or "(結果なし)"
                        if getattr(r, "is_error", None) or getattr(r, "isError", None):
                            text = "error: " + text
                        else:
                            text = shape_result(name, text)
                    except Exception as e:  # noqa: BLE001
                        text = f"error: {e}"
                    if len(text) > MAX_TOOL_CHARS:
                        text = text[:MAX_TOOL_CHARS] + f"…(全 {len(text)} 文字のうち先頭のみ)"
                    calls_left -= 1
                    text += f"\n(残りのツール呼び出し: {max(calls_left, 0)} 回)"
                    log(f"  {name}({json.dumps(args, ensure_ascii=False)[:120]}) -> {len(text)} chars")
                    messages.append({"role": "tool", "tool_call_id": tc.id, "content": text})
            else:
                result = messages[-1].get("content") or "(上限ターン数に達したので中断)"
    if "</think>" in result:
        # llama.cpp が思考部分を分離し損ねたとき、本文だけ残す
        result = result.split("</think>", 1)[1].lstrip()
    result = dedupe_paragraphs(result)
    return {
        "type": "result", "subtype": "success", "is_error": False,
        "duration_ms": int((time.monotonic() - t0) * 1000),
        "num_turns": turns, "result": result, "usage": usage,
        "tool_calls": tool_log, "model": model,
    }


def main() -> int:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")
    parser = argparse.ArgumentParser()
    parser.add_argument("--model", default=os.environ.get("CLAUDE_MODEL", "qwen3.6-35b-a3b"))
    parser.add_argument("--mcp-config", default="mcp.json")
    args, _unknown = parser.parse_known_args()   # claude -p 互換の他の引数は無視する
    with open(args.mcp_config, encoding="utf-8") as fh:
        servers = json.load(fh)["mcpServers"]
    mcp_url = next(iter(servers.values()))["url"]
    prompt = sys.stdin.read()
    out = asyncio.run(run(args.model, mcp_url, prompt))
    print(json.dumps(out, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    sys.exit(main())