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 };
 }
}

まとめ

この記事では、手順を追って動かしながら重要な確認ポイントを押さえることが大切です。まずは学習用の最小構成で動作確認し、本番利用する前にセキュリティ、権限、エラー処理、運用手順を見直してください。

参考リソース

• 公式ドキュメント - OpenTelemetry実践 - 分散トレーシングとメトリクス計装 を確認するための一次情報

関連記事

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

×

PR

×

PR