Introduction

A deployed Machine Learning model is part of a larger production system. A prediction request may pass through an API gateway, authentication service, feature store, inference service, model runtime, database, message broker, and downstream business application.

When a prediction is slow, incorrect, unavailable, or inconsistent, an operations team needs more than a generic error message. The team must be able to determine which model version ran, which application version handled the request, where time was spent, which dependency failed, whether feature values were missing or stale, and whether similar requests were affected.

Model observability is the ability to understand the internal state and behavior of an ML system by examining the telemetry it produces. Logging is one important part of observability, but complete observability also includes metrics, traces, events, version metadata, model-quality signals, and business outcomes.

Learning Objectives

Prerequisites

python -m pip install \
  fastapi \
  uvicorn \
  opentelemetry-api \
  opentelemetry-sdk \
  opentelemetry-exporter-otlp \
  opentelemetry-instrumentation-fastapi \
  prometheus-client

What Is Observability?

Observability is an engineering property that allows an operator to understand a system's internal behavior using the signals emitted by that system.

Monitoring vs Observability

Layers of ML Observability

Logs

A log is a timestamped record describing an event that occurred within an application, service, pipeline, or infrastructure component.

Structured Logging

Structured logging records events as named fields rather than unstructured text. This makes logs easier to search, filter, aggregate, correlate, validate, and process automatically.

{
    "timestamp": "recorded-event-time",
    "level": "ERROR",
    "event_name": "feature_retrieval_failed",
    "service_name": "customer-churn-api",
    "application_version": "2.1.0",
    "model_name": "customer-churn-classifier",
    "model_version": "1.0.0",
    "request_id": "generated-request-id",
    "trace_id": "generated-trace-id",
    "dependency": "feature-store",
    "error_type": "timeout",
    "duration_ms": 1200,
    "fallback_applied": true
}

Log Levels

Create a JSON Log Formatter

import json
import logging
from datetime import datetime
from datetime import timezone


class JsonLogFormatter(
    logging.Formatter
):

    def format(
        self,
        record
    ):
        log_record = {
            "timestamp": (
                datetime.now(
                    timezone.utc
                ).isoformat()
            ),
            "level": record.levelname,
            "logger": record.name,
            "message": (
                record.getMessage()
            )
        }

        approved_fields = [
            "event_name",
            "service_name",
            "application_version",
            "model_name",
            "model_version",
            "request_id",
            "trace_id",
            "span_id",
            "schema_version",
            "duration_ms",
            "status",
            "error_type"
        ]

        for field_name in approved_fields:

            field_value = getattr(
                record,
                field_name,
                None
            )

            if field_value is not None:

                log_record[
                    field_name
                ] = field_value

        if record.exc_info:

            log_record[
                "exception"
            ] = self.formatException(
                record.exc_info
            )

        return json.dumps(
            log_record,
            ensure_ascii=False
        )

Configure Structured Logging

import logging
import sys


def configure_logging():

    log_handler = logging.StreamHandler(
        sys.stdout
    )

    log_handler.setFormatter(
        JsonLogFormatter()
    )

    root_logger = logging.getLogger()

    root_logger.handlers.clear()

    root_logger.addHandler(
        log_handler
    )

    root_logger.setLevel(
        logging.INFO
    )


configure_logging()


logger = logging.getLogger(
    "customer-churn-api"
)

Log Model Startup

from pathlib import Path

import joblib


def load_model_artifact(
    model_path,
    model_name,
    model_version
):
    model_path = Path(
        model_path
    )

    try:
        model = joblib.load(
            model_path
        )

        logger.info(
            "Model loaded successfully.",
            extra={
                "event_name": "model_loaded",
                "service_name": (
                    "customer-churn-api"
                ),
                "model_name": model_name,
                "model_version": model_version,
                "status": "success"
            }
        )

        return model

    except Exception:

        logger.critical(
            "Model loading failed.",
            extra={
                "event_name": (
                    "model_loading_failed"
                ),
                "service_name": (
                    "customer-churn-api"
                ),
                "model_name": model_name,
                "model_version": model_version,
                "status": "failed"
            },
            exc_info=True
        )

        raise

Correlation Identifiers

Correlation identifiers connect telemetry produced by different components while processing one logical request.

Create Request Middleware

import time
import uuid

from fastapi import FastAPI
from fastapi import Request


app = FastAPI(
    title="Customer Churn API",
    version="2.1.0"
)


