[trace_id=ab12...c9d3]
chat-edge (root span)
  └── gateway (child span)
        └── workflow (child span)
              ├── vector-search (child span)
              ├── embedding-serving (generation span) [model=bge-m3, tokens=128]
              └── llm-serving (generation span) [model=gpt-4o, input=1520, output=380]

traceparent: 00-{trace_id 32hex}-{span_id 16hex}-{flags 2hex}
예) traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01

tracestate: vendor1=value1,vendor2=value2

baggage: userId=u123,sessionId=s456,tenant=acme

┌─ Application ──────────────────────┐
│  OTel API  (vendor-neutral 계측)    │
│  OTel SDK  (TracerProvider,         │
│             SpanProcessor,          │
│             Exporter)               │
└──────┬──────────────────────────────┘
       │ OTLP/HTTP or OTLP/gRPC
       ▼
┌─ OTel Collector (선택) ──────────────┐
│  Receivers → Processors → Exporters │
│  (배치, 필터링, 변환, fan-out)        │
└──────┬──────────────────────────────┘
       │ 여러 백엔드로 라우팅
       ▼
  Jaeger / Tempo / Datadog / Langfuse / ...

message ExportTraceServiceRequest {
  repeated ResourceSpans resource_spans = 1;
}
message ResourceSpans {
  Resource resource = 1;          // service.name, k8s.pod.name, ...
  repeated ScopeSpans scope_spans = 2;
}
message ScopeSpans {
  InstrumentationScope scope = 1; // tracer 이름·버전
  repeated Span spans = 2;
}
message Span {
  bytes trace_id = 1;             // 16 bytes
  bytes span_id = 2;              // 8 bytes
  bytes parent_span_id = 4;
  string name = 5;
  fixed64 start_time_unix_nano = 7;
  fixed64 end_time_unix_nano = 8;
  repeated KeyValue attributes = 9;
  Status status = 15;
  // ...
}

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.propagate import set_global_textmap
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator
from opentelemetry.propagators.composite import CompositePropagator

# 1) Resource — 모든 span에 붙는 프로세스 차원의 속성
resource = Resource.create({
    "service.name": "edge-api",
    "deployment.environment": "prod",
})

# 2) TracerProvider — Span 발급 공장
provider = TracerProvider(resource=resource)

# 3) Exporter — 어디로 보낼지
exporter = OTLPSpanExporter(endpoint="http://otel-backend:4318/v1/traces")

# 4) SpanProcessor — 어떻게 보낼지 (Batch 권장)
provider.add_span_processor(BatchSpanProcessor(exporter))

trace.set_tracer_provider(provider)

# 5) Propagator — 어떻게 전파할지
set_global_textmap(CompositePropagator([
    TraceContextTextMapPropagator(),
    W3CBaggagePropagator(),
]))

from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.aiohttp_client import AioHttpClientInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

FastAPIInstrumentor.instrument_app(app, excluded_urls="health,healthcheck")
AioHttpClientInstrumentor().instrument()
HTTPXClientInstrumentor().instrument()
RequestsInstrumentor().instrument()

tracer = trace.get_tracer("my.module")

with tracer.start_as_current_span("vector-search") as span:
    span.set_attribute("vdb.collection", collection_name)
    span.set_attribute("vdb.query", query)
    docs = await vdb.search(query)
    span.set_attribute("vdb.result_count", len(docs))

from opentelemetry.processor.baggage import BaggageSpanProcessor, ALLOW_ALL_BAGGAGE_KEYS

provider.add_span_processor(BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS))
provider.add_span_processor(BatchSpanProcessor(exporter))

FastAPIInstrumentor.instrument_app(
    app,
    excluded_urls=",".join(["health", "healthcheck", "/auth/", "/metrics"]),
    exclude_spans=["receive", "send"],  # ASGI internal 0ms span 제거
)

