Context as Code
Documentation

Performance Monitoring

The PerformanceMonitor port, in-memory adapter, and resilient production adapters.

Show:

Related: 14-observability/02-metrics-patterns.md · 14-observability/01-logging-standards.md Framework implementation: For FastAPI see 18-framework-implementations/fastapi/11-monitoring-and-performance.md.

Why an Abstract Interface

Monitoring backends change — CloudWatch today, Datadog or Prometheus tomorrow. Application code that calls a specific SDK directly cannot be tested without credentials and cannot be swapped without touching every call site.

The PerformanceMonitor port decouples metric recording from the monitoring backend. Application and middleware code records metrics through the interface; the concrete adapter handles delivery to whatever backend is configured.

The PerformanceMonitor Port

from abc import ABC, abstractmethod
 
 
class PerformanceMonitor(ABC):
 
    @abstractmethod
    async def record_request(
        self,
        endpoint: str,
        method: str,
        status_code: int,
        duration_ms: float,
    ) -> None:
        """Record a completed HTTP request with its outcome and duration."""
 
    @abstractmethod
    async def record_error(
        self,
        endpoint: str,
        error_type: str,
    ) -> None:
        """Record an application error at a given endpoint."""
 
    async def startup(self) -> None:
        """Called at application startup. Override to initialise connections."""
 
    async def shutdown(self) -> None:
        """Called at application shutdown. Override to flush buffered metrics."""

The interface lives in the application layer (application/monitoring/performance.py). Infrastructure adapters implement it. Application and middleware code never imports a concrete adapter.

Adapters

AdapterPurpose
In-memoryTests — collects calls in lists, zero external dependencies
CloudWatchProduction on AWS — batches metrics, circuit breaker included
No-opProduction services that don't require metrics yet

The adapter is selected by settings.performance_monitor_type and wired at startup via a factory.

In-Memory Adapter (Testing)

The in-memory adapter is used in all tests. It satisfies the interface without network calls:

class InMemoryPerformanceMonitor(PerformanceMonitor):
    def __init__(self):
        self.requests: list[dict] = []
        self.errors: list[dict] = []
 
    async def record_request(self, endpoint, method, status_code, duration_ms):
        self.requests.append({
            "endpoint": endpoint, "method": method,
            "status_code": status_code, "duration_ms": duration_ms,
        })
 
    async def record_error(self, endpoint, error_type):
        self.errors.append({"endpoint": endpoint, "error_type": error_type})

Tests assert on monitor.requests and monitor.errors directly — no mocking needed.

Where Metrics Are Recorded

Metrics are recorded in middleware, not in route handlers or application code. Middleware intercepts every request, measures wall-clock duration, and calls monitor.record_request() after the response is produced.

This keeps application handlers free of observability concerns. They perform their domain operation; middleware wraps and measures it.

Resilience Considerations

Production adapters that make external network calls must not allow monitoring failures to affect the service:

  • Circuit breaker: Stop attempting delivery after repeated failures; reset after a cooldown period. Monitoring is non-critical — a broken metric pipeline must never cascade.
  • Batching: Buffer metrics and flush in batches to reduce API call volume and cost.
  • Graceful shutdown: Flush buffered metrics on shutdown to avoid data loss.

Application code never handles monitoring failures — the adapter absorbs them silently (and may log a warning).

Rules

  • PerformanceMonitor is defined in the application layer — never in infrastructure.
  • Metrics are recorded in middleware — application handlers do not call the monitor.
  • The in-memory adapter is used in tests — never mock the interface.
  • Production adapters must not propagate exceptions to callers.
  • Startup/shutdown hooks initialise and flush adapters with external connections.
  • The monitoring adapter failing must not fail the request — observability is non-critical infrastructure.

Framework implementations

Abstract reference: 14-observability/

Monitoring uses the same adapter pattern as events and caching. The application layer defines an abstract interface. Infrastructure provides in-memory (development/testing) and CloudWatch (production) adapters.

Interface (application/monitoring/performance.py)

from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
from typing import AsyncIterator
 
 
class PerformanceMonitor(ABC):
    @asynccontextmanager
    @abstractmethod
    async def time_operation(
        self, operation_name: str, context: dict | None = None
    ) -> AsyncIterator[None]: ...
 
    @abstractmethod
    async def record_metric(
        self, metric_name: str, value: float, unit: str = "None",
        context: dict | None = None
    ) -> None: ...
 
    @abstractmethod
    async def record_counter(
        self, counter_name: str, increment: int = 1,
        context: dict | None = None
    ) -> None: ...
 
    async def start(self) -> None:
        pass
 
    async def shutdown(self) -> None:
        pass

Adapters

In-Memory (infrastructure/monitoring/adapters/inmemory.py)

import time
from contextlib import asynccontextmanager
from application.monitoring.performance import PerformanceMonitor
 
 
class InMemoryPerformanceMonitor(PerformanceMonitor):
    def __init__(self):
        self.metrics: list[dict] = []
        self.counters: dict[str, int] = {}
 
    @asynccontextmanager
    async def time_operation(self, operation_name: str, context=None):
        start = time.perf_counter()
        try:
            yield
        finally:
            duration_ms = (time.perf_counter() - start) * 1000
            self.metrics.append({
                "name": operation_name,
                "value": duration_ms,
                "unit": "Milliseconds",
                "context": context or {},
            })
 
    async def record_metric(self, metric_name, value, unit="None", context=None):
        self.metrics.append({"name": metric_name, "value": value, "unit": unit})
 
    async def record_counter(self, counter_name, increment=1, context=None):
        self.counters[counter_name] = self.counters.get(counter_name, 0) + increment

