Monitoring & Observability

import sentry_sdk

# Initialize Sentry
sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    traces_sample_rate=1.0,  # 100% of transactions for tracing
    profiles_sample_rate=1.0,  # Profile 100% of sampled transactions
    environment="production",
    release="myapp@1.2.3",
)

# Automatic exception capture
def divide(a, b):
    return a / b

try:
    result = divide(10, 0)
except Exception as e:
    # Sentry automatically captures this
    sentry_sdk.capture_exception(e)

# Manual error reporting
sentry_sdk.capture_message("Something went wrong", level="error")

# Add context
sentry_sdk.set_user({"id": 123, "email": "user@example.com"})
sentry_sdk.set_tag("payment_method", "stripe")
sentry_sdk.set_context("order", {"order_id": "ORDER-123", "total": 99.99})

# Breadcrumbs (what happened before error)
sentry_sdk.add_breadcrumb(
    category="auth",
    message="User logged in",
    level="info"
)

from flask import Flask
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration

sentry_sdk.init(
    dsn="https://...",
    integrations=[FlaskIntegration()],
    traces_sample_rate=0.1,  # 10% sampling in production
    environment="production",
)

app = Flask(__name__)

@app.route('/api/users/<user_id>')
def get_user(user_id):
    # Sentry automatically creates transactions for each request
    user = db.query(User).get(user_id)

    if not user:
        # This will be captured by Sentry
        raise ValueError(f"User {user_id} not found")

    return {"user": user.to_dict()}

# settings.py
import sentry_sdk
from sentry_sdk.integrations.django import DjangoIntegration

sentry_sdk.init(
    dsn="https://...",
    integrations=[DjangoIntegration()],
    traces_sample_rate=0.1,
    send_default_pii=True,  # Send user IP, user agent
    environment="production",
    release="myapp@1.2.3",
)

# views.py
from django.http import JsonResponse
from sentry_sdk import capture_message, set_tag

def checkout(request):
    set_tag("payment_provider", "stripe")

    try:
        result = process_payment(request.POST)
        return JsonResponse({"success": True})
    except PaymentError as e:
        # Automatically captured by Sentry
        capture_message("Payment failed", level="error")
        return JsonResponse({"error": str(e)}, status=400)

from prometheus_client import Counter, Histogram, Gauge, Summary
from prometheus_client import start_http_server
import time
import random

# Define metrics
http_requests_total = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

http_request_duration_seconds = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint']
)

active_connections = Gauge(
    'active_connections',
    'Number of active connections'
)

database_query_duration = Summary(
    'database_query_duration_seconds',
    'Database query duration'
)

# Use metrics
def handle_request(method, endpoint):
    # Increment counter
    http_requests_total.labels(method=method, endpoint=endpoint, status=200).inc()

    # Track duration with histogram
    with http_request_duration_seconds.labels(method=method, endpoint=endpoint).time():
        time.sleep(random.random() * 0.1)  # Simulate work

    # Update gauge
    active_connections.inc()
    # ... do work ...
    active_connections.dec()

    # Track database query
    with database_query_duration.time():
        # Execute query
        time.sleep(random.random() * 0.05)

# Start metrics server on :8000/metrics
if __name__ == '__main__':
    start_http_server(8000)

    # Simulate traffic
    while True:
        handle_request('GET', '/api/users')
        handle_request('POST', '/api/orders')
        time.sleep(1)

from flask import Flask
from prometheus_flask_exporter import PrometheusMetrics

app = Flask(__name__)

# Automatic instrumentation
metrics = PrometheusMetrics(app)

# Custom metrics
by_endpoint_counter = metrics.counter(
    'by_endpoint_counter',
    'Request count by endpoint',
    labels={'endpoint': lambda: request.endpoint}
)

@app.route('/api/users')
@by_endpoint_counter
def get_users():
    return {"users": []}

@app.route('/api/orders')
def create_order():
    # Automatically tracked
    return {"order_id": 123}

# Custom metric
orders_created = Counter('orders_created_total', 'Total orders created')

@app.route('/api/checkout', methods=['POST'])
def checkout():
    # ... process order ...
    orders_created.inc()
    return {"success": True}

# Metrics available at /metrics endpoint

import logging
from pythonjsonlogger import jsonlogger

# Setup JSON formatter
logger = logging.getLogger()
logger.setLevel(logging.INFO)

logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    '%(asctime)s %(name)s %(levelname)s %(message)s'
)
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)

# Log with structured data
logger.info(
    "User login",
    extra={
        "user_id": 12345,
        "email": "user@example.com",
        "ip_address": "192.168.1.1",
        "login_method": "oauth"
    }
)