┌─ Langfuse Stack ────────────────────────────────────┐
│                                                     │
│  ┌─────────────┐         ┌──────────────┐           │
│  │ langfuse-web│         │langfuse-worker│          │
│  │  (Next.js)  │         │  (Node.js)   │           │
│  │  UI + API   │         │  Queue worker│           │
│  └──────┬──────┘         └──────┬───────┘           │
│         │                       │                   │
│  ┌──────┴──────┐ ┌────────┐    │  ┌─────────────┐   │
│  │ PostgreSQL  │ │ Redis  │◀───┘  │ ClickHouse  │   │
│  │ (사용자/세션   │ │  (큐)  │       │ (telemetry) │   │
│  │  /트랜잭션)   │ └────────┘       │  3 replica  │   │
│  └─────────────┘                  └─────────────┘   │
│                                                     │
│            ┌────────────────┐                       │
│            │     MinIO      │ ← 대용량 input/output   │
│            │  (S3 호환)      │                       │
│            └────────────────┘                       │
└─────────────────────────────────────────────────────┘

Application (OTel SDK, BatchSpanProcessor)
  │  POST {langfuse-web}/api/public/otel/v1/traces  (OTLP/HTTP, batch)
  ▼
langfuse-web (Next.js OTLP receiver)
  │  파싱 → 정규화 → 큐 publish
  ▼
MinIO (raw temp store)             ← 대용량 input/output 원본
  │
  ▼
Redis (queue)                       ← 인입 이벤트 버퍼링
  │
  ▼
langfuse-worker
  │  큐 consume → attribute 매핑
  ▼
┌─────────────────┐     ┌──────────────────┐
│ ClickHouse      │     │ PostgreSQL       │
│ traces /        │     │ session 트랜잭션   │
│ observations /  │     │ (필요한 경우)       │
│ scores          │     │                  │
└─────────────────┘     └──────────────────┘

span.set_attribute("langfuse.observation.type", "generation")

# OTel GenAI semantic convention
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.response.model", actual_model)
span.set_attribute("gen_ai.usage.input_tokens", prompt_tokens)
span.set_attribute("gen_ai.usage.output_tokens", completion_tokens)

# Langfuse native (둘 다 인식, 병행해도 OK)
span.set_attribute("langfuse.observation.model.name", model)
span.set_attribute("langfuse.observation.usage_details",
                   json.dumps({"input": prompt_tokens, "output": completion_tokens}))

span.set_attribute("langfuse.observation.input", json.dumps(request_body))
span.set_attribute("langfuse.observation.output", json.dumps(response_body))

@@@langfuseMedia:type=text/plain|id={22-char-base62}|source=file@@@

                ┌──────── Distributed table ────────┐
                │  (no data, only metadata)         │
                └────┬─────────────┬─────────────┬──┘
                     ▼             ▼             ▼
              ┌─ Shard 1 ─┐  ┌─ Shard 2 ─┐ ┌─ Shard 3 ─┐
              │ Replica A │  │ Replica A │ │ Replica A │
              │ Replica B │  │ Replica B │ │ Replica B │
              │ Replica C │  │ Replica C │ │ Replica C │
              └───────────┘  └───────────┘ └───────────┘
                     ↑             ↑             ↑
                     └──── Keeper / ZooKeeper ────┘

SELECT id, trace_id, start_time, end_time, metadata
FROM observations FINAL
WHERE trace_id = 'abc...'

-- Langfuse getCostForTraces (observations.ts:1202)
-- "Wrapping the query in a CTE allows us to skip FINAL
--  which allows Clickhouse to use skip indexes."
WITH selected_observations AS (
    SELECT o.total_cost
    FROM observations o                 -- FINAL 없음
    WHERE o.project_id = {projectId}
      AND o.trace_id IN ({traceIds})
      AND o.start_time >= {ts} - INTERVAL ...
    ORDER BY o.event_ts DESC
    LIMIT 1 BY o.id, o.project_id        -- 같은 id 중 event_ts 최신 1행만
)
SELECT sum(total_cost) FROM selected_observations

// shouldSkipObservationsFinal(projectId)
//   - env LANGFUSE_API_CLICKHOUSE_DISABLE_OBSERVATIONS_FINAL=true → skip
//   - isProjectOtelUser(projectId)                                → skip
// 주석: "OTel projects use immutable spans - no need for deduplication"

SELECT ...
FROM observations FINAL
WHERE project_id = 'p123'        -- 필수
  AND type = 'GENERATION'        -- 두 번째 ordering key
  AND start_time >= '2026-05-01' -- partition pruning

SELECT count(), sum(toInt64OrZero(usage_details['input']))
FROM observations FINAL
WHERE project_id = 'p123'
  AND metadata['service'] = 'serving'
  AND metadata['resource_id'] = '484'
  AND start_time BETWEEN '2026-05-01' AND '2026-05-15'

