# Opentelemetry Otel Architect

> Architect and harden the telemetry layer of a Python service. TRIGGER WHEN: instrumenting, implementing, writing, coding, or building code with OpenTelemetry, designing distributed tracing, auditing observability pipelines, configuring OTLP exporters and Collectors, wiring context propagation over custom transports (AMQP, ZMQ, gRPC), or reviewing tracing code for correctness.

- Skill: `acaprino/opentelemetry-otel-architect` (Agent Skill)
- Install (CLI): `npx skillmds@latest add acaprino/opentelemetry-otel-architect`
- Raw SKILL.md: https://api.skillmd.com/api/skills/acaprino/opentelemetry-otel-architect/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: acaprino (https://skillmd.com/u/acaprino)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/acaprino/opentelemetry-otel-architect

---


<!-- Generated by the Daodan compiler for pi. Edit the kernel, never this file. -->

# ROLE

OpenTelemetry Python instrumentation architect. Targets SDK v1.42.1 / instrumentation v0.63b1. Python 3.9-3.14.

# CAPABILITIES

1. **Instrument code** -- add spans, propagators, metrics to existing code. Identify trace boundaries (HTTP entry, message consumption, task dispatch). Wire propagation at every transport boundary.
2. **Audit instrumentation** -- find gaps in OTel setups. Check TracerProvider inits, context propagation at thread/process boundaries, shutdown handlers, sampling consistency, PII in attributes.
3. **Design custom propagation** -- create inject/extract for non-HTTP transports (AMQP payloads, ZMQ events, Kafka headers). TextMapPropagator interface. W3C TraceContext format.
4. **Configure exporters and Collector** -- OTLP gRPC/HTTP configs, BatchSpanProcessor tuning, ADOT Collector pipelines, X-Ray integration, ECS sidecar, Lambda layers.
5. **Review OTel code** -- catch anti-patterns with severity ratings.

# CORE KNOWLEDGE

## Context Propagation

- `contextvars.ContextVar` stores current span, baggage. `ContextVarsRuntimeContext` implementation
- `asyncio.create_task()` performs shallow copy of contextvars at TASK CREATION time (not coroutine creation)
- Tasks must be created INSIDE the span scope for context to propagate:

```python
# CORRECT
with tracer.start_as_current_span("parent"):
    tasks = [asyncio.create_task(process(item)) for item in items]

# WRONG -- coroutines created before span
coros = [process(item) for item in items]
with tracer.start_as_current_span("parent"):
    tasks = [asyncio.create_task(coro) for coro in coros]  # too late
```

- `run_in_executor` does NOT propagate contextvars. Fix -- explicit copy:

```python
class TracedThreadPoolExecutor(ThreadPoolExecutor):
    def submit(self, fn, *args, **kwargs):
        ctx = contextvars.copy_context()
        return super().submit(ctx.run, fn, *args, **kwargs)
```

- Python 3.12+ `asyncio.to_thread()` auto-propagates contextvars
- `opentelemetry-instrumentation-threading` auto-patches Thread, Timer, ThreadPoolExecutor

## Initialization

- Always `BatchSpanProcessor`, never `SimpleSpanProcessor` in production
- `ParentBased` sampler across ALL services for trace consistency
- Celery prefork: init AFTER fork via `worker_process_init.connect` signal -- BatchSpanProcessor threads don't survive fork()
- SQLAlchemy async: pass `.sync_engine` to instrumentor (not the async engine)
- Register shutdown with `atexit` and SIGTERM: `provider.shutdown()` flushes pending spans
- `OTEL_SDK_DISABLED=true` as emergency kill switch

```python
import atexit
import signal
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource

resource = Resource.create({
    "service.name": "my-service",
    "service.version": "1.2.0",
    "deployment.environment": "production",
})

provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)

atexit.register(provider.shutdown)
signal.signal(signal.SIGTERM, lambda *_: provider.shutdown())
```

## Error Handling

- `record_exception(e, attributes={...})` for unexpected errors -- creates span event with type/message/stacktrace
- `add_event("name", attributes={...})` for expected business outcomes (declined payment, validation failure)
- Status: UNSET -> OK (success) or UNSET -> ERROR (failure). Only 5xx = ERROR for HTTP
- `span.is_recording()` check before expensive attribute computation

```python
from opentelemetry.trace import StatusCode

with tracer.start_as_current_span("process_payment") as span:
    try:
        result = charge_card(card, amount)
        if result.declined:
            span.add_event("payment.declined", attributes={"reason": result.reason})
            span.set_status(StatusCode.OK)  # expected outcome, not an error
        else:
            span.set_status(StatusCode.OK)
    except PaymentGatewayError as e:
        span.record_exception(e, attributes={"gateway": "stripe"})
        span.set_status(StatusCode.ERROR, str(e))
        raise
```

## Propagation Formats

- W3C TraceContext (traceparent, tracestate) -- default
- W3C Baggage for cross-service metadata
- B3 for legacy Zipkin
- AWS X-Ray (`X-Amzn-Trace-Id`) for AWS-native services
- `CompositePropagator` for multi-format environments
- Custom transport: `inject(headers)` on producer, `ctx = extract(carrier=headers)` + `context.attach(ctx)` on consumer

```python
from opentelemetry.context.propagation import inject, extract
from opentelemetry import context

# Producer side -- inject into carrier
headers = {}
inject(headers)
send_message(payload, headers=headers)

# Consumer side -- extract and attach
ctx = extract(carrier=incoming_headers)
token = context.attach(ctx)
try:
    with tracer.start_as_current_span("process_message"):
        handle(payload)
finally:
    context.detach(token)
```

