Building and Managing Data Pipeline SLAs: Defining, Measuring, and Enforcing Freshness and Latency Guarantees in Production

Introduction

It's 8:47 AM on a Tuesday and your VP of Sales is presenting the weekly revenue dashboard to the executive team. Halfway through the meeting, someone notices that the numbers haven't moved since Friday. Your Slack lights up. The pipeline ran — the orchestrator shows green — but the data is three days stale because an upstream API started throttling requests and your pipeline silently swallowed the partial load without complaint. No alert fired. No SLA was breached, technically, because you never formally defined one.

This is the gap that data pipeline SLAs are designed to close. Not the gap between "did the pipeline run?" and "did the pipeline fail?" — your orchestrator already handles that — but the gap between technical success and business value. A pipeline that runs on schedule but produces stale, incomplete, or late data has failed in the way that actually matters. SLAs give you the vocabulary, the measurements, and the enforcement mechanisms to catch that kind of failure before your VP's presentation, not during it.

By the end of this lesson, you'll be able to define precise SLA contracts for your data pipelines, instrument your pipelines to measure against those contracts in production, build alerting logic that catches real failures rather than just job crashes, and establish a review cadence that keeps SLAs relevant as your systems evolve.

What you'll learn:

• How to define freshness, latency, completeness, and availability SLAs as measurable contracts
• How to instrument pipeline code to capture the metrics you need for SLA evaluation
• How to build an SLA monitoring layer on top of your existing orchestration
• How to implement tiered alerting that escalates appropriately without creating alert fatigue
• How to conduct SLA reviews and iterate on thresholds based on production data

---

Prerequisites

You should already be comfortable with:

• Running data pipelines in a production orchestration environment (Airflow, Prefect, Dagster, or similar)
• Basic SQL for querying pipeline metadata and audit tables
• Python at an intermediate level — you'll read and write real code
• General familiarity with observability concepts like metrics, logs, and alerts

---

