6장: OTel SDK 계측 실전

# 자동 계측 패키지 설치
pip install opentelemetry-distro opentelemetry-exporter-otlp
 
# 사용 중인 라이브러리의 계측 패키지 자동 설치
opentelemetry-bootstrap -a install
 
# 자동 계측으로 애플리케이션 실행
opentelemetry-instrument \
    --service_name user-service \
    --exporter_otlp_endpoint http://collector:4317 \
    python app.py

from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
import logging
 
 
def init_telemetry(
    service_name: str,
    service_version: str,
    collector_endpoint: str = "http://collector:4317",
    environment: str = "production",
):
    """OpenTelemetry 3대 신호(트레이스, 메트릭, 로그) 통합 초기화"""
    
    resource = Resource.create({
        "service.name": service_name,
        "service.version": service_version,
        "deployment.environment": environment,
    })
 
    # --- Traces ---
    trace_provider = TracerProvider(resource=resource)
    trace_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(endpoint=collector_endpoint),
            max_queue_size=2048,
            max_export_batch_size=512,
        )
    )
    trace.set_tracer_provider(trace_provider)
 
    # --- Metrics ---
    metric_reader = PeriodicExportingMetricReader(
        OTLPMetricExporter(endpoint=collector_endpoint),
        export_interval_millis=15000,
    )
    metric_provider = MeterProvider(
        resource=resource,
        metric_readers=[metric_reader],
    )
    metrics.set_meter_provider(metric_provider)
 
    # --- Logs ---
    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(
        BatchLogRecordProcessor(
            OTLPLogExporter(endpoint=collector_endpoint)
        )
    )
    handler = LoggingHandler(level=logging.INFO, logger_provider=logger_provider)
    logging.getLogger().addHandler(handler)
 
    # --- 자동 계측 ---
    FlaskInstrumentor().instrument()
    RequestsInstrumentor().instrument()
    SQLAlchemyInstrumentor().instrument()
 
    return trace_provider, metric_provider, logger_provider

from flask import Flask, jsonify, request
from telemetry import init_telemetry
from opentelemetry import trace, metrics
import logging
 
# 텔레메트리 초기화
init_telemetry(
    service_name="order-service",
    service_version="2.0.0",
    environment="production",
)
 
app = Flask(__name__)
tracer = trace.get_tracer("order-service.handlers")
meter = metrics.get_meter("order-service.business")
logger = logging.getLogger("order-service")
 
# 비즈니스 메트릭
order_counter = meter.create_counter("orders.created.count")
order_amount = meter.create_histogram("orders.amount")
 
 
@app.route("/orders", methods=["POST"])
def create_order():
    data = request.json
    
    with tracer.start_as_current_span("validate-order") as span:
        span.set_attribute("order.item_count", len(data["items"]))
        validate_order(data)
    
    with tracer.start_as_current_span("process-payment") as span:
        span.set_attribute("payment.method", data["payment_method"])
        span.set_attribute("payment.amount", data["total"])
        result = process_payment(data)
    
    # 비즈니스 메트릭 기록
    order_counter.add(1, {"payment.method": data["payment_method"]})
    order_amount.record(data["total"], {"currency": "KRW"})
    
    logger.info("주문 생성 완료", extra={
        "order_id": result["order_id"],
        "total": data["total"],
    })
    
    return jsonify(result), 201

import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-grpc";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-grpc";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { HttpInstrumentation } from "@opentelemetry/instrumentation-http";
import { ExpressInstrumentation } from "@opentelemetry/instrumentation-express";
import { PgInstrumentation } from "@opentelemetry/instrumentation-pg";
import { Resource } from "@opentelemetry/resources";
import {
  ATTR_SERVICE_NAME,
  ATTR_SERVICE_VERSION,
} from "@opentelemetry/semantic-conventions";
 
const sdk = new NodeSDK({
  resource: new Resource({
    [ATTR_SERVICE_NAME]: "user-service",
    [ATTR_SERVICE_VERSION]: "1.0.0",
    "deployment.environment": "production",
  }),
  traceExporter: new OTLPTraceExporter({
    url: "http://collector:4317",
  }),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: "http://collector:4317",
    }),
    exportIntervalMillis: 15000,
  }),
  instrumentations: [
    new HttpInstrumentation(),
    new ExpressInstrumentation(),
    new PgInstrumentation(),
  ],
});
 
sdk.start();
 
// 종료 시 정리
process.on("SIGTERM", () => {
  sdk.shutdown().then(() => process.exit(0));
});

node --require ./telemetry.js app.js

import { trace, metrics, SpanStatusCode } from "@opentelemetry/api";
 
const tracer = trace.getTracer("user-service.handlers");
const meter = metrics.getMeter("user-service.business");
 
const loginCounter = meter.createCounter("user.login.count");
 