CloudWatch (infrastructure/monitoring/adapters/cloudwatch.py)

Key features:

  • Batches metrics (default 20) before flushing to CloudWatch to reduce API calls
  • Background flush task runs every 60 seconds
  • Circuit breaker disables CloudWatch on credential or persistent network errors
import asyncio
import time
import logging
import boto3
from contextlib import asynccontextmanager
from application.monitoring.performance import PerformanceMonitor
 
logger = logging.getLogger(__name__)
 
 
class CloudWatchPerformanceMonitor(PerformanceMonitor):
    def __init__(
        self,
        namespace: str,
        region: str,
        batch_size: int = 20,
        flush_interval: float = 60.0,
    ):
        self._namespace = namespace
        self._client = boto3.client("cloudwatch", region_name=region)
        self._batch_size = batch_size
        self._flush_interval = flush_interval
        self._buffer: list[dict] = []
        self._circuit_open = False
        self._circuit_reset_at: float = 0.0
        self._flush_task: asyncio.Task | None = None
 
    @asynccontextmanager
    async def time_operation(self, operation_name, context=None):
        start = time.perf_counter()
        try:
            yield
        finally:
            duration_ms = (time.perf_counter() - start) * 1000
            await self.record_metric(operation_name, duration_ms, "Milliseconds", context)
 
    async def record_metric(self, metric_name, value, unit="None", context=None):
        if self._circuit_open and time.time() < self._circuit_reset_at:
            return
        self._circuit_open = False
 
        dimensions = [{"Name": k, "Value": str(v)} for k, v in (context or {}).items()]
        self._buffer.append({
            "MetricName": metric_name,
            "Value": value,
            "Unit": unit,
            "Dimensions": dimensions,
        })
        if len(self._buffer) >= self._batch_size:
            await self._flush()
 
    async def record_counter(self, counter_name, increment=1, context=None):
        await self.record_metric(counter_name, float(increment), "Count", context)
 
    async def _flush(self) -> None:
        if not self._buffer:
            return
        batch = self._buffer[:self._batch_size]
        self._buffer = self._buffer[self._batch_size:]
        try:
            await asyncio.to_thread(
                self._client.put_metric_data,
                Namespace=self._namespace,
                MetricData=batch,
            )
        except self._client.exceptions.NoCredentialsError:
            logger.warning("cloudwatch.no_credentials — disabling for 5 minutes")
            self._circuit_open = True
            self._circuit_reset_at = time.time() + 300
        except Exception as exc:
            logger.warning("cloudwatch.flush_failed", extra={"error": str(exc)})
 
    async def start(self) -> None:
        self._flush_task = asyncio.create_task(self._periodic_flush())
 
    async def shutdown(self) -> None:
        if self._flush_task:
            self._flush_task.cancel()
        await self._flush()  # Final flush on shutdown
 
    async def _periodic_flush(self) -> None:
        while True:
            await asyncio.sleep(self._flush_interval)
            await self._flush()

Factory (infrastructure/monitoring/factories.py)

from config.settings import Settings
from application.monitoring.performance import PerformanceMonitor
from .adapters.inmemory import InMemoryPerformanceMonitor
from .adapters.cloudwatch import CloudWatchPerformanceMonitor
 
 
def build_performance_monitor(settings: Settings) -> PerformanceMonitor:
    match settings.monitoring_adapter:
        case "inmemory":
            return InMemoryPerformanceMonitor()
        case "cloudwatch":
            return CloudWatchPerformanceMonitor(
                namespace=settings.monitoring_namespace,
                region=settings.aws_region,
                batch_size=settings.monitoring_batch_size,
                flush_interval=settings.monitoring_flush_interval,
            )
        case _:
            raise ValueError(f"Unknown monitoring adapter: {settings.monitoring_adapter}")

Usage in Application / Infrastructure

# In a repository (timing database operations):
class SQLAlchemyAccountRepository(AccountRepository):
    def __init__(self, monitor: PerformanceMonitor):
        self._monitor = monitor
 
    async def find_by_id(self, session, account_id):
        async with self._monitor.time_operation(
            "db.account.find_by_id",
            context={"account_id": str(account_id)},
        ):
            result = await session.execute(...)
            return account_to_domain(result.scalar_one_or_none())
 
# In an application handler (timing a full use case):
async def execute(self, command):
    async with self._monitor.time_operation("handler.create_account"):
        async with self._uow as session:
            ...
    await self._monitor.record_counter("accounts.created")

Lifecycle

Register start() and shutdown() in api/lifespan.py:

@asynccontextmanager
async def lifespan(app):
    monitor = get_performance_monitor()
    await monitor.start()  # Starts background flush task (CloudWatch)
    yield
    await monitor.shutdown()  # Final flush, cancel background task

Rules

  • Use time_operation for latency measurements — it's always a context manager, never manual start/stop.
  • Metric names are dot-separated: db.account.find_by_id, handler.create_account, messages.received.
  • The circuit breaker in the CloudWatch adapter prevents cascading failures when AWS credentials are missing or the service is unavailable.
  • The in-memory adapter's metrics and counters are inspectable in tests — assert on them to verify monitoring calls.
Logging StandardsDependency Direction Violations