What an SLA Actually Is (and What It Isn't)

Before building anything, let's be precise about terminology, because the data industry has a habit of using "SLA," "SLO," and "SLI" interchangeably in ways that cause real operational confusion.

Borrowing from Google's Site Reliability Engineering framework:

• An SLI (Service Level Indicator) is a specific, measurable metric. "The percentage of pipeline runs where the target table was updated within 30 minutes of the scheduled trigger time" is an SLI.
• An SLO (Service Level Objective) is the target you set for that indicator. "95% of pipeline runs should meet the 30-minute latency threshold" is an SLO.
• An SLA (Service Level Agreement) is the formal contract, usually with business consequences when SLOs are missed. "If we miss the 95% target for two consecutive weeks, the data engineering team commits to a root cause analysis and architecture review."

In practice, most data teams conflate these, and that's fine as long as the person writing the pipeline and the person relying on its output share the same understanding. What matters is that your "SLA" answer four questions:

1. What is being measured? (freshness? latency? completeness? availability?)
2. What is the acceptable threshold?
3. Over what time window is it evaluated?
4. What happens when it's violated?

Without answers to all four, you don't have an SLA — you have a wish.

---

The Four Dimensions of Data Pipeline SLAs

Most pipelines need SLAs across four dimensions. Understanding each one helps you choose the right measurements.

Freshness

Freshness measures how recent the data in a target system is relative to some reference point, usually either wall clock time or the timestamp of the most recent source event.

A freshness SLA might read: "The orders table in the analytics warehouse should reflect events that occurred no more than 4 hours ago, measured at any point during business hours."

This is different from a latency SLA. Freshness is a property of the data at rest; latency is a property of the pipeline's processing speed. You can have fresh data delivered with high latency (if source data is slow to arrive) and you can have stale data despite low-latency pipelines (if the pipeline is failing silently).

Latency

Latency measures the time between when data enters the pipeline and when it becomes available to consumers downstream. It's typically measured as end-to-end pipeline duration, but you can also measure it per stage if you need to diagnose bottlenecks.

A latency SLA might read: "95% of CDC events captured from the production PostgreSQL database should be reflected in the data warehouse within 15 minutes of being committed to the source."

Completeness

Completeness measures whether all expected data arrived. This is subtle: a pipeline can run successfully, finish on time, and produce fresh-looking data while silently dropping 30% of records due to a schema mismatch or API pagination bug.

A completeness SLA might read: "The daily revenue reconciliation pipeline should produce a row count within ±2% of the prior 7-day average. Deviations beyond ±5% should trigger an immediate alert."

Availability

Availability measures what fraction of the time your pipeline produces usable output. This is most relevant for pipelines that feed real-time dashboards or operational systems.

An availability SLA might read: "The customer-facing analytics endpoint should return data no older than 1 hour for 99% of requests between 6 AM and 10 PM UTC."

Why all four matter: A common mistake is to monitor only latency (because it's easy to measure from job logs) while ignoring freshness and completeness. The highest-severity incidents at most data teams involve completeness failures — not job crashes.

---

Designing SLA Contracts: A Practical Framework

SLA design is a negotiation between what the business needs and what the system can reliably deliver. Here's how to structure that conversation.

Start with the Consumer, Not the Pipeline

The most common mistake in SLA design is starting with what the pipeline currently does and working backward to a threshold. That approach bakes in existing limitations as permanent constraints.

Instead, start by asking the data consumer: "What's the worst staleness you could tolerate before the data becomes misleading or harmful?" For a fraud detection model, the answer might be 5 minutes. For a quarterly board report, it might be 24 hours.

Once you have the consumer requirement, work backward through your pipeline architecture to identify whether you can actually meet it, and at what cost.

Define a Criticality Tier

Not every pipeline deserves the same SLA rigor. Establish tiers:

Tier	Description	Example	Typical Freshness SLA
P0	Business-critical, customer-facing	Payment processing feed	< 5 minutes
P1	Executive/operational reporting	Daily revenue dashboard	< 1 hour
P2	Internal analytics	Marketing attribution model	< 4 hours
P3	Data science feature engineering	Historical training datasets	< 24 hours

The tier determines how aggressively you monitor, how quickly you escalate, and how much engineering investment you make in reliability.

Capture SLA Definitions as Code

Don't leave SLA definitions in a Confluence page that goes stale. Encode them in a configuration file that your monitoring code actually reads.

# pipeline_slas.yaml
pipelines:
  orders_cdc_to_warehouse:
    tier: P1
    description: "CDC replication from orders PostgreSQL to BigQuery"
    owner: "data-engineering@company.com"
    escalation_channel: "#data-alerts-urgent"
    slas:
      freshness:
        max_lag_minutes: 60
        evaluation_window: "business_hours"  # 6am-10pm UTC
        measurement: "max(event_timestamp) in target vs wall_clock"
      latency:
        p95_minutes: 15
        p99_minutes: 30
        window_hours: 24
      completeness:
        row_count_tolerance_pct: 5.0
        comparison_window_days: 7

  daily_revenue_reconciliation:
    tier: P1
    description: "Nightly roll-up of revenue by product and region"
    owner: "data-engineering@company.com"
    escalation_channel: "#data-alerts-urgent"
    slas:
      freshness:
        max_lag_hours: 2
        evaluation_time: "08:00 UTC"
        measurement: "pipeline completion timestamp vs scheduled time"
      completeness:
        row_count_tolerance_pct: 2.0
        comparison_window_days: 14
        minimum_rows: 10000

This YAML becomes the ground truth. Your monitoring system reads it; your on-call runbook references it; your SLA review process updates it.

---

Instrumenting Your Pipelines for SLA Measurement

You can't measure what you don't capture. Most orchestrators give you job-level start/end times, but SLA measurement requires deeper telemetry.

Building a Pipeline Audit Table

The foundation of SLA monitoring is a persistent audit table that every pipeline writes to. Here's a schema that handles the four dimensions we defined:

CREATE TABLE pipeline_audit (
    run_id           VARCHAR(64)   NOT NULL,
    pipeline_name    VARCHAR(255)  NOT NULL,
    pipeline_version VARCHAR(32),
    triggered_at     TIMESTAMPTZ   NOT NULL,
    started_at       TIMESTAMPTZ,
    completed_at     TIMESTAMPTZ,
    status           VARCHAR(32)   NOT NULL, -- 'running', 'success', 'failed', 'partial'
    records_read     BIGINT,
    records_written  BIGINT,
    records_failed   BIGINT,
    max_source_ts    TIMESTAMPTZ,  -- freshest event timestamp in source batch
    min_source_ts    TIMESTAMPTZ,  -- oldest event timestamp in source batch
    target_table     VARCHAR(255),
    error_message    TEXT,
    metadata         JSONB,        -- pipeline-specific key/value pairs
    PRIMARY KEY (run_id)
);

CREATE INDEX idx_audit_pipeline_name ON pipeline_audit(pipeline_name, completed_at DESC);
CREATE INDEX idx_audit_status ON pipeline_audit(status, completed_at DESC);

Notice the status column includes 'partial' — this is critical. A pipeline that wrote 60% of expected records and exited cleanly is not a success. You need a status value that captures this condition.

A Python Audit Context Manager

Rather than manually writing audit records in every pipeline, wrap the logic in a context manager that handles both success and failure cases:

import uuid
import json
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Optional
import psycopg2


class PipelineAuditContext:
    """
    Context manager for capturing pipeline run metadata to the audit table.
    
    Usage:
        with PipelineAuditContext("orders_cdc_to_warehouse", conn) as audit:
            records = fetch_from_source()
            audit.set_records_read(len(records))
            
            written = load_to_warehouse(records)
            audit.set_records_written(written)
            audit.set_max_source_ts(records[-1]["event_timestamp"])
    """

    def __init__(
        self,
        pipeline_name: str,
        db_conn,
        pipeline_version: str = "1.0.0",
        target_table: Optional[str] = None,
        triggered_at: Optional[datetime] = None,
    ):
        self.pipeline_name = pipeline_name
        self.db_conn = db_conn
        self.pipeline_version = pipeline_version
        self.target_table = target_table
        self.run_id = str(uuid.uuid4())
        self.triggered_at = triggered_at or datetime.now(timezone.utc)
        self.started_at: Optional[datetime] = None
        self.completed_at: Optional[datetime] = None
        self.status = "running"
        self.records_read: Optional[int] = None
        self.records_written: Optional[int] = None
        self.records_failed: Optional[int] = None
        self.max_source_ts: Optional[datetime] = None
        self.min_source_ts: Optional[datetime] = None
        self.error_message: Optional[str] = None
        self.metadata: dict = {}

    def set_records_read(self, count: int):
        self.records_read = count

    def set_records_written(self, count: int):
        self.records_written = count

    def set_records_failed(self, count: int):
        self.records_failed = count

    def set_max_source_ts(self, ts: datetime):
        self.max_source_ts = ts

    def set_min_source_ts(self, ts: datetime):
        self.min_source_ts = ts

    def add_metadata(self, key: str, value):
        self.metadata[key] = value

    def _determine_final_status(self) -> str:
        """
        Determine whether a technically successful run actually delivered
        complete data. This is where silent failures get caught.
        """
        if self.status == "failed":
            return "failed"

        if self.records_read is not None and self.records_written is not None:
            if self.records_read > 0:
                write_ratio = self.records_written / self.records_read
                if write_ratio < 0.95:
                    # More than 5% of records were dropped
                    return "partial"

        if self.records_failed is not None and self.records_failed > 0:
            total = (self.records_written or 0) + self.records_failed
            if total > 0 and self.records_failed / total > 0.05:
                return "partial"

        return "success"

    def _write_audit_record(self):
        cursor = self.db_conn.cursor()
        cursor.execute(
            """
            INSERT INTO pipeline_audit (
                run_id, pipeline_name, pipeline_version, triggered_at,
                started_at, completed_at, status, records_read,
                records_written, records_failed, max_source_ts,
                min_source_ts, target_table, error_message, metadata
            ) VALUES (
                %(run_id)s, %(pipeline_name)s, %(pipeline_version)s,
                %(triggered_at)s, %(started_at)s, %(completed_at)s,
                %(status)s, %(records_read)s, %(records_written)s,
                %(records_failed)s, %(max_source_ts)s, %(min_source_ts)s,
                %(target_table)s, %(error_message)s, %(metadata)s
            )
            ON CONFLICT (run_id) DO UPDATE SET
                completed_at = EXCLUDED.completed_at,
                status = EXCLUDED.status,
                records_read = EXCLUDED.records_read,
                records_written = EXCLUDED.records_written,
                records_failed = EXCLUDED.records_failed,
                max_source_ts = EXCLUDED.max_source_ts,
                error_message = EXCLUDED.error_message,
                metadata = EXCLUDED.metadata
            """,
            {
                "run_id": self.run_id,
                "pipeline_name": self.pipeline_name,
                "pipeline_version": self.pipeline_version,
                "triggered_at": self.triggered_at,
                "started_at": self.started_at,
                "completed_at": self.completed_at,
                "status": self.status,
                "records_read": self.records_read,
                "records_written": self.records_written,
                "records_failed": self.records_failed,
                "max_source_ts": self.max_source_ts,
                "min_source_ts": self.min_source_ts,
                "target_table": self.target_table,
                "error_message": self.error_message,
                "metadata": json.dumps(self.metadata),
            },
        )
        self.db_conn.commit()
        cursor.close()

    def __enter__(self):
        self.started_at = datetime.now(timezone.utc)
        self.status = "running"
        self._write_audit_record()  # Write initial record so we can track stuck pipelines
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.completed_at = datetime.now(timezone.utc)

        if exc_type is not None:
            self.status = "failed"
            self.error_message = f"{exc_type.__name__}: {str(exc_val)}"
        else:
            self.status = self._determine_final_status()

        self._write_audit_record()
        return False  # Don't suppress exceptions

With this in place, every pipeline run — whether it succeeds, fails, or produces partial output — leaves a complete, queryable record.

---

Building the SLA Monitoring Layer

The audit table is your raw material. Now you need a monitoring process that reads it and evaluates it against your SLA definitions.

The SLA Evaluator

import yaml
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass
from typing import List, Optional
import psycopg2.extras


@dataclass
class SLAViolation:
    pipeline_name: str
    tier: str
    violation_type: str  # 'freshness', 'latency', 'completeness', 'availability'
    description: str
    severity: str        # 'warning', 'critical'
    measured_value: float
    threshold_value: float
    run_id: Optional[str] = None
    detected_at: datetime = None

    def __post_init__(self):
        if self.detected_at is None:
            self.detected_at = datetime.now(timezone.utc)


class SLAEvaluator:
    def __init__(self, sla_config_path: str, db_conn):
        with open(sla_config_path) as f:
            self.config = yaml.safe_load(f)
        self.db_conn = db_conn

    def evaluate_all(self) -> List[SLAViolation]:
        violations = []
        for pipeline_name, pipeline_config in self.config["pipelines"].items():
            violations.extend(self.evaluate_pipeline(pipeline_name, pipeline_config))
        return violations

    def evaluate_pipeline(
        self, pipeline_name: str, config: dict
    ) -> List[SLAViolation]:
        violations = []
        slas = config.get("slas", {})

        if "freshness" in slas:
            v = self._check_freshness(pipeline_name, config, slas["freshness"])
            if v:
                violations.append(v)

        if "latency" in slas:
            violations.extend(
                self._check_latency(pipeline_name, config, slas["latency"])
            )

        if "completeness" in slas:
            v = self._check_completeness(pipeline_name, config, slas["completeness"])
            if v:
                violations.append(v)

        return violations

    def _check_freshness(
        self, pipeline_name: str, config: dict, freshness_sla: dict
    ) -> Optional[SLAViolation]:
        """
        Check if the most recent successful run's max_source_ts is within
        the freshness threshold.
        """
        cursor = self.db_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
        cursor.execute(
            """
            SELECT run_id, completed_at, max_source_ts
            FROM pipeline_audit
            WHERE pipeline_name = %s
              AND status = 'success'
            ORDER BY completed_at DESC
            LIMIT 1
            """,
            (pipeline_name,),
        )
        row = cursor.fetchone()

        if row is None:
            return SLAViolation(
                pipeline_name=pipeline_name,
                tier=config["tier"],
                violation_type="freshness",
                description=f"No successful runs found for {pipeline_name}",
                severity="critical",
                measured_value=float("inf"),
                threshold_value=freshness_sla.get("max_lag_minutes", 
                                freshness_sla.get("max_lag_hours", 0) * 60),
            )

        now = datetime.now(timezone.utc)
        reference_ts = row["max_source_ts"] or row["completed_at"]
        lag_minutes = (now - reference_ts).total_seconds() / 60

        threshold_minutes = freshness_sla.get(
            "max_lag_minutes", freshness_sla.get("max_lag_hours", 1) * 60
        )

        if lag_minutes > threshold_minutes:
            severity = "critical" if lag_minutes > threshold_minutes * 2 else "warning"
            return SLAViolation(
                pipeline_name=pipeline_name,
                tier=config["tier"],
                violation_type="freshness",
                description=(
                    f"Data is {lag_minutes:.1f} minutes stale "
                    f"(threshold: {threshold_minutes} minutes). "
                    f"Last successful run: {row['completed_at'].isoformat()}"
                ),
                severity=severity,
                measured_value=lag_minutes,
                threshold_value=threshold_minutes,
                run_id=row["run_id"],
            )
        return None

    def _check_latency(
        self, pipeline_name: str, config: dict, latency_sla: dict
    ) -> List[SLAViolation]:
        """
        Calculate p95 and p99 pipeline durations over the last 24 hours
        and compare against thresholds.
        """
        cursor = self.db_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
        cursor.execute(
            """
            SELECT 
                EXTRACT(EPOCH FROM (completed_at - started_at)) / 60 AS duration_minutes,
                PERCENTILE_CONT(0.95) WITHIN GROUP (
                    ORDER BY EXTRACT(EPOCH FROM (completed_at - started_at))
                ) OVER () / 60 AS p95_minutes,
                PERCENTILE_CONT(0.99) WITHIN GROUP (
                    ORDER BY EXTRACT(EPOCH FROM (completed_at - started_at))
                ) OVER () / 60 AS p99_minutes,
                COUNT(*) OVER () AS run_count
            FROM pipeline_audit
            WHERE pipeline_name = %s
              AND status IN ('success', 'partial')
              AND completed_at > NOW() - INTERVAL '24 hours'
            ORDER BY completed_at DESC
            LIMIT 1
            """,
            (pipeline_name,),
        )
        row = cursor.fetchone()

        violations = []
        if row is None or row["run_count"] < 3:
            return violations  # Not enough data to evaluate

        if "p95_minutes" in latency_sla and row["p95_minutes"]:
            if row["p95_minutes"] > latency_sla["p95_minutes"]:
                violations.append(
                    SLAViolation(
                        pipeline_name=pipeline_name,
                        tier=config["tier"],
                        violation_type="latency",
                        description=(
                            f"p95 latency is {row['p95_minutes']:.1f} min "
                            f"(threshold: {latency_sla['p95_minutes']} min) "
                            f"over last 24 hours"
                        ),
                        severity="warning",
                        measured_value=row["p95_minutes"],
                        threshold_value=latency_sla["p95_minutes"],
                    )
                )

        return violations

    def _check_completeness(
        self, pipeline_name: str, config: dict, completeness_sla: dict
    ) -> Optional[SLAViolation]:
        """
        Compare the most recent run's record count against the rolling
        average of recent successful runs to detect unexpected drops.
        """
        cursor = self.db_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
        window_days = completeness_sla.get("comparison_window_days", 7)
        tolerance_pct = completeness_sla.get("row_count_tolerance_pct", 5.0)
        minimum_rows = completeness_sla.get("minimum_rows", 0)

        cursor.execute(
            """
            WITH recent_runs AS (
                SELECT records_written, completed_at,
                       AVG(records_written) OVER (
                           ORDER BY completed_at
                           ROWS BETWEEN %s PRECEDING AND 1 PRECEDING
                       ) AS rolling_avg
                FROM pipeline_audit
                WHERE pipeline_name = %s
                  AND status = 'success'
                  AND completed_at > NOW() - (%s || ' days')::INTERVAL
                ORDER BY completed_at DESC
            )
            SELECT records_written, rolling_avg
            FROM recent_runs
            LIMIT 1
            """,
            (window_days, pipeline_name, window_days),
        )
        row = cursor.fetchone()

        if row is None or row["rolling_avg"] is None:
            return None  # Not enough history

        current = row["records_written"] or 0
        avg = float(row["rolling_avg"])

        if current < minimum_rows:
            return SLAViolation(
                pipeline_name=pipeline_name,
                tier=config["tier"],
                violation_type="completeness",
                description=(
                    f"Latest run wrote {current:,} rows, below minimum of "
                    f"{minimum_rows:,}"
                ),
                severity="critical",
                measured_value=current,
                threshold_value=minimum_rows,
            )

        if avg > 0:
            deviation_pct = abs(current - avg) / avg * 100
            if deviation_pct > tolerance_pct:
                severity = "critical" if deviation_pct > tolerance_pct * 2 else "warning"
                return SLAViolation(
                    pipeline_name=pipeline_name,
                    tier=config["tier"],
                    violation_type="completeness",
                    description=(
                        f"Row count deviation: {deviation_pct:.1f}% from "
                        f"{window_days}-day average of {avg:,.0f} rows "
                        f"(current: {current:,}, tolerance: ±{tolerance_pct}%)"
                    ),
                    severity=severity,
                    measured_value=deviation_pct,
                    threshold_value=tolerance_pct,
                )

        return None

Wiring the Evaluator to Alerting

The evaluator produces SLAViolation objects. Now you need to route them to the right people through the right channels:

import requests
from typing import List


class SLAAlertRouter:
    def __init__(self, slack_webhook_url: str, pagerduty_routing_key: str = None):
        self.slack_webhook_url = slack_webhook_url
        self.pagerduty_routing_key = pagerduty_routing_key

    def route(self, violations: List[SLAViolation], sla_config: dict):
        for violation in violations:
            pipeline_config = sla_config["pipelines"].get(violation.pipeline_name, {})
            channel = pipeline_config.get("escalation_channel", "#data-alerts")

            # Always send to Slack
            self._send_slack_alert(violation, channel)

            # PagerDuty for critical P0/P1 violations
            if (
                violation.severity == "critical"
                and violation.tier in ("P0", "P1")
                and self.pagerduty_routing_key
            ):
                self._send_pagerduty_alert(violation)

    def _send_slack_alert(self, violation: SLAViolation, channel: str):
        emoji = "🔴" if violation.severity == "critical" else "🟡"
        color = "#FF0000" if violation.severity == "critical" else "#FFA500"

        payload = {
            "channel": channel,
            "attachments": [
                {
                    "color": color,
                    "title": f"{emoji} SLA Violation: {violation.pipeline_name}",
                    "fields": [
                        {"title": "Type", "value": violation.violation_type, "short": True},
                        {"title": "Tier", "value": violation.tier, "short": True},
                        {"title": "Severity", "value": violation.severity.upper(), "short": True},
                        {
                            "title": "Measured",
                            "value": f"{violation.measured_value:.2f}",
                            "short": True,
                        },
                        {
                            "title": "Threshold",
                            "value": f"{violation.threshold_value:.2f}",
                            "short": True,
                        },
                        {"title": "Details", "value": violation.description, "short": False},
                    ],
                    "footer": f"Detected at {violation.detected_at.strftime('%Y-%m-%d %H:%M UTC')}",
                }
            ],
        }
        requests.post(self.slack_webhook_url, json=payload, timeout=5)

    def _send_pagerduty_alert(self, violation: SLAViolation):
        payload = {
            "routing_key": self.pagerduty_routing_key,
            "event_action": "trigger",
            "dedup_key": f"{violation.pipeline_name}_{violation.violation_type}",
            "payload": {
                "summary": f"SLA Violation [{violation.tier}]: {violation.pipeline_name} - {violation.violation_type}",
                "severity": "critical",
                "source": "data-pipeline-sla-monitor",
                "custom_details": {
                    "pipeline": violation.pipeline_name,
                    "violation_type": violation.violation_type,
                    "description": violation.description,
                    "measured_value": violation.measured_value,
                    "threshold": violation.threshold_value,
                },
            },
        }
        requests.post(
            "https://events.pagerduty.com/v2/enqueue",
            json=payload,
            timeout=5,
        )

---

Tracking SLA Compliance Over Time

Individual violation alerts are tactical. You also need strategic visibility: are you getting better or worse at meeting your SLAs?

A Compliance Reporting Query

Store SLA violations in their own table (not just alert them), and run weekly compliance reports:

-- SLA compliance summary for the last 30 days
WITH pipeline_runs AS (
    SELECT
        pipeline_name,
        DATE_TRUNC('day', completed_at) AS run_date,
        COUNT(*) AS total_runs,
        SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS successful_runs,
        SUM(CASE WHEN status = 'partial' THEN 1 ELSE 0 END) AS partial_runs,
        SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed_runs,
        AVG(
            EXTRACT(EPOCH FROM (completed_at - started_at)) / 60
        ) AS avg_latency_minutes,
        PERCENTILE_CONT(0.95) WITHIN GROUP (
            ORDER BY EXTRACT(EPOCH FROM (completed_at - started_at)) / 60
        ) AS p95_latency_minutes
    FROM pipeline_audit
    WHERE completed_at > NOW() - INTERVAL '30 days'
      AND completed_at IS NOT NULL
    GROUP BY 1, 2
),
sla_violations AS (
    SELECT
        pipeline_name,
        DATE_TRUNC('day', detected_at) AS violation_date,
        COUNT(*) AS violations,
        SUM(CASE WHEN severity = 'critical' THEN 1 ELSE 0 END) AS critical_violations
    FROM sla_violation_log
    WHERE detected_at > NOW() - INTERVAL '30 days'
    GROUP BY 1, 2
)
SELECT
    r.pipeline_name,
    SUM(r.total_runs) AS total_runs_30d,
    SUM(r.successful_runs)::FLOAT / NULLIF(SUM(r.total_runs), 0) * 100 AS success_rate_pct,
    AVG(r.p95_latency_minutes) AS avg_p95_latency_minutes,
    COALESCE(SUM(v.violations), 0) AS total_sla_violations,
    COALESCE(SUM(v.critical_violations), 0) AS critical_violations,
    (30 - COUNT(DISTINCT v.violation_date))::FLOAT / 30 * 100 AS clean_day_pct
FROM pipeline_runs r
LEFT JOIN sla_violations v ON r.pipeline_name = v.pipeline_name
    AND r.run_date = v.violation_date
GROUP BY 1
ORDER BY critical_violations DESC, success_rate_pct ASC;

Run this query weekly and put the results in front of your team. Pipelines with deteriorating compliance become candidates for the next architecture sprint; pipelines with consistently clean records can have their SLA thresholds tightened.

---

Hands-On Exercise

You're a data engineer at a B2B SaaS company. The customer_usage_events pipeline runs every 30 minutes, reads from a Kafka consumer, and writes to a Snowflake table that feeds a real-time customer health dashboard. The sales team uses this dashboard to identify customers at risk of churning.

Your task:

1. Define an appropriate SLA for this pipeline across all four dimensions. Use the YAML schema from earlier. Assume business hours are 7 AM–8 PM in the customer's local timezone (you can simplify to UTC-8 to UTC+5 range, so effectively 24/7 monitoring is needed).

2. Add the PipelineAuditContext to a hypothetical pipeline function (you can write the skeleton with placeholder comments for the actual processing logic):

def run_customer_usage_pipeline(kafka_consumer, snowflake_conn, audit_db_conn):
    """
    Consume customer usage events from Kafka, transform them,
    and write to Snowflake customer_usage_events table.
    """
    # Your implementation here — use the PipelineAuditContext
    pass

3. Write the freshness check specifically for this pipeline. The freshness SLA should measure the lag between the Kafka message timestamp (stored in max_source_ts) and wall clock time, not the pipeline completion time. Explain in a comment why this distinction matters for customer health scoring.

4. Define what a "partial" run means for this specific pipeline — what percentage of record loss would you tolerate before escalating from warning to critical, and why?

Stretch goal: Extend the SLAEvaluator to detect a "stuck pipeline" scenario — where the pipeline appears to be running (status = 'running' in the audit table) but has been running for longer than the p99 latency threshold. This is distinct from a failed pipeline and requires a different response.

---

Common Mistakes & Troubleshooting

Measuring pipeline completion time instead of data event time

This is the single most common SLA instrumentation mistake. If your freshness SLA says "data should be no more than 1 hour old" but you measure pipeline completion time, you'll miss scenarios where the source system itself is delayed. A pipeline that runs perfectly on schedule but receives data that's already 50 minutes old will pass your freshness check while your consumers receive 110-minute-old data.

Fix: Always capture max_source_ts — the highest event timestamp in the batch — and use that as your reference point for freshness calculation.

SLA thresholds that were never calibrated against production data

Setting a p95 latency threshold of 5 minutes because "that sounds fast" when your actual p95 is 12 minutes means you'll be in perpetual violation from day one. Your team starts ignoring alerts, the alerting system loses credibility, and you're back to the Tuesday morning dashboard incident from the introduction.

Fix: Before setting thresholds, run your evaluator in "observation mode" for two weeks — capture what your pipelines actually do, then set thresholds at the 90th percentile of observed performance. Tighten them quarterly.

Not accounting for expected row count seasonality

A completeness check that compares Monday's row count against a 7-day rolling average will always fire on Mondays, because weekday volumes are typically higher than weekend volumes. The check becomes noise.

Fix: Compare against the same day of week. In your rolling average query, filter WHERE EXTRACT(DOW FROM completed_at) = EXTRACT(DOW FROM NOW()).

Alerting on every individual violation instead of sustained violations

If your pipeline runs every 5 minutes and your SLA monitor runs every 5 minutes, a 20-minute degradation period will fire four consecutive alerts. Your on-call engineer acknowledges the first, then gets paged three more times while actively investigating.

Fix: Implement deduplication using the dedup_key field in PagerDuty (shown above), or add a "first violation at" timestamp to your violation log and suppress re-alerts until the violation is resolved or a cooldown window expires.

Forgetting to monitor the SLA monitor itself

Your SLA monitoring job is a pipeline. It can fail. If it fails silently, you go dark — no violations fire even when pipelines are broken.

Fix: Implement a dead man's switch: your SLA monitor should write a heartbeat record every time it runs successfully. A separate, simple alerting rule fires if no heartbeat arrives within 2x the monitoring interval.

---

Summary & Next Steps

Building effective pipeline SLAs is a discipline that sits at the intersection of engineering and business communication. The technical work — audit tables, context managers, evaluators, alert routers — is genuinely straightforward once you commit to the framework. The harder work is the ongoing negotiation between what your consumers need and what your systems can reliably provide.

Here's what you built in this lesson:

• A definitional framework that distinguishes freshness, latency, completeness, and availability, and ties each to business impact
• A YAML-based SLA contract format that lives in version control alongside your pipeline code
• An audit infrastructure that captures the telemetry you need for all four SLA dimensions, including the critical distinction between success and partial run states
• An SLA evaluator that reads your contracts and produces structured violation objects
• A tiered alert router that escalates appropriately based on severity and tier without creating noise

Where to go from here:

The logical next step is integrating this system with your data catalog or observability platform. Tools like Monte Carlo, Soda Core, and Great Expectations offer complementary approaches — particularly for data quality checks that go deeper than row counts. Soda Core, notably, has an open-source SLA/freshness checking framework that's worth evaluating against the custom approach built here.

You should also explore how SLA metadata can feed into your data catalog as machine-readable lineage. When a consumer sees a table in your catalog, they should be able to click through to its SLA definition, its current compliance status, and its recent violation history. That kind of transparency is what turns a data engineering team from a black box into a trustworthy infrastructure provider.

Finally, revisit your SLAs quarterly. Business priorities shift, data volumes grow, and pipeline architectures evolve. An SLA that was perfectly calibrated six months ago may be either too loose (masking real degradation) or too tight (generating chronic noise) today. Treat your SLA thresholds the way a good engineer treats any configuration: version-controlled, reviewed, and deliberately updated based on evidence.