Day 57: Logging, Secrets, Concurrency and Run History · Applied Decision Intelligence & Data Architecture

Written lesson

Time	Block
0:00 – 0:25	57.1 Logging in depth
0:25 – 0:45	57.2 Secrets and runtime configuration in Prefect
0:45 – 0:53	Break
0:53 – 1:15	57.3 Concurrency limits: one DuckDB writer, enforced
1:15 – 1:40	57.4 Run history as data
1:40 – 1:48	Break
1:48 – 2:05	57.5 A watchdog: alerting when nothing happens
2:05 – 2:30	57.6 Capstone: `daily_refresh` v3, end to end

57.1 Logging in Depth (25 min)

Levels from subprocess output

Since Day 52, every line of every tool's output is logged at INFO, so a dbt error and a progress message look the same in the UI. Map levels from the content:

import re

ERROR_PAT = re.compile(r"(\bERROR\b|✗|⛔|Traceback|FAIL)", re.IGNORECASE)
WARN_PAT = re.compile(r"(\bWARN(ING)?\b|⚠)", re.IGNORECASE)


def log_line(logger, name: str, line: str):
    text = f"[{name}] {line.rstrip()}"
    if ERROR_PAT.search(line):
        logger.error(text)
    elif WARN_PAT.search(line):
        logger.warning(text)
    else:
        logger.info(text)

Use log_line(logger, name, line) in run_step's streaming loop. Now the UI's level filter shows you only the problems from a 3,000-line run.

What makes a good log message?

✗ Poor	✓ Good
`done`	`wap_statcan_tables: published statcan_lfs_full (release 2026-09-05T08:30), 5.4M rows, 212s`
`error`	`ingest_api attempt 2/3: TransientStepError exit 1 — retrying in 600s`
`fingerprint ok`	`warehouse fingerprint 3fa1c09b… unchanged since 2026-09-29 17:41 — dbt skipped`

Include what, which object, how much, how long, and what happens next. Never log secrets (API keys, webhook URLs) or large data dumps.

Log levels for the whole platform

prefect config set PREFECT_LOGGING_LEVEL=INFO            # DEBUG while developing

Libraries you call in-process (e.g. duckdb or requests inside a task) log to their own Python loggers. To see them in Prefect, list them:

prefect config set PREFECT_LOGGING_EXTRA_LOGGERS="urllib3"

(Our main tools run as subprocesses, so their output already arrives through the streaming in run_step.)

Logging outside a run

Code shared by flows and ordinary scripts can't always call get_run_logger(), which only works inside a run. Use a fallback:

from prefect.exceptions import MissingContextError
from prefect.logging import get_logger, get_run_logger


def logger_for_context():
    try:
        return get_run_logger()
    except MissingContextError:
        return get_logger("macro")

57.2 Secrets and Runtime Configuration in Prefect (20 min)

Secret blocks

Blocks are Prefect's stored configuration objects. A Secret block stores a sensitive value encrypted on the server, masked in the UI, and loadable by any flow:

from prefect.blocks.system import Secret

Secret(value="https://hooks.example.com/services/XXX").save("notifications-webhook", overwrite=True)

Run that once, interactively, never in committed code. Then use it:

from prefect.blocks.system import Secret


def webhook_url() -> str | None:
    try:
        return Secret.load("notifications-webhook").get()
    except ValueError:
        return None                                     # not configured: print alerts instead

Update the alert hook (Day 53) to post to webhook_url(), falling back to Day 29's environment variable. Scheduled runs started by the worker then don't depend on your shell's environment variables.

Store	For	Example
Secret block	sensitive values	webhook URL, FRED API key, cloud credentials
Variable	small, non-sensitive runtime settings editable in the UI	`dbt_last_fingerprint`, `max_staleness_hours`
`prefect.yaml` parameters	per-deployment defaults, reviewed in Git	`force_transform: true` for weekly rebuilds
`.dlt/config.toml`, dbt `vars`	settings owned by each tool	series lists, panel start date

Each setting should have one home. Don't copy the dbt start date into a Prefect Variable "for convenience". When the two drift apart, you'll spend an afternoon finding out why.

Using a Variable for an operational threshold

from prefect.variables import Variable

Variable.set("refresh_max_age_hours", 30, overwrite=True)      # once, or edit in the UI

The watchdog (57.5) reads it. An operator can relax the threshold during a planned outage from the UI, without a code change.

57.3 Concurrency Limits: One DuckDB Writer, Enforced (23 min)

concurrency_limit: 1 (Day 56) stops two runs of the same deployment overlapping. But prod-morning, prod-evening and weekly-full-rebuild are different deployments, and a manual run of ingest.py could also collide with them. The real constraint is one writer to macro.duckdb at a time, so let's enforce exactly that.

A global concurrency limit

prefect gcl create duckdb-writer --limit 1
prefect gcl ls

(gcl is short for global-concurrency-limit.) In code, wrap every writing step:

from prefect.concurrency.sync import concurrency

with concurrency("duckdb-writer", occupy=1):
    result = run_step.fn(name, cmd, cwd)          # only one holder at a time, across all runs

Mark which steps write in STEP_POLICIES (writes_duckdb=True for ingest_api, wap_statcan_tables and dbt_build) and apply the context manager only to those. Readers (GX, briefing) don't need the slot; DuckDB allows concurrent readers.

Test it

Trigger prod-morning and weekly-full-rebuild at the same time (UI or CLI).
Watch the logs: one run's writing step acquires duckdb-writer; the other's waits until the slot is released, then proceeds.
Inspect the limit in the UI (Concurrency), where its active slots are visible while runs hold them.

Mechanism	Scope	Protects against
deployment `concurrency_limit`	one deployment	a slow run overlapping its own next scheduled run
global concurrency limit `duckdb-writer`	any code, any flow, any deployment	two writers on one DuckDB file
sequential steps inside a flow (Day 50)	one run	the flow fighting itself

57.4 Run History as Data (25 min)

The Prefect UI shows runs, but you've learned the value of history as a table (GX history, Day 47): trends, KPIs, dbt models. Apply the same pattern to orchestration.

Create orchestration/tasks/history.py:

"""Append one row per step per run to Parquet (data/orchestration/runs/)."""
from datetime import datetime, timezone

import pandas as pd
from prefect import task
from prefect.runtime import flow_run

from orchestration.settings import REPO

HISTORY_DIR = REPO / "data" / "orchestration" / "runs"


@task(cache_policy=None)
def record_run_history(summary: dict, details: dict, final_state_name: str):
    now = datetime.now(timezone.utc)
    rows = [{
        "recorded_at": now,
        "flow_run_id": str(flow_run.id),
        "flow_run_name": flow_run.name,
        "deployment": flow_run.deployment_id and str(flow_run.deployment_id),
        "scheduled_start": flow_run.scheduled_start_time,
        "step": step,
        "outcome": outcome,
        "duration_s": details.get(step, {}).get("duration_s"),
        "exit_code": details.get(step, {}).get("exit_code"),
        "run_final_state": final_state_name,
    } for step, outcome in summary.items()]
    HISTORY_DIR.mkdir(parents=True, exist_ok=True)
    pd.DataFrame(rows).to_parquet(HISTORY_DIR / f"run_{now:%Y%m%dT%H%M%S}.parquet", index=False)

prefect.runtime.flow_run exposes the current run's ID, name, deployment and scheduled start time, all useful for joining with the UI later.

Call it at the end of daily_refresh, before returning the final state (compute the final state's name first), so failed runs are recorded too. (If you also need pandas in the pure-Prefect environment, it's already installed with dlt.)

Questions the history answers

-- How long does each step take, and is it getting slower?
select step,
       date_trunc('week', recorded_at)       as week,
       round(avg(duration_s))                as avg_s,
       round(max(duration_s))                as max_s
from 'data/orchestration/runs/*.parquet'
group by all
order by step, week desc;

-- How often did each step need attention?
select step, outcome, count(*) as runs
from 'data/orchestration/runs/*.parquet'
group by all
order by step, outcome;

Into dbt (the same pattern as Day 50)

Add a dbt source ops with external_location: "../data/orchestration/runs/*.parquet", a staging model stg_ops__pipeline_runs, and a mart ops_refresh_status: last successful refresh time, typical duration, and the steps that failed in the last 7 days. Add a KPI to macro_kpis (Day 39): "Hours since last successful refresh". The cockpit now shows economy, data quality and pipeline health side by side.

57.5 A Watchdog: Alerting When Nothing Happens (18 min)

Hooks alert when a run fails. But what if no run happens at all: the worker died, the Codespace stopped, someone paused a schedule and forgot? No failure means no alert. You need a check that looks for the absence of success.

Create orchestration/flows/watchdog.py:

"""Alert if daily-refresh hasn't completed successfully recently."""
from datetime import datetime, timedelta, timezone

import duckdb
from prefect import flow, get_run_logger
from prefect.variables import Variable

from macro_ingest.resilience import notify
from orchestration.settings import REPO


@flow(name="watchdog")
def watchdog():
    logger = get_run_logger()
    max_age = timedelta(hours=float(Variable.get("refresh_max_age_hours", default=30)))
    pattern = str(REPO / "data" / "orchestration" / "runs" / "*.parquet")

    try:
        last_ok = duckdb.sql(f"""
            select max(recorded_at) from '{pattern}'
            where run_final_state in ('Completed', 'CompletedWithWarnings')
        """).fetchone()[0]
    except Exception as e:
        last_ok = None
        logger.warning(f"No run history readable yet: {e}")

    now = datetime.now(timezone.utc)
    if last_ok is None or now - last_ok.replace(tzinfo=timezone.utc) > max_age:
        msg = (f"🕳 No successful daily-refresh for more than {max_age}. "
               f"Last success: {last_ok or 'never'}. Check the worker and deployment schedules.")
        logger.error(msg)
        notify(msg)
    else:
        logger.info(f"✓ Last successful refresh {now - last_ok.replace(tzinfo=timezone.utc)} ago")


if __name__ == "__main__":
    watchdog()

Add it to prefect.yaml with an hourly schedule, and redeploy (prefect deploy --all). Test it by setting the Variable refresh_max_age_hours to 0.01 in the UI, waiting for the next watchdog run, confirming the alert, then restoring 30.

A watchdog running on the same machine as the worker can't detect that the machine itself is down. In real production, run the watchdog elsewhere (a different host, or a managed monitoring service), or use Prefect Cloud's built-in automations for "deployment hasn't run". Note this limitation in DEPLOYMENT.md.

57.6 Capstone: `daily_refresh` v3, End to End (25 min)

Assemble Days 51–57 into the final flow for this module.

Requirements:

Every step goes through step() with its STEP_POLICIES entry: retries only for transient codes, backoff with jitter, timeouts, and the duckdb-writer concurrency limit for writers.
transform_flow skips dbt when the content fingerprint is unchanged, unless force_transform=True, and records the fingerprint only after success.
Subprocess output is logged with levels.
The final state follows STATE_POLICY.md (Day 53, updated on Day 54 for exit code 3).
Hooks alert through the Secret-block webhook; artifacts publish the step table and macro briefing.
Run history is recorded for every run, including failed ones.
Deployments prod-morning, prod-evening, weekly-full-rebuild, cpi-briefing and watchdog are defined in prefect.yaml; a worker is running.

Acceptance drills

#	Scenario	How	Expected
1	clean run	trigger `prod-evening`	Completed (or CompletedWithWarnings if GX warns); history row; briefing artifact
2	immediate re-run	trigger again	dbt Skipped by fingerprint; much faster
3	transient ingestion failure	custom run with `drill={"ingest_api": {"code": 1, "attempts": 1}}`	one retry, then Completed
4	quarantine	`drill={"wap_statcan_tables": {"code": 3, "attempts": 99}}`	not retried; policy outcome recorded
5	overlapping runs	trigger `prod-morning` and `weekly-full-rebuild` together	writers serialised by `duckdb-writer`
6	silent stop	pause both prod schedules; lower `refresh_max_age_hours`	watchdog alerts

Record the evidence (run names, screenshots or log excerpts) in orchestration/ACCEPTANCE.md.

Update the runbook

Add to RUNBOOK.md:

## Orchestration
- UI: port 4200 (Codespaces). Deployments: prod-morning (09:15 ET), prod-evening (17:30 ET),
  weekly-full-rebuild (Sun 07:00 ET), cpi-briefing, watchdog (hourly).
- Worker: `prefect worker start --pool local-process` (in .venv). No worker → no scheduled runs.
- Force a rebuild: run prod-evening with force_transform=true.
- Maintenance: pause schedules in the UI; resume afterwards; the watchdog will alert if you forget.
- Stuck "waiting" runs: check the `duckdb-writer` concurrency limit for a held slot (UI → Concurrency).
- Exit codes and retry rules: orchestration/EXIT_CODES.md, settings.STEP_POLICIES.

Recap: Days 54–57

Capability	Built
retries at the right level	task retries with backoff and jitter; HTTP retries kept; flow retries avoided
retrying only what time can fix	`retry_condition_fn`, `TransientStepError`, unambiguous exit codes
bounded runtimes	per-step timeouts; a retry budget derived from the deadline
not repeating work	release-keyed caches for fetches; content fingerprints for dbt
running by itself	deployments in `prefect.yaml`, `America/Toronto` schedules aligned with releases, a worker
safety between runs	deployment concurrency limits; a global `duckdb-writer` limit
operability	levelled logs, Secret blocks, Variables, run history, a watchdog, pipeline-health KPIs

Looking ahead: Days 58–60

The daily refresh of the macro dashboard pipeline, end to end: dlt extraction, DuckDB staging, dbt transformations and GX validation, as the curriculum describes, with the notebook dashboards regenerated from the marts.
Event-driven runs: StatCan release detection triggering the WAP flow as soon as new data appears, instead of waiting for the next schedule.
A final production-readiness review of the whole platform, Modules 1–6.

Day 57 checklist

Tool output is logged with correct levels; messages say what, which, how much, how long, what next
Secrets live in Secret blocks; operational thresholds in Variables; each setting has one home
A global duckdb-writer limit serialises every writer, across all deployments
Every run appends to a queryable history, which feeds a pipeline-health KPI
A watchdog alerts when successful runs stop happening, and its limitations are documented
daily_refresh v3 passes all six acceptance drills

git add . && git commit -m "Day 57: logging, secrets, concurrency limits, run history, watchdog" && git push

Logging, Secrets, Concurrency and Run History

What you'll learn today

Today's plan