async function handleLogin(req: Request, res: Response) {
  const span = tracer.startSpan("authenticate-user");
  
  try {
    span.setAttribute("user.email_domain", extractDomain(req.body.email));
    
    const user = await authenticateUser(req.body);
    
    loginCounter.add(1, { method: req.body.auth_method });
    span.setStatus({ code: SpanStatusCode.OK });
    
    res.json({ token: generateToken(user) });
  } catch (error) {
    span.setStatus({
      code: SpanStatusCode.ERROR,
      message: error.message,
    });
    span.recordException(error);
    res.status(401).json({ error: "Authentication failed" });
  } finally {
    span.end();
  }
}

package telemetry
 
import (
    "context"
    "time"
 
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
    "go.opentelemetry.io/otel/sdk/metric"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
)
 
func InitTelemetry(ctx context.Context, serviceName, serviceVersion string) (func(), error) {
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName(serviceName),
            semconv.ServiceVersion(serviceVersion),
            semconv.DeploymentEnvironment("production"),
        ),
    )
    if err != nil {
        return nil, err
    }
 
    // Trace Exporter
    traceExporter, err := otlptracegrpc.New(ctx,
        otlptracegrpc.WithEndpoint("collector:4317"),
        otlptracegrpc.WithInsecure(),
    )
    if err != nil {
        return nil, err
    }
 
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithResource(res),
        sdktrace.WithBatcher(traceExporter,
            sdktrace.WithBatchTimeout(5*time.Second),
            sdktrace.WithMaxQueueSize(2048),
        ),
    )
    otel.SetTracerProvider(tp)
 
    // Metric Exporter
    metricExporter, err := otlpmetricgrpc.New(ctx,
        otlpmetricgrpc.WithEndpoint("collector:4317"),
        otlpmetricgrpc.WithInsecure(),
    )
    if err != nil {
        return nil, err
    }
 
    mp := metric.NewMeterProvider(
        metric.WithResource(res),
        metric.WithReader(
            metric.NewPeriodicReader(metricExporter,
                metric.WithInterval(15*time.Second),
            ),
        ),
    )
    otel.SetMeterProvider(mp)
 
    // 종료 함수 반환
    shutdown := func() {
        tp.Shutdown(ctx)
        mp.Shutdown(ctx)
    }
    return shutdown, nil
}

package handlers
 
import (
    "context"
    "net/http"
 
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/metric"
)
 
var (
    tracer       = otel.Tracer("order-service.handlers")
    meter        = otel.Meter("order-service.business")
    orderCounter metric.Int64Counter
)
 
func init() {
    var err error
    orderCounter, err = meter.Int64Counter("orders.processed.count",
        metric.WithDescription("Total processed orders"),
    )
    if err != nil {
        panic(err)
    }
}
 
func HandleCreateOrder(ctx context.Context, order Order) error {
    ctx, span := tracer.Start(ctx, "create-order")
    defer span.End()
 
    span.SetAttributes(
        attribute.String("order.id", order.ID),
        attribute.Float64("order.total", order.Total),
        attribute.Int("order.item_count", len(order.Items)),
    )
 
    if err := validateOrder(ctx, order); err != nil {
        span.SetStatus(codes.Error, err.Error())
        span.RecordError(err)
        return err
    }
 
    if err := processPayment(ctx, order); err != nil {
        span.SetStatus(codes.Error, err.Error())
        span.RecordError(err)
        return err
    }
 
    orderCounter.Add(ctx, 1,
        metric.WithAttributes(
            attribute.String("payment.method", order.PaymentMethod),
        ),
    )
 
    span.SetStatus(codes.Ok, "")
    return nil
}

java -javaagent:opentelemetry-javaagent.jar \
     -Dotel.service.name=payment-service \
     -Dotel.exporter.otlp.endpoint=http://collector:4317 \
     -Dotel.metrics.exporter=otlp \
     -Dotel.logs.exporter=otlp \
     -jar payment-service.jar

from opentelemetry import trace, context
from opentelemetry.propagate import inject, extract
 
tracer = trace.get_tracer("propagation-example")
 
# --- 발신측: 컨텍스트를 캐리어에 주입 ---
def send_message(message: dict):
    with tracer.start_as_current_span("send-message") as span:
        # 현재 컨텍스트를 메시지 헤더에 주입
        headers = {}
        inject(headers)
        
        message["_trace_headers"] = headers
        queue.publish(message)
 
# --- 수신측: 캐리어에서 컨텍스트를 추출 ---
def receive_message(message: dict):
    # 메시지 헤더에서 컨텍스트 추출
    ctx = extract(message.get("_trace_headers", {}))
    
    # 추출된 컨텍스트를 부모로 하는 새 스팬 생성
    with tracer.start_as_current_span(
        "process-message",
        context=ctx,
    ) as span:
        span.set_attribute("message.type", message["type"])
        process(message)