Lab 09 — CI/CD & Observability

The goal

So far mini-GridDP is correct when you run it. Two gaps separate it from a real platform:

• Nobody checks your changes before they land. You could merge a broken model at midnight and not find out until the dashboard is wrong. We fix this with CI — Continuous Integration — that runs your dbt tests automatically on every pull request and blocks the merge if anything fails.
• You can't see what your pipelines did. When a run is slow or empty, you have no record of how many rows moved or how long it took. We fix this with observability — structured logs, run metrics, and freshness checks surfaced in the Dagster UI.

This lab adds the last two boxes from the architecture diagram — the QUALITY and OBSERVABILITY rows at the bottom:

A CI workflow on every PR

GitHub Actions runs a workflow — a YAML file describing steps to execute on an event. Our event is pull_request. The steps: check out the code, set up Python with uv (the same tool from Course 5 ch.05), install the project from its lockfile, seed a tiny throwaway DuckDB, then run dbt build, which runs the tests too. If any test fails, the step exits non-zero and the whole check goes red.

The key idea: CI runs against a throwaway database seeded with a tiny sample, not your real warehouse. That makes the run fast, deterministic, and safe — it never touches production data. dbt's seeds are perfect for this: CSVs that dbt loads as tables.

First, give CI a sample to chew on. The seeds have to look exactly like bronze — same table names, same columns, loaded_at included — because the staging models will read them through the same source('bronze', …) declarations. Five small files, a handful of rows each; note the H100 price change on 2 April, so the SCD2 join gets exercised too:

customer_id,name,tier,created_at,loaded_at
1,Acme GPU Co,pro,2026-01-04 09:00:00+00,2026-04-03 06:00:00+00
2,Northwind AI,free,2026-02-11 09:00:00+00,2026-04-03 06:00:00+00
3,Globex Render,enterprise,2026-03-02 09:00:00+00,2026-04-03 06:00:00+00

gpu_id,model,vram_gb,loaded_at
1,NVIDIA H100 80GB,80,2026-04-03 06:00:00+00
2,NVIDIA A100 80GB,80,2026-04-03 06:00:00+00

gpu_id,price_per_hour,valid_from,loaded_at
1,4.20,2026-01-01 00:00:00+00,2026-04-03 06:00:00+00
1,4.50,2026-04-02 00:00:00+00,2026-04-03 06:00:00+00
2,2.40,2026-01-01 00:00:00+00,2026-04-03 06:00:00+00

rental_id,customer_id,gpu_id,hours,started_at,loaded_at
1001,1,1,4.00,2026-04-01 10:00:00+00,2026-04-03 06:00:00+00
1002,2,2,1.00,2026-04-01 11:30:00+00,2026-04-03 06:00:00+00
1003,1,1,2.50,2026-04-02 09:00:00+00,2026-04-03 06:00:00+00

gpu_id,ts,util_pct,temp_c,power_w,loaded_at
1,2026-04-02 09:00:05+00,81.0,82.5,330.0,2026-04-03 06:00:00+00
1,2026-04-02 09:00:10+00,79.0,81.9,325.0,2026-04-03 06:00:00+00
2,2026-04-02 09:00:05+00,55.0,70.1,260.0,2026-04-03 06:00:00+00

Now make dbt load them as bronze, and only in CI. Two lines of config in dbt_project.yml do it: the seeds land in the bronze schema (the Lab 03 macro keeps that name as-is), and they are enabled only when the target is ci, so a local dbt build can never overwrite your real bronze tables with three rows. Column types are pinned so timestamps stay timestamps:

seed-paths: ["seeds"]

# CI fixtures: tiny CSVs that stand in for bronze. Enabled ONLY for the ci
# target, so a local `dbt build` can never overwrite your real bronze tables.
seeds:
  mini_griddp:
    +enabled: "{{ target.name == 'ci' }}"
    +schema: bronze
    customers:
      +column_types: {created_at: timestamptz, loaded_at: timestamptz}
    gpus:
      +column_types: {loaded_at: timestamptz}
    gpu_prices:
      +column_types: {price_per_hour: "decimal(6,2)", valid_from: timestamptz, loaded_at: timestamptz}
    rentals:
      +column_types: {hours: "decimal(8,2)", started_at: timestamptz, loaded_at: timestamptz}
    telemetry:
      +column_types: {ts: timestamptz, loaded_at: timestamptz}

