Observabilidade Distribuída

import os
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
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased

SERVICE_NAME  = os.getenv("SERVICE_NAME",  "agent-service")
OTLP_ENDPOINT = os.getenv("OTLP_ENDPOINT", "http://localhost:4317")
SAMPLE_RATE   = float(os.getenv("TRACE_SAMPLE_RATE", "1.0"))


def setup_tracing() -> trace.Tracer:
    resource = Resource.create({
        "service.name":          SERVICE_NAME,
        "service.version":       os.getenv("APP_VERSION", "unknown"),
        "deployment.environment": os.getenv("ENVIRONMENT", "dev"),
    })

    # ParentBased: se o pai já foi sampleado, o filho segue — veja seção de sampling.
    sampler  = ParentBased(root=TraceIdRatioBased(SAMPLE_RATE))
    provider = TracerProvider(resource=resource, sampler=sampler)
    exporter = OTLPSpanExporter(endpoint=OTLP_ENDPOINT, insecure=True)

    # BatchSpanProcessor: acumula spans e envia em lotes — muito mais eficiente
    # que SimpleSpanProcessor que faz I/O para cada span individualmente.
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return trace.get_tracer(SERVICE_NAME)


tracer = setup_tracing()  # tracer global — importado pelos outros módulos

from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx   import HTTPXClientInstrumentor

app = FastAPI()

# FastAPIInstrumentor: cria root span para cada request.
# Se o request tiver traceparent no header, continua aquele trace.
# Se não tiver, cria um novo trace com trace_id fresco.
FastAPIInstrumentor.instrument_app(app)

# HTTPXClientInstrumentor: injeta traceparent em todas as saídas httpx.
# Isso cobre automaticamente chamadas A2A — sem código adicional por endpoint.
HTTPXClientInstrumentor().instrument()

from functools import wraps
from opentelemetry.trace import SpanKind, Status, StatusCode


def traced_node(node_name: str, attributes: dict | None = None):
    def decorator(func):
        @wraps(func)
        async def wrapper(state: dict, config: dict, *args, **kwargs):
            thread_id = config.get("configurable", {}).get("thread_id", "unknown")

            # start_as_current_span: cria o span e o define como current no contexto.
            # Spans filhos criados dentro desta função herdarão este como pai.
            # kind=INTERNAL: não é chamada HTTP — é processamento interno do grafo.
            with tracer.start_as_current_span(
                f"langgraph.node.{node_name}",
                kind=SpanKind.INTERNAL,
            ) as span:
                span.set_attribute("langgraph.node",      node_name)
                span.set_attribute("langgraph.thread_id", thread_id)

                if attributes:
                    for k, v in attributes.items():
                        span.set_attribute(k, str(v))

                try:
                    result = await func(state, config, *args, **kwargs)
                    # Anota campos escalares do resultado — visíveis no trace do X-Ray.
                    # Útil para ver booking_id ou status diretamente no span.
                    if isinstance(result, dict):
                        for k, v in result.items():
                            if isinstance(v, (str, int, float, bool)):
                                span.set_attribute(f"langgraph.output.{k}", v)
                    span.set_status(Status(StatusCode.OK))
                    return result
                except Exception as exc:
                    # record_exception: captura stack trace no span.
                    # Visível no X-Ray como "exceptions" no span — filtráveis.
                    span.record_exception(exc)
                    span.set_status(Status(StatusCode.ERROR, str(exc)))
                    raise
        return wrapper
    return decorator

# Ordem de execução (de fora para dentro na declaração):
# traced_node → idempotent_node → função real
# O span cobre tudo — incluindo o check de cache do idempotent_node.

@traced_node("book_hotel", attributes={"agent.capability": "booking"})
@idempotent_node(redis, "book_hotel", key_fields=["hotel_id", "checkin"])
async def book_hotel_node(state: AgentState, config: dict) -> dict:
    return await with_retry(
        hotel_api.create_booking, hotel_id=state["hotel_id"],
        config=RETRY_CONFIGS["hotel_api"],
        circuit_breaker=circuit_registry.get("hotel_api"),
        service_name="hotel_api",
    )

