feat(telemetry): W3C trace context extraction and SSE/transport observability (#416)

* feat(telemetry): extract W3C trace context from upstream and add SSE/transport observability

- Call propagation.extract() in middleware so server span becomes a child
  of the upstream gateway (Envoy) span, connecting distributed traces
- Rename server span to 'http.server.request' (OTel/Datadog convention)
- Add http.transport (sse|jsonrpc) attribute: GET /mcp = SSE long-lived
  stream, everything else = one-shot JSON-RPC
- Add http.response.close_reason (completed|client_disconnect): tracks
  which event fired first — finish (server-side complete) vs close
  (client disconnect), making SSE max-stream timeouts distinguishable
  from genuine app slowness that triggers envoy upstream_rq_timeout alerts
- Add mcp.sse.connection.duration histogram (seconds) for SSE connection
  lifetime, separate from http.server.request.duration to avoid SSE
  long-tail skewing the one-shot request distribution
- Expand propagator to CompositePropagator([W3CTraceContext, W3CBaggage])
- Add deployment.environment resource attribute (OTEL_DEPLOYMENT_ENVIRONMENT)

* refactor(telemetry): collapse closeReason intermediate to a single mutable

- Default closeReason to 'completed' instead of undefined; the close
  listener flips it to 'client_disconnect' only when relevant
- Drop the finalCloseReason intermediate const and the ?? 'completed'
  fallback — they were tracking the same logical state
- finish listener becomes a bare flush reference; close listener stays
  a small arrow that mutates before flushing

flushOnce() in the flush function continues to guard against double
emission, so a close-after-finish overwrite never reaches a downstream
consumer.

* refactor(telemetry): extract createDefaultPropagator helper

Centralize the W3C TraceContext + Baggage composite propagator chain
so test setup and initTelemetry register the exact same propagator —
preventing drift where one side adds a propagator the other lacks.

Author: JeongJaeSoon

.changeset/distributed-trace-context-extract.md

@@ -0,0 +1,11 @@
+---
+"freee-mcp": patch
+---
+
+Remote モードで W3C `traceparent` を抽出して上流ゲートウェイの trace に server span を接続するようにし、SSE と JSON-RPC を区別できる観測ラベルを追加。
+
+- middleware で `propagation.extract` を呼び、サーバー span を上流（Envoy/Istio）の child として紐付け
+- span 名を `http.server.request` に変更（Datadog operation 命名に整合）
+- `http.transport` (`sse` | `jsonrpc`)、`http.response.close_reason` (`completed` | `client_disconnect`) を span attribute と canonical log に追加
+- 新ヒストグラム `mcp.sse.connection.duration` を追加 — SSE 接続寿命を専用バケットで観測
+- propagator を `CompositePropagator([W3CTraceContext, W3CBaggage])` に拡張、resource に `deployment.environment` を付与

src/server/request-context.ts

@@ -86,6 +86,31 @@ export interface RequestRecorderContext {
   path: string;
 }
 
+/**
+ * Inbound transport classification for a single MCP request.
+ *
+ * - `sse`: a long-lived SSE GET (Streamable-HTTP transport). Duration tracks
+ *   the connection lifetime, not the time to produce a JSON-RPC response.
+ * - `jsonrpc`: a short POST that yields a single JSON-RPC response.
+ *
+ * This distinction matters operationally because a 9-minute SSE connection
+ * looks identical to a 9-minute slow JSON-RPC call when filtering on
+ * duration alone.
+ */
+export type CanonicalRequestTransport = 'sse' | 'jsonrpc';
+
+/**
+ * How the response stream actually ended.
+ *
+ * - `completed`: server emitted the full response and `res.on('finish')` fired.
+ * - `client_disconnect`: client closed the socket before completion;
+ *   `res.on('close')` fired without prior `finish`.
+ *
+ * Distinguishes legitimate SSE max-stream timeouts (`completed` at the route's
+ * stream limit) from client-side aborts.
+ */
+export type CanonicalCloseReason = 'completed' | 'client_disconnect';
+
 /**
  * Canonical log line: the complete payload emitted as one JSON log entry
  * per HTTP request at `res.on('finish')`. Consumers (pino, Datadog) see
@@ -114,6 +139,8 @@ export interface CanonicalLogPayload {
     path: string;
     status: number;
     duration_ms: number;
+    transport: CanonicalRequestTransport;
+    close_reason: CanonicalCloseReason;
   };
   mcp: {
     tool_calls: ToolCallInfo[];
@@ -185,7 +212,12 @@ export class RequestRecorder {
   }
 
   /** Builds the canonical log payload; caller passes it to pino. */
-  buildPayload(http: { status: number; duration_ms: number }): CanonicalLogPayload {
+  buildPayload(http: {
+    status: number;
+    duration_ms: number;
+    transport: CanonicalRequestTransport;
+    close_reason: CanonicalCloseReason;
+  }): CanonicalLogPayload {
     return {
       request_id: this.context.request_id,
       source_ip: this.context.source_ip,
@@ -197,6 +229,8 @@ export class RequestRecorder {
         path: this.context.path,
         status: http.status,
         duration_ms: http.duration_ms,
+        transport: http.transport,
+        close_reason: http.close_reason,
       },
       mcp: {
         tool_calls: this.toolCalls,

src/telemetry/init.ts

@@ -1,6 +1,10 @@
 import { SpanKind, SpanStatusCode, context, metrics, propagation, trace } from '@opentelemetry/api';
 import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks';
-import { W3CTraceContextPropagator } from '@opentelemetry/core';
+import {
+  CompositePropagator,
+  W3CBaggagePropagator,
+  W3CTraceContextPropagator,
+} from '@opentelemetry/core';
 import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
 import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
 import { resourceFromAttributes } from '@opentelemetry/resources';
@@ -21,6 +25,17 @@ export function isOtelEnabled(): boolean {
   return _enabled;
 }
 
+/**
+ * Build the propagator stack the server uses for both incoming extract and
+ * outgoing inject. Exported so test setup can register the exact same
+ * propagator chain — keeping prod and test invariants in sync.
+ */
+export function createDefaultPropagator(): CompositePropagator {
+  return new CompositePropagator({
+    propagators: [new W3CTraceContextPropagator(), new W3CBaggagePropagator()],
+  });
+}
+
 const SENSITIVE_PARAMS = new Set([
   'code',
   'code_verifier',
@@ -114,9 +129,13 @@ export function initTelemetry(serviceVersion: string): OtelHandle | null {
   const serviceName = process.env.OTEL_SERVICE_NAME || 'freee-mcp';
   const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT || 'http://localhost:4318';
 
+  const deploymentEnv =
+    process.env.OTEL_DEPLOYMENT_ENVIRONMENT ?? process.env.NODE_ENV ?? 'unknown';
+
   const resource = resourceFromAttributes({
     'service.name': serviceName,
     'service.version': serviceVersion,
+    'deployment.environment': deploymentEnv,
   });
 
   const exporter = new OTLPTraceExporter({
@@ -139,10 +158,13 @@ export function initTelemetry(serviceVersion: string): OtelHandle | null {
     spanProcessors: [new BatchSpanProcessor(exporter)],
   });
 
-  // Register global context manager, propagator, and tracer provider
+  // Register global context manager, propagator, and tracer provider.
+  // CompositePropagator wraps W3C Trace Context (parent linkage) + W3C Baggage
+  // (cross-cutting attributes) so future propagators can be added without
+  // changing call sites.
   const contextManager = new AsyncLocalStorageContextManager();
   context.setGlobalContextManager(contextManager);
-  propagation.setGlobalPropagator(new W3CTraceContextPropagator());
+  propagation.setGlobalPropagator(createDefaultPropagator());
   trace.setGlobalTracerProvider(provider);
 
   // Initialize MeterProvider for metrics export

src/telemetry/metrics.test.ts

@@ -9,6 +9,7 @@ import { afterEach, describe, expect, it } from 'vitest';
 import {
   getHttpRequestDuration,
   getHttpRequestErrorCount,
+  getMcpSseConnectionDuration,
   getToolErrorCount,
   getToolInvocationDuration,
 } from './metrics.js';
@@ -57,6 +58,29 @@ describe('HTTP metrics', () => {
 
     await provider.shutdown();
   });
+
+  it('records mcp.sse.connection.duration with seconds unit', async () => {
+    // The new SSE-only histogram is the signal that distinguishes a normal
+    // long-lived stream (ends at the platform stream timeout) from an
+    // anomalous one (ends much earlier or later). Locking the unit to "s"
+    // keeps the Datadog/Grafana axis legible.
+    const { reader, provider } = setupTestMetrics();
+
+    getMcpSseConnectionDuration().record(599.9, {
+      path: '/mcp',
+      status: '200',
+      close_reason: 'completed',
+    });
+
+    const result = await reader.collect();
+    const metric = result.resourceMetrics.scopeMetrics[0]?.metrics.find(
+      (m) => m.descriptor.name === 'mcp.sse.connection.duration',
+    );
+    expect(metric).toBeDefined();
+    expect(metric?.descriptor.unit).toBe('s');
+
+    await provider.shutdown();
+  });
 });
 
 describe('MCP tool metrics', () => {

src/telemetry/metrics.ts

@@ -6,14 +6,15 @@ const METER_NAME = 'freee-mcp';
 // HTTP server metrics
 let _httpRequestDuration: Histogram | null = null;
 let _httpRequestErrorCount: Counter | null = null;
+let _mcpSseConnectionDuration: Histogram | null = null;
 
 // MCP tool metrics
 let _toolInvocationDuration: Histogram | null = null;
 let _toolErrorCount: Counter | null = null;
 
 /**
  * HTTP request duration histogram (seconds).
- * Labels: method, path, status
+ * Labels: method, path, status, transport, close_reason
  */
 export function getHttpRequestDuration(): Histogram {
   if (!_httpRequestDuration) {
@@ -25,6 +26,28 @@ export function getHttpRequestDuration(): Histogram {
   return _httpRequestDuration;
 }
 
+/**
+ * SSE (Streamable-HTTP) connection lifetime histogram (seconds).
+ *
+ * Recorded only for `transport=sse` requests at connection close. The expected
+ * distribution clusters near the route's `streamIdleTimeout` (e.g. ~600s on
+ * Istio's default HTTPRoute), so the p99 reveals whether SSE clients are
+ * disconnecting early or running until the platform max — useful when
+ * triaging `envoy.cluster.upstream_rq_timeout` alerts which alone cannot
+ * distinguish "long but normal SSE" from "slow JSON-RPC".
+ *
+ * Labels: path, status, close_reason
+ */
+export function getMcpSseConnectionDuration(): Histogram {
+  if (!_mcpSseConnectionDuration) {
+    _mcpSseConnectionDuration = metrics.getMeter(METER_NAME).createHistogram('mcp.sse.connection.duration', {
+      description: 'SSE (Streamable-HTTP) connection lifetime',
+      unit: 's',
+    });
+  }
+  return _mcpSseConnectionDuration;
+}
+
 /**
  * HTTP server error counter (5xx responses).
  * Labels: method, path, status