SELECT o.id, o.trace_id, o.start_time, t.session_id, t.user_id
FROM observations FINAL AS o
LEFT JOIN traces FINAL AS t ON o.trace_id = t.id
WHERE o.project_id = 'p123'
  AND o.start_time BETWEEN '2026-05-01' AND '2026-05-15'

┌─ Service Layer (서비스 로직) ────────────────┐
│  - 대시보드 토큰 합산                          │
│  - CSV 다운로드                               │
│  - 사용자별·모델별 통계                        │
└────────────┬────────────────────────────────┘
             │ SQL (Vendor-independent)
             ▼
       ┌──────────────┐
       │ ClickHouse   │←─ Langfuse worker가 적재
       └──────────────┘
             ▲
             │ OTLP/HTTP
       ┌─────┴────────┐
       │ Application  │ — OTel SDK만 사용
       │ (FastAPI)    │
       └──────────────┘

Browser/admin-front (iframe)
       │
       ▼
 decrypt-proxy (FastAPI)  ← 응답에서 input/output 평문 복호화
       │
       ▼
 langfuse-web (Next.js)   ← 암호문 그대로 응답
       │
       ▼
 ClickHouse (암호문 저장)

# 비-LLM 서비스(채팅 메시지 등)는 민감 필드만 값 암호화
SENSITIVE_KEYS = {"question", "text", "message", "pageContent"}

def encrypt_payload(data, keys):
    if isinstance(data, dict):
        return {
            k: crypter.encrypt(v) if k in keys and isinstance(v, str)
               else encrypt_payload(v, keys)
            for k, v in data.items()
        }
    if isinstance(data, list):
        return [encrypt_payload(x, keys) for x in data]
    return data

# otel_setup.py
import base64
import os
from fastapi import FastAPI
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.aiohttp_client import AioHttpClientInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.processor.baggage import ALLOW_ALL_BAGGAGE_KEYS, BaggageSpanProcessor
from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.composite import CompositePropagator
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator


def setup_otel(app: FastAPI) -> None:
    service_name = os.environ["OTEL_SERVICE_NAME"]
    langfuse_base = os.environ["LANGFUSE_BASE_URL"]
    pk = os.environ["LANGFUSE_PUBLIC_KEY"]
    sk = os.environ["LANGFUSE_SECRET_KEY"]

    # 1) Resource
    resource = Resource.create({
        "service.name": service_name,
        "deployment.environment": os.environ.get("ENV", "dev"),
    })

    # 2) Exporter — Langfuse OTLP/HTTP 직접 송신
    auth = base64.b64encode(f"{pk}:{sk}".encode()).decode()
    exporter = OTLPSpanExporter(
        endpoint=f"{langfuse_base}/api/public/otel/v1/traces",
        headers={"Authorization": f"Basic {auth}"},
    )

    # 3) Propagator
    set_global_textmap(CompositePropagator([
        TraceContextTextMapPropagator(),
        W3CBaggagePropagator(),
    ]))

    # 4) TracerProvider
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BaggageSpanProcessor(ALLOW_ALL_BAGGAGE_KEYS))
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

    # 5) Auto-instrumentation
    FastAPIInstrumentor.instrument_app(
        app,
        excluded_urls="health,healthcheck,/auth",
        exclude_spans=["receive", "send"],
    )
    AioHttpClientInstrumentor().instrument()
    HTTPXClientInstrumentor().instrument()

# otel_middleware.py
import json
from opentelemetry import baggage, trace
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request


class LangfuseAttributeMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, trace_name: str = None, collect_io: bool = False):
        super().__init__(app)
        self.trace_name = trace_name
        self.collect_io = collect_io

    async def dispatch(self, request: Request, call_next):
        span = trace.get_current_span()
        if not span or not span.is_recording():
            return await call_next(request)

        # Root span에만 trace 이름 부여
        if span.parent is None and self.trace_name:
            span.set_attribute("langfuse.trace.name", self.trace_name)

        # 1급 메타 — 검색·필터 가능
        span.set_attribute("langfuse.observation.metadata.http_path", request.url.path)
        span.set_attribute("langfuse.observation.metadata.http_method", request.method)

        # baggage → 1급 메타 승격
        if (s := baggage.get_baggage("session.id")):
            span.set_attribute("langfuse.observation.metadata.session_id", str(s))
        if (u := baggage.get_baggage("user.id")):
            span.set_attribute("langfuse.observation.metadata.user_id", str(u))

        # Request body 캡처 (옵션)
        input_text = None
        if self.collect_io:
            try:
                body = await request.body()
                input_text = body.decode("utf-8", errors="replace")
            except Exception:
                pass

        response = await call_next(request)

        if self.collect_io and input_text:
            span.set_attribute("langfuse.observation.input", input_text)
            # response body 캡처는 StreamingResponse 처리 등 추가 작업 필요

        return response

