OpenTelemetry実践 - 分散トレーシングとメトリクス計装

OpenTelemetryとは

OpenTelemetry（OTel）は、ベンダー中立の観測性フレームワークです。トレース・メトリクス・ログを統一的な方法で計装・収集・エクスポートできます。

主要コンポーネント

自動計装（Node.js）

# 依存関係のインストール
npm install @opentelemetry/sdk-node \
  @opentelemetry/auto-instrumentations-node \
  @opentelemetry/exporter-trace-otlp-http \
  @opentelemetry/exporter-metrics-otlp-http

// tracing.ts
import { NodeSDK } from '@opentelemetry/sdk-node';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics';
import { Resource } from '@opentelemetry/resources';
import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';

const sdk = new NodeSDK({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: 'my-app',
    [SemanticResourceAttributes.SERVICE_VERSION]: '1.0.0',
    [SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: process.env.NODE_ENV || 'development',
  }),
  traceExporter: new OTLPTraceExporter({
    url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT || 'http://localhost:4318/v1/traces',
  }),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT || 'http://localhost:4318/v1/metrics',
    }),
    exportIntervalMillis: 60000, // 1分ごと
  }),
  instrumentations: [
    getNodeAutoInstrumentations({
      '@opentelemetry/instrumentation-http': {
        enabled: true,
      },
      '@opentelemetry/instrumentation-express': {
        enabled: true,
      },
      '@opentelemetry/instrumentation-pg': {
        enabled: true,
      },
      '@opentelemetry/instrumentation-redis': {
        enabled: true,
      },
    }),
  ],
});

sdk.start();

// シャットダウン処理
process.on('SIGTERM', () => {
  sdk
    .shutdown()
    .then(() => console.log('Tracing terminated'))
    .catch((error) => console.error('Error terminating tracing', error))
    .finally(() => process.exit(0));
});

export default sdk;

// app.ts
import './tracing'; // 最初にインポート
import express from 'express';

const app = express();

app.get('/api/users/:id', async (req, res) => {
  const user = await fetchUser(req.params.id);
  res.json(user);
});

app.listen(3000, () => {
  console.log('Server running on port 3000');
});

手動計装（Node.js）

// instrumentation.ts
import { trace, context, SpanStatusCode } from '@opentelemetry/api';
import { metrics } from '@opentelemetry/api';

// Tracerの取得
const tracer = trace.getTracer('my-app', '1.0.0');

// Meterの取得
const meter = metrics.getMeter('my-app', '1.0.0');

// カスタムメトリクス
const requestCounter = meter.createCounter('http_requests_total', {
  description: 'Total number of HTTP requests',
});

const requestDuration = meter.createHistogram('http_request_duration_seconds', {
  description: 'HTTP request duration',
  unit: 'seconds',
});

export async function fetchUser(userId: string) {
  // Spanの作成
  return tracer.startActiveSpan('fetchUser', async (span) => {
    try {
      span.setAttributes({
        'user.id': userId,
        'db.system': 'postgresql',
      });

      const user = await db.query('SELECT * FROM users WHERE id = $1', [userId]);

      if (!user) {
        span.setStatus({ code: SpanStatusCode.ERROR, message: 'User not found' });
        span.recordException(new Error('User not found'));
        throw new Error('User not found');
      }

      span.setStatus({ code: SpanStatusCode.OK });
      return user;
    } catch (error) {
      span.setStatus({ code: SpanStatusCode.ERROR, message: error.message });
      span.recordException(error);
      throw error;
    } finally {
      span.end();
    }
  });
}

export function recordHttpRequest(method: string, path: string, statusCode: number, duration: number) {
  // メトリクスの記録
  requestCounter.add(1, {
    method,
    path,
    status: statusCode.toString(),
  });

  requestDuration.record(duration / 1000, {
    method,
    path,
    status: statusCode.toString(),
  });
}

// middleware.ts
import { trace, context } from '@opentelemetry/api';
import { recordHttpRequest } from './instrumentation';

export function tracingMiddleware(req, res, next) {
  const tracer = trace.getTracer('my-app');
  const span = tracer.startSpan(`${req.method} ${req.path}`);

  const start = Date.now();

  // レスポンス完了時
  res.on('finish', () => {
    const duration = Date.now() - start;
    
    span.setAttributes({
      'http.method': req.method,
      'http.url': req.url,
      'http.status_code': res.statusCode,
      'http.route': req.route?.path,
    });

    span.setStatus({ 
      code: res.statusCode >= 500 ? SpanStatusCode.ERROR : SpanStatusCode.OK 
    });

    recordHttpRequest(req.method, req.path, res.statusCode, duration);
    span.end();
  });

  // Contextを伝播
  context.with(trace.setSpan(context.active(), span), () => {
    next();
  });
}

