Automating Data Freshness SLAs: Defining, Measuring, and Alerting on Staleness Across Your Entire Modern Data Stack

Introduction

It's 9:15 AM on a Monday. Your head of sales is presenting pipeline metrics to the board, and the numbers look wrong — off by about 40%. You dig in and discover that the Salesforce sync broke on Friday afternoon, the dbt models that depend on it ran successfully (against stale data), and nobody got paged because technically nothing failed. The data was just... old. Three days old, to be precise. And your observability tooling had no idea, because it only watches for pipeline errors, not data age.

This is the data freshness problem, and it's quietly responsible for a large share of the trust issues between data teams and their stakeholders. Pipelines succeed. Models compile. Dashboards load. And yet the numbers are wrong, because the underlying data stopped being refreshed at some point and nobody noticed. The solution isn't more manual checking — it's a systematic approach to defining what "fresh enough" means for each asset in your stack, measuring actual freshness continuously, and routing alerts to the right people before stakeholders discover problems themselves.

By the end of this lesson, you'll have a production-ready system for managing data freshness SLAs across your entire data stack — from ingestion through transformation to serving. We'll go from first principles (what does freshness actually mean?) all the way to working code for multi-layer monitoring, PagerDuty integration, and freshness dashboards your stakeholders can see.

What you'll learn:

• How to formally define freshness SLAs that account for the different freshness requirements of different data consumers and pipeline architectures
• How to instrument freshness checks at every layer of the modern data stack — ingestion (Fivetran/Airbyte), transformation (dbt), and serving (Snowflake/BigQuery/Databricks)
• How to build a centralized freshness metadata table that aggregates staleness signals across all layers
• How to write alerting logic that distinguishes between expected latency, SLA warnings, and hard breaches — and routes them appropriately
• How to expose freshness status to stakeholders through a self-service dashboard so they can check data age without pinging you

---

Prerequisites

You should be comfortable with:

• SQL at the intermediate level (window functions, CTEs)
• dbt — you've run models and written tests before
• At least one cloud data warehouse (Snowflake, BigQuery, or Databricks)
• Basic Python for scripting
• A conceptual understanding of ELT pipelines (Fivetran, Airbyte, or similar)

You don't need to have built a monitoring system before, but this lesson assumes you're working with a real data stack, not a sandbox.

---