import httpx
from opentelemetry.propagate import inject
from cascade_timeout import remaining_budget
from token_manager import token_manager


async def call_search_agent(endpoint: str, payload: dict) -> dict:
    token  = await token_manager.get_token()
    budget = remaining_budget()

    headers = {
        "Authorization":   f"Bearer {token}",
        "X-Timeout-Budget": str(budget),
        "Content-Type":    "application/json",
    }
    # inject() adiciona traceparent (e tracestate) ao dict de headers.
    # Modifica in-place — os headers do JWT e budget já estão presentes e
    # não são sobrescritos porque inject() só adiciona chaves de trace.
    inject(headers)

    async with httpx.AsyncClient(timeout=budget) as client:
        resp = await client.post(endpoint, json=payload, headers=headers)
        resp.raise_for_status()
        return resp.json()

from opentelemetry.propagate import extract
from opentelemetry.trace import SpanKind

# FastAPIInstrumentor faz isso automaticamente.
# Mostrado aqui para clareza sobre o mecanismo interno:
@app.post("/search/invoke")
async def search_invoke(request: Request, body: SearchRequest,
                         claims=Depends(require_scope("search:invoke"))):
    # extract() lê o traceparent do header e cria um SpanContext com o
    # trace_id do Agente A. O span criado aqui é filho daquele span.
    ctx = extract(dict(request.headers))

    with tracer.start_as_current_span("search_agent.invoke",
                                       context=ctx, kind=SpanKind.SERVER) as span:
        span.set_attribute("agent.name",         "search-agent")
        span.set_attribute("langgraph.thread_id", body.thread_id)
        return await run_search_graph(body)

from opentelemetry.sdk.trace.sampling import (
    Sampler, SamplingResult, Decision, TraceIdRatioBased
)


class ErrorAwareSampler(Sampler):
    def __init__(self, ratio: float):
        self._ratio = TraceIdRatioBased(ratio)

    def should_sample(self, parent_context, trace_id, name, kind, attributes, links):
        # Se vier de um pai já sampleado (cadeia A2A), segue a decisão do pai.
        # Garante consistência na cadeia — mesmo comportamento do ParentBased.
        parent_span = trace.get_current_span(parent_context)
        if parent_span and parent_span.is_recording():
            return SamplingResult(Decision.RECORD_AND_SAMPLE)

        # Atributo "error" definido antes do sampling — samplea sempre que há erro.
        if attributes and attributes.get("error"):
            return SamplingResult(Decision.RECORD_AND_SAMPLE)

        return self._ratio.should_sample(
            parent_context, trace_id, name, kind, attributes, links)

    def get_description(self) -> str:
        return "ErrorAwareSampler"

{
  name      = "xray-daemon"
  image     = "amazon/aws-xray-daemon:latest"
  essential = false  # falha do sidecar não derruba o agente
  cpu       = 32
  memory    = 256

  portMappings = [
    { containerPort = 4317, protocol = "tcp" },  # OTLP gRPC — recebe do agente
    { containerPort = 2000, protocol = "udp" },  # X-Ray nativo — SDKs legados
  ]
  # Sem hostPort: só acessível via localhost dentro do task
}

class QueueMonitor:
    async def run(self):
        while True:
            try:
                metrics = {
                    "recovery_queue_size":        await self.redis.llen("recovery:queue"),
                    "recovery_queue_utilization": await self._utilization(),
                    "dlq_size":                   await self.redis.llen("recovery:dlq"),
                    "threads_in_progress":        await self.redis.scard("threads:in_progress"),
                }
                # JSON no log — Logs Insights parseia e indexa cada campo.
                # Query: fields @timestamp, recovery_queue_size, dlq_size
                #        | filter ispresent(recovery_queue_size)
                #        | sort @timestamp desc
                logger.info("METRICS %s", json.dumps(metrics))

                if metrics["dlq_size"] > 0:
                    logger.error("DLQ contém %d item(ns). Intervenção necessária.",
                                 metrics["dlq_size"])
            except Exception:
                logger.exception("Erro no QueueMonitor")
            await asyncio.sleep(30)