Python自動計装

# インストール
pip install opentelemetry-distro \
  opentelemetry-exporter-otlp \
  opentelemetry-instrumentation-flask \
  opentelemetry-instrumentation-requests \
  opentelemetry-instrumentation-sqlalchemy

# 自動計装ライブラリのインストール
opentelemetry-bootstrap -a install

# app.py
from flask import Flask
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.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# リソースの設定
resource = Resource.create({
    "service.name": "my-python-app",
    "service.version": "1.0.0",
    "deployment.environment": "production",
})

# TracerProviderの設定
provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Flaskアプリ
app = Flask(__name__)

# 自動計装
FlaskInstrumentor().instrument_app(app)
RequestsInstrumentor().instrument()

@app.route('/api/data')
def get_data():
    tracer = trace.get_tracer(__name__)
    
    with tracer.start_as_current_span("process_data") as span:
        span.set_attribute("user.id", "123")
        # ビジネスロジック
        data = {"result": "success"}
        return data

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

# 環境変数で起動
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
export OTEL_SERVICE_NAME=my-python-app

opentelemetry-instrument python app.py

OpenTelemetry Collector

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 10s
    send_batch_size: 1024
  
  memory_limiter:
    check_interval: 1s
    limit_mib: 512
  
  resource:
    attributes:
      - key: cluster.name
        value: production
        action: insert

exporters:
  # Jaegerへのエクスポート
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true
  
  # Tempoへのエクスポート
  otlp/tempo:
    endpoint: tempo:4317
    tls:
      insecure: true
  
  # Prometheusへのエクスポート
  prometheus:
    endpoint: 0.0.0.0:8889
  
  # ログ出力（デバッグ用）
  logging:
    loglevel: debug

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch, resource]
      exporters: [otlp/jaeger, otlp/tempo, logging]
    
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, batch, resource]
      exporters: [prometheus, logging]

# docker-compose.yml
services:
  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.96.0
    command: ["--config=/etc/otel-collector-config.yaml"]
    volumes:
      - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml
    ports:
      - "4317:4317"   # OTLP gRPC
      - "4318:4318"   # OTLP HTTP
      - "8889:8889"   # Prometheus metrics
    networks:
      - monitoring

  jaeger:
    image: jaegertracing/all-in-one:1.55
    ports:
      - "16686:16686"  # Jaeger UI
      - "4317:4317"    # OTLP gRPC
    environment:
      - COLLECTOR_OTLP_ENABLED=true
    networks:
      - monitoring

  tempo:
    image: grafana/tempo:2.4.0
    command: ["-config.file=/etc/tempo.yaml"]
    volumes:
      - ./tempo.yaml:/etc/tempo.yaml
      - tempo-data:/tmp/tempo
    ports:
      - "3200:3200"   # Tempo
      - "4317"        # OTLP gRPC
    networks:
      - monitoring

networks:
  monitoring:

volumes:
  tempo-data:

Context Propagation（コンテキスト伝播）

// frontend.ts (Next.js)
import { trace, context, propagation } from '@opentelemetry/api';

export async function fetchData() {
  const tracer = trace.getTracer('frontend');
  
  return tracer.startActiveSpan('fetchData', async (span) => {
    const headers = {};
    
    // トレースコンテキストをHTTPヘッダーに注入
    propagation.inject(context.active(), headers);
    
    const response = await fetch('http://backend:3000/api/data', {
      headers,
    });
    
    const data = await response.json();
    span.end();
    return data;
  });
}

// backend.ts (Express)
import { propagation, context, trace } from '@opentelemetry/api';

app.use((req, res, next) => {
  // HTTPヘッダーからトレースコンテキストを抽出
  const extractedContext = propagation.extract(context.active(), req.headers);
  
  context.with(extractedContext, () => {
    const tracer = trace.getTracer('backend');
    const span = tracer.startSpan('handleRequest');
    
    // スパン情報をリクエストに保存
    req.span = span;
    
    res.on('finish', () => {
      span.end();
    });
    
    next();
  });
});

Semantic Conventions

// セマンティック属性の使用
import { SemanticAttributes } from '@opentelemetry/semantic-conventions';

