Grafana Observability and Security for LLM Agents: OpenTelemetry on AWS Bedrock, GCP Vertex AI and Microsoft Foundry

# llm_tracing.py
import time
from contextlib import contextmanager
from opentelemetry import trace, metrics

tracer = trace.get_tracer("agent.llm")
meter = metrics.get_meter("agent.llm")

token_usage = meter.create_histogram(
    "gen_ai.client.token.usage", unit="{token}",
    description="Input and output tokens per call")
op_duration = meter.create_histogram(
    "gen_ai.client.operation.duration", unit="s",
    description="End to end model call duration")
guardrail_hits = meter.create_counter(
    "agent.guardrail.interventions", description="Guardrail blocks and redactions")

@contextmanager
def llm_span(provider: str, model: str, operation: str = "chat"):
    start = time.monotonic()
    attrs = {"gen_ai.operation.name": operation,
             "gen_ai.provider.name": provider,
             "gen_ai.request.model": model}
    with tracer.start_as_current_span(f"{operation} {model}", attributes=attrs) as span:
        try:
            yield span
        except Exception as exc:
            span.record_exception(exc)
            span.set_status(trace.StatusCode.ERROR, type(exc).__name__)
            span.set_attribute("error.type", type(exc).__name__)
            raise
        finally:
            op_duration.record(time.monotonic() - start, attrs)

def record_usage(span, provider, model, input_tokens, output_tokens, finish):
    span.set_attribute("gen_ai.usage.input_tokens", input_tokens)
    span.set_attribute("gen_ai.usage.output_tokens", output_tokens)
    span.set_attribute("gen_ai.response.finish_reasons", [finish])
    base = {"gen_ai.provider.name": provider, "gen_ai.request.model": model}
    token_usage.record(input_tokens, {**base, "gen_ai.token.type": "input"})
    token_usage.record(output_tokens, {**base, "gen_ai.token.type": "output"})

import boto3
from llm_tracing import llm_span, record_usage, guardrail_hits

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

def call_bedrock(model_id: str, messages: list, guardrail_id: str | None = None):
    with llm_span("aws.bedrock", model_id) as span:
        kwargs = dict(modelId=model_id, messages=messages,
                      inferenceConfig={"maxTokens": 1024, "temperature": 0.2})
        if guardrail_id:
            kwargs["guardrailConfig"] = {"guardrailIdentifier": guardrail_id,
                                         "guardrailVersion": "DRAFT", "trace": "enabled"}
        resp = bedrock.converse(**kwargs)
        usage = resp["usage"]
        record_usage(span, "aws.bedrock", model_id,
                     usage["inputTokens"], usage["outputTokens"], resp["stopReason"])
        if resp["stopReason"] == "guardrail_intervened":
            guardrail_hits.add(1, {"provider": "aws.bedrock", "action": "block"})
        return resp

from google import genai
from llm_tracing import llm_span, record_usage

client = genai.Client(vertexai=True, project="my-project", location="us-central1")

def call_vertex(model: str, prompt: str):
    with llm_span("gcp.vertex_ai", model) as span:
        resp = client.models.generate_content(model=model, contents=prompt)
        u = resp.usage_metadata
        finish = resp.candidates[0].finish_reason.name if resp.candidates else "none"
        record_usage(span, "gcp.vertex_ai", model,
                     u.prompt_token_count, u.candidates_token_count, finish)
        return resp

import os
from azure.ai.inference import ChatCompletionsClient
from azure.core.credentials import AzureKeyCredential
from llm_tracing import llm_span, record_usage

foundry = ChatCompletionsClient(
    endpoint=os.environ["FOUNDRY_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["FOUNDRY_KEY"]))

def call_foundry(model: str, messages: list):
    with llm_span("azure.ai.inference", model) as span:
        resp = foundry.complete(model=model, messages=messages)
        record_usage(span, "azure.ai.inference", model,
                     resp.usage.prompt_tokens, resp.usage.completion_tokens,
                     resp.choices[0].finish_reason)
        return resp

from opentelemetry import trace
tracer = trace.get_tracer("agent.loop")

def run_agent(task: str, max_steps: int = 8):
    with tracer.start_as_current_span("agent.run", attributes={"agent.name": "support-bot"}) as run:
        for step in range(max_steps):
            reply = call_model(task)                    # llm_span inside
            run.set_attribute("agent.steps", step + 1)
            if not reply.tool_calls:
                return reply.text
            for call in reply.tool_calls:
                with tracer.start_as_current_span(f"tool {call.name}",
                        attributes={"gen_ai.operation.name": "execute_tool",
                                    "gen_ai.tool.name": call.name}) as tspan:
                    try:
                        task = invoke_tool(call)
                    except Exception as exc:
                        tspan.record_exception(exc)
                        tspan.set_status(trace.StatusCode.ERROR)
                        task = f"tool error: {exc}"
        run.set_attribute("agent.stopped_reason", "max_steps")
        return "I could not finish this task."

# otel-collector.yaml
receivers:
  otlp:
    protocols: { grpc: { endpoint: 0.0.0.0:4317 } }

processors:
  memory_limiter: { check_interval: 1s, limit_mib: 512 }
  # Drop content attributes even if a developer turns capture on by mistake
  attributes/strip_content:
    actions:
      - { key: gen_ai.prompt,        action: delete }
      - { key: gen_ai.completion,    action: delete }
      - { key: gen_ai.input.messages,  action: delete }
      - { key: gen_ai.output.messages, action: delete }
  # Hash user identifiers instead of storing them
  attributes/hash_user:
    actions:
      - { key: enduser.id, action: hash }
  batch: {}

exporters:
  otlphttp/grafana:
    endpoint: https://otlp-gateway-prod-us-east-0.grafana.net/otlp
    headers: { Authorization: "Basic ${env:GRAFANA_OTLP_TOKEN}" }

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, attributes/strip_content, attributes/hash_user, batch]
      exporters: [otlphttp/grafana]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [otlphttp/grafana]

// Agent runs that hit the step cap
{ name = "agent.run" && span.agent.stopped_reason = "max_steps" }

// Slow tool calls inside long runs
{ name = "agent.run" && duration > 15s } >> { span.gen_ai.operation.name = "execute_tool" && duration > 3s }

// Any model call that returned a guardrail intervention
{ span.gen_ai.response.finish_reasons = "guardrail_intervened" }

groups:
  - name: llm-agents
    rules:
      - alert: LLMSpendSpike
        expr: llm:spend_per_hour:sum > 2 * avg_over_time(llm:spend_per_hour:sum[7d] offset 1h)
        for: 30m
        labels: { severity: page }
      - alert: GuardrailBlockSpike
        expr: sum(increase(agent_guardrail_interventions_total[15m])) > 3 * sum(increase(agent_guardrail_interventions_total[15m] offset 1d)) and sum(increase(agent_guardrail_interventions_total[15m])) > 10
        for: 10m
        labels: { severity: ticket }
        annotations:
          summary: "Guardrail blocks are 3x yesterday. Possible prompt-injection attempt or a bad prompt release."
      - alert: ProviderErrorRateHigh
        expr: llm:error_ratio:rate5m > 0.05
        for: 10m
        labels: { severity: page }