# Observability

Comprehensive observability for agent systems: logging, metrics, tracing, alerting, and debugging.

---

## Observability Pillars

| Pillar | Purpose | Tools |
|---|---|---|
| **Logs** | Structured events, debugging | Loki, ELK, CloudWatch, Datadog |
| **Metrics** | Quantitative measurements | Prometheus, Datadog, CloudWatch |
| **Traces** | Request flow, latency analysis | Tempo, Jaeger, Zipkin, X-Ray |
| **Profiles** | CPU, memory, CPU/GPU | PySpy, PyInstrument, pprof |
| **Alerting** | Proactive notification | Alertmanager, PagerDuty, Opsgenie |

---

## Logging

### Structured Logging
```python
import structlog
import logging

# Configure structlog
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

# Usage with context
logger = logger.bind(agent_id="agent-123", session_id="session-456")

async def process_request(request: Request):
    logger.info("request_received", 
                method=request.method,
                path=request.path,
                user_id=request.user.id)
    
    try:
        result = await process(request)
        logger.info("request_completed", 
                    duration_ms=duration,
                    status="success")
        return result
    except Exception as e:
        logger.exception("request_failed",
                         error=str(e),
                         error_type=type(e).__name__)
        raise
```

### Log Levels & Sampling
```python
# Dynamic log levels
import logging
from functools import wraps

def sampled_logging(sample_rate: float = 0.1):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if random.random() < sample_rate:
                return await func(*args, **kwargs)
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Structured fields for querying
logger.info("tool_called",
    tool_name="search_web",
    tool_args={"query": "AI agents"},
    duration_ms=245,
    success=True,
    tokens_used=150,
    cost_usd=0.0003)
```

### Log Aggregation (Loki)
```yaml
# promtail-config.yaml
clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
- job_name: agent-logs
  static_configs:
  - targets:
      - localhost
    labels:
      job: agent-service
      environment: production
      agent: agent-service
    labels:
      agent: agent-service
      environment: production

  pipeline_stages:
  - json:
      expressions:
        agent_id: agent_id
        session_id: session_id
        tool_name: tool_name
        duration_ms: duration_ms
        success: success
  - labels:
        agent_id:
        session_id:
        tool_name:
        success:
```

---

## Metrics

### Prometheus Metrics
```python
from prometheus_client import Counter, Histogram, Gauge, Summary

# Counters
AGENT_REQUESTS_TOTAL = Counter(
    'agent_requests_total',
    'Total requests processed',
    ['agent', 'status', 'tool']
)

AGENT_TOOL_CALLS_TOTAL = Counter(
    'agent_tool_calls_total',
    'Total tool calls',
    ['agent', 'tool', 'status']
)

AGENT_ERRORS_TOTAL = Counter(
    'agent_errors_total',
    'Total errors',
    ['agent', 'error_type', 'tool']
)

# Histograms
AGENT_REQUEST_DURATION = Histogram(
    'agent_request_duration_seconds',
    'Request latency in seconds',
    ['agent', 'operation'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60]
)

AGENT_TOKEN_USAGE = Histogram(
    'agent_tokens_total',
    'Tokens used',
    ['agent', 'type'],  # input, output, total
    buckets=[10, 50, 100, 500, 1000, 5000, 10000, 50000, 100000]
)

AGENT_COST_USD = Histogram(
    'agent_cost_usd',
    'Cost in USD',
    ['agent', 'operation'],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10, 50, 100]
)

# Gauges
AGENT_ACTIVE_SESSIONS = Gauge(
    'agent_active_sessions',
    'Currently active sessions',
    ['agent']
)

AGENT_QUEUE_DEPTH = Gauge(
    'agent_queue_depth',
    'Tasks waiting in queue',
    ['agent', 'queue']
)

AGENT_ACTIVE_CONNECTIONS = Gauge(
    'agent_active_connections',
    'Active WebSocket/HTTP connections',
    ['agent']
)
```