# llm_serving_wrapper.py
import json
from opentelemetry import trace
from opentelemetry.trace import SpanKind

tracer = trace.get_tracer("llm.serving")


async def call_llm_with_tracing(model: str, messages: list, *, serving_id: int):
    with tracer.start_as_current_span("llm.completion", kind=SpanKind.CLIENT) as span:
        span.set_attribute("langfuse.observation.type", "generation")
        span.set_attribute("langfuse.observation.metadata.serving_id", str(serving_id))

        # OTel GenAI 표준
        span.set_attribute("gen_ai.request.model", model)

        # 호출
        response = await http_client.post(
            f"http://gateway/serving/{serving_id}/v1/chat/completions",
            json={"model": model, "messages": messages},
        )
        body = response.json()

        # 응답 enrich
        usage = body.get("usage", {})
        span.set_attribute("gen_ai.response.model", body.get("model", model))
        span.set_attribute("gen_ai.usage.input_tokens", int(usage.get("prompt_tokens", 0)))
        span.set_attribute("gen_ai.usage.output_tokens", int(usage.get("completion_tokens", 0)))
        span.set_attribute("langfuse.observation.input", json.dumps(messages))
        span.set_attribute("langfuse.observation.output", json.dumps(body))
        return body

# token_aggregator.py
import clickhouse_connect

CH = clickhouse_connect.get_client(
    host="clickhouse",
    port=8123,
    username="default",
    password="...",
    database="default",
)


def aggregate_serving_tokens(
    project_id: str,
    serving_id: int,
    revision_id: int,
    start_iso: str,
    end_iso: str,
) -> dict:
    """특정 LLM 서빙 + 리비전의 기간 토큰·요청 수 합산."""
    sql = """
    SELECT
        count() AS request_count,
        sum(toInt64OrZero(usage_details['input']))  AS input_tokens,
        sum(toInt64OrZero(usage_details['output'])) AS output_tokens,
        sum(toInt64OrZero(usage_details['total']))  AS total_tokens
    FROM observations FINAL
    WHERE project_id = {project_id:String}
      AND type = 'GENERATION'
      AND metadata['service'] = 'serving'
      AND metadata['serving_id'] = {serving_id:String}
      AND metadata['serving_revision_id'] = {revision_id:String}
      AND start_time BETWEEN {start:DateTime64(3)} AND {end:DateTime64(3)}
    """
    row = CH.query(
        sql,
        parameters={
            "project_id": project_id,
            "serving_id": str(serving_id),
            "revision_id": str(revision_id),
            "start": start_iso,
            "end": end_iso,
        },
    ).result_rows[0]
    return {
        "request_count": row[0],
        "input_tokens":  row[1],
        "output_tokens": row[2],
        "total_tokens":  row[3],
    }

def get_session_chat_log(project_id: str, session_id: str) -> list[dict]:
    """한 채팅 세션의 모든 trace를 시간순으로 — root observation의 input/output만 사용."""
    sql = """
    SELECT
        t.id            AS trace_id,
        t.timestamp     AS created_at,
        o.input         AS request_data,
        o.output        AS response_data
    FROM traces FINAL AS t
    LEFT JOIN observations FINAL AS o
        ON o.trace_id = t.id
        AND o.parent_observation_id IS NULL  -- root observation만
    WHERE t.project_id = {project_id:String}
      AND t.session_id = {session_id:String}
      AND t.timestamp >= now() - INTERVAL 30 DAY
    ORDER BY t.timestamp ASC
    """
    rows = CH.query(
        sql,
        parameters={"project_id": project_id, "session_id": session_id},
    ).result_rows
    return [
        {
            "trace_id": r[0],
            "created_at": r[1].isoformat(),
            "request": r[2],
            "response": r[3],
        }
        for r in rows
    ]