@app.middleware(
    "http"
)
async def request_observability(
    request: Request,
    call_next
):
    request_id = (
        request.headers.get(
            "X-Request-ID"
        )
        or str(
            uuid.uuid4()
        )
    )

    request.state.request_id = (
        request_id
    )

    started_at = (
        time.perf_counter()
    )

    try:
        response = await call_next(
            request
        )

        duration_ms = (
            time.perf_counter()
            -
            started_at
        ) * 1000

        logger.info(
            "Request completed.",
            extra={
                "event_name": (
                    "request_completed"
                ),
                "service_name": (
                    "customer-churn-api"
                ),
                "application_version": (
                    "2.1.0"
                ),
                "request_id": request_id,
                "duration_ms": round(
                    duration_ms,
                    3
                ),
                "status": str(
                    response.status_code
                )
            }
        )

        response.headers[
            "X-Request-ID"
        ] = request_id

        return response

    except Exception:

        duration_ms = (
            time.perf_counter()
            -
            started_at
        ) * 1000

        logger.error(
            "Request failed.",
            extra={
                "event_name": (
                    "request_failed"
                ),
                "service_name": (
                    "customer-churn-api"
                ),
                "application_version": (
                    "2.1.0"
                ),
                "request_id": request_id,
                "duration_ms": round(
                    duration_ms,
                    3
                ),
                "status": "failed"
            },
            exc_info=True
        )

        raise

What Must Not Be Logged?

Redact Sensitive Fields

SENSITIVE_KEYS = {
    "password",
    "access_token",
    "api_key",
    "authorization",
    "cookie",
    "secret",
    "private_key"
}


def redact_value(
    value
):
    if isinstance(
        value,
        dict
    ):
        return {
            key: (
                "[REDACTED]"
                if key.lower()
                in SENSITIVE_KEYS
                else redact_value(
                    nested_value
                )
            )
            for key, nested_value
            in value.items()
        }

    if isinstance(
        value,
        list
    ):
        return [
            redact_value(
                item
            )
            for item in value
        ]

    return value

Metrics

Metrics are numerical measurements aggregated over time. They are efficient for dashboards, thresholds, trend analysis, service-level objectives, and alerts.

Operational Metrics for ML APIs

ML-Specific Metrics

Instrument Metrics with Prometheus

from prometheus_client import Counter
from prometheus_client import Histogram


prediction_requests = Counter(
    "ml_prediction_requests_total",
    "Total prediction requests",
    [
        "model_name",
        "model_version",
        "status"
    ]
)


prediction_latency = Histogram(
    "ml_prediction_duration_seconds",
    "Prediction request duration",
    [
        "model_name",
        "model_version"
    ]
)


validation_failures = Counter(
    "ml_validation_failures_total",
    "Rejected prediction inputs",
    [
        "schema_version",
        "failure_type"
    ]
)

Record Prediction Metrics

import time


def run_observed_prediction(
    model,
    feature_frame,
    model_name,
    model_version
):
    started_at = (
        time.perf_counter()
    )

    try:
        probabilities = model.predict_proba(
            feature_frame
        )[:, 1]

        prediction_requests.labels(
            model_name=model_name,
            model_version=model_version,
            status="success"
        ).inc()

        return probabilities

    except Exception:

        prediction_requests.labels(
            model_name=model_name,
            model_version=model_version,
            status="failed"
        ).inc()

        raise

    finally:
        duration_seconds = (
            time.perf_counter()
            -
            started_at
        )

        prediction_latency.labels(
            model_name=model_name,
            model_version=model_version
        ).observe(
            duration_seconds
        )

Traces

A distributed trace represents the path of one request across services and operations. A trace consists of spans. Each span represents one timed operation such as authentication, feature retrieval, preprocessing, model inference, or persistence.

Configure OpenTelemetry Tracing

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


resource = Resource.create(
    {
        "service.name": (
            "customer-churn-api"
        ),
        "service.version": "2.1.0",
        "deployment.environment.name": (
            "development"
        )
    }
)


trace_provider = TracerProvider(
    resource=resource
)


trace_exporter = OTLPSpanExporter(
    endpoint=(
        "replace-with-approved-otlp-endpoint"
    ),
    insecure=False
)


trace_provider.add_span_processor(
    BatchSpanProcessor(
        trace_exporter
    )
)


trace.set_tracer_provider(
    trace_provider
)


tracer = trace.get_tracer(
    "customer-churn-api"
)

Instrument FastAPI

from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor


FastAPIInstrumentor.instrument_app(
    app
)

Create Custom ML Spans

from opentelemetry.trace import Status
from opentelemetry.trace import StatusCode