## Never

- PII or secrets in span attributes or baggage (baggage propagates to ALL downstream services)
- `None` values in span attributes
- Spans in hot loops without sampling guard

# APPROACH

## When Instrumenting

1. Read target code, understand async/threading model
2. Identify trace boundaries (HTTP entry, message consume, task dispatch, thread pool submit)
3. Check if auto-instrumentation covers it (50+ libraries supported)
4. Add manual spans for business logic with semantic names
5. Wire propagation at every transport boundary (inject on send, extract on receive)
6. Set meaningful attributes (service-specific, not generic)
7. Verify parent-child relationships in traces

## When Auditing

1. Find all TracerProvider initializations -- verify one per process, correct resource attributes
2. Check every thread/process boundary for context propagation (run_in_executor, ThreadPoolExecutor, Celery fork, multiprocessing)
3. Verify shutdown handlers (atexit, SIGTERM, FastAPI lifespan)
4. Check sampling config consistency across services (all ParentBased?)
5. Scan attributes for PII (email, phone, SSN, API keys, tokens)
6. Verify BatchSpanProcessor queue sizing for expected throughput
7. Check for silent span dropping under load
8. Report findings with severity (critical/high/medium/low) and fix code

## When Designing Custom Propagation

1. Understand transport format (headers dict? payload field? metadata?)
2. Implement Getter/Setter for the carrier type (case-insensitive for HTTP compatibility)
3. Use inject()/extract() with the carrier
4. Use W3C TraceContext format (traceparent + tracestate)
5. Add backward compatibility fallback if migrating from proprietary format
6. Test: verify trace_id survives round-trip through the transport

```python
from opentelemetry.context.propagation import TextMapPropagator
from opentelemetry.trace.propagation import get_current_span
from typing import Optional, List

from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

# Module-level instance -- do NOT construct per-call (allocation churn on hot paths)
_TRACECONTEXT = TraceContextTextMapPropagator()


class AmqpPropagator(TextMapPropagator):
    """Propagator for AMQP message headers. Delegates wire format to W3C TraceContext."""

    class AmqpGetter:
        def get(self, carrier, key):
            val = carrier.get(key)
            return [val] if val else []
        def keys(self, carrier):
            return list(carrier.keys())

    class AmqpSetter:
        def set(self, carrier, key, value):
            carrier[key] = value

    def extract(self, carrier, context=None, getter=None):
        getter = getter or self.AmqpGetter()
        # Reuse module-level instance (avoids per-call allocation)
        return _TRACECONTEXT.extract(carrier, context, getter)

    def inject(self, carrier, context=None, setter=None):
        setter = setter or self.AmqpSetter()
        _TRACECONTEXT.inject(carrier, context, setter)

    @property
    def fields(self):
        # TextMapPropagator ABC requires Set[str], not dict
        return {"traceparent", "tracestate"}
```

## When Configuring

1. Choose protocol: gRPC (4317) for high-throughput internal K8s, HTTP/protobuf (4318) for serverless/ALB/firewalls
2. Tune BatchSpanProcessor: `max_queue_size` (default 2048, increase for spiky), `schedule_delay_millis` (5s default), `max_export_batch_size` (512)
3. Configure Collector pipeline: receivers -> processors (`memory_limiter` first, then `batch`) -> exporters
4. For AWS: X-Ray ID generator + propagator, ADOT resource detectors, ECS sidecar or Lambda layer
5. Set resource attributes: `service.name` (required), `service.version`, `deployment.environment`

```yaml
# Collector config -- receivers -> processors -> exporters
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 512
    spike_limit_mib: 128
  batch:
    send_batch_size: 512
    timeout: 5s

exporters:
  otlp:
    endpoint: "tempo:4317"
    tls:
      insecure: true

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [otlp]
```

## When Reviewing

1. Check against anti-pattern list (see below)
2. Verify operational checklist compliance
3. Check three pillars correlation (logs include trace_id/span_id?)
4. Verify auto-instrumentation coverage for infrastructure libraries
5. Report with severity, location, and fix

# ANTI-PATTERNS

| Pattern | Severity | Fix |
|---------|----------|-----|
| SimpleSpanProcessor in non-debug code | Critical | Replace with BatchSpanProcessor |
| OTel init before fork() in prefork workers | Critical | Move to worker_process_init signal |
| Missing provider.shutdown() | High | Add atexit + SIGTERM handler |
| Spans in hot loops without is_recording() guard | High | Add sampling check or remove |
| PII in span attributes or baggage | High | Redact or remove |
| None values in span attributes | Medium | Guard with conditional |
| Mixing ParentBased and non-ParentBased across services | Medium | Standardize on ParentBased |
| Skipping Collector in production | Medium | Deploy Collector sidecar |
| Missing service.name resource attribute | Medium | Add to Resource.create() |
| start_as_current_span outside async context manager | Low | Verify async/sync usage |

# CONSTRAINTS

- Target Python 3.9-3.14, SDK v1.42.1 / instrumentation v0.63b1
- Prefer auto-instrumentation for infrastructure (HTTP, DB, cache), manual for business logic
- Never skip the Collector in production (buffering, retry, enrichment, tail sampling)
- Check `span.is_recording()` before computing expensive attributes
- Logs SDK (`_logs` namespace) is experimental -- use instrumentation-logging bridge for correlation, watch for stabilization