### Custom Metrics Wrapper
```python
class MetricsCollector:
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
    
    @contextmanager
    def measure(self, operation: str, labels: dict = None):
        labels = labels or {}
        labels['agent'] = self.agent_name
        labels['operation'] = operation
        
        start = time.time()
        success = True
        try:
            yield
        except Exception as e:
            AGENT_ERRORS_TOTAL.labels(
                agent=self.agent_name,
                error_type=type(e).__name__,
                operation=operation
            ).inc()
            raise
        finally:
            duration = time.time() - start
            AGENT_REQUEST_DURATION.labels(**labels).observe(duration)
            AGENT_REQUESTS_TOTAL.labels(
                agent=self.agent_name,
                status='success' if success else 'error',
                operation=operation
            ).inc()
    
    def record_tokens(self, input_tokens: int, output_tokens: int, 
                      model: str = "gpt-4o"):
        AGENT_TOKEN_USAGE.labels(agent=self.agent_name, type='input').observe(input_tokens)
        AGENT_TOKEN_USAGE.labels(agent=self.agent_name, type='output').observe(output_tokens)
        AGENT_TOKEN_USAGE.labels(agent=self.agent_name, type='total').observe(input_tokens + output_tokens)
        
        # Cost calculation
        pricing = MODEL_PRICING.get(model, {"input": 0.01, "output": 0.03})
        cost = (input_tokens * pricing['input'] + output_tokens * pricing['output']) / 1000
        AGENT_COST_USD.labels(agent=self.agent_name, operation='llm').observe(cost)
```

---

## Distributed Tracing

### OpenTelemetry Setup
```python
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# Setup
resource = Resource.create({
    "service.name": "agent-service",
    "service.version": "1.2.3",
    "deployment.environment": "production"
})

provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://tempo:4317", insecure=True)
))
trace.set_tracer_provider(provider)

# Auto-instrumentation
OpenAIInstrumentor().instrument()
RequestsInstrumentor().instrument()

# Get tracer
tracer = trace.get_tracer(__name__)
```

### Custom Spans for Agent Operations
```python
from opentelemetry.trace import SpanKind, Status, StatusCode

class TracedAgent:
    def __init__(self, name: str):
        self.name = name
        self.tracer = trace.get_tracer(__name__)
    
    async def run(self, input: str) -> str:
        with self.tracer.start_as_current_span(
            "agent.run",
            kind=SpanKind.SERVER,
            attributes={
                "agent.name": self.name,
                "input.length": len(input)
            }
        ) as span:
            try:
                # Planning
                with self.tracer.start_as_current_span("agent.plan") as plan_span:
                    plan = await self.plan(input)
                    plan_span.set_attribute("plan.steps", len(plan.steps))
                
                # Execution
                with self.tracer.start_as_current_span("agent.execute") as exec_span:
                    result = await self.execute(plan)
                    exec_span.set_attribute("result.length", len(result))
                
                span.set_status(Status(StatusCode.OK))
                return result
            except Exception as e:
                span.record_exception(e)
                span.set_status(Status(StatusCode.ERROR, str(e)))
                raise
    
    @contextmanager
    def tool_span(self, tool_name: str, args: dict):
        with self.tracer.start_as_current_span(
            f"tool.{tool_name}",
            kind=SpanKind.CLIENT,
            attributes={
                "tool.name": tool_name,
                "tool.args": json.dumps(args)[:1000]
            }
        ) as span:
            try:
                yield span
                span.set_status(Status(StatusCode.OK))
            except Exception as e:
                span.record_exception(e)
                span.set_status(Status(StatusCode.ERROR, str(e)))
                raise
```

### Context Propagation
```python
# Inject context into outgoing requests
from opentelemetry.propagate import inject, extract
from opentelemetry.propagators.textmap import TextMapPropagator

def inject_context(headers: dict) -> dict:
    inject(headers)
    return headers

def extract_context(headers: dict) -> Context:
    return extract(headers)

# In HTTP client
async def make_request(url: str, data: dict, headers: dict = None):
    headers = headers or {}
    inject(headers)  # Inject trace context
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=data, headers=headers)
        return response
```

---

## Alerting