def traced_prediction(
    model,
    feature_frame,
    model_name,
    model_version,
    feature_service_version
):
    with tracer.start_as_current_span(
        "ml.predict"
    ) as prediction_span:

        prediction_span.set_attribute(
            "ml.model.name",
            model_name
        )

        prediction_span.set_attribute(
            "ml.model.version",
            model_version
        )

        prediction_span.set_attribute(
            "ml.feature_service.version",
            feature_service_version
        )

        prediction_span.set_attribute(
            "ml.batch_size",
            len(
                feature_frame
            )
        )

        try:
            probabilities = model.predict_proba(
                feature_frame
            )[:, 1]

            prediction_span.set_status(
                Status(
                    StatusCode.OK
                )
            )

            return probabilities

        except Exception as error:

            prediction_span.record_exception(
                error
            )

            prediction_span.set_status(
                Status(
                    StatusCode.ERROR,
                    str(
                        error
                    )
                )
            )

            raise

Trace Feature Retrieval

def retrieve_features_with_trace(
    feature_store,
    feature_service,
    entity_rows
):
    with tracer.start_as_current_span(
        "feature_store.get_online_features"
    ) as feature_span:

        feature_span.set_attribute(
            "feature_store.service",
            feature_service
        )

        feature_span.set_attribute(
            "feature_store.entity_count",
            len(
                entity_rows
            )
        )

        try:
            result = (
                feature_store.get_online_features(
                    features=feature_service,
                    entity_rows=entity_rows
                )
            )

            response = result.to_dict()

            feature_span.set_attribute(
                "feature_store.status",
                "success"
            )

            return response

        except Exception as error:

            feature_span.set_attribute(
                "feature_store.status",
                "failed"
            )

            feature_span.record_exception(
                error
            )

            raise

Correlate Logs with Traces

from opentelemetry import trace


def get_trace_context():

    current_span = (
        trace.get_current_span()
    )

    span_context = (
        current_span.get_span_context()
    )

    if not span_context.is_valid:

        return {
            "trace_id": None,
            "span_id": None
        }

    return {
        "trace_id": format(
            span_context.trace_id,
            "032x"
        ),
        "span_id": format(
            span_context.span_id,
            "016x"
        )
    }

trace_context = (
    get_trace_context()
)


logger.info(
    "Prediction completed.",
    extra={
        "event_name": (
            "prediction_completed"
        ),
        "service_name": (
            "customer-churn-api"
        ),
        "model_name": (
            "customer-churn-classifier"
        ),
        "model_version": "1.0.0",
        "trace_id": (
            trace_context[
                "trace_id"
            ]
        ),
        "span_id": (
            trace_context[
                "span_id"
            ]
        ),
        "status": "success"
    }
)

OpenTelemetry Collector

An OpenTelemetry Collector can receive telemetry, process it, and export it to one or more approved observability destinations.

receivers:
  otlp:
    protocols:
      grpc:
      http:

processors:
  batch:

exporters:
  otlp:
    endpoint: replace-with-approved-backend
    tls:
      insecure: false

service:
  pipelines:
    traces:
      receivers:
        - otlp
      processors:
        - batch
      exporters:
        - otlp

Telemetry Sampling

Sampling controls which trace or event records are retained. It can reduce telemetry volume and cost, but insufficient sampling may remove rare failures or important operational evidence.

Metric Cardinality

Metric cardinality is the number of distinct label combinations associated with a metric. Unbounded labels can make monitoring expensive and difficult to operate.

Observability Dashboard

Actionable Alerts

Reduce Alert Fatigue

Incident Investigation

Fallback and Rollback Observability

Test Logging Behavior

import json
import logging


def test_json_log_formatter():

    formatter = JsonLogFormatter()

    record = logging.LogRecord(
        name="test-logger",
        level=logging.INFO,
        pathname=__file__,
        lineno=1,
        msg="Prediction completed.",
        args=(),
        exc_info=None
    )

    record.event_name = (
        "prediction_completed"
    )

    record.model_version = "1.0.0"

    record.request_id = (
        "test-request-id"
    )

    serialized_log = formatter.format(
        record
    )

    parsed_log = json.loads(
        serialized_log
    )

    assert parsed_log[
        "level"
    ] == "INFO"

    assert parsed_log[
        "event_name"
    ] == "prediction_completed"

    assert parsed_log[
        "model_version"
    ] == "1.0.0"

    assert parsed_log[
        "request_id"
    ] == "test-request-id"

Test Redaction

def test_sensitive_values_are_redacted():

    payload = {
        "username": "approved-test-user",
        "password": "example-password",
        "nested": {
            "api_key": "example-key",
            "status": "active"
        }
    }

    redacted = redact_value(
        payload
    )

    assert redacted[
        "password"
    ] == "[REDACTED]"

    assert redacted[
        "nested"
    ][
        "api_key"
    ] == "[REDACTED]"

    assert redacted[
        "nested"
    ][
        "status"
    ] == "active"

python -m pytest -v

Telemetry Retention

Observability Cost Management

Responsible Model Observability

Common Observability Mistakes

Recommended Observability Workflow

Hands-On Practice

Knowledge Check

Observability Readiness Checklist

Lesson Summary