What Freshness Actually Means (And Why It's Harder Than It Looks)

Before writing a single line of alerting code, you need to get precise about what you're measuring. "Freshness" sounds intuitive, but in a layered data stack it fragments into several distinct concepts that require different measurement strategies.

Event time vs. processing time vs. load time are the three timestamps that matter. Event time is when something happened in the source system — when the order was placed, when the user clicked, when the sensor fired. Processing time is when your pipeline picked it up and transformed it. Load time is when it landed in the warehouse. The gap between event time and load time is your end-to-end latency. The gap between load time and now is your data age. These are not the same thing, and confusing them leads to monitoring that misses the real problem.

Consider a Salesforce opportunity table. Fivetran syncs it every 15 minutes and finishes reliably. Load time is recent. But if a rep updates an opportunity at 4:30 PM on a Friday and your dbt models run at 2 AM, the processed data reflects a state that's already 9.5 hours behind real life — before you even add the weekend. Measuring "when did Fivetran last run?" tells you almost nothing meaningful about whether your sales pipeline report is trustworthy.

The practical framework that works in production has three components:

1. Freshness SLA: the maximum acceptable age of data for a specific use case, measured from event time to consumption time. Defined per-asset and per-consumer.
2. Freshness measurement: the actual, continuously-updated age of each asset, computed from metadata or watermark columns.
3. Freshness breach alerting: a tiered notification system that warns before an SLA is violated and escalates when it is.

Tip: Different consumers of the same table often have different SLA requirements. Finance closing reports can tolerate T+1 data. The real-time ops dashboard cannot. Your SLA system needs to account for this — a single freshness threshold per table will either be too strict (causing alert fatigue) or too lax (missing real problems).

---

Step 1: Defining Your SLA Contract

The worst freshness SLAs are the implicit ones — "it should update pretty often." Your first job is to make the contract explicit and machine-readable.

Create a configuration file (YAML works well for this) that lives in your dbt project or a dedicated data-ops repo. Each entry captures who owns the data, who depends on it, and what the SLA actually is.

# freshness_slas.yml
slas:
  - asset_id: salesforce.opportunity
    layer: ingestion
    connector: fivetran
    schema: raw_salesforce
    table: opportunity
    watermark_column: _fivetran_synced
    event_time_column: last_modified_date
    sla_warn_minutes: 30
    sla_breach_minutes: 60
    consumer_slas:
      - consumer: sales_pipeline_dashboard
        sla_breach_minutes: 120
        contact: "#sales-data-alerts"
      - consumer: finance_close_report
        sla_breach_minutes: 1440  # 24 hours is fine for this consumer
        contact: "#finance-data"
    owner: data-engineering
    escalation_contact: "oncall-data@company.com"
    notes: "Critical for Monday board review. Escalate immediately on weekend breaches."

  - asset_id: dbt.fct_orders
    layer: transformation
    schema: analytics
    table: fct_orders
    watermark_column: dbt_updated_at
    event_time_column: order_created_at
    sla_warn_minutes: 90
    sla_breach_minutes: 180
    upstream_dependencies:
      - salesforce.opportunity
      - stripe.charge
    owner: analytics-engineering
    escalation_contact: "analytics-lead@company.com"

  - asset_id: dbt.fct_daily_revenue
    layer: transformation
    schema: analytics
    table: fct_daily_revenue
    watermark_column: dbt_updated_at
    event_time_column: revenue_date
    sla_warn_minutes: 360   # 6 hours
    sla_breach_minutes: 480 # 8 hours — must be ready before market open
    owner: analytics-engineering
    escalation_contact: "finance-data@company.com"

This YAML becomes the single source of truth for your freshness program. Notice a few intentional design choices: we track both watermark_column (load time) and event_time_column (event time) because we want to measure both. We have separate warn and breach thresholds so alerts are tiered. And we have consumer-level overrides so the same table can have different SLAs for different use cases.

Warning: Don't define SLAs based on what your pipelines currently achieve. Define them based on what your business actually needs, then work backwards to determine whether your current pipeline architecture can meet them. You may discover gaps immediately — that's useful information.

---

Step 2: Instrumenting Freshness at the Ingestion Layer

Your ingestion tools (Fivetran, Airbyte, Stitch) have their own metadata tables that you can query directly to measure connector freshness.

Fivetran Metadata

Fivetran populates a fivetran_log schema in your warehouse. The most useful table for freshness monitoring is connector combined with sync_log. Here's a query that computes current freshness for all active connectors:

-- Fivetran connector freshness
-- Run this in Snowflake/BigQuery against your fivetran_log schema
WITH latest_syncs AS (
    SELECT
        connector_id,
        connector_name,
        MAX(sync_start) AS last_sync_start,
        MAX(CASE WHEN status = 'SUCCESSFUL' THEN sync_start END) AS last_successful_sync,
        MAX(CASE WHEN status = 'SUCCESSFUL' THEN sync_end END) AS last_successful_sync_end
    FROM fivetran_log.sync_log
    WHERE sync_start >= DATEADD('day', -7, CURRENT_TIMESTAMP())
    GROUP BY connector_id, connector_name
),
freshness AS (
    SELECT
        connector_id,
        connector_name,
        last_successful_sync,
        last_successful_sync_end,
        DATEDIFF('minute', last_successful_sync_end, CURRENT_TIMESTAMP()) AS minutes_since_last_success,
        CASE
            WHEN DATEDIFF('minute', last_successful_sync_end, CURRENT_TIMESTAMP()) > 120 THEN 'BREACH'
            WHEN DATEDIFF('minute', last_successful_sync_end, CURRENT_TIMESTAMP()) > 60  THEN 'WARN'
            ELSE 'OK'
        END AS freshness_status
    FROM latest_syncs
)
SELECT * FROM freshness
ORDER BY minutes_since_last_success DESC;

This gives you a real-time freshness reading at the connector level. But connector freshness is only half the picture — you also want to measure freshness at the table level using the actual watermark columns.

Table-Level Watermark Queries

For tables with a reliable watermark column like _fivetran_synced, updated_at, or _airbyte_emitted_at, you can query the table directly:

-- Generic watermark freshness check
-- Parameterize this with your SLA config values
SELECT
    'raw_salesforce.opportunity' AS asset_id,
    MAX(_fivetran_synced) AS last_watermark,
    MAX(last_modified_date) AS latest_event_time,
    DATEDIFF('minute', MAX(_fivetran_synced), CURRENT_TIMESTAMP()) AS load_age_minutes,
    DATEDIFF('minute', MAX(last_modified_date), CURRENT_TIMESTAMP()) AS event_age_minutes,
    COUNT(*) AS total_rows,
    COUNTIF(_fivetran_synced >= DATEADD('hour', -1, CURRENT_TIMESTAMP())) AS rows_loaded_last_hour
FROM raw_salesforce.opportunity;

The rows_loaded_last_hour metric is valuable because it catches a subtle failure mode: the watermark can look fresh if even a single row was loaded recently, but if you're expecting thousands of rows per hour and only got one, something is wrong. Volume anomalies and freshness are related problems.

Centralizing Ingestion Freshness with Python

Rather than running these queries manually, you want a scheduled job that writes freshness readings into a central metadata table. Here's a Python script that reads your YAML config, queries each asset, and writes results to a data_observability.freshness_readings table:

import yaml
import snowflake.connector
from datetime import datetime, timezone
import json

def load_sla_config(config_path: str) -> dict:
    with open(config_path, 'r') as f:
        return yaml.safe_load(f)

def measure_table_freshness(conn, sla: dict) -> dict:
    """
    Query a single table's watermark and compute freshness metrics.
    Returns a dict ready for insertion into the freshness_readings table.
    """
    schema = sla['schema']
    table = sla['table']
    watermark_col = sla['watermark_column']
    event_time_col = sla.get('event_time_column')

    event_time_select = f"MAX({event_time_col}) AS latest_event_time," if event_time_col else "NULL AS latest_event_time,"

    query = f"""
        SELECT
            MAX({watermark_col}) AS last_watermark,
            {event_time_select}
            COUNT(*) AS total_rows,
            COUNT(CASE WHEN {watermark_col} >= DATEADD('hour', -1, CURRENT_TIMESTAMP()) THEN 1 END) AS rows_last_hour,
            CURRENT_TIMESTAMP() AS measured_at
        FROM {schema}.{table}
    """

    cursor = conn.cursor()
    cursor.execute(query)
    row = cursor.fetchone()

    last_watermark, latest_event_time, total_rows, rows_last_hour, measured_at = row

    load_age_minutes = None
    event_age_minutes = None

    if last_watermark:
        delta = measured_at - last_watermark.replace(tzinfo=timezone.utc) if last_watermark.tzinfo is None else measured_at - last_watermark
        load_age_minutes = delta.total_seconds() / 60

    if latest_event_time:
        delta = measured_at - latest_event_time.replace(tzinfo=timezone.utc) if latest_event_time.tzinfo is None else measured_at - latest_event_time
        event_age_minutes = delta.total_seconds() / 60

    breach_minutes = sla['sla_breach_minutes']
    warn_minutes = sla['sla_warn_minutes']

    age_for_status = load_age_minutes or 99999
    if age_for_status >= breach_minutes:
        status = 'BREACH'
    elif age_for_status >= warn_minutes:
        status = 'WARN'
    else:
        status = 'OK'

    return {
        'asset_id': sla['asset_id'],
        'layer': sla['layer'],
        'schema_name': schema,
        'table_name': table,
        'last_watermark': last_watermark,
        'latest_event_time': latest_event_time,
        'load_age_minutes': round(load_age_minutes, 2) if load_age_minutes else None,
        'event_age_minutes': round(event_age_minutes, 2) if event_age_minutes else None,
        'rows_last_hour': rows_last_hour,
        'sla_warn_minutes': warn_minutes,
        'sla_breach_minutes': breach_minutes,
        'freshness_status': status,
        'measured_at': measured_at,
        'owner': sla['owner'],
        'consumer_slas': json.dumps(sla.get('consumer_slas', []))
    }

def write_freshness_reading(conn, reading: dict):
    cursor = conn.cursor()
    cursor.execute("""
        INSERT INTO data_observability.freshness_readings (
            asset_id, layer, schema_name, table_name,
            last_watermark, latest_event_time,
            load_age_minutes, event_age_minutes,
            rows_last_hour, sla_warn_minutes, sla_breach_minutes,
            freshness_status, measured_at, owner, consumer_slas
        ) VALUES (
            %(asset_id)s, %(layer)s, %(schema_name)s, %(table_name)s,
            %(last_watermark)s, %(latest_event_time)s,
            %(load_age_minutes)s, %(event_age_minutes)s,
            %(rows_last_hour)s, %(sla_warn_minutes)s, %(sla_breach_minutes)s,
            %(freshness_status)s, %(measured_at)s, %(owner)s, %(consumer_slas)s
        )
    """, reading)
    conn.commit()

def run_freshness_checks(config_path: str, conn):
    config = load_sla_config(config_path)
    results = []

    for sla in config['slas']:
        try:
            reading = measure_table_freshness(conn, sla)
            write_freshness_reading(conn, reading)
            results.append(reading)
            print(f"[{reading['freshness_status']}] {reading['asset_id']} — {reading['load_age_minutes']:.1f} min old")
        except Exception as e:
            print(f"ERROR measuring {sla['asset_id']}: {e}")

    return results

Schedule this script to run every 5–15 minutes via Airflow, Prefect, or even a simple cron job on a cloud VM. The freshness_readings table becomes your observability backbone — everything downstream (alerting, dashboards) reads from it.

---

Step 3: Instrumenting Freshness in dbt

dbt has built-in freshness testing via source freshness, but it has important limitations: it only runs when you explicitly trigger it, it only covers sources (not models), and it doesn't write results to a queryable table by default. You need to extend it.

Using dbt's Built-In Source Freshness

In your sources.yml, define freshness expectations:

# models/staging/sources.yml
version: 2

sources:
  - name: raw_salesforce
    database: your_db
    schema: raw_salesforce
    freshness:
      warn_after: {count: 30, period: minute}
      error_after: {count: 60, period: minute}
    loaded_at_field: _fivetran_synced
    tables:
      - name: opportunity
        freshness:
          warn_after: {count: 20, period: minute}
          error_after: {count: 45, period: minute}
      - name: account
        freshness:
          warn_after: {count: 30, period: minute}
          error_after: {count: 60, period: minute}

  - name: raw_stripe
    database: your_db
    schema: raw_stripe
    freshness:
      warn_after: {count: 15, period: minute}
      error_after: {count: 30, period: minute}
    loaded_at_field: _airbyte_emitted_at
    tables:
      - name: charge
      - name: customer

Run dbt source freshness and dbt will write results to target/sources.json. You can capture this in CI/CD and ship it to your observability table.

Adding `dbt_updated_at` to Every Model

The most durable pattern for model-level freshness is to add a dbt_updated_at column to every model:

-- models/marts/fct_orders.sql
{{
  config(
    materialized='table',
    tags=['daily', 'finance']
  )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_stripe__charges') }}
),

order_items AS (
    SELECT * FROM {{ ref('stg_postgres__order_items') }}
),

joined AS (
    SELECT
        o.charge_id,
        o.customer_id,
        o.amount_usd,
        o.status,
        o.created_at AS order_created_at,
        oi.product_id,
        oi.quantity,
        oi.unit_price_usd,
        -- Every mart model gets this column
        CURRENT_TIMESTAMP() AS dbt_updated_at,
        '{{ invocation_id }}' AS dbt_invocation_id
    FROM orders o
    JOIN order_items oi ON o.charge_id = oi.charge_id
)

SELECT * FROM joined

Now you can query MAX(dbt_updated_at) on any model to know exactly when it was last successfully rebuilt. The dbt_invocation_id is a bonus — it lets you correlate model runs back to specific dbt invocations for debugging.

The dbt Meta Block for SLA Annotation

Use dbt's meta config to embed SLA information directly in the model definition. This keeps SLA context co-located with the model itself:

# models/marts/schema.yml
version: 2

models:
  - name: fct_orders
    description: "One row per order. Source of truth for revenue reporting."
    meta:
      sla_warn_minutes: 90
      sla_breach_minutes: 180
      owner: analytics-engineering
      consumers:
        - name: revenue_dashboard
          sla_breach_minutes: 180
        - name: finance_close
          sla_breach_minutes: 1440
    columns:
      - name: charge_id
        description: "Stripe charge ID. Primary key."
        tests:
          - unique
          - not_null
      - name: dbt_updated_at
        description: "Timestamp when this model was last rebuilt."

You can extract these meta fields via the dbt artifacts API or by querying your warehouse's information schema — more on that in the dashboard section.

---

Step 4: Building the Centralized Freshness Table

Now that you're collecting readings from multiple layers, you need a unified view. Create the target table first:

-- Run once to create the observability schema and table
CREATE SCHEMA IF NOT EXISTS data_observability;

CREATE TABLE IF NOT EXISTS data_observability.freshness_readings (
    reading_id        VARCHAR DEFAULT UUID_STRING(),
    asset_id          VARCHAR NOT NULL,
    layer             VARCHAR NOT NULL,  -- 'ingestion', 'transformation', 'serving'
    schema_name       VARCHAR,
    table_name        VARCHAR,
    last_watermark    TIMESTAMP_TZ,
    latest_event_time TIMESTAMP_TZ,
    load_age_minutes  FLOAT,
    event_age_minutes FLOAT,
    rows_last_hour    INTEGER,
    sla_warn_minutes  INTEGER,
    sla_breach_minutes INTEGER,
    freshness_status  VARCHAR,  -- 'OK', 'WARN', 'BREACH'
    measured_at       TIMESTAMP_TZ DEFAULT CURRENT_TIMESTAMP(),
    owner             VARCHAR,
    consumer_slas     VARIANT  -- JSON blob for per-consumer SLA details
);

Then build a view that shows the current (most recent) freshness reading for each asset:

-- data_observability.v_freshness_current
CREATE OR REPLACE VIEW data_observability.v_freshness_current AS
WITH latest_readings AS (
    SELECT
        *,
        ROW_NUMBER() OVER (PARTITION BY asset_id ORDER BY measured_at DESC) AS rn
    FROM data_observability.freshness_readings
    WHERE measured_at >= DATEADD('hour', -6, CURRENT_TIMESTAMP())
)
SELECT
    asset_id,
    layer,
    schema_name,
    table_name,
    last_watermark,
    latest_event_time,
    load_age_minutes,
    event_age_minutes,
    rows_last_hour,
    sla_warn_minutes,
    sla_breach_minutes,
    freshness_status,
    measured_at,
    owner,
    consumer_slas,
    -- Helpful derived columns
    CASE
        WHEN freshness_status = 'BREACH' THEN load_age_minutes - sla_breach_minutes
        ELSE 0
    END AS minutes_past_breach,
    ROUND(load_age_minutes / sla_breach_minutes * 100, 1) AS pct_of_sla_used
FROM latest_readings
WHERE rn = 1;

Tip: The pct_of_sla_used column is surprisingly useful for dashboards. Seeing "72% of SLA used" for a table that's been idle for two hours is much more actionable than just seeing "OK" status.

---

Step 5: Alerting Logic That Doesn't Cry Wolf

Alert fatigue is real, and badly designed freshness alerting is a major cause of it. Here's the alerting architecture that works in production:

Three alert tiers:

1. WARN: SLA warn threshold crossed. Creates a ticket or Slack message. No page.
2. BREACH: SLA breach threshold crossed. Pages the on-call engineer. Creates an incident.
3. ESCALATION: Breach has persisted beyond 2x the breach threshold. Escalates to the team lead and notifies data consumers.

Alert suppression rules prevent a single sustained outage from generating hundreds of pages:

• Only fire a new alert if the asset wasn't already in a breached state during the previous measurement window.
• For recurring pipelines (like daily dbt jobs), suppress WARN alerts during the expected run window (e.g., suppress between 1 AM and 3 AM for a pipeline that runs at 2 AM).

Here's the alerting function that implements this:

import requests
import json
from datetime import datetime, timezone, timedelta
from typing import Optional

PAGERDUTY_ROUTING_KEY = "your_pagerduty_routing_key"
SLACK_WEBHOOK_URL = "your_slack_webhook_url"

def get_previous_status(conn, asset_id: str, lookback_minutes: int = 20) -> Optional[str]:
    """
    Returns the freshness status from the previous measurement window.
    Used to detect state *changes* rather than re-alerting on sustained issues.
    """
    cursor = conn.cursor()
    cursor.execute("""
        SELECT freshness_status
        FROM data_observability.freshness_readings
        WHERE asset_id = %s
          AND measured_at < DATEADD('minute', -%s, CURRENT_TIMESTAMP())
          AND measured_at >= DATEADD('minute', -%s, CURRENT_TIMESTAMP())
        ORDER BY measured_at DESC
        LIMIT 1
    """, (asset_id, lookback_minutes // 2, lookback_minutes * 2))
    row = cursor.fetchone()
    return row[0] if row else None

def should_alert(current_status: str, previous_status: Optional[str]) -> bool:
    """
    Only alert when we transition INTO a bad state, not every measurement cycle.
    """
    if current_status == 'OK':
        return False
    if previous_status in ('WARN', 'BREACH'):
        return False  # Already alerted on this issue
    return True

def send_slack_alert(asset_id: str, status: str, reading: dict):
    color = "#FFA500" if status == "WARN" else "#FF0000"
    message = {
        "attachments": [{
            "color": color,
            "title": f"Data Freshness {status}: {asset_id}",
            "fields": [
                {"title": "Current Age", "value": f"{reading['load_age_minutes']:.1f} minutes", "short": True},
                {"title": "SLA Threshold", "value": f"{reading['sla_breach_minutes']} minutes", "short": True},
                {"title": "% of SLA Used", "value": f"{reading['load_age_minutes'] / reading['sla_breach_minutes'] * 100:.0f}%", "short": True},
                {"title": "Owner", "value": reading['owner'], "short": True},
                {"title": "Last Successful Load", "value": str(reading['last_watermark']), "short": False},
            ],
            "footer": "Data Observability | Wicked Smart Data",
            "ts": int(datetime.now(timezone.utc).timestamp())
        }]
    }
    requests.post(SLACK_WEBHOOK_URL, json=message)

def send_pagerduty_alert(asset_id: str, reading: dict):
    payload = {
        "routing_key": PAGERDUTY_ROUTING_KEY,
        "event_action": "trigger",
        "dedup_key": f"freshness-breach-{asset_id}",  # Prevents duplicate incidents
        "payload": {
            "summary": f"Data freshness SLA BREACH: {asset_id} is {reading['load_age_minutes']:.0f} min old (SLA: {reading['sla_breach_minutes']} min)",
            "severity": "critical",
            "source": "data-observability",
            "component": asset_id,
            "group": reading['layer'],
            "custom_details": {
                "asset_id": asset_id,
                "load_age_minutes": reading['load_age_minutes'],
                "event_age_minutes": reading['event_age_minutes'],
                "last_watermark": str(reading['last_watermark']),
                "owner": reading['owner'],
                "rows_last_hour": reading['rows_last_hour'],
            }
        }
    }
    response = requests.post(
        "https://events.pagerduty.com/v2/enqueue",
        headers={"Content-Type": "application/json"},
        data=json.dumps(payload)
    )
    return response.status_code

def resolve_pagerduty_alert(asset_id: str):
    """Call this when a BREACH resolves to auto-close the PD incident."""
    payload = {
        "routing_key": PAGERDUTY_ROUTING_KEY,
        "event_action": "resolve",
        "dedup_key": f"freshness-breach-{asset_id}",
    }
    requests.post(
        "https://events.pagerduty.com/v2/enqueue",
        headers={"Content-Type": "application/json"},
        data=json.dumps(payload)
    )

def process_alerts(conn, readings: list):
    for reading in readings:
        asset_id = reading['asset_id']
        current_status = reading['freshness_status']
        previous_status = get_previous_status(conn, asset_id)

        if current_status == 'OK' and previous_status == 'BREACH':
            # Auto-resolve the PagerDuty incident when freshness recovers
            resolve_pagerduty_alert(asset_id)
            send_slack_alert(asset_id, "RESOLVED", reading)
            continue

        if not should_alert(current_status, previous_status):
            continue

        # New WARN or BREACH — fire the appropriate alert
        send_slack_alert(asset_id, current_status, reading)

        if current_status == 'BREACH':
            send_pagerduty_alert(asset_id, reading)

The dedup_key in the PagerDuty payload is critical — it ensures that if the same asset triggers a breach on five consecutive measurement cycles, you still only get one incident. And the resolve_pagerduty_alert function auto-closes the incident when the data freshens up, which keeps your incident history clean.

---

Step 6: The Freshness Dashboard Stakeholders Can Trust

Your freshness system is only complete when stakeholders can see it themselves. A self-service freshness dashboard eliminates the "is the data current?" Slack messages that interrupt your team every morning.

Build a dbt model that powers the dashboard, pulling from your freshness metadata:

-- models/observability/dashboard_freshness_status.sql
{{
  config(
    materialized='view',
    tags=['observability']
  )
}}

WITH current_status AS (
    SELECT
        asset_id,
        layer,
        table_name,
        load_age_minutes,
        event_age_minutes,
        sla_breach_minutes,
        freshness_status,
        last_watermark,
        measured_at,
        owner,
        ROUND(load_age_minutes / NULLIF(sla_breach_minutes, 0) * 100, 0) AS pct_sla_used,
        -- Human-readable age
        CASE
            WHEN load_age_minutes < 60 THEN CONCAT(ROUND(load_age_minutes, 0)::VARCHAR, ' minutes ago')
            WHEN load_age_minutes < 1440 THEN CONCAT(ROUND(load_age_minutes / 60, 1)::VARCHAR, ' hours ago')
            ELSE CONCAT(ROUND(load_age_minutes / 1440, 1)::VARCHAR, ' days ago')
        END AS human_readable_age,
        -- Traffic light for BI tools
        CASE freshness_status
            WHEN 'OK' THEN '🟢'
            WHEN 'WARN' THEN '🟡'
            WHEN 'BREACH' THEN '🔴'
            ELSE '⚪'
        END AS status_icon
    FROM data_observability.v_freshness_current
),

-- Add 24h breach history to show trend
breach_history AS (
    SELECT
        asset_id,
        COUNTIF(freshness_status = 'BREACH') AS breach_count_24h,
        COUNTIF(freshness_status = 'WARN') AS warn_count_24h,
        MIN(CASE WHEN freshness_status != 'OK' THEN measured_at END) AS first_issue_at
    FROM data_observability.freshness_readings
    WHERE measured_at >= DATEADD('hour', -24, CURRENT_TIMESTAMP())
    GROUP BY asset_id
)

SELECT
    cs.*,
    COALESCE(bh.breach_count_24h, 0) AS breach_count_24h,
    COALESCE(bh.warn_count_24h, 0) AS warn_count_24h,
    bh.first_issue_at
FROM current_status cs
LEFT JOIN breach_history bh ON cs.asset_id = bh.asset_id
ORDER BY
    CASE cs.freshness_status WHEN 'BREACH' THEN 1 WHEN 'WARN' THEN 2 ELSE 3 END,
    cs.pct_sla_used DESC

Connect this model to Looker, Metabase, or Tableau with a Slack notification schedule — stakeholders get a morning digest of the freshness status for every dataset they care about. The status_icon column renders beautifully in any BI tool. Sorting by breach status first and then by pct_sla_used descending means the most urgent issues always appear at the top.

---

Hands-On Exercise

You're going to build a working freshness monitoring system for a three-table stack. Here's the scenario: your company tracks customer orders through a Postgres → Snowflake pipeline (via Airbyte), and dbt transforms the raw data into two mart tables.

Step 1: Create the observability schema and freshness_readings table from the SQL in Step 4.

Step 2: Create a test table that simulates a source with a watermark column:

CREATE TABLE IF NOT EXISTS raw_postgres.orders (
    order_id VARCHAR,
    customer_id VARCHAR,
    amount_usd FLOAT,
    status VARCHAR,
    created_at TIMESTAMP_TZ,
    _airbyte_emitted_at TIMESTAMP_TZ DEFAULT CURRENT_TIMESTAMP()
);

-- Insert some test data with a recent watermark
INSERT INTO raw_postgres.orders VALUES
('ord_001', 'cust_a', 99.99, 'completed', DATEADD('hour', -2, CURRENT_TIMESTAMP()), DATEADD('hour', -2, CURRENT_TIMESTAMP())),
('ord_002', 'cust_b', 149.50, 'pending', DATEADD('hour', -1, CURRENT_TIMESTAMP()), DATEADD('hour', -1, CURRENT_TIMESTAMP())),
('ord_003', 'cust_c', 75.00, 'completed', DATEADD('minute', -45, CURRENT_TIMESTAMP()), DATEADD('minute', -45, CURRENT_TIMESTAMP()));

Step 3: Write a freshness_slas.yml that defines an SLA for raw_postgres.orders with a 30-minute warn and 60-minute breach threshold.

Step 4: Run the Python run_freshness_checks function against your config and verify a reading appears in freshness_readings.

Step 5: Simulate a breach by updating your YAML to set sla_breach_minutes: 30 (your data is already 45 minutes old), re-running the check, and verifying the status flips to BREACH.

Step 6: Build the v_freshness_current view and query it. Confirm pct_sla_used and minutes_past_breach compute correctly.

Challenge extension: Modify process_alerts to also write to a data_observability.alert_log table every time an alert fires. Query that table to count how many alerts fired in the last 24 hours, grouped by asset and status.

---

Common Mistakes & Troubleshooting

Using MAX(updated_at) on a table with soft deletes. If your source system uses soft deletes and the deleted records have recent updated_at values, your watermark will look fresh even when no new business events have occurred. Use MAX(updated_at) WHERE deleted_at IS NULL or use a dedicated _etl_loaded_at column instead.

Setting SLA thresholds based on current pipeline performance, not business requirements. If your pipeline runs every hour and you set a 90-minute breach threshold "to give it some buffer," you've essentially made your SLA meaningless. A real business SLA should reflect when stale data causes a business problem. Define the SLA first; fix the pipeline second.

Not accounting for time zones in watermark comparisons. If your warehouse stores timestamps in UTC but your source system stores them in US/Eastern, DATEDIFF comparisons will be off by 4–5 hours. Always normalize to UTC in your ETL process and enforce TIMESTAMP_TZ column types in Snowflake (or TIMESTAMP WITH TIME ZONE in BigQuery/Postgres).

Alerting on load age when event age is what matters. For tables with bursty load patterns (large batches every few hours), load age can look fine while event age is dangerously high. Always instrument both metrics and define your SLA breach condition based on whichever is more relevant to the consumer.

Forgetting to handle the case where a table is empty. If a new table is created but no data has landed yet, MAX(watermark_column) returns NULL. Your freshness checker needs to handle NULL watermarks explicitly — treat them as infinite age (status: BREACH) unless the table was intentionally created empty.

# Null-safe status computation
def compute_status(load_age_minutes, warn_threshold, breach_threshold):
    if load_age_minutes is None:
        return 'BREACH'  # No data is always a breach
    if load_age_minutes >= breach_threshold:
        return 'BREACH'
    if load_age_minutes >= warn_threshold:
        return 'WARN'
    return 'OK'

Running freshness checks too infrequently for short SLAs. If you have a 15-minute SLA and run checks every 30 minutes, you can miss the breach window entirely. Your measurement interval should be no longer than one-quarter of your tightest SLA threshold.

---

Summary & Next Steps

You've built a complete data freshness SLA system from first principles. Let's recap what you now have:

• A machine-readable SLA contract in YAML that defines warn and breach thresholds per asset and per consumer
• Multi-layer instrumentation that measures freshness at ingestion (via connector metadata and watermark queries) and transformation (via dbt's source freshness and dbt_updated_at columns)
• A centralized freshness_readings table that aggregates all measurements into a queryable observability backend
• Tiered alerting with state-change detection to prevent alert fatigue, PagerDuty integration with dedup keys, and automatic incident resolution
• A stakeholder-facing dashboard that shows current freshness status, human-readable data age, and 24-hour breach history

The natural evolutions from here are:

1. Anomaly detection on row volumes — freshness is one signal, but a table can be fresh and still wrong. Add volume anomaly detection to catch partial loads and silent data loss.

2. Lineage-aware alerting — when a source table breaches its SLA, automatically compute which downstream models and dashboards are affected and include that impact list in the alert. dbt's manifest.json exposes the full lineage graph for exactly this purpose.

3. Freshness SLA reporting to stakeholders — generate a weekly digest of SLA adherence rates (uptime-style, e.g., "fct_orders met its freshness SLA 99.2% of the time this week") so you can have data quality conversations with business partners grounded in actual numbers.

4. Integration with data catalog tools — if you use Atlan, Alation, or DataHub, push freshness readings into the catalog so every dataset's page shows current freshness status alongside column descriptions and lineage.

The data reliability engineering discipline is still young, but the patterns in this lesson are battle-tested. Once you have freshness monitoring running in production, you'll wonder how you ever operated without it.