### Alert Rules (Prometheus)
```yaml
# alerts.yml
groups:
- name: agent-alerts
  rules:
  # High error rate
  - alert: AgentHighErrorRate
    expr: |
      sum(rate(agent_errors_total[5m])) by (agent) 
      / sum(rate(agent_requests_total[5m])) by (agent) > 0.05
    for: 2m
    labels:
      severity: critical
    annotations:
      summary: "High error rate for {{ $labels.agent }}"
      description: "Error rate is {{ $value | humanizePercentage }}"
  
  # High latency
  - alert: AgentHighLatency
    expr: |
      histogram_quantile(0.95, 
        sum(rate(agent_request_duration_seconds_bucket[5m])) by (agent, le)
      ) > 10
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "High latency for {{ $labels.agent }}"
      description: "P95 latency is {{ $value }}s"
  
  # Queue backlog
  - alert: AgentQueueBacklog
    expr: |
      agent_queue_depth > 1000
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Queue backlog for {{ $labels.agent }}"
  
  # High cost
  - alert: AgentHighCost
    expr: |
      sum(rate(agent_cost_usd_sum[1h])) by (agent) > 10
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "High cost for {{ $labels.agent }}"
      description: "Hourly cost is ${{ $value }}"
  
  # Low success rate
  - alert: AgentLowSuccessRate
    expr: |
      sum(rate(agent_requests_total{status="success"}[5m])) by (agent)
      / sum(rate(agent_requests_total[5m])) by (agent) < 0.95
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Low success rate for {{ $labels.agent }}"
```

### Alertmanager Config
```yaml
# alertmanager.yml
global:
  resolve_timeout: 5m
  smtp_smarthost: 'smtp.gmail.com:587'
  smtp_from: 'alerts@example.com'
  smtp_auth_username: 'alerts@example.com'
  smtp_auth_password: '${SMTP_PASSWORD}'

route:
  group_by: ['alertname', 'agent']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  receiver: 'default'
  routes:
  - match:
      severity: critical
    receiver: 'critical'
    continue: true
  - match:
      severity: warning
    receiver: 'warning'

receivers:
- name: 'default'
  email_configs:
  - to: 'team@example.com'
    send_resolved: true
- name: 'critical'
  pagerduty_configs:
  - service_key: '${PAGERDUTY_KEY}'
    severity: critical
- name: 'warning'
  slack_configs:
  - api_url: '${SLACK_WEBHOOK}'
    channel: '#alerts-warning'
    send_resolved: true
    title: '{{ .GroupLabels.alertname }}'
    text: '{{ range .Alerts }}{{ .Annotations.summary }}\n{{ end }}'
```

---

## Debugging & Profiling

### Profiling
```python
# PySpy for CPU profiling
# pip install py-spy
# py-spy record -o profile.svg --pid <PID>

# Or programmatically
import pyinstrument

profiler = pyinstrument.Profiler()
profiler.start()

# Your agent code here
result = await agent.run(input)

profiler.stop()
profiler.print()
profiler.write_html("profile.html")
```

### Memory Profiling
```python
# Memory profiling with memory_profiler
from memory_profiler import profile

@profile
async def memory_intensive_function():
    # Your code
    pass

# Or with objgraph
import objgraph

def count_objects():
    print(f"Total objects: {objgraph.count()}")
    print("Top types:", objgraph.most_common_types(limit=10))
    objgraph.show_most_common_types(limit=10)
```

### Memory Leak Detection
```python
import tracemalloc

tracemalloc.start()

# ... your agent code ...

snapshot1 = tracemalloc.take_snapshot()
# ... run agent ...
snapshot2 = tracemalloc.take_snapshot()

top_stats = snapshot2.compare_to(snapshot1, 'lineno')
for stat in top_stats[:10]:
    print(stat)
```

---

## Log Analysis Queries

