Lab 09 — CI/CD & Observability

customer_id,name,region,signed_up_at
1,Acme GPU Co,us-east,2026-01-04
2,Northwind AI,eu-west,2026-02-11
3,Globex Render,us-west,2026-03-02

rental_id,customer_id,gpu_id,started_at,ended_at,hourly_price
1001,1,h100-01,2026-04-01T10:00:00,2026-04-01T14:00:00,2.50
1002,2,a100-07,2026-04-01T11:30:00,2026-04-01T12:30:00,1.80
1003,1,h100-02,2026-04-02T09:00:00,,2.50

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
    defaults:
      run:
        working-directory: dbt        # all steps run inside the dbt project
    env:
      # dbt reads the profile from the repo, not ~/.dbt, so CI is self-contained.
      DBT_PROFILES_DIR: .
    steps:
      - name: Check out the code
        uses: actions/checkout@v4

      - name: Set up uv (Python toolchain)
        uses: astral-sh/setup-uv@v5
        with:
          python-version: "3.12"

      - name: Install dbt + the DuckDB adapter
        run: uv pip install --system dbt-core dbt-duckdb

      - name: Show dbt can see the project
        run: dbt debug

      - name: Load the tiny sample (throwaway DuckDB)
        run: dbt seed --full-refresh

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

mini_griddp:
  target: ci
  outputs:
    ci:
      type: duckdb
      path: ci_warehouse.duckdb    # created fresh on the runner, discarded after
      threads: 4

git checkout -b break-ci-on-purpose

# Add a duplicate rental_id to the seed so the unique test must fail.
printf '1003,2,a100-09,2026-04-02T09:00:00,,1.80\n' >> dbt/seeds/sample_rentals.csv

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

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

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

"""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)
        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})

from obs import get_logger, timed_run

log = get_logger("ingest.postgres")

def extract_table(con, table):
    with timed_run(log, f"extract:{table}") as m:
        rows = con.execute(f"SELECT * FROM {table}").fetchall()
        write_to_bronze(table, rows)   # your existing landing logic
        m["rows"] = len(rows)          # <-- the run metric we care about
        m["table"] = table
    return rows


if __name__ == "__main__":
    con = connect_postgres()
    for table in ("customers", "gpus", "rentals", "prices"):
        extract_table(con, table)

python ingest/extract_postgres.py

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

import datetime as dt
from dagster import asset_check, AssetCheckResult, AssetKey


@asset_check(asset=AssetKey("fct_telemetry_hourly"))
def telemetry_is_fresh(duckdb) -> AssetCheckResult:
    """Fail if the newest telemetry hour is more than 2 hours old."""
    with duckdb.get_connection() as con:
        newest = con.execute(
            "SELECT max(hour) FROM fct_telemetry_hourly"
        ).fetchone()[0]

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

    age_hours = (dt.datetime.utcnow() - newest).total_seconds() / 3600
    return AssetCheckResult(
        passed=age_hours <= 2,
        metadata={
            "newest_hour": str(newest),
            "age_hours": round(age_hours, 2),
            "threshold_hours": 2,
        },
    )

dagster dev    # open http://localhost:3000

git checkout -b lab-09-cicd-observability
git add .github/workflows/ci.yml dbt/seeds dbt/profiles.yml \
        ingest/obs.py ingest/extract_postgres.py dagster_app/checks.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