Event-Driven Microservices with Observability and Data Governance: A Complete Integration Tutorial¶

"""Observability-driven development: telemetry from day one."""
import time
import logging
from contextlib import contextmanager
from typing import Optional, Dict, Any
from functools import wraps

from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor
from opentelemetry.instrumentation.redis import RedisInstrumentor
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response

# Structured logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s - '
    'trace_id=%(trace_id)s span_id=%(span_id)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)

# OpenTelemetry setup
resource = Resource.create({"service.name": "order-service"})
trace.set_tracer_provider(TracerProvider(resource=resource))
jaeger_exporter = JaegerExporter(
    agent_host_name=os.getenv("JAEGER_AGENT_HOST", "localhost"),
    agent_port=int(os.getenv("JAEGER_AGENT_PORT", 6831)),
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)
tracer = trace.get_tracer(__name__)

# Prometheus metrics
orders_created = Counter(
    'orders_created_total',
    'Total number of orders created',
    ['status']
)
order_processing_duration = Histogram(
    'order_processing_duration_seconds',
    'Time spent processing orders',
    ['operation']
)
active_orders = Gauge(
    'active_orders',
    'Number of active orders'
)

# Instrumentation
Psycopg2Instrumentor().instrument()
RedisInstrumentor().instrument()


@contextmanager
def trace_operation(operation_name: str, **attributes):
    """Context manager for tracing operations with attributes."""
    span = tracer.start_span(operation_name)
    for key, value in attributes.items():
        span.set_attribute(key, str(value))
    try:
        yield span
    except Exception as e:
        span.record_exception(e)
        span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
        raise
    finally:
        span.end()


def observe_function(func):
    """Decorator for automatic observability."""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        operation = f"{func.__module__}.{func.__name__}"
        start_time = time.time()

        with trace_operation(operation):
            logger.info(f"Starting {operation}", extra={
                "operation": operation,
                "args": str(args),
                "kwargs": str(kwargs)
            })

            try:
                result = await func(*args, **kwargs)
                duration = time.time() - start_time
                order_processing_duration.labels(operation=operation).observe(duration)

                logger.info(f"Completed {operation}", extra={
                    "operation": operation,
                    "duration_seconds": duration,
                    "success": True
                })
                return result
            except Exception as e:
                duration = time.time() - start_time
                logger.error(f"Failed {operation}", extra={
                    "operation": operation,
                    "duration_seconds": duration,
                    "error": str(e),
                    "error_type": type(e).__name__
                }, exc_info=True)
                raise

    return wrapper

"""Data validation and contract governance."""
from datetime import datetime
from decimal import Decimal
from typing import Optional, List
from enum import Enum

from pydantic import BaseModel, Field, validator, root_validator
from pydantic.dataclasses import dataclass


class OrderStatus(str, Enum):
    """Order status enumeration."""
    PENDING = "pending"
    CONFIRMED = "confirmed"
    PROCESSING = "processing"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"


class OrderItem(BaseModel):
    """Order item with validation."""
    product_id: str = Field(..., min_length=1, max_length=100)
    quantity: int = Field(..., gt=0, le=1000)
    price: Decimal = Field(..., gt=0, decimal_places=2)

    @validator('product_id')
    def validate_product_id(cls, v):
        """Validate product ID format."""
        if not v.isalnum():
            raise ValueError("product_id must be alphanumeric")
        return v

    @property
    def total(self) -> Decimal:
        """Calculate item total."""
        return self.quantity * self.price


class OrderCreate(BaseModel):
    """Order creation contract."""
    customer_id: str = Field(..., min_length=1, max_length=100)
    items: List[OrderItem] = Field(..., min_items=1, max_items=100)
    shipping_address: str = Field(..., min_length=10, max_length=500)

    @validator('customer_id')
    def validate_customer_id(cls, v):
        """Validate customer ID format."""
        if not v.isalnum():
            raise ValueError("customer_id must be alphanumeric")
        return v

    @root_validator
    def validate_order_total(cls, values):
        """Validate order business rules."""
        items = values.get('items', [])
        if not items:
            raise ValueError("Order must have at least one item")

        total = sum(item.total for item in items)
        if total > Decimal('100000'):
            raise ValueError("Order total cannot exceed $100,000")

        return values


class OrderResponse(BaseModel):
    """Order response contract."""
    order_id: str
    customer_id: str
    status: OrderStatus
    items: List[OrderItem]
    total: Decimal
    created_at: datetime
    updated_at: datetime

    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat(),
            Decimal: lambda v: float(v)
        }