### Loki Queries (LogQL)
```logql
# Error rate by agent
sum by (agent) (rate({job="agent-service"} |= "ERROR" [5m]))

# Latency percentiles
histogram_quantile(0.95, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))

# Tool usage
sum by (tool) (rate(agent_tool_calls_total[5m]))

# Error patterns
{job="agent-service"} |= "ERROR" | json | level="error" | tool_name != ""

# Slow queries
{job="agent-service"} | json | duration_ms > 5000

# Specific tool errors
{job="agent-service"} | json | tool_name="search_web" | level="error"

# Session tracking
{job="agent-service"} | json | session_id="abc123"

# Rate limited
{job="agent-service"} | json | status_code="429"
```

---

## Dashboards (Grafana)

### Agent Overview Dashboard
```json
{
  "dashboard": {
    "title": "Agent Service Overview",
    "panels": [
      {
        "title": "Request Rate",
        "type": "graph",
        "targets": [{
          "expr": "sum(rate(agent_requests_total[5m])) by (agent)",
          "legendFormat": "{{agent}}"
        }]
      },
      {
        "title": "Error Rate",
        "targets": [{
          "expr": "sum(rate(agent_errors_total[5m])) by (agent) / sum(rate(agent_requests_total[5m])) by (agent)",
          "legendFormat": "{{agent}}"
        }]
      },
      {
        "title": "Latency (P50, P95, P99)",
        "targets": [
          {"expr": "histogram_quantile(0.50, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))", "legendFormat": "P50 {{agent}}"},
          {"expr": "histogram_quantile(0.95, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))", "legendFormat": "P95 {{agent}}"},
          {"expr": "histogram_quantile(0.99, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))", "legendFormat": "P99 {{agent}}"}
        ]
      },
      {
        "title": "Token Usage & Cost",
        "targets": [
          {"expr": "sum(rate(agent_tokens_total{type=\"total\"}[5m])) by (agent)", "legendFormat": "Tokens/sec {{agent}}"},
          {"expr": "sum(rate(agent_cost_usd_sum[1h])) by (agent)", "legendFormat": "Cost/hr {{agent}}"}
        ]
      },
      {
        "title": "Queue Depth",
        "targets": [{
          "expr": "agent_queue_depth",
          "legendFormat": "{{agent}}"
        }]
      },
      {
        "title": "Tool Usage",
        "targets": [{
          "expr": "sum(rate(agent_tool_calls_total[5m])) by (tool, agent)",
          "legendFormat": "{{agent}} - {{tool}}"
        }]
      }
    ]
  }
}
```

---

## Related Notes

- [[Production/Deployment-Architecture.md]] — Deployment patterns
- [[Production/Scaling-Strategies.md]] — Scaling strategies
- [[Production/Reliability.md]] — Circuit breakers, retries
- [[Production/Security.md]] — Security
- [[Production/Cost-Optimization.md]] — Cost optimization
- [[Production/Compliance.md]] — Compliance
- [[Production/Deployment-Architecture.md]] — Deployment
- [[DevOps/Observability.md]] — DevOps observability
- [[DevOps/CI-CD-Best-Practices.md]] — CI/CD
- [[DevOps/Infrastructure-as-Code.md]] — IaC
- [[Security/Supply-Chain-Security.md]] — Supply chain security
- [[Production/Reliability.md]] — Reliability
- [[Production/Cost-Optimization.md]] — Cost optimization
- [[Production/Compliance.md]] — Compliance
- [[Frontiers/Research-Directions.md]] — Observability research

---

*Sources: Prometheus, Grafana, Loki, Tempo, Jaeger, OpenTelemetry, OpenTelemetry Python, PyInstrument, py-spy, memory_profiler, objgraph, tracemalloc, Prometheus Client, Grafana, Loki, Tempo, Jaeger, Zipkin, Zipkin, Zipkin, Alertmanager, PagerDuty, Opsgenie, Grafana OnCall, Prometheus Alerting, Prometheus Recording Rules, Prometheus Federation, Thanos, Cortex, Mimir, VictoriaMetrics, Grafana Loki, Grafana Tempo, Grafana Pyroscope, Grafana Faro, Grafana Beyla, eBPF, bpftrace, perf, perfetto, Sentry, Datadog, New Relic, Honeycomb, Lightstep, Elastic APM, Stackdriver.*