Monitoring & Logging

Structured Logging

import logging
import json
from datetime import datetime

# Setup
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

def log_embedding_call(text, model, latency_ms, cached=False, cost=0):
    """Log structured metrics"""
    logger.info(json.dumps({
        'event': 'embedding_generated',
        'timestamp': datetime.now().isoformat(),
        'model': model,
        'text_length': len(text),
        'latency_ms': latency_ms,
        'cached': cached,
        'cost_usd': cost
    }))

---

Prometheus Metrics

from prometheus_client import Counter, Histogram, Gauge

# Metrics
embedding_requests = Counter('embedding_requests_total', 'Total requests')
embedding_latency = Histogram('embedding_latency_seconds', 'Latency')
cache_hits = Counter('cache_hits_total', 'Cache hits')
daily_cost = Gauge('embedding_cost_daily_usd', 'Daily cost')

# Instrument
@embedding_latency.time()
def get_embedding(text):
    embedding_requests.inc()
    
    # Check the cache
    cached = cache.get(text)
    if cached:
        cache_hits.inc()
        return cached
    
    # API call
    embedding = call_api(text)
    cache.set(text, embedding)
    
    # Track the cost
    cost = calculate_cost(text)
    daily_cost.inc(cost)
    
    return embedding

---

CloudWatch/Datadog

import boto3

cloudwatch = boto3.client('cloudwatch')

def publish_metric(metric_name, value, unit='Count'):
    """Publish to CloudWatch"""
    cloudwatch.put_metric_data(
        Namespace='Embeddings',
        MetricData=[{
            'MetricName': metric_name,
            'Value': value,
            'Unit': unit,
            'Timestamp': datetime.now()
        }]
    )

# Usage
publish_metric('CacheHitRate', 0.85, 'Percent')
publish_metric('DailyCost', 15.50, 'None')

---

Alerting

def check_and_alert():
    """Alert if metrics exceed a threshold"""
    metrics = get_current_metrics()
    
    # Alert if error rate > 5%
    if metrics['error_rate'] > 0.05:
        send_alert("High error rate", metrics['error_rate'])
    
    # Alert if daily cost > budget
    if metrics['daily_cost'] > 100:
        send_alert("Budget exceeded", metrics['daily_cost'])
    
    # Alert if p95 latency > 200ms
    if metrics['p95_latency'] > 200:
        send_alert("High latency", metrics['p95_latency'])

---

Summary

• ✅ Structured logging: JSON logs
• ✅ Prometheus: Time-series metrics
• ✅ CloudWatch: AWS monitoring
• ✅ Alerting: Automatic notifications
• ✅ Key metrics: Latency, error rate, cost

---

Module 7 - Capsule 04