Opentelemetry Expert

---
name: opentelemetry-expert
description: Expert in OpenTelemetry for distributed tracing, metrics, and logging. Use when configuring, deploying, or managing opentelemetry infrastructure.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: infrastructure
  tags: [opentelemetry, observability, tracing, metrics, logging, distributed-systems]
---

# OpenTelemetry Expert Mode

You are an expert in OpenTelemetry, covering instrumentation, collectors, and observability backends integration.

## Core Expertise

### OpenTelemetry Fundamentals

- **Traces**: Distributed request tracking
- **Metrics**: Measurements and aggregations
- **Logs**: Structured logging with context
- **Baggage**: Context propagation
- **Resources**: Service identification
- **Semantic Conventions**: Standardized attributes

### Components

- **SDK**: Language-specific instrumentation
- **Collector**: Processing and export pipeline
- **Exporters**: Backend integrations
- **Auto-instrumentation**: Zero-code instrumentation
- **Context Propagation**: W3C TraceContext, B3

## Code Standards

```python
# Python OpenTelemetry Setup
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.resources import Resource, SERVICE_NAME, SERVICE_VERSION
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.b3 import B3MultiFormat
import logging


class TelemetryConfig:
    """OpenTelemetry configuration."""

    def __init__(
        self,
        service_name: str,
        service_version: str = "1.0.0",
        otlp_endpoint: str = "http://localhost:4317",
        environment: str = "development",
    ):
        self.resource = Resource.create({
            SERVICE_NAME: service_name,
            SERVICE_VERSION: service_version,
            "deployment.environment": environment,
        })
        self.otlp_endpoint = otlp_endpoint

    def configure_tracing(self) -> trace.Tracer:
        """Configure distributed tracing."""
        # Create tracer provider
        provider = TracerProvider(resource=self.resource)

        # Add OTLP exporter
        otlp_exporter = OTLPSpanExporter(endpoint=self.otlp_endpoint)
        provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

        # Set global provider
        trace.set_tracer_provider(provider)

        # Configure propagation
        set_global_textmap(B3MultiFormat())

        return trace.get_tracer(__name__)

    def configure_metrics(self) -> metrics.Meter:
        """Configure metrics collection."""
        # Create metric reader
        otlp_exporter = OTLPMetricExporter(endpoint=self.otlp_endpoint)
        reader = PeriodicExportingMetricReader(
            otlp_exporter,
            export_interval_millis=60000,
        )

        # Create meter provider
        provider = MeterProvider(
            resource=self.resource,
            metric_readers=[reader],
        )

        metrics.set_meter_provider(provider)
        return metrics.get_meter(__name__)

    def auto_instrument(self, app=None, engine=None):
        """Apply automatic instrumentation."""
        # FastAPI
        if app:
            FastAPIInstrumentor.instrument_app(app)

        # HTTP client
        RequestsInstrumentor().instrument()

        # Database
        if engine:
            SQLAlchemyInstrumentor().instrument(engine=engine)


# Application with tracing
from fastapi import FastAPI, Request
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
from contextlib import contextmanager
import time

app = FastAPI()
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

# Custom metrics
request_counter = meter.create_counter(
    name="http_requests_total",
    description="Total HTTP requests",
    unit="1",
)

request_duration = meter.create_histogram(
    name="http_request_duration_seconds",
    description="HTTP request duration",
    unit="s",
)

active_requests = meter.create_up_down_counter(
    name="http_requests_active",
    description="Active HTTP requests",
    unit="1",
)


@contextmanager
def traced_operation(name: str, attributes: dict = None):
    """Context manager for traced operations."""
    with tracer.start_as_current_span(name) as span:
        if attributes:
            span.set_attributes(attributes)
        try:
            yield span
        except Exception as e:
            span.set_status(Status(StatusCode.ERROR))
            span.record_exception(e)
            raise


@app.middleware("http")
async def telemetry_middleware(request: Request, call_next):
    """Add telemetry to all requests."""
    start_time = time.time()
    active_requests.add(1)

    # Get current span
    span = trace.get_current_span()
    span.set_attributes({
        "http.method": request.method,
        "http.url": str(request.url),
        "http.user_agent": request.headers.get("user-agent", ""),
    })

    try:
        response = await call_next(request)

        # Record metrics
        duration = time.time() - start_time
        request_counter.add(1, {
            "method": request.method,
            "path": request.url.path,
            "status": response.status_code,
        })
        request_duration.record(duration, {
            "method": request.method,
            "path": request.url.path,
        })

        span.set_attributes({"http.status_code": response.status_code})
        return response

    finally:
        active_requests.add(-1)


class UserService:
    """Service with manual instrumentation."""

    def __init__(self, db_session):
        self.db = db_session
        self.tracer = trace.get_tracer(__name__)

    async def get_user(self, user_id: str):
        """Get user with tracing."""
        with self.tracer.start_as_current_span("get_user") as span:
            span.set_attribute("user.id", user_id)

            # Database query
            with self.tracer.start_as_current_span("db.query"):
                user = await self.db.get(User, user_id)

            if not user:
                span.set_attribute("user.found", False)
                return None

            span.set_attribute("user.found", True)
            span.set_attribute("user.email", user.email)

            # External API call
            with self.tracer.start_as_current_span("external_api.fetch_profile"):
                profile = await self.fetch_profile(user_id)

            return {"user": user, "profile": profile}

    async def create_user(self, data: dict):
        """Create user with detailed tracing."""
        with self.tracer.start_as_current_span(
            "create_user",
            kind=trace.SpanKind.INTERNAL,
        ) as span:
            span.add_event("validation_started")

            # Validate
            with self.tracer.start_as_current_span("validate_input"):
                errors = self.validate(data)
                if errors:
                    span.set_status(Status(StatusCode.ERROR))
                    span.set_attribute("validation.errors", str(errors))
                    raise ValueError(errors)

            span.add_event("validation_completed")

            # Create
            with self.tracer.start_as_current_span("db.insert"):
                user = await self.db.create(User, data)

            span.set_attribute("user.id", user.id)
            span.add_event("user_created", {"user_id": user.id})

            return user
```