# Output (JSON):
# {
#   "asctime": "2024-01-20 10:30:00",
#   "name": "root",
#   "levelname": "INFO",
#   "message": "User login",
#   "user_id": 12345,
#   "email": "user@example.com",
#   "ip_address": "192.168.1.1",
#   "login_method": "oauth"
# }

from elasticsearch import Elasticsearch
import logging
import datetime

class ElasticsearchHandler(logging.Handler):
    def __init__(self, es_host='localhost:9200', index_name='app-logs'):
        super().__init__()
        self.es = Elasticsearch([es_host])
        self.index_name = index_name

    def emit(self, record):
        log_entry = {
            'timestamp': datetime.datetime.now(datetime.timezone.utc).isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
            'logger': record.name,
            'path': record.pathname,
            'line': record.lineno,
        }

        # Add extra fields
        if hasattr(record, 'user_id'):
            log_entry['user_id'] = record.user_id
        if hasattr(record, 'request_id'):
            log_entry['request_id'] = record.request_id

        # Send to Elasticsearch
        self.es.index(
            index=f"{self.index_name}-{datetime.date.today().strftime('%Y.%m.%d')}",
            document=log_entry
        )

# Setup
logger = logging.getLogger()
logger.addHandler(ElasticsearchHandler())

# Use
logger.info("Order created", extra={"user_id": 123, "order_id": 456})

from flask import Flask, request, g
import logging
import uuid

app = Flask(__name__)

# Configure structured logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.before_request
def before_request():
    # Generate request ID
    g.request_id = str(uuid.uuid4())

    # Log request
    logger.info(
        "Request started",
        extra={
            "request_id": g.request_id,
            "method": request.method,
            "path": request.path,
            "ip": request.remote_addr,
            "user_agent": request.headers.get('User-Agent')
        }
    )

@app.after_request
def after_request(response):
    # Log response
    logger.info(
        "Request completed",
        extra={
            "request_id": g.request_id,
            "status_code": response.status_code,
            "content_length": response.content_length
        }
    )
    return response

@app.route('/api/users/<user_id>')
def get_user(user_id):
    logger.info(
        "Fetching user",
        extra={"request_id": g.request_id, "user_id": user_id}
    )
    return {"user_id": user_id}

# Run your app with ddtrace
# ddtrace-run python app.py

from ddtrace import tracer
from datadog import initialize, statsd

# Initialize Datadog
initialize(api_key='your-api-key')

# Custom metrics
statsd.increment('page.views', tags=['page:home'])
statsd.gauge('users.active', 100)
statsd.histogram('request.duration', 0.5)

# Custom tracing
with tracer.trace("database.query") as span:
    span.set_tag("query.type", "SELECT")
    span.set_tag("table", "users")
    result = db.execute("SELECT * FROM users")

# Distributed tracing (automatic with ddtrace-run)
# Traces follow requests across services

from flask import Flask
from ddtrace import patch_all

# Patch all supported libraries
patch_all()

app = Flask(__name__)

@app.route('/api/users')
def get_users():
    # Automatically traced by Datadog
    return {"users": []}

# Run with: ddtrace-run flask run
# Datadog automatically:
# - Traces HTTP requests
# - Traces database queries (if using supported ORMs)
# - Captures errors
# - Measures latency

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

# Setup tracer
resource = Resource.create({"service.name": "my-service"})
trace.set_tracer_provider(TracerProvider(resource=resource))
tracer = trace.get_tracer(__name__)

# Export to Jaeger via OTLP (Jaeger natively supports OTLP)
otlp_exporter = OTLPSpanExporter(
    endpoint="localhost:4317",
    insecure=True,
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(otlp_exporter)
)

# Create custom spans
def process_order(order_id):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order_id", order_id)

        # Child span
        with tracer.start_as_current_span("validate_order"):
            validate(order_id)

        with tracer.start_as_current_span("charge_payment"):
            charge_payment(order_id)

        with tracer.start_as_current_span("send_confirmation"):
            send_email(order_id)

        span.set_attribute("status", "completed")
        return True

from flask import Flask
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

app = Flask(__name__)

# Auto-instrument Flask
FlaskInstrumentor().instrument_app(app)

# Auto-instrument requests library
RequestsInstrumentor().instrument()

@app.route('/api/users/<user_id>')
def get_user(user_id):
    # This HTTP request is automatically traced

    # Calls to external services are also traced
    import requests
    response = requests.get(f"http://auth-service/verify/{user_id}")

    return {"user_id": user_id}

# Every request automatically creates a trace
# with spans for HTTP request, database queries, external calls