span.setAttributes({
  // HTTP
  [SemanticAttributes.HTTP_METHOD]: 'GET',
  [SemanticAttributes.HTTP_URL]: 'https://api.example.com/users',
  [SemanticAttributes.HTTP_STATUS_CODE]: 200,
  
  // データベース
  [SemanticAttributes.DB_SYSTEM]: 'postgresql',
  [SemanticAttributes.DB_NAME]: 'myapp',
  [SemanticAttributes.DB_STATEMENT]: 'SELECT * FROM users WHERE id = $1',
  [SemanticAttributes.DB_USER]: 'app_user',
  
  // RPC
  [SemanticAttributes.RPC_SERVICE]: 'UserService',
  [SemanticAttributes.RPC_METHOD]: 'GetUser',
  
  // メッセージング
  [SemanticAttributes.MESSAGING_SYSTEM]: 'kafka',
  [SemanticAttributes.MESSAGING_DESTINATION]: 'user-events',
  [SemanticAttributes.MESSAGING_OPERATION]: 'publish',
});

ログとトレースの統合

// logger.ts
import pino from 'pino';
import { trace, context } from '@opentelemetry/api';

const logger = pino({
  mixin() {
    const span = trace.getSpan(context.active());
    if (!span) return {};
    
    const spanContext = span.spanContext();
    return {
      trace_id: spanContext.traceId,
      span_id: spanContext.spanId,
      trace_flags: spanContext.traceFlags,
    };
  },
});

export default logger;

// 使用例
import logger from './logger';

app.get('/api/users/:id', async (req, res) => {
  logger.info({ userId: req.params.id }, 'Fetching user');
  
  try {
    const user = await fetchUser(req.params.id);
    res.json(user);
  } catch (error) {
    logger.error({ error, userId: req.params.id }, 'Failed to fetch user');
    res.status(500).json({ error: 'Internal Server Error' });
  }
});

カスタムメトリクス

// metrics.ts
import { metrics } from '@opentelemetry/api';

const meter = metrics.getMeter('my-app', '1.0.0');

// Counter（単調増加）
export const httpRequestsTotal = meter.createCounter('http_requests_total', {
  description: 'Total number of HTTP requests',
});

// Histogram（分布）
export const httpRequestDuration = meter.createHistogram('http_request_duration_seconds', {
  description: 'HTTP request duration in seconds',
  unit: 'seconds',
});

// UpDownCounter（増減可能）
export const activeConnections = meter.createUpDownCounter('active_connections', {
  description: 'Number of active connections',
});

// Gauge（非同期観測）
meter.createObservableGauge('process_memory_usage_bytes', {
  description: 'Process memory usage',
  unit: 'bytes',
}).addCallback((observableResult) => {
  const memUsage = process.memoryUsage();
  observableResult.observe(memUsage.heapUsed, { type: 'heap' });
  observableResult.observe(memUsage.rss, { type: 'rss' });
});

// 使用例
import { httpRequestsTotal, httpRequestDuration } from './metrics';

app.use((req, res, next) => {
  const start = Date.now();
  
  res.on('finish', () => {
    const duration = (Date.now() - start) / 1000;
    
    httpRequestsTotal.add(1, {
      method: req.method,
      path: req.route?.path || req.path,
      status: res.statusCode.toString(),
    });
    
    httpRequestDuration.record(duration, {
      method: req.method,
      path: req.route?.path || req.path,
    });
  });
  
  next();
});

サンプリング設定

// sampling.ts
import { ParentBasedSampler, TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-base';

// 10%のトレースをサンプリング
const sampler = new ParentBasedSampler({
  root: new TraceIdRatioBasedSampler(0.1),
});

const sdk = new NodeSDK({
  sampler,
  // その他の設定
});

// カスタムサンプラー
import { Sampler, SamplingDecision } from '@opentelemetry/sdk-trace-base';

class ErrorSampler implements Sampler {
  shouldSample(context, traceId, spanName, spanKind, attributes) {
    // エラーの場合は必ずサンプリング
    if (attributes['http.status_code'] >= 500) {
      return { decision: SamplingDecision.RECORD_AND_SAMPLED };
    }
    
    // それ以外は10%
    return Math.random() < 0.1
      ? { decision: SamplingDecision.RECORD_AND_SAMPLED }
      : { decision: SamplingDecision.NOT_RECORD };
  }
}

関連記事

• Grafana で観測性ダッシュボードを作る
• 観測性の3本柱 - メトリクス・ログ・トレース