"""Event-driven architecture: event publishing."""
import json
import logging
from datetime import datetime
from typing import Dict, Any

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField

from app.schemas import OrderResponse, OrderStatus

logger = logging.getLogger(__name__)


class EventPublisher:
    """Publishes events to Kafka with schema validation."""

    def __init__(self, kafka_bootstrap_servers: str, schema_registry_url: str):
        self.producer = Producer({
            'bootstrap.servers': kafka_bootstrap_servers,
            'acks': 'all',  # Wait for all replicas
            'retries': 3,
            'max.in.flight.requests.per.connection': 1,  # Ensure ordering
        })

        self.schema_registry = SchemaRegistryClient({'url': schema_registry_url})
        self._load_schemas()

    def _load_schemas(self):
        """Load Avro schemas for events."""
        # In production, load from schema registry or files
        self.order_created_schema = {
            "type": "record",
            "name": "OrderCreated",
            "fields": [
                {"name": "order_id", "type": "string"},
                {"name": "customer_id", "type": "string"},
                {"name": "total", "type": "double"},
                {"name": "timestamp", "type": "long", "logicalType": "timestamp-millis"}
            ]
        }

        self.order_serializer = AvroSerializer(
            self.schema_registry,
            json.dumps(self.order_created_schema),
            lambda order, ctx: {
                'order_id': order['order_id'],
                'customer_id': order['customer_id'],
                'total': float(order['total']),
                'timestamp': int(datetime.now().timestamp() * 1000)
            }
        )

    def publish_order_created(self, order: OrderResponse):
        """Publish order created event."""
        try:
            event_data = {
                'order_id': order.order_id,
                'customer_id': order.customer_id,
                'total': order.total,
                'timestamp': int(datetime.now().timestamp() * 1000)
            }

            self.producer.produce(
                topic='order.created',
                value=self.order_serializer(
                    event_data,
                    SerializationContext('order.created', MessageField.VALUE)
                ),
                key=order.order_id,
                callback=self._delivery_callback
            )

            logger.info("Published order.created event", extra={
                "order_id": order.order_id,
                "customer_id": order.customer_id
            })

        except Exception as e:
            logger.error("Failed to publish order.created event", extra={
                "order_id": order.order_id,
                "error": str(e)
            }, exc_info=True)
            raise

    def _delivery_callback(self, err, msg):
        """Handle message delivery callback."""
        if err:
            logger.error(f"Message delivery failed: {err}")
        else:
            logger.debug(f"Message delivered to {msg.topic()} [{msg.partition()}]")

"""Order service main application."""
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import Response
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.observability import observe_function, orders_created, active_orders
from app.schemas import OrderCreate, OrderResponse, OrderStatus
from app.events import EventPublisher
from app.models import Order, OrderItem as OrderItemModel
from app.handlers import OrderHandler

# Database setup
DATABASE_URL = os.getenv("DATABASE_URL")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Event publisher
event_publisher = EventPublisher(
    kafka_bootstrap_servers=os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092"),
    schema_registry_url=os.getenv("SCHEMA_REGISTRY_URL", "http://localhost:8081")
)

# Order handler
order_handler = OrderHandler(SessionLocal, event_publisher)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan management."""
    # Startup
    yield
    # Shutdown
    event_publisher.producer.flush()


app = FastAPI(
    title="Order Service",
    description="Order processing microservice with observability",
    lifespan=lifespan
)

# Instrument FastAPI
from app.observability import FastAPIInstrumentor
FastAPIInstrumentor.instrument_app(app)


@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy", "service": "order-service"}


@app.get("/metrics")
async def metrics():
    """Prometheus metrics endpoint."""
    from app.observability import generate_latest
    return Response(content=generate_latest(), media_type="text/plain")


@app.post("/orders", response_model=OrderResponse)
@observe_function
async def create_order(order_data: OrderCreate):
    """Create a new order with validation and event publishing."""
    try:
        order = await order_handler.create_order(order_data)
        orders_created.labels(status=order.status.value).inc()
        active_orders.inc()
        return order
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail="Internal server error")


@app.get("/orders/{order_id}", response_model=OrderResponse)
@observe_function
async def get_order(order_id: str):
    """Get order by ID."""
    order = await order_handler.get_order(order_id)
    if not order:
        raise HTTPException(status_code=404, detail="Order not found")
    return order