One subtlety decides the order of the CI steps: dbt does not know that a seed stands in for a source, so on an empty database dbt build would start creating views before the seeds they read exist. Seed first, then build. Now the workflow:

name: CI

# Run on every PR targeting main, and on pushes to main itself.
on:
  pull_request:
    branches: [main]
  push:
    branches: [main]

jobs:
  dbt-build-and-test:
    runs-on: ubuntu-latest
    steps:
      - name: Check out the code
        uses: actions/checkout@v4

      - name: Set up uv (Python toolchain)
        uses: astral-sh/setup-uv@v5

      - name: Install the project from the lockfile (same versions as your laptop)
        run: uv sync --frozen

      - name: Show dbt can see the project
        run: uv run dbt debug --target ci
        working-directory: dbt

      - name: Load the tiny sample (throwaway DuckDB)
        run: uv run dbt seed --full-refresh --target ci
        working-directory: dbt

      - name: Build models and run ALL tests
        # dbt build runs seeds, models AND tests in dependency order.
        # If any test fails, dbt exits non-zero and this step (and the PR) goes red.
        run: uv run dbt build --target ci
        working-directory: dbt

Every dbt step runs inside dbt/, where dbt finds profiles.yml on its own, and names the ci target. Add that target to the Lab 03 profile next to dev, not instead of it; it writes a throwaway file that the runner discards when the job ends:

mini_griddp:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: ../warehouse.duckdb   # the file from Lab 02, relative to dbt/
      schema: main
      threads: 4
    ci:
      type: duckdb
      path: ci_warehouse.duckdb   # created fresh on the CI runner, discarded after
      schema: main
      threads: 4

You don't need GitHub to see what CI will see. The same two commands run on your laptop against the same throwaway file (*.duckdb is already gitignored):

cd dbt
uv run dbt seed --full-refresh --target ci
uv run dbt build --target ci
cd ..

Watch CI go red, then green

Tests you never see fail are tests you don't trust. Let's break something on purpose and watch the gate catch it. Pick a test to break: fct_rentals.customer_id carries a relationships test to dim_customer (Lab 06). On a new branch, add a rental for a customer that does not exist:

git checkout -b break-ci-on-purpose

# Add a rental for customer 9, who is not in customers.csv, so the relationships test must fail.
printf '1004,9,1,1.00,2026-04-02 12:00:00+00,2026-04-03 06:00:00+00\n' >> dbt/seeds/rentals.csv

git add -A
git commit -m "TEMP: orphan rental to prove CI catches it"
git push -u origin break-ci-on-purpose
gh pr create --fill        # opens the PR

Now fix it. Remove the orphan row, commit, and push to the same branch — CI re-runs automatically:

# Drop the last (orphan) line we appended.
sed -i.bak '$ d' dbt/seeds/rentals.csv && rm dbt/seeds/rentals.csv.bak

git add -A
git commit -m "Fix: remove the orphan rental"
git push                   # same branch → CI runs again on the PR

Structured logs & run metrics

CI guards your code. Logging tells you what your runs actually did. Right now your ingest and rollup scripts probably print() a few things, which is invisible the moment you're not watching. Replace prints with structured logging: one JSON object per event, so a human can read it and a machine can parse it.

Create a tiny shared helper so every script logs the same way:

"""Shared structured-logging helper for mini-GridDP scripts."""
import json
import logging
import sys
import time
from contextlib import contextmanager


class JsonFormatter(logging.Formatter):
    """Render each log record as a single line of JSON."""
    def format(self, record):
        payload = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        # Anything passed via logger.info("...", extra={"fields": {...}})
        if hasattr(record, "fields"):
            payload.update(record.fields)
        if record.exc_info:                 # keep the traceback when logger.exception() is used
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)


def get_logger(name):
    logger = logging.getLogger(name)
    if not logger.handlers:            # avoid duplicate handlers on re-import
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(JsonFormatter())
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
    return logger


@contextmanager
def timed_run(logger, step):
    """Log start/finish of a step and emit duration + whatever metrics you set."""
    metrics = {}                       # mutate this inside the `with` block
    start = time.perf_counter()
    logger.info(f"{step} started", extra={"fields": {"step": step, "event": "start"}})
    try:
        yield metrics
    except Exception:
        logger.exception(f"{step} FAILED", extra={"fields": {"step": step, "event": "error"}})
        raise
    else:
        metrics["duration_s"] = round(time.perf_counter() - start, 3)
        metrics["step"] = step
        metrics["event"] = "finish"
        logger.info(f"{step} finished", extra={"fields": metrics})

Now wire it into the Postgres extractor from Lab 02. The pattern: open a timed_run, do the work, and record rows into the metrics dict — duration is captured for you. Three small edits: full_refresh ends with return n, incremental_rentals ends with return new_rows, and main() wraps each step:

from obs import get_logger, timed_run

log = get_logger("ingest.postgres")

# ... connect(), full_refresh() (now ends with `return n`),
#     incremental_rentals() (now ends with `return new_rows`) unchanged otherwise ...


def main():
    con = connect()
    ensure_watermark_store(con)
    print(f"Landing bronze into {WAREHOUSE} ...")
    for t in FULL_REFRESH:
        with timed_run(log, f"extract:{t}") as m:
            m["rows"] = full_refresh(con, t)        # <-- the run metric we care about
    with timed_run(log, "extract:rentals") as m:
        m["rows"] = incremental_rentals(con)
    con.close()
    print("Done.")

Run it and watch the output. The JSON lines sit between the script's own print lines, so tell jq to skip anything that isn't JSON:

uv run python ingest/extract_postgres.py

# Want only the finished-step metrics? Pipe through jq (-R + fromjson? skips the plain lines):
uv run python ingest/extract_postgres.py | jq -R 'fromjson? | select(.event == "finish") | {step, rows, duration_s}'

The observability surface

Logs are the raw material; you still need a surface to see them on. For mini-GridDP that surface is the Dagster UI you stood up in Lab 05 (and the orchestration ideas from Course 5 ch.07). Every asset materialization shows up as a run with a timeline, the captured logs, and — once we add them — checks.

Dagster asset checks let an asset assert things about itself after it materializes. Add a freshness check that flags stale telemetry — if the newest row in fct_telemetry_hourly is older than two hours, the data has gone stale and the check fails loudly in the UI:

from datetime import datetime, timezone

import duckdb
from dagster import asset_check, AssetCheckResult, AssetKey

from dagster_app.assets import WAREHOUSE


@asset_check(asset=AssetKey("fct_telemetry_hourly"))
def telemetry_is_fresh() -> AssetCheckResult:
    """Fail if the newest telemetry hour is more than 2 hours old."""
    con = duckdb.connect(str(WAREHOUSE), read_only=True)
    newest = con.execute("select max(hour) from gold.fct_telemetry_hourly").fetchone()[0]
    con.close()

    if newest is None:
        return AssetCheckResult(passed=False, metadata={"reason": "table is empty"})

    # `hour` is TIMESTAMPTZ (Lab 04), so DuckDB hands back an aware datetime; compare like with like.
    age_hours = (datetime.now(timezone.utc) - newest).total_seconds() / 3600
    return AssetCheckResult(
        passed=age_hours <= 2,
        metadata={
            "newest_hour": str(newest),
            "age_hours": round(age_hours, 2),
            "threshold_hours": 2,
        },
    )

Checks live in their own module, so tell definitions.py to load them from both files, then run the pipeline:

from dagster_app import assets, checks

all_assets = load_assets_from_modules([assets])
all_checks = load_asset_checks_from_modules([assets, checks])

uv run dagster job execute -j full_pipeline -m dagster_app.definitions   # headless, or:
uv run dagster dev    # open http://localhost:3000 (stop the Lab 01 placeholder first)

What you'd actually alert on, in priority order:

Commit Lab 09

You've added the CI workflow, sample seeds, a CI profile, the logging helper, and asset checks. Land it all on main through a clean PR — and this time, let your own green check approve the merge:

git checkout -b lab-09-cicd-observability
git add .github/workflows/ci.yml dbt/seeds dbt/profiles.yml dbt/dbt_project.yml \
        ingest/obs.py ingest/extract_postgres.py dagster_app/checks.py dagster_app/definitions.py
git commit -m "Lab 09: CI on every PR + structured logs + freshness checks"
git push -u origin lab-09-cicd-observability
gh pr create --fill
# wait for the green check, then:
gh pr merge --squash --delete-branch

✓ Check yourself

Answer from memory before opening each one.

1. CI is red on a PR, yet the merge button still works and a teammate uses it. What was never set up?

   Show answer

   Branch protection. A failing check is only advisory until main has a rule requiring the dbt-build-and-test status check to pass; with that rule, GitHub itself refuses the merge. See Make it a real gate.

2. A teammate proposes running CI against the real warehouse.duckdb "so the tests see real data". What stands in the way, and what do you concede?

   Show answer

   The runner does not have it — the warehouse is a gitignored build artifact — and even if it did, the run would be slow, non-deterministic (green today, red tomorrow with no code change) and touching real data. Seeds give a small fixed sample, so a red check means the code broke. The concession: a three-row sample never contains the odd rows production does. CI guards code; the same tests running inside the scheduled pipeline (Lab 06) guard data, and you need both. See A CI workflow on every PR.

3. Last night's log contains extract:customers finished with "rows": 0, and the run is green. Why did nothing fail, and what should have happened?

   Show answer

   Selecting from an empty table is a perfectly successful query — no exception, so timed_run logs a normal finish. A green status says the code ran, not that data arrived. That is why rows is emitted as a metric: rows == 0 on an ingest step is one of the signals worth alerting on, before an empty dimension makes every downstream model wrong. See Structured logs and what you'd alert on.

4. An exception is raised inside a with timed_run(log, "extract:gpus") as m: block, after m["rows"] was set. Which log lines exist for that step, and what does the jq 'select(.event == "finish")' filter show for it?

   Show answer

   A start line and an ERROR line with "event": "error"; then the exception is re-raised. There is no finish line, so no duration_s, and the rows you set is never emitted. The jq filter shows nothing at all for the step — a view of finished steps hides failures, so watch for error events, or a start with no matching finish. See timed_run.

5. The telemetry producer died yesterday afternoon. This morning's scheduled run is green from end to end. Which signal tells you something is wrong, and why is run status silent?

   Show answer

   The telemetry_is_fresh asset check: the newest hour in fct_telemetry_hourly is more than two hours old, so it fails and reports age_hours. The run succeeded honestly — rolling up yesterday's readings again works fine. Staleness is missing data, not failing code, so only a check that compares the data's own timestamp with the clock can see it. See The observability surface.

6. A teammate routes every signal in the alerting table — including "duration 3× the norm" — to a page at any hour. What would you change, and on what principle?

   Show answer

   Page for what a human must act on now: a failed scheduled run, a red freshness check (dashboards quietly showing old numbers). A slow but successful run is an early warning — log it and review it in daylight, escalating only if it starts to threaten freshness. Alerts that fire with nothing to do teach the team to ignore alerts, and that costs you the one that matters. See Alert on symptoms, not noise.

      - name: Check source freshness
        # The seeds' loaded_at stamps are months old, so this reports ERROR STALE
        # and exits 1. continue-on-error keeps the step informational; drop that
        # line once you want freshness to gate merges (with fixtures that are fresh).
        run: uv run dbt source freshness --target ci
        working-directory: dbt
        continue-on-error: true

from obs import get_logger, timed_run

log = get_logger("ingest.telemetry")


def main():
    # ... connect, subscribe as before ...
    landed = 0
    with timed_run(log, "land:telemetry") as m:
        # ... the poll loop as before, but count what flush() inserts:
        #     landed += len(buffer) just before each flush(con, buffer)
        m["rows"] = landed


if __name__ == "__main__":
    main()

Takeaway

• CI runs dbt build on every pull request against a throwaway DuckDB file seeded with a tiny sample, so it is fast, repeatable and cannot touch production.
• The seeds must look exactly like bronze and be enabled only for the ci target, and CI must run dbt seed before dbt build because dbt does not know a seed stands in for a source.
• A red check is advisory until branch protection requires dbt-build-and-test, and then GitHub itself blocks the merge.
• Prove the gate by breaking the relationships test on a branch: the check goes red, then green once you fix it.
• Structured JSON logs with the step, rows and duration are queryable with jq, while a print is invisible the moment you stop watching.
• A Dagster asset check can flag telemetry older than two hours, and you alert only on what a human must act on, such as a failed run or stale data.

The platform is built, tested in CI and observable, so the last lab packages it so anyone can run and understand it. → Lab 10 — Ship & Document

Further reading

• GitHub Actions: Understanding GitHub Actions — GitHub docs, accessed 2026. Workflows, events, jobs and runners.
• GitHub: About protected branches — GitHub docs, accessed 2026. Required status checks as a merge gate.
• dbt: Seeds — dbt Labs docs, accessed 2026. Loading CSVs as tables and enabling them per target.
• Dagster: Testing assets with asset checks — Dagster Labs docs, accessed 2026. The freshness check on the rollup asset.
• jq Manual — jq project. Filtering the JSON log lines, including skipping non-JSON lines.