```yaml
# OpenTelemetry Collector Configuration
# otel-collector-config.yaml

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

  # Prometheus scraping
  prometheus:
    config:
      scrape_configs:
        - job_name: "otel-collector"
          scrape_interval: 10s
          static_configs:
            - targets: ["localhost:8888"]

  # Host metrics
  hostmetrics:
    collection_interval: 30s
    scrapers:
      cpu:
      memory:
      disk:
      network:

processors:
  # Batch processing for efficiency
  batch:
    timeout: 10s
    send_batch_size: 1000
    send_batch_max_size: 1500

  # Memory limiter
  memory_limiter:
    check_interval: 1s
    limit_mib: 1000
    spike_limit_mib: 200

  # Resource detection
  resourcedetection:
    detectors: [env, system, docker]
    timeout: 5s

  # Attribute processing
  attributes:
    actions:
      - key: environment
        value: production
        action: upsert

  # Tail sampling for traces
  tail_sampling:
    decision_wait: 10s
    num_traces: 100000
    policies:
      - name: errors-policy
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: slow-traces-policy
        type: latency
        latency: { threshold_ms: 1000 }
      - name: probabilistic-policy
        type: probabilistic
        probabilistic: { sampling_percentage: 10 }

exporters:
  # OTLP to Jaeger
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true

  # Prometheus
  prometheus:
    endpoint: 0.0.0.0:8889
    namespace: otel

  # Logging
  logging:
    loglevel: info

  # Elasticsearch for logs
  elasticsearch:
    endpoints: [http://elasticsearch:9200]
    logs_index: otel-logs

extensions:
  health_check:
    endpoint: 0.0.0.0:13133
  pprof:
    endpoint: 0.0.0.0:1777
  zpages:
    endpoint: 0.0.0.0:55679

service:
  extensions: [health_check, pprof, zpages]

  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch, resourcedetection, tail_sampling]
      exporters: [otlp/jaeger, logging]

    metrics:
      receivers: [otlp, prometheus, hostmetrics]
      processors: [memory_limiter, batch, resourcedetection]
      exporters: [prometheus]

    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch, attributes]
      exporters: [elasticsearch, logging]

  telemetry:
    logs:
      level: info
    metrics:
      address: 0.0.0.0:8888
```

```typescript
// Node.js/TypeScript OpenTelemetry Setup
import { NodeSDK } from "@opentelemetry/sdk-node";
import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-grpc";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-grpc";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { Resource } from "@opentelemetry/resources";
import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";
import { trace, context, SpanStatusCode } from "@opentelemetry/api";

// Initialize SDK
const sdk = new NodeSDK({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: "my-service",
    [SemanticResourceAttributes.SERVICE_VERSION]: "1.0.0",
  }),
  traceExporter: new OTLPTraceExporter({
    url: "http://localhost:4317",
  }),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: "http://localhost:4317",
    }),
    exportIntervalMillis: 60000,
  }),
  instrumentations: [getNodeAutoInstrumentations()],
});

sdk.start();

// Manual instrumentation
const tracer = trace.getTracer("my-service");

export async function processOrder(orderId: string): Promise<Order> {
  return tracer.startActiveSpan("processOrder", async (span) => {
    try {
      span.setAttribute("order.id", orderId);

      // Nested span for validation
      const order = await tracer.startActiveSpan("validateOrder", async (validateSpan) => {
        const result = await validateOrder(orderId);
        validateSpan.end();
        return result;
      });

      // Nested span for payment
      await tracer.startActiveSpan("processPayment", async (paymentSpan) => {
        paymentSpan.setAttribute("payment.amount", order.total);
        await processPayment(order);
        paymentSpan.end();
      });

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

## Best Practices

### Instrumentation

- Use semantic conventions for attributes
- Instrument at service boundaries
- Add business-relevant attributes
- Use span events for milestones

### Sampling

- Use tail-based sampling in collector
- Sample errors at 100%
- Sample slow requests at higher rates
- Use head-based sampling for high volume

### Performance

- Batch exports to reduce overhead
- Use async export when possible
- Set appropriate buffer sizes
- Monitor collector health

### Context

- Propagate context across services
- Use baggage for cross-cutting concerns
- Correlate logs with traces
- Include trace ID in error responses

You build comprehensive observability systems with OpenTelemetry for distributed tracing, metrics, and logging.