Distributed Tracing: spans, baggage e sampling strategies

# Estrutura de um span OpenTelemetry
{
    "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",  # 128 bits, único por trace
    "span_id": "i9j0c2d4e6f8a0b2",                  # 64 bits, único por span
    "parent_span_id": "a1b2c3d4e5f6a7b8",            # None = root span
    "name": "process-payment",
    "kind": "SERVER",              # CLIENT | SERVER | PRODUCER | CONSUMER | INTERNAL
    "start_time": 1705329791.350,
    "end_time": 1705329791.690,    # duration = 340ms
    "status": {"code": "ERROR", "message": "Stripe timeout"},

    # Atributos — dados de contexto do span
    "attributes": {
        "service.name": "payment-service",
        "http.method": "POST",
        "http.url": "https://api.stripe.com/v1/charges",
        "http.status_code": 408,
        "payment.amount": 9990,
        "payment.currency": "BRL",
    },

    # Eventos — logs point-in-time dentro do span
    "events": [
        {
            "name": "exception",
            "timestamp": 1705329791.645,
            "attributes": {
                "exception.type": "stripe.error.APIConnectionError",
                "exception.message": "Connection timeout after 300s",
                "exception.stacktrace": "...",
            }
        }
    ],

    # Links — referências a outros traces (ex: mensagem de fila)
    "links": [],
    "resource": {"service.name": "payment-service", "service.version": "2.1.0"}
}

# traceparent — obrigatório
# Formato: versão-trace_id-parent_span_id-flags
traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01
#             ^^  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^ ^^
#             v0  trace_id (32 hex chars / 128 bits) parent_span_id  sampled=1

# tracestate — opcional, vendor-specific key-value
tracestate: rojo=00f067aa0ba902b7,congo=t61rcWkgMzE

# Flags (último byte):
# 01 = sampled (este trace está sendo coletado)
# 00 = not sampled (propaga contexto mas sem coletar dados)

# Python: propagação manual (útil em filas, workers assíncronos)
from opentelemetry import trace, propagate
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

propagator = TraceContextTextMapPropagator()

# === PRODUTOR (quem cria a mensagem) ===
def publish_to_queue(payload: dict, queue):
    carrier = {}
    propagator.inject(carrier)  # Extrai contexto ativo → dict

    message = {
        "payload": payload,
        "otel_context": carrier,  # {"traceparent": "00-4bf9...", "tracestate": ""}
    }
    queue.publish(message)

# === CONSUMIDOR (worker que processa a mensagem) ===
def process_queue_message(message: dict):
    carrier = message.get("otel_context", {})
    ctx = propagator.extract(carrier)  # Restaura contexto do produtor

    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span(
        "process-order-event",
        context=ctx,  # Continua o trace do produtor
        kind=trace.SpanKind.CONSUMER,
    ) as span:
        span.set_attribute("messaging.system", "rabbitmq")
        span.set_attribute("messaging.destination", "orders")
        process(message["payload"])

from opentelemetry import baggage, context

# === Injetar no início da requisição (ex: middleware de auth) ===
ctx = baggage.set_baggage("user.tier", "premium")
ctx = baggage.set_baggage("tenant.id", "acme-corp", context=ctx)
ctx = baggage.set_baggage("feature.flags", "new-checkout", context=ctx)

# Attach ao contexto da thread/coroutine
token = context.attach(ctx)

# === Ler em qualquer serviço downstream ===
user_tier = baggage.get_baggage("user.tier")      # "premium"
tenant_id = baggage.get_baggage("tenant.id")      # "acme-corp"

# Usar para decisões sem parâmetro explícito
if user_tier == "premium":
    rate_limit = 1000  # req/min
else:
    rate_limit = 100

# Baggage é propagado no header:
# baggage: user.tier=premium,tenant.id=acme-corp,feature.flags=new-checkout

from opentelemetry import trace
from opentelemetry.trace import StatusCode

tracer = trace.get_tracer(__name__, schema_url="https://semconv.org/")

def process_checkout(cart_id: str, user_id: str) -> Order:
    with tracer.start_as_current_span("checkout.process") as span:
        # Atributos descrevem o que o span representa
        span.set_attribute("checkout.cart_id", cart_id)
        span.set_attribute("checkout.user_id", user_id)
        span.set_attribute("checkout.item_count", len(cart.items))

        try:
            # Sub-operação — span filho automático
            with tracer.start_as_current_span("checkout.validate_inventory") as inv_span:
                inventory = check_inventory(cart.items)
                inv_span.set_attribute("inventory.items_available", len(inventory))

            # Evento point-in-time dentro do span pai
            span.add_event("inventory.validated", {"available": len(inventory)})

            order = create_order(cart, user_id)
            span.set_attribute("checkout.order_id", order.id)
            span.set_status(StatusCode.OK)
            return order

        except InventoryError as e:
            # Registrar exceção — aparece na timeline do span no Jaeger/Tempo
            span.record_exception(e)
            span.set_status(StatusCode.ERROR, f"Inventory unavailable: {e}")
            raise

        except Exception as e:
            span.record_exception(e)
            span.set_status(StatusCode.ERROR, str(e))
            raise

// Node.js: instrumentação manual
import { trace, SpanStatusCode, SpanKind } from '@opentelemetry/api';

const tracer = trace.getTracer('checkout-service', '1.0.0');

async function processCheckout(cartId: string, userId: string): Promise<Order> {
  return tracer.startActiveSpan(
    'checkout.process',
    {
      kind: SpanKind.SERVER,
      attributes: { 'checkout.cart_id': cartId, 'checkout.user_id': userId },
    },
    async (span) => {
      try {
        const order = await createOrder(cartId, userId);
        span.setAttributes({ 'checkout.order_id': order.id });
        span.setStatus({ code: SpanStatusCode.OK });
        return order;
      } catch (err) {
        span.recordException(err as Error);
        span.setStatus({ code: SpanStatusCode.ERROR, message: String(err) });
        throw err;
      } finally {
        span.end(); // SEMPRE chamar end() — mesmo em erro
      }
    }
  );
}

# Tail sampling no OTel Collector
processors:
  tail_sampling:
    decision_wait: 30s          # Aguarda trace completar
    num_traces: 100000          # Buffer de traces em memória
    expected_new_traces_per_sec: 1000

    policies:
      # SEMPRE guardar traces com erro
      - name: errors-policy
        type: status_code
        status_code: {status_codes: [ERROR]}

      # SEMPRE guardar traces lentos (> 2s)
      - name: slow-traces-policy
        type: latency
        latency: {threshold_ms: 2000}

      # SEMPRE guardar traces com header de debug
      - name: debug-policy
        type: string_attribute
        string_attribute:
          key: "sampling.debug"
          values: ["true"]

      # 5% dos restantes aleatoriamente
      - name: baseline-policy
        type: probabilistic
        probabilistic: {sampling_percentage: 5}

      # Composta: todos os spans de um trace devem satisfazer
      - name: composite-policy
        type: composite
        composite:
          max_total_spans_per_second: 10000
          policy_order: [errors-policy, slow-traces-policy, debug-policy, baseline-policy]

# TraceQL — exemplos práticos

# Todos os traces com erro no payment-service
{service.name="payment-service" && status=error}

# Traces com duração > 2s no checkout
{span.name="checkout.process" && duration > 2s}

# Traces onde payment falhou com status HTTP 500
{service.name="payment-service" && http.status_code=500}

# Traces de usuário premium com latência > 1s (via baggage/atributo)
{span.user.tier="premium" && duration > 1s}

# Buscar por atributo de negócio
{span.checkout.order_id="ORD-12345"}

# Agrupar por serviço + contar erros (Tempo 2.4+)
{status=error} | by(service.name) | count() > 10

# Rate de erros por serviço
rate({status=error}[5m]) by (service.name)