diff --git a/ASSETS.md b/ASSETS.md index 3cbcc2e..99ab02f 100644 --- a/ASSETS.md +++ b/ASSETS.md @@ -22,7 +22,7 @@ Run the command from the root of your bundle project. The CLI will prompt for an | `monitoring-sql-warehouse` | Small, dedicated serverless SQL warehouse (2X-Small, `auto_stop_mins: 1`) for scheduled Databricks Alerts and monitoring queries. Keeps cost proportional to actual query time instead of idle warm-up. | Stable | [README](assets/monitoring-sql-warehouse/README.md) | | `sdp-quarantine-pattern` | Lakeflow SDP pipeline demonstrating the inverse-expectations quarantine pattern on `samples.nyctaxi.trips`: critical (drop) expectations route bad rows into a separate quarantine table (silver schema), valid rows flow to silver, warn expectations log to a queryable event log. NULL-safe predicates keep the silver/quarantine split a clean partition. Ships a companion agent skill (`SKILL.md`) that adapts the pattern to your own dataset and self-verifies it. | Stable | [README](assets/sdp-quarantine-pattern/README.md) | | `pyspark-test-runner` | Single-file Python wrapper around `pytest` for local PySpark suites that prints a bounded, agent-friendly digest (counts, runnable failing node ids, failures deduplicated by signature) and keeps full output in a log file, so a suite that floods with repetitive failures does not burn a coding agent's context window. Ships a `SKILL.md` for agent integration. | Stable | [README](assets/pyspark-test-runner/README.md) | -| `sdp-expectation-notifications` | Per-expectation data-quality notification for Lakeflow SDP as a validated pair: a native event hook notifies the moment a WARN expectation result is logged (fast, best-effort by platform design), and one DABs-managed Alert v2 sweeps the published pipeline event log over a past-time window on a schedule (guaranteed). Demo pipeline on `samples.nyctaxi.trips` fires both paths on the first run. Ships a companion agent skill (`SKILL.md`) that wires the pattern into your own SDP pipeline. | Stable | [README](assets/sdp-expectation-notifications/README.md) | +| `sdp-expectation-notifications` | Per-expectation data-quality notification for Lakeflow SDP as a validated pair: a native event hook notifies the moment a WARN expectation result is logged (fast, best-effort by platform design, throttled to at most one notification per expectation per window via a two-layer time-aware de-dup with optional UC Volume marker state), and one DABs-managed Alert v2 sweeps the published pipeline event log over a past-time window on a schedule (guaranteed, one email per state change). Webhook payloads for Slack, Teams, or generic receivers, with secret-scope URL resolution. Demo pipeline on `samples.nyctaxi.trips` fires both paths on the first run. Ships a companion agent skill (`SKILL.md`) that wires the pattern into your own SDP pipelines. | Stable | [README](assets/sdp-expectation-notifications/README.md) | ## What an asset is not diff --git a/CHANGELOG.md b/CHANGELOG.md index 47171da..026bb30 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Changed +- **Asset `sdp-expectation-notifications`**: notification hygiene revision, grounded in a two-day live investigation (2026-07-17/18, serverless SDP on a non-development-mode target). The v1.11.0 hook notified on every qualifying `flow_progress` event, and the platform emits expectation counts once per microbatch, so one failing expectation notified once per file on a multi-file backlog (measured: 4 notifications for one expectation in one update, counts summing exactly to the single-batch total), re-notified everything on full refresh, and fired within ~70s of every arriving file on a continuous pipeline. Wired to a real channel, that floods. + - **The hook now throttles: at most one notification per (pipeline, dataset, expectation) per `dq_notify.throttle_seconds` (default 3600).** Two time-aware layers, both fail-open (any state error notifies rather than stays silent): an in-memory last-notified map (always on; timestamp-based, not a seen-set, because continuous pipelines keep one Python process alive and a notify-once set would go silent forever; validated live NOTIFY -> SUPPRESS in window -> NOTIFY after window on a continuous Auto Loader pipeline) and optional durable marker files under `/dq_notify_state//__.json` in an existing UC Volume (one human-readable JSON per expectation, overwritten on notify, never growing; validated across full refreshes and process restarts: 12 raw events across two consecutive full refreshes reduced to 2 notifications). State scopes per pipeline automatically from the event's own `origin` (`pipeline_name`/`pipeline_id`, observed present in hook events), so the same code serves any number of pipelines and cannot cross-suppress (validated with two pipelines sharing one state root). The asset deliberately ships no volume resource: state that describes operations should outlive any one bundle, so a new `state_volume_path` prompt (placeholder default = in-memory only) points under a volume the user already owns. `dq_notify.throttle_seconds: "0"` restores the old per-microbatch behavior. + - **Webhook channels and secrets.** `dq_notify.channel_format` selects the hook's payload: `slack` (validated end to end: secret scope -> `dbutils.secrets.get` -> throttled hook -> channel message, HTTP 200), `teams` (documented Workflows Adaptive Card envelope, marked doc-confirmed, not live-tested), or `generic` (plain JSON with the violation fields). Webhook URLs are credentials, so the shipped path resolves them from a secret scope via `dbutils.secrets.get` at module level, which works in serverless SDP pipeline Python (validated live), with two traps documented from observation: `{{secrets/scope/key}}` in pipeline configuration is NOT interpolated (the literal string arrives), and a missing scope raises `IllegalArgumentException`, which the code catches to degrade to print-only without touching the pipeline (validated). Plain `dq_notify.webhook_url` remains as the labeled demo-only fallback. + - **Backstop alert defaults and documentation.** `notify_on_ok: true` is the new shipped default: Alerts v2 notifies exactly once per state transition (measured: 9 consecutive TRIGGERED evaluations, 1 email), which means persistent failures mask new ones until recovery, and the single OK email (measured: exactly one) closes that loop. `retrigger_seconds` ships commented with measured semantics (re-send at the first evaluation past each window; a deliberate escalation knob, suggested 86400 for critical pipelines) and a commented `destination_id` line documents Slack/Teams/PagerDuty delivery via workspace notification destinations. New measured operational note: a broken sweep query flips the alert to ERROR and emails on EVERY evaluation until fixed (5 emails in 4 minutes observed), so the backstop cannot die silently but its failure mode is itself a flood. + - **README corrections and additions from measurement:** `hook_progress` records ENABLED, FAILED, and DISABLED transitions (v1.11.0 claimed enable-state only); the post-update teardown grace budget is seconds (a ~1s/event hook drained 5 queued events with zero loss; a 20s/event hook lost 6 of 6 under natural teardown), closing two formerly open questions; the in-hook `spark.sql()` write door is definitively shut (`INSERT INTO` fails with `UNSUPPORTED_SPARK_SQL_COMMAND`; allowlist: SELECT, DESCRIBE, SHOW variants, USE) while UC Volume file I/O from a hook works (create/overwrite/read/exists/listdir/stat; append fails with `OSError Errno 29`), which is exactly the surface the durable throttle uses; an update that processes no new rows emits no expectation events (re-running on static data never re-notified); continuous pipelines auto-start on `bundle deploy`. The skill grows a third non-negotiable rule (the throttle stays time-aware and fail-open), the adaptation reference covers throttle/window/state/channel/secret decisions and multi-pipeline rollout, and self-verify gains a marker-file check (suppression proof: a rerun inside the window must not advance `last_notified_at`). + - Tests extended accordingly: the throttle decision logic is exercised offline by exec-ing the pure helpers from the installed source against a temp state dir (suppress within window, durable across simulated process restarts, per-pipeline scoping, marker overwrite, `0` disables, empty state dir writes nothing), payload formats and guarded secret resolution are asserted, and the no-`spark.sql()`-calls AST guard still holds over the reworked source. + ## [1.11.0] - 2026-07-11 ### Added diff --git a/assets/sdp-expectation-notifications/README.md b/assets/sdp-expectation-notifications/README.md index d3c1f01..604c680 100644 --- a/assets/sdp-expectation-notifications/README.md +++ b/assets/sdp-expectation-notifications/README.md @@ -1,6 +1,6 @@ # sdp-expectation-notifications -Per-expectation data-quality notification for Lakeflow Spark Declarative Pipelines (SDP), as a pair: a native **event hook** fires the moment a WARN expectation result is logged (the fast path), and exactly one DABs-managed **Alert v2** sweeps the published pipeline event log on a schedule (the guaranteed path). The pair exists because the platform explicitly does not guarantee hook delivery; each half covers the other's gap. +Per-expectation data-quality notification for Lakeflow Spark Declarative Pipelines (SDP), as a pair: a scheduled DABs-managed **Alert v2** sweeps the published pipeline event log (the guaranteed path, throttled by the platform to one email per state change), and a native **event hook** posts the moment a WARN expectation result is logged (the fast path, throttled by this asset to one notification per expectation per window). The pair exists because the platform explicitly does not guarantee hook delivery; each half covers the other's gap. ## Install @@ -9,14 +9,14 @@ databricks bundle init https://github.com/vmariiechko/databricks-bundle-template --template-dir assets/sdp-expectation-notifications ``` -You will be prompted for `target_dir`, `pipeline_resource_key`, `pipeline_name`, `catalog`, `schema`, `warehouse_id`, `notification_email`, and `skill_dir` (all with safe defaults; the warehouse id and email default to placeholders you replace before deploying). The asset installs the pipeline source (with the event hook inside), event-log parsing queries, the DABs pipeline resource, the backstop alert resource, an in-bundle usage doc, and a companion agent skill. +You will be prompted for `target_dir`, `pipeline_resource_key`, `pipeline_name`, `catalog`, `schema`, `warehouse_id`, `notification_email`, `state_volume_path`, and `skill_dir` (all with safe defaults; the warehouse id, email, and volume path default to placeholders you replace before deploying). The asset installs the pipeline source (with the throttled event hook inside), event-log parsing queries, the DABs pipeline resource, the backstop alert resource, an in-bundle usage doc, and a companion agent skill. ## Two doors This is a dual-door asset: -- **Door 1, the reference pipeline.** A self-contained demo on the public `samples.nyctaxi.trips` dataset: one streaming table with a deliberately tight WARN expectation, the notifier hook, and the backstop alert. Deploy it, run it, and watch both notification paths fire on the first update. This README documents it. -- **Door 2, the companion agent skill** at `/skills/sdp-expectation-notifications/SKILL.md`. Point a coding agent (Claude Code, Codex, or Genie Code on Databricks) at it and it wires the hook-plus-backstop pattern into your own SDP pipeline: it reads your pipeline source and expectations, adds the hook and the event-log publication, adapts the backstop alert to your event log and recipients, and verifies the result with read-only queries. It proposes and confirms; it is not an auto-rewriter. +- **Door 1, the reference pipeline.** A self-contained demo on the public `samples.nyctaxi.trips` dataset: one streaming table with a deliberately tight WARN expectation, the throttled notifier hook, and the backstop alert. Deploy it, run it, and watch both notification paths fire on the first update. This README documents it. +- **Door 2, the companion agent skill** at `/skills/sdp-expectation-notifications/SKILL.md`. Point a coding agent (Claude Code, Codex, or Genie Code on Databricks) at it and it wires the pattern into your own SDP pipelines: the same hook code runs unchanged in any number of pipelines because it scopes its throttle state by the pipeline identity carried in every event. It proposes and confirms; it is not an auto-rewriter. ## Architecture @@ -26,62 +26,91 @@ samples.nyctaxi.trips ▼ monitored_trips @dp.expect (WARN, fails by design on the healthy sample) │ - ├─► event hook (in-pipeline) fires at the moment the flow_progress - │ print + optional webhook event is logged; best-effort + ├─► event hook (in-pipeline) fires when the flow_progress event + │ two-layer time-aware throttle is logged; best-effort delivery; + │ print + optional webhook at most 1 notification per + │ (Slack / Teams / generic) expectation per throttle window + │ │ + │ └─ state markers (optional): /dq_notify_state/ + │ /__.json │ └─► published event log (UC table: dq_notifications_event_log) ▲ - └── backstop Alert v2 scheduled sweep over a trailing - email subscription 1-day window; guaranteed evaluation + └── backstop Alert v2 scheduled sweep over a trailing + email / destinations 1-day window; guaranteed + 1 email per state change evaluation; notify_on_ok closes + the loop on recovery ``` -The demo WARN rule `trip_distance < 5` is a tripwire chosen to fail on a healthy public dataset (plenty of NYC trips exceed 5 miles); it is not presented as a sensible quality rule. WARN keeps every row flowing while still recording per-expectation passed/failed counts in the event log, which is exactly the payload both notification paths consume. +The demo WARN rule `trip_distance < 5` is a tripwire chosen to fail on a healthy public dataset; it is not presented as a sensible quality rule. WARN keeps every row flowing while still recording per-expectation passed/failed counts in the event log, which is exactly the payload both notification paths consume. -## Why a pair and not just the hook +## Why the hook is throttled (measured, not assumed) -The event-hooks documentation states the limitation plainly: SDP "attempts" to run each hook on every emitted event, waits "a fixed, non-configurable period" before terminating the pipeline's compute, and "there is no guarantee that all hooks will be triggered on all events before compute termination." The events this pattern cares about (`flow_progress` with expectation counts) are emitted as flows complete, late in the update, adjacent to that termination window. So the hook is the right fast path and the wrong only path. +The v1.11.0 hook notified on every qualifying `flow_progress` event. Measured live (2026-07-17, serverless, non-development-mode target): expectation counts arrive **once per microbatch**, so one failing expectation on a 4-file backlog notified 4 times in a single update (failed counts 708, 682, 636, 686, summing exactly to the single-batch 2,712); a full refresh re-notified everything; a continuous pipeline emitted a fresh event within about 70 seconds of every arriving file. Wire that to a real channel on a failing scheduled pipeline and it floods. -The backstop is one scheduled Alert v2 over the published event log. A scheduled SQL query either runs or visibly fails, so it provides the delivery guarantee the hook lacks. Its query aggregates over a past-time window (trailing 1 day), not just the latest row: a latest-row read silently misses violations whenever alert cadence differs from pipeline cadence. If you change the alert's cron, change the `INTERVAL` in its query to stay at least as long as the cadence. +The revised hook throttles at most one notification per (pipeline, dataset, expectation) per `dq_notify.throttle_seconds` (default 3600), in two time-aware layers: -This replaces the fleet-of-scheduled-alerts approach (one alert per expectation, plus jobs to trigger them, plus a deployment script) with one hook function and one YAML block. +- **In-memory (always on):** kills repeats within one Python process: later microbatches of the same update, and a continuous pipeline's whole run. Time-aware on purpose: continuous compute lives for days, so a notify-once-per-process set would go silent forever. Validated on a continuous pipeline: NOTIFY, SUPPRESS at +89s (inside a 180s test window), NOTIFY again at +239s. +- **Durable markers (optional):** small human-readable JSON files under `/dq_notify_state//__.json` in a UC Volume you already own, carrying the throttle across updates and restarts. One file per expectation, overwritten on notify, never growing; reading one answers "when did this last notify and how bad was it". Validated across full refreshes and process restarts: two consecutive full refreshes produced 12 raw events and 2 notifications, and a rerun minutes later produced 0. +- **Multi-pipeline safe by construction:** state is scoped per pipeline (identity read from the event's own `origin`, no configuration), so pipelines cannot cross-suppress; validated with two pipelines sharing one state root. +- **Fail-open:** any state error means notify, never stay silent. Empty `dq_notify.state_dir` (the placeholder default) runs in-memory only. `dq_notify.throttle_seconds: "0"` restores the old per-microbatch behavior. + +An update that processes no new rows emits no expectation events at all (measured), so re-running a pipeline on static data does not re-notify even unthrottled; the flood mechanisms are new data, full refreshes, retries, and continuous mode. + +## Notification channels + +The two lanes get channels in two different ways: + +- **Backstop (email and beyond, zero code):** the alert subscribes an email by default. For Slack, MS Teams, PagerDuty, or a generic webhook, create a workspace notification destination (admin settings) and reference it via the commented `destination_id` line in the alert YAML. Platform feature, nothing to implement. +- **Hook (webhook formats):** `dq_notify.channel_format` selects the payload: `slack` (`{"text": ...}`, validated end to end: secret scope to channel message, HTTP 200), `teams` (Workflows Adaptive Card envelope, documented format, not live-tested by this asset), or `generic` (plain JSON with the violation fields). Route hook posts to chat-style channels that tolerate an occasional repeat; email belongs to the backstop, and a hook cannot send email anyway (no email surface exists inside a hook). + +**Webhook URLs are credentials** (a Slack incoming webhook grants posting rights). Real setups put the URL in a secret scope and set `dq_notify.webhook_secret_scope`/`dq_notify.webhook_secret_key`; the pipeline resolves it via `dbutils.secrets.get` at startup, which works at module level in serverless SDP Python (validated live), and degrades to print-only if the scope is missing (also validated). Two things that do NOT work, so you do not have to rediscover them: `{{secrets/scope/key}}` in pipeline configuration is not interpolated (the literal string arrives), and the plain `dq_notify.webhook_url` config is a demo-only convenience. ## The rules of the hook (learned the honest way) -These were established on a live serverless SDP run (validated 2026-07-11, Databricks CLI v0.297.2), not just read from docs: +Established on live serverless SDP runs (2026-07-11 and 2026-07-17/18, CLI v0.297.2): -- **Do not write Delta from inside a hook.** SDP restricts `spark.sql()` in pipeline Python to a read-oriented command allowlist: `CREATE TABLE` from a hook fails with `UNSUPPORTED_SPARK_SQL_COMMAND` (the error lists the allowlist: SELECT, DESCRIBE, SHOW variants, USE), and the SDP-patched `spark.sql()` wrapper rejects the standard PySpark parameterized-query `args=` kwarg outright. The supported notification surface inside a hook is `print` and HTTP calls such as `requests`, which is also what the official docs examples use. -- **Keep the hook fast.** Hooks run serialized, one at a time; a slow call delays every other hook and widens the window in which queued events are lost at compute termination. In a probe run with a deliberately slow (2-second) hook, forcing compute teardown mid-queue lost 7 of 18 STABLE events (39%) for that hook. The loss mechanism is real, not just quotable. -- **`hook_progress` tells you less than it sounds like.** It records only enable/disable state (`{"hook_name": ..., "state": "ENABLED"}`), one row per hook per update, at registration time. It does not record which events a hook processed, counts, timings, or failures. Per-invocation output (the printed notifications) exists only in the pipeline compute's driver log, which has no CLI or SQL surface. -- **`max_allowable_consecutive_failures` is a tradeoff, not a setting to forget.** A finite value silently disables a flaky hook until the next pipeline restart; `None` (this asset's choice) lets a permanently broken hook fail forever. That is why the optional webhook call catches and prints its own failures instead of raising. -- **`mode: development` masks the teardown behavior.** A DABs target in development mode propagates `development: true` to the pipeline, which keeps pipeline compute warm across updates instead of tearing it down after the grace period. If you are trying to observe the hook delivery gap, a dev-mode target will hide it. -- **Inside the hook, `event["details"]` arrives as a native Python dict.** The JSON-string form only appears when querying the persisted event log table. The shipped `_event_details` helper keeps a defensive string branch anyway; it costs nothing. +- **Do not write Delta from inside a hook.** `spark.sql()` in SDP pipeline Python is restricted to a read allowlist: `INSERT INTO` fails with `UNSUPPORTED_SPARK_SQL_COMMAND` (the error lists the allowlist: SELECT, DESCRIBE, SHOW variants, USE), `CREATE TABLE` fails the same way, and the SDP-patched `spark.sql()` rejects the parameterized-query `args=` kwarg. All three write paths are closed; this is definitive. +- **Durable state goes through UC Volume file I/O instead.** Plain Python `open()` against `/Volumes/...` works from inside a hook: create, overwrite, read, `os.path.exists`, `os.listdir`, `os.stat`, `os.makedirs` all validated live. The one gap: append to an existing file fails (`OSError [Errno 29] Illegal seek`), which is why the markers are overwrite-only. +- **Reads are fine.** `spark.sql("SELECT ...").collect()` works reliably from a hook (validated across many invocations), so a hook can consult a small gate table if you need a manual mute switch. +- **Keep the hook fast; the grace budget is seconds.** Hooks run serialized, and after an update completes the platform terminates compute within roughly 20 seconds. A hook needing about 1s per event drained 5 queued events with zero loss; a hook needing 20s per event lost **6 of 6** notifications under normal, non-forced teardown. No retries, tight webhook timeout (5s), marker I/O only (~100ms). +- **`hook_progress` records ENABLED, FAILED, and DISABLED transitions** (correcting v1.11.0, which claimed enable-state only). Hook health is event-log-observable and alertable. It still does not record per-invocation output; the printed notifications exist only in the pipeline compute's driver log, which has no CLI or SQL surface. +- **`max_allowable_consecutive_failures` is a tradeoff.** A finite value disables a flaky hook until the next restart (visible as `DISABLED` in `hook_progress`, but notifications silently stop); `None` (this asset's choice) lets a broken hook fail forever, which is why every failure path in the shipped hook prints instead of raising. +- **`mode: development` masks the teardown behavior.** A DABs development-mode target keeps pipeline compute warm across updates, hiding both the delivery gap and the fresh-process-per-update behavior (in non-dev mode each update gets a new Python process; measured via per-process ids). +- **Inside the hook, `event["details"]` arrives as a native Python dict**, and `event["origin"]` carries `pipeline_name` and `pipeline_id` (the basis for zero-config multi-pipeline state scoping). -## Validation results +## The backstop's notification behavior (measured) -The pattern was validated end to end on a live workspace (serverless SDP, Free Edition, 2026-07-11, CLI v0.297.2) against `samples.nyctaxi.trips`: +- **Default: exactly one email per state transition.** Nine consecutive TRIGGERED evaluations on an every-minute cron produced exactly one email; the state change is what notifies, not the evaluation. The masking consequence: while the alert stays TRIGGERED, new failures inside the sweep window send nothing, which is why `notify_on_ok: true` ships as the default: the single recovery email (also measured: exactly one) closes the loop and re-arms attention. +- **`retrigger_seconds` is deliberate re-nagging:** re-sends at the first evaluation past each window (measured with 120s: gaps of 120s, 180s, 179s, aligned to evaluation ticks). Off by default; the commented block suggests 86400 as a daily "still broken" escalation for critical pipelines. +- **ERROR does not throttle.** If the sweep query itself fails, the alert flips to ERROR and emails on **every evaluation** until fixed (measured: 5 emails in 4 minutes). The backstop cannot die silently; treat an `(ERROR)` email as a page. +- The sweep query aggregates over a past-time window (trailing 1 day), not just the latest row; keep the window at least as long as the cadence if you change the cron. + +## Validation results -- **Hooks run on serverless SDP pipelines.** Confirmed via `hook_progress` rows in the published event log on every update; the docs are silent on this, so it is a lived finding with the date stamped. -- **The hook fires per-expectation with exact counts.** The driver log line `EXPECTATION VIOLATION | dataset=...monitored_trips | expectation=demo_short_trip | failed=3003 | passed=18929` matched the event log's `flow_progress` counts for the same `update_id` exactly (3,003 failed, 18,929 passed of 21,932 sample rows), reproduced across multiple full-refresh updates. -- **The backstop alert delivers end to end.** The sweep query, the scheduled execution (confirmed via query history), the alert state (`TRIGGERED` via `alerts-v2 get-alert`), and the actual email delivery (from `noreply@databricks.com`, evaluated value matching the independently computed sum exactly) all agreed. -- **`comparison_operator: GREATER_THAN` validates and deploys** in the Alerts v2 bundle resource, and **the `event_log` block is accepted** by the DABs pipeline resource as written here. -- **Alerts v2 bundle resources update in place on rename**; changing `display_name` and redeploying kept the same alert id with no orphaned duplicate. +v1.11.0's pattern was validated end to end on 2026-07-11 (hook fires with exact counts matching the event log; backstop delivers email; resources deploy and update in place). The v1.12 revision was validated on 2026-07-17/18 (Free Edition serverless, CLI v0.297.2, non-development-mode target): -The shipped asset itself (installed via `bundle init --template-dir`, wrapped in a fresh bundle) was then re-validated end to end on the same workspace and date: install produced exactly the expected files, `bundle validate` passed first try, the pipeline update completed on serverless with the hook registered (`hook_progress: ENABLED`), the driver log carried the exact `failed=3003 | passed=18929` notification line, and the backstop alert evaluated on its cron to `TRIGGERED` with the email delivered (evaluated value 3003 matching the sweep query run independently). +- Per-microbatch multiplication measured (4 events for one expectation on a 4-file backlog; sums match single-batch exactly: 708+682+636+686 = 2,712). +- Two-layer throttle: 12 raw events across two consecutive full refreshes reduced to 2 notifications; cross-update suppression through marker files after a process restart; multi-pipeline isolation with two pipelines sharing one state root. +- Continuous mode with Auto Loader: notify, suppress inside the window, re-notify after it, in one long-lived process; continuous pipelines auto-start on `bundle deploy`. +- Slack end to end: secret scope, `dbutils.secrets.get` at module level, throttled hook, two channel messages with HTTP 200 and counts matching the hook records exactly; missing secret scope degraded to print-only with the pipeline unaffected. +- Alerts v2: one email per state transition (9 TRIGGERED evaluations, 1 email), `retrigger_seconds` re-send alignment, exactly one `notify_on_ok` recovery email, ERROR emailing every evaluation. ## Honest limits and open questions -- Hook delivery is best-effort by design; this asset treats that as an architectural input (hence the backstop), not a problem it solves. -- The backstop inherits scheduled-scan limits at miniature scale: latency bounded by its cron cadence, and a SQL warehouse in the loop. The `monitoring-sql-warehouse` asset (2X-Small serverless, `auto_stop_mins: 1`) exists for exactly this workload shape. +- Hook delivery is best-effort by design; the throttle makes the fast lane quiet, not reliable. The backstop remains the guaranteed path. +- The two lanes de-duplicate independently and share no state: one violation can produce one hook message and one backstop email. Different audiences, different guarantees; by design. - The hook cannot notice "the pipeline has not run at all": no update, no events, no hook. Pair with a freshness check (for example Unity Catalog data quality monitoring's anomaly detection) if that failure mode matters. -- Still open, not settled by the validation run: whether the fixed grace period drops queued hook events automatically under default (non-development) pipeline teardown (the observed loss was under a forced stop); the exact size of that grace period; whether a literal non-parameterized `INSERT` from a hook would pass the `spark.sql()` allowlist (untested; two other write patterns failed for two different reasons); and the full `comparison_operator` enum beyond the confirmed `EQUAL` and `GREATER_THAN`. +- The backstop inherits scheduled-scan limits: latency bounded by its cron cadence, and a SQL warehouse in the loop. The `monitoring-sql-warehouse` asset (2X-Small serverless, `auto_stop_mins: 1`) exists for exactly this workload shape. +- Still open: the Teams payload ships doc-confirmed, not live-tested; concurrent marker writes from two simultaneously running pipelines are untested (the per-pipeline folders give them no shared file to race on, but Free Edition cannot run two updates at once to prove it); the exact numeric grace period (bounded to roughly 8 to 20-plus seconds of post-update hook budget by measurement). ## Inspecting the results -`/event_log_queries.sql` ships three queries against the published event log table (`..dq_notifications_event_log`): per-expectation passed/failed counts for the latest update (the numbers the hook prints), the `hook_progress` registration states, and the backstop alert's exact sweep query so you can see the value the alert compares against its threshold. +`/event_log_queries.sql` ships queries against the published event log table (`..dq_notifications_event_log`): per-expectation passed/failed counts for the latest update (the numbers the hook prints), the `hook_progress` states, and the backstop alert's exact sweep query. The throttle's durable state, when enabled, is directly browsable: `/dq_notify_state//`, one JSON per expectation with the last-notified time and counts. ## Tests -Repo-level tests (`tests/assets/test_sdp_expectation_notifications.py`) verify the asset installs the expected files, the pipeline source parses and contains the hook wired to the supported surface (and no `spark.sql` calls anywhere in it), the resource YAMLs are valid (event-log publication, alert query shape, past-time-window aggregation, threshold, subscription, schedule), and custom prompt values flow through to filenames, resource keys, and paths. End-to-end firing is verified live on a workspace (see above). +Repo-level tests (`tests/assets/test_sdp_expectation_notifications.py`) verify the asset installs the expected files, the pipeline source parses and contains the hook wired to the supported surface (and no `spark.sql` write calls anywhere in it), the throttle and payload functions behave (pure-function tests, offline), the resource YAMLs are valid, and custom prompt values flow through to filenames, resource keys, and paths. End-to-end firing is verified live on a workspace (see above). ## What this asset is @@ -96,3 +125,4 @@ It demonstrates one notification pattern with its tradeoffs documented, not a ge 3. [Monitor pipelines with the event log](https://docs.databricks.com/aws/en/ldp/monitor-event-logs) 4. [`event_log` table-valued function](https://docs.databricks.com/aws/en/sql/language-manual/functions/event_log) 5. [DABs alert resource (Alerts v2)](https://docs.databricks.com/aws/en/dev-tools/bundles/resources) +6. [Notification destinations](https://docs.databricks.com/aws/en/admin/workspace-settings/notification-destinations) diff --git a/assets/sdp-expectation-notifications/databricks_template_schema.json b/assets/sdp-expectation-notifications/databricks_template_schema.json index 71598c1..496d09a 100644 --- a/assets/sdp-expectation-notifications/databricks_template_schema.json +++ b/assets/sdp-expectation-notifications/databricks_template_schema.json @@ -1,5 +1,5 @@ { - "welcome_message": "\nSDP Expectation Notifications: Asset Installer\n\nInstalls two things that ship together:\n 1. A reference Lakeflow Spark Declarative Pipeline demonstrating\n per-expectation data-quality notification: a native event hook fires the\n moment a WARN expectation result is logged, on the public\n samples.nyctaxi.trips dataset (the worked example).\n 2. A companion agent skill that instructs a coding agent to wire the same\n hook-plus-backstop pattern into YOUR own SDP pipeline.\n\nThe pattern is a pair, not a single mechanism: event hook delivery is\nexplicitly best-effort (the platform does not guarantee every hook runs\nbefore pipeline compute terminates), so the asset also installs exactly one\nDABs-managed Alert v2 that sweeps the published pipeline event log on a\nschedule as the guaranteed path. Hook = fast, alert = guaranteed.\n\nThe demo WARN expectation is a deliberate tripwire chosen to fail on the\nhealthy public sample, so the hook fires and the backstop alert triggers on\nthe first run with no extra setup.\n\nLet's pick the install settings...\n", + "welcome_message": "\nSDP Expectation Notifications: Asset Installer\n\nInstalls two things that ship together:\n 1. A reference Lakeflow Spark Declarative Pipeline demonstrating\n per-expectation data-quality notification: a native event hook fires the\n moment a WARN expectation result is logged, on the public\n samples.nyctaxi.trips dataset (the worked example).\n 2. A companion agent skill that instructs a coding agent to wire the same\n hook-plus-backstop pattern into YOUR own SDP pipeline.\n\nThe pattern is a pair, not a single mechanism: event hook delivery is\nexplicitly best-effort (the platform does not guarantee every hook runs\nbefore pipeline compute terminates), so the asset also installs exactly one\nDABs-managed Alert v2 that sweeps the published pipeline event log on a\nschedule as the guaranteed path. Hook = fast, alert = guaranteed.\n\nThe hook is throttled by design: without it, one failing expectation\nnotifies once per microbatch (measured live), which floods on multi-batch\nand continuous pipelines. A time-aware two-layer throttle (in-memory always,\ndurable volume markers optionally) caps it at one notification per\nexpectation per window.\n\nThe demo WARN expectation is a deliberate tripwire chosen to fail on the\nhealthy public sample, so the hook fires and the backstop alert triggers on\nthe first run with no extra setup.\n\nLet's pick the install settings...\n", "properties": { "target_dir": { @@ -65,17 +65,26 @@ "pattern_match_failure_message": "Notification email cannot be empty or contain whitespace." }, + "state_volume_path": { + "type": "string", + "default": "VOLUME_PATH_PLACEHOLDER", + "description": "\n========================================\n\nPath under an EXISTING Unity Catalog volume where the hook keeps its\nnotification throttle state, for example /Volumes/main/dq_notifications/state.\nMarker files land in /dq_notify_state//, one small JSON per\nmonitored expectation, overwritten in place (the folder never grows).\n\nThe asset deliberately does not create or manage a volume; state that\ndescribes operations should outlive any one bundle.\n\nKeep the default placeholder to run with the in-memory throttle only (still\nremoves per-microbatch repeats within an update); set the path later in\nresources/.pipeline.yml (dq_notify.state_dir) to add\ncross-update throttling.\nstate_volume_path", + "order": 8, + "pattern": "^\\S+$", + "pattern_match_failure_message": "Volume path cannot be empty or contain whitespace." + }, + "skill_dir": { "type": "string", "default": ".agents", "description": "\n========================================\n\nTarget directory for the companion agent skill (relative to your bundle root).\nThe skill installs at /skills/sdp-expectation-notifications/ and\ninstructs a coding agent to wire the hook-plus-backstop pattern into your own\nSDP pipeline.\n\nDefault `.agents` is vendor-neutral. If you use one agent and want\nauto-discovery, override with its native folder:\n - Claude Code: .claude\n - Codex: .codex\n - Cursor: .cursor\n - Gemini CLI: .gemini\n\nPick a path whose `skills/sdp-expectation-notifications/` subfolder doesn't\nalready exist in your project.\nskill_dir", - "order": 8, + "order": 9, "pattern": "^[A-Za-z0-9_.][A-Za-z0-9_./-]*$", "pattern_match_failure_message": "Skill directory must start with a letter, number, underscore, or dot and contain only letters, numbers, underscores, slashes, hyphens, or dots." } }, - "success_message": "\n========================================\n\nExpectation notifications installed (reference pipeline + backstop alert + companion skill):\n - Pipeline source: {{.target_dir}}/expectation_notifications_pipeline.py (event hook inside)\n - Event log SQL: {{.target_dir}}/event_log_queries.sql\n - Pipeline: resources/{{.pipeline_resource_key}}.pipeline.yml\n - Backstop alert: resources/{{.pipeline_resource_key}}_backstop.alert.yml\n - Usage doc: docs/sdp-expectation-notifications/README.md\n - Agent skill: {{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md\n\nRun the reference as-is:\n\n1. Confirm your `databricks.yml` includes the resource files. Most bundles\n already include `resources/*.yml` by default; verify:\n include:\n - resources/*.yml\n\n2. If you kept a placeholder for the warehouse id or the notification email,\n replace it in resources/{{.pipeline_resource_key}}_backstop.alert.yml.\n\n3. Validate, deploy, run:\n databricks bundle validate -t \n databricks bundle deploy -t \n databricks bundle run {{.pipeline_resource_key}} -t \n\n4. The demo WARN expectation fails on the healthy public sample by design, so\n the hook prints its notification during the first update (visible in the\n pipeline compute's driver log) and the backstop alert triggers on its next\n scheduled sweep (daily 06:00 UTC by default; open the alert in the UI and\n run it manually to see it immediately).\n\nThe demo table lands at {{.catalog}}.{{.schema}}.monitored_trips; the event\nlog publishes to {{.catalog}}.{{.schema}}.dq_notifications_event_log. Inspect\nper-expectation counts with {{.target_dir}}/event_log_queries.sql. Design\nnotes and the honest caveats (hook delivery is best-effort; that is why the\nbackstop exists) live in 'docs/sdp-expectation-notifications/README.md'.\n\nWire the pattern into YOUR pipeline with the companion skill:\n\n Point your coding agent at the skill and ask it to add the hook-plus-backstop\n pattern to your own SDP pipeline. Wire it the way your agent expects:\n - Claude Code: auto-discovered if {{.skill_dir}} is `.claude`; otherwise\n add to CLAUDE.md: 'Use the skill at\n {{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md'.\n - Codex: reference that SKILL.md path from AGENTS.md.\n - Other / on Databricks (Genie): point the agent at the SKILL.md path.\n\n A ready-to-paste kickoff prompt is in 'docs/sdp-expectation-notifications/README.md'.\n", + "success_message": "\n========================================\n\nExpectation notifications installed (reference pipeline + backstop alert + companion skill):\n - Pipeline source: {{.target_dir}}/expectation_notifications_pipeline.py (event hook inside)\n - Event log SQL: {{.target_dir}}/event_log_queries.sql\n - Pipeline: resources/{{.pipeline_resource_key}}.pipeline.yml\n - Backstop alert: resources/{{.pipeline_resource_key}}_backstop.alert.yml\n - Usage doc: docs/sdp-expectation-notifications/README.md\n - Agent skill: {{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md\n\nRun the reference as-is:\n\n1. Confirm your `databricks.yml` includes the resource files. Most bundles\n already include `resources/*.yml` by default; verify:\n include:\n - resources/*.yml\n\n2. If you kept a placeholder for the warehouse id or the notification email,\n replace it in resources/{{.pipeline_resource_key}}_backstop.alert.yml.\n If you kept the volume-path placeholder, the throttle runs in-memory only;\n set dq_notify.state_dir in resources/{{.pipeline_resource_key}}.pipeline.yml\n later to add cross-update throttling. For a real webhook, put the URL in a\n secret scope (it is a credential):\n databricks secrets create-scope dq-notifications\n databricks secrets put-secret dq-notifications slack_webhook_url --string-value \"\"\n then set dq_notify.webhook_secret_scope: dq-notifications in the pipeline\n configuration.\n\n3. Validate, deploy, run:\n databricks bundle validate -t \n databricks bundle deploy -t \n databricks bundle run {{.pipeline_resource_key}} -t \n\n4. The demo WARN expectation fails on the healthy public sample by design, so\n the hook prints its notification during the first update (visible in the\n pipeline compute's driver log) and the backstop alert triggers on its next\n scheduled sweep (daily 06:00 UTC by default; open the alert in the UI and\n run it manually to see it immediately).\n\nThe demo table lands at {{.catalog}}.{{.schema}}.monitored_trips; the event\nlog publishes to {{.catalog}}.{{.schema}}.dq_notifications_event_log. Inspect\nper-expectation counts with {{.target_dir}}/event_log_queries.sql. Design\nnotes and the honest caveats (hook delivery is best-effort; that is why the\nbackstop exists) live in 'docs/sdp-expectation-notifications/README.md'.\n\nWire the pattern into YOUR pipeline with the companion skill:\n\n Point your coding agent at the skill and ask it to add the hook-plus-backstop\n pattern to your own SDP pipeline. Wire it the way your agent expects:\n - Claude Code: auto-discovered if {{.skill_dir}} is `.claude`; otherwise\n add to CLAUDE.md: 'Use the skill at\n {{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md'.\n - Codex: reference that SKILL.md path from AGENTS.md.\n - Other / on Databricks (Genie): point the agent at the SKILL.md path.\n\n A ready-to-paste kickoff prompt is in 'docs/sdp-expectation-notifications/README.md'.\n", "min_databricks_cli_version": "v0.296.0", "version": 1 diff --git a/assets/sdp-expectation-notifications/template/docs/sdp-expectation-notifications/README.md b/assets/sdp-expectation-notifications/template/docs/sdp-expectation-notifications/README.md index e08fc3b..4e31c47 100644 --- a/assets/sdp-expectation-notifications/template/docs/sdp-expectation-notifications/README.md +++ b/assets/sdp-expectation-notifications/template/docs/sdp-expectation-notifications/README.md @@ -1,6 +1,6 @@ # SDP Expectation Notifications (in-bundle usage) -This asset installs per-expectation data-quality notification for a Lakeflow Spark Declarative Pipeline as a pair: a native event hook fires the moment a WARN expectation result is logged (fast, best-effort), and one DABs-managed Alert v2 sweeps the published pipeline event log on a schedule (guaranteed). The demo pipeline runs on the public `samples.nyctaxi.trips` dataset with a deliberately tight WARN expectation, so both paths fire on the first run. +This asset installs per-expectation data-quality notification for a Lakeflow Spark Declarative Pipeline as a pair: a native event hook fires the moment a WARN expectation result is logged (fast, best-effort, throttled to at most one notification per expectation per window), and one DABs-managed Alert v2 sweeps the published pipeline event log on a schedule (guaranteed, one email per state change). The demo pipeline runs on the public `samples.nyctaxi.trips` dataset with a deliberately tight WARN expectation, so both paths fire on the first run. The full design rationale, the validated facts, and the honest caveats live in the asset README in the [databricks-bundle-template repo](https://github.com/vmariiechko/databricks-bundle-template/tree/main/assets/sdp-expectation-notifications). This file covers what you do after install. @@ -17,7 +17,7 @@ The full design rationale, the validated facts, and the honest caveats live in t ## Before you deploy -1. **Placeholders.** If you kept the defaults for `warehouse_id` or `notification_email` at install, `resources/_backstop.alert.yml` contains `WAREHOUSE_ID_PLACEHOLDER` or `EMAIL_PLACEHOLDER`. Replace them: find a warehouse id with `databricks warehouses list`, and use an email that belongs to a workspace user. If you installed the `monitoring-sql-warehouse` asset from the same repo, you can reference its warehouse instead: `${resources.sql_warehouses.monitoring_sql_warehouse.id}`. +1. **Placeholders.** If you kept the defaults for `warehouse_id` or `notification_email` at install, `resources/_backstop.alert.yml` contains `WAREHOUSE_ID_PLACEHOLDER` or `EMAIL_PLACEHOLDER`. Replace them: find a warehouse id with `databricks warehouses list`, and use an email that belongs to a workspace user. If you installed the `monitoring-sql-warehouse` asset from the same repo, you can reference its warehouse instead: `${resources.sql_warehouses.monitoring_sql_warehouse.id}`. If you kept the `state_volume_path` placeholder, the hook's throttle runs in-memory only (still removes per-microbatch repeats within an update); to add cross-update throttling, set `dq_notify.state_dir` in the pipeline resource to a path under an existing UC Volume. 2. **Bundle integration.** Most bundles already include `resources/*.yml` from `databricks.yml`: ```yaml @@ -41,12 +41,22 @@ The first update ingests the full sample table, so the demo WARN expectation (`d - **The hook's notification** is printed to the pipeline compute's driver log: open the pipeline in the workspace UI, go to the update's compute, open Logs, and search for `EXPECTATION VIOLATION`. There is no CLI or SQL surface for driver-log output; this is a UI observation. - **The durable record** is the published event log table `..dq_notifications_event_log`. Run the first query in `/event_log_queries.sql` to see per-expectation passed/failed counts for the latest update; they match the numbers in the hook's printed line. -- **Hook registration state** is the second query (`hook_progress` events). Note what it does and does not tell you: one row per hook per update recording enable/disable state only, not per-invocation execution. +- **Hook registration state** is the second query (`hook_progress` events). It records ENABLED, FAILED, and DISABLED transitions (so a failing hook is visible here), but not per-invocation execution. +- **Throttle state** (when `dq_notify.state_dir` is set) is browsable: `/dq_notify_state//`, one human-readable JSON marker per expectation with the last-notified time and counts, overwritten in place. Expect one notification per expectation per `dq_notify.throttle_seconds` window, not one per microbatch; a rerun inside the window notifies nothing, by design. - **The backstop alert** evaluates daily at 06:00 UTC by default. To see it immediately, open the alert in the workspace UI (SQL > Alerts) and run it manually, or temporarily tighten `quartz_cron_schedule`. On trigger it emails the configured subscription from `noreply@databricks.com`. The third query in `event_log_queries.sql` is the alert's exact sweep query, so you can preview the evaluated value. -## Optional webhook +## Optional webhook (Slack, Teams, or generic) -Set `dq_notify.webhook_url` in the pipeline resource's `configuration` block to a webhook endpoint (a Slack incoming webhook or similar) and redeploy; the hook then posts each violation message as JSON (`{"text": ...}`) in addition to printing. Keep the endpoint fast: hooks run one at a time, and a slow call delays every other hook. For authenticated endpoints, resolve tokens from a Databricks secret scope at module level, the way the official event-hooks docs example does. +The hook posts each due notification to a webhook when one is configured, in the format selected by `dq_notify.channel_format` (`slack`, `teams`, or `generic`). Route hook posts to chat-style channels; email belongs to the backstop alert. + +A webhook URL is a credential (it grants posting rights), so the real setup goes through a secret scope: + +```bash +databricks secrets create-scope dq-notifications +databricks secrets put-secret dq-notifications slack_webhook_url --string-value "" +``` + +Then set `dq_notify.webhook_secret_scope: dq-notifications` in the pipeline resource's `configuration` and redeploy; the pipeline resolves the URL at startup and, if the scope is missing, prints a notice and runs print-only (your pipeline is never blocked by notification config). The plain `dq_notify.webhook_url` config exists as a demo-only shortcut. Do not use `{{secrets/scope/key}}` in pipeline configuration; it is not interpolated and the literal string arrives. ## Wire the pattern into your own pipeline (companion skill) @@ -58,9 +68,12 @@ To start, paste a prompt like this into your agent and fill the placeholders: Use the sdp-expectation-notifications skill to add per-expectation notifications to my SDP pipeline. -- Pipeline source file(s): +- Pipeline source file(s): - Event log: -- Notify via: > +- Notify via: +- Throttle: - Backstop alert: warehouse , recipient , cadence - Deploy mode: @@ -72,7 +85,9 @@ then apply them and guide me through (or do) the deploy and verification. ## Adjusting the backstop - **Cadence and window are coupled.** The alert query aggregates the trailing `INTERVAL 1 DAY`; the schedule is daily. If you change the cron, change the interval to stay at least as long as the cadence, or violations can fall between sweeps. -- **Recipients** live in `evaluation.notification.subscriptions` (add more `user_email` entries, or a `destination_id` for a workspace notification destination). +- **Notification behavior (measured live):** one email per state transition, however many evaluations stay TRIGGERED; `notify_on_ok: true` (the shipped default) adds exactly one recovery email; the commented `retrigger_seconds` re-sends while TRIGGERED at most once per window, a deliberate escalation knob (for example 86400 for a daily reminder on a critical pipeline). While the alert stays TRIGGERED, new failures inside the window send nothing; the recovery email is what re-arms attention. +- **If the sweep query itself breaks** (dropped table, deleted warehouse), the alert flips to ERROR and emails on every evaluation until fixed. It cannot die silently; treat an `(ERROR)` email as a page. +- **Recipients** live in `evaluation.notification.subscriptions` (add more `user_email` entries, or a `destination_id` for a workspace notification destination, which is how Slack, MS Teams, PagerDuty, and generic webhooks attach to the backstop with zero code). - **Threshold** is `failed_records > 0`. Raise the threshold or filter the query to specific expectation names if the demo tripwire pattern is too chatty for your rules. ## References diff --git a/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}.pipeline.yml.tmpl b/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}.pipeline.yml.tmpl index 61aaeb8..7e888ca 100644 --- a/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}.pipeline.yml.tmpl +++ b/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}.pipeline.yml.tmpl @@ -40,9 +40,27 @@ resources: # The source table the demo ingests. Point this at a curated test # table to exercise the hook against known violation counts. dq_notify.source: samples.nyctaxi.trips - # Optional webhook the hook posts violations to (Slack incoming - # webhook or similar). Empty disables it; the hook always prints to - # the driver log. + # Throttle window: at most one notification per (pipeline, dataset, + # expectation) per window, across microbatches and (with state_dir + # set) across updates. 0 restores unthrottled per-microbatch firing. + dq_notify.throttle_seconds: "3600" + # Durable throttle state root: a path under an EXISTING UC Volume + # (markers land in /dq_notify_state//). Empty + # keeps the throttle in-memory only (still kills per-microbatch + # repeats within an update and within a continuous run). + dq_notify.state_dir: {{if eq .state_volume_path "VOLUME_PATH_PLACEHOLDER"}}""{{else}}{{.state_volume_path}}{{end}} + # Webhook payload format: slack | teams | generic. + dq_notify.channel_format: slack + # Real setups: put the webhook URL in a secret scope (it is a + # credential) and name the scope/key here; the pipeline resolves it + # via dbutils.secrets.get at startup and degrades to print-only if + # the scope is missing. Create it with: + # databricks secrets create-scope dq-notifications + # databricks secrets put-secret dq-notifications slack_webhook_url --string-value "" + dq_notify.webhook_secret_scope: "" + dq_notify.webhook_secret_key: slack_webhook_url + # Demo-only fallback: a plain webhook URL. Empty disables posting; + # the hook always prints to the driver log. dq_notify.webhook_url: "" # Traceability back to the originating asset (forwarded as cluster tags). diff --git a/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}_backstop.alert.yml.tmpl b/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}_backstop.alert.yml.tmpl index 9c4e075..e37d8c0 100644 --- a/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}_backstop.alert.yml.tmpl +++ b/assets/sdp-expectation-notifications/template/resources/{{.pipeline_resource_key}}_backstop.alert.yml.tmpl @@ -43,9 +43,27 @@ resources: value: double_value: 0 notification: - notify_on_ok: false + # One email per state transition (validated live 2026-07-18: nine + # consecutive TRIGGERED evaluations produced exactly one email). + # notify_on_ok adds exactly one recovery email on TRIGGERED -> OK, + # closing the loop after the masking window (while the alert stays + # TRIGGERED, new failures inside the sweep window send nothing). + notify_on_ok: true + # Deliberate re-nagging while the alert stays TRIGGERED: re-sends at + # the first evaluation past each window (validated live). Off by + # default; a real use is escalation on critical pipelines, for + # example 86400 for a daily "still broken" reminder. + # retrigger_seconds: 86400 subscriptions: - user_email: {{.notification_email}} + # Slack / MS Teams / PagerDuty / webhook delivery: create a + # workspace notification destination (admin settings) and + # reference it here instead of, or alongside, the email: + # - destination_id: + # Operational note (validated live): if this alert's query itself + # fails (dropped table, deleted warehouse), the alert flips to ERROR + # and emails on EVERY evaluation until fixed. It cannot die silently; + # treat an (ERROR) email as a page. schedule: pause_status: UNPAUSED quartz_cron_schedule: '0 0 6 * * ?' diff --git a/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md.tmpl b/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md.tmpl index 8e48bad..e5af4c8 100644 --- a/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md.tmpl +++ b/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/SKILL.md.tmpl @@ -1,13 +1,13 @@ --- name: sdp-expectation-notifications -description: Use this skill to add per-expectation data-quality notifications to the user's own Lakeflow Spark Declarative Pipeline (SDP). The pattern is a pair, an in-pipeline event hook that notifies the moment an expectation result with failures is logged (fast, best-effort) plus exactly one DABs-managed Alert v2 that sweeps the published pipeline event log on a schedule (guaranteed). Use it whenever the user wants to be notified when expectations fail, wants alerting on WARN expectations that would otherwise only sit in the event log, or asks for event-hook-based monitoring. The skill adapts the shipped samples.nyctaxi.trips reference to the user's pipeline, keeps the hook on the supported notification surface (print and HTTP, never spark.sql writes), wires the backstop alert to the user's event log and recipients, and verifies the result with read-only queries. It proposes changes and confirms with the user; it is not an auto-rewriter. +description: Use this skill to add per-expectation data-quality notifications to the user's own Lakeflow Spark Declarative Pipeline (SDP). The pattern is a pair, an in-pipeline event hook that notifies the moment an expectation result with failures is logged (fast, best-effort, throttled to at most one notification per expectation per window) plus exactly one DABs-managed Alert v2 that sweeps the published pipeline event log on a schedule (guaranteed, one email per state change). Use it whenever the user wants to be notified when expectations fail, wants alerting on WARN expectations that would otherwise only sit in the event log, or asks for event-hook-based monitoring, especially when they worry about notification noise or floods. The skill adapts the shipped samples.nyctaxi.trips reference to the user's pipelines (the same hook code works unchanged across any number of pipelines), keeps the hook on the supported notification surface (print and HTTP, never spark.sql writes; durable throttle state via UC Volume marker files), wires the backstop alert to the user's event log and recipients, and verifies the result with read-only queries. It proposes changes and confirms with the user; it is not an auto-rewriter. --- # sdp-expectation-notifications Add per-expectation notification to the user's own Lakeflow Spark Declarative Pipeline (SDP), then prove both paths work. -The pattern: an event hook in the pipeline source filters `flow_progress` events for expectation results with `failed_records > 0` and notifies immediately (driver-log print, optionally a webhook). Because hook delivery is explicitly best-effort (the platform does not guarantee every hook runs on every event before pipeline compute terminates), one scheduled Alert v2 sweeps the published event log over a trailing time window as the guaranteed path. Hook = fast, alert = guaranteed. Neither replaces the other. +The pattern: an event hook in the pipeline source filters `flow_progress` events for expectation results with `failed_records > 0`, throttles (expectation counts arrive once per microbatch, so an unthrottled hook floods on multi-batch and continuous pipelines; measured live), and notifies (driver-log print, optionally a webhook in Slack, Teams, or generic format). Because hook delivery is explicitly best-effort (the platform does not guarantee every hook runs on every event before pipeline compute terminates), one scheduled Alert v2 sweeps the published event log over a trailing time window as the guaranteed path. Hook = fast and quiet, alert = guaranteed. Neither replaces the other. This asset shipped a validated reference implementation alongside this skill. Read it first; it is the worked example you adapt away from: @@ -17,10 +17,11 @@ This asset shipped a validated reference implementation alongside this skill. Re Paths in this skill are relative to the bundle root. -## The two rules you must not break +## The three rules you must not break 1. **The hook is never the only notification path.** The docs state there is no guarantee all hooks run on all events before compute termination, and expectation events are emitted late in the update, closest to that window. Always keep (or add) the backstop alert. If the user asks you to drop it, explain the delivery gap and let them decide explicitly. -2. **Never write Delta from inside a hook.** `spark.sql()` inside SDP pipeline Python is restricted to a read-oriented command allowlist: `CREATE TABLE` fails with `UNSUPPORTED_SPARK_SQL_COMMAND`, and the SDP-patched `spark.sql()` rejects the parameterized-query `args=` kwarg (both observed live, 2026-07-11). The supported surface inside a hook is `print` and HTTP calls such as `requests`, which is what the official docs examples use. If the user wants violations in a table, they already have one: the published event log. +2. **Never write Delta from inside a hook.** `spark.sql()` inside SDP pipeline Python is restricted to a read-oriented command allowlist: `INSERT INTO` and `CREATE TABLE` fail with `UNSUPPORTED_SPARK_SQL_COMMAND` (allowed: SELECT, DESCRIBE, SHOW variants, USE), and the SDP-patched `spark.sql()` rejects the parameterized-query `args=` kwarg (all observed live, 2026-07-11 and 2026-07-18). The supported surface inside a hook is `print`, HTTP calls such as `requests`, and plain-Python file I/O against a UC Volume (create, overwrite, read, exists, listdir all work; append does not: `OSError Errno 29`). The throttle's durable state uses exactly that volume surface; if the user wants violations in a table, they already have one: the published event log. +3. **Keep the throttle time-aware and fail-open.** Never replace the timestamp-based throttle with a notify-once set (continuous pipelines keep one process alive for days and would go silent forever; measured live), and never let a state error suppress a notification: on any marker read or write failure the code notifies and prints the error. Notification code must never raise; a raising hook risks disable semantics. ## Workflow @@ -29,7 +30,7 @@ The work splits at a trust boundary. **Phase A** is non-mutating: investigate th ### Phase A (autonomous, no workspace changes) 1. **Understand the target pipeline.** Read the user's pipeline source and DABs resources. Establish: where expectations are declared (the hook needs no changes for new expectations, it reacts to all of them); whether the pipeline is Python or SQL (hooks are Python-only; a SQL pipeline needs a small Python file added to its libraries to carry the hook); and whether the pipeline already publishes its event log to a UC table (the `event_log` block in the pipeline resource). -2. **Propose the wiring and confirm it.** Present the user a short plan: where the hook function goes, what it will notify through (print only, or also a webhook and which endpoint), where the event log will publish, and the backstop alert's warehouse, recipient, and cadence. Do not silently pick notification endpoints, recipients, or cadences; those are the user's calls. +2. **Propose the wiring and confirm it.** Present the user a short plan: where the hook function goes; what it will notify through (print only, or a webhook: which channel format, and the secret scope/key holding the URL, since webhook URLs are credentials); the throttle window (`dq_notify.throttle_seconds`, default 3600) and whether cross-update state gets a path under an existing UC Volume (`dq_notify.state_dir`, empty = in-memory only; never create a volume for this, ask which existing one to use); where the event log will publish; and the backstop alert's warehouse, recipient, and cadence. When adapting for multiple pipelines, the same hook code and one shared state root are correct: state scopes per pipeline automatically from the event's own origin. Do not silently pick notification endpoints, recipients, cadences, or throttle windows; those are the user's calls. 3. **Adapt the files.** Copy the hook from the reference into the user's pipeline source (or a new hooks file for SQL pipelines), add the `event_log` block if missing, and adapt the backstop alert resource: event log table FQN in the query, warehouse id, subscriptions, schedule, and the query window kept at least as long as the cadence. Mechanics in `references/adapt-the-pattern.md`. 4. **Validate.** `databricks bundle validate` must pass. Also re-read the hook body against the two rules above. @@ -65,7 +66,8 @@ These steps are deliberately high-level. The user's setup will have specifics yo Do not report success until all of these hold: -- [ ] The hook is in the user's pipeline source, filters `flow_progress` events, and stays on the supported surface (print / HTTP; no `spark.sql()` writes, no other workspace mutation from inside the hook). +- [ ] The hook is in the user's pipeline source, filters `flow_progress` events, and stays on the supported surface (print / HTTP / volume file I/O; no `spark.sql()` writes, no other workspace mutation from inside the hook). +- [ ] The throttle is time-aware and fail-open, the user chose the window (or accepted the default), and durable state (if enabled) points under an existing UC Volume the user named. - [ ] The pipeline publishes its event log to a UC table, and the backstop alert's query points at that table with a past-time window at least as long as the alert cadence. - [ ] The user confirmed the notification endpoint(s), recipient(s), and cadence. - [ ] `databricks bundle validate` is clean. diff --git a/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/adapt-the-pattern.md.tmpl b/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/adapt-the-pattern.md.tmpl index e2bff05..248aebf 100644 --- a/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/adapt-the-pattern.md.tmpl +++ b/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/adapt-the-pattern.md.tmpl @@ -6,17 +6,19 @@ Load this when you are editing the user's pipeline and bundle. It assumes you ha Three pieces, and only the first lives inside pipeline execution: -- **The hook** (`notify_on_expectation_violation` in `{{.target_dir}}/expectation_notifications_pipeline.py`): a `@dp.on_event_hook(max_allowable_consecutive_failures=None)` function that ignores everything except `flow_progress` events, walks `details.flow_progress.data_quality.expectations`, and notifies for each entry with `failed_records > 0`. Inside the hook, `event["details"]` arrives as a native dict; the `_event_details` helper keeps a defensive JSON-string branch anyway. +- **The hook** (`notify_on_expectation_violation` in `{{.target_dir}}/expectation_notifications_pipeline.py`): a `@dp.on_event_hook(max_allowable_consecutive_failures=None)` function that ignores everything except `flow_progress` events, walks `details.flow_progress.data_quality.expectations`, throttles via `_should_notify` (two time-aware layers: in-memory always, volume marker files when `dq_notify.state_dir` is set), and notifies for each entry with `failed_records > 0` that is due. Inside the hook, `event["details"]` arrives as a native dict (the `_event_details` helper keeps a defensive JSON-string branch), and `event["origin"]` carries `pipeline_name`/`pipeline_id`, which is what scopes throttle state per pipeline with no configuration. - **The event-log publication** (the `event_log` block in the pipeline resource): publishes the pipeline event log to a UC table. This is the durable record of every expectation result, whether or not the hook got to run, and it is the backstop's scan target. - **The backstop alert** (`resources/_backstop.alert.yml`): one Alert v2 whose query sums `failed_records` from `flow_progress` events over a trailing window, with `comparison_operator: GREATER_THAN` and threshold `0`, its own quartz cron schedule, and email subscriptions. ## What to change for the user's pipeline -1. **The hook function.** Copy it into the user's pipeline source close to verbatim; it needs no per-expectation configuration because it reacts to every expectation the pipeline declares, current and future. Decisions to confirm with the user: - - Notification surface: print only, or also a webhook. For a webhook, take the URL from pipeline `configuration` (as the reference does with `dq_notify.webhook_url`) and keep the timeout short; hooks run one at a time, so a slow endpoint delays every other hook. - - Secrets: for authenticated endpoints, resolve tokens at module level from a Databricks secret scope (the docs' Slack example does exactly this). Never hardcode tokens in pipeline source or resource YAML. +1. **The hook function.** Copy it into the user's pipeline source close to verbatim (helpers included: `_should_notify`, `_sanitize`, `_payload`, `_event_details`); it needs no per-expectation configuration because it reacts to every expectation the pipeline declares, current and future, and no per-pipeline configuration because state scopes from the event origin. Decisions to confirm with the user: + - Throttle window: `dq_notify.throttle_seconds` (default 3600) is the "at most one notification per expectation per" window. `0` disables throttling (one notification per failing expectation per microbatch; only sensible for pipelines that process exactly one batch per run and never full-refresh). + - Durable state: `dq_notify.state_dir` empty keeps the throttle in-memory (kills repeats within an update and within a continuous run; forgets across updates). A path under an EXISTING UC Volume adds cross-update throttling via marker files (`/dq_notify_state//__.json`, overwritten in place, human-readable). Never create a volume for this; ask which existing volume to use. Multiple pipelines share one state root safely. + - Notification surface: print only, or also a webhook, and which payload format (`dq_notify.channel_format`: `slack` validated live; `teams` documented Adaptive Card envelope, not live-tested; `generic` plain JSON). Keep the timeout short (the reference uses 5s) and no retries; hooks run one at a time and the post-update grace budget is seconds. + - Secrets: webhook URLs are credentials. Put the URL in a secret scope and set `dq_notify.webhook_secret_scope`/`dq_notify.webhook_secret_key`; the reference resolves it via `dbutils.secrets.get` at module level (works in serverless SDP Python, validated live) and degrades to print-only on a missing scope. Two traps, both observed live: `{{"{{secrets/scope/key}}"}}` in pipeline configuration is NOT interpolated (the literal string arrives), and the plain `dq_notify.webhook_url` config is a demo-only convenience. Setup commands: `databricks secrets create-scope ` then `databricks secrets put-secret --string-value ""` (run by the user so the credential stays out of your transcript). - Filtering: if the user only wants certain expectations or datasets to notify, filter on `exp["name"]` / `exp["dataset"]` inside the loop. Keep the default broad; per-rule routing is the alert layer's job if it grows complicated. - - `max_allowable_consecutive_failures`: keep `None` unless the user prefers auto-disable semantics; explain the tradeoff (finite = a flaky endpoint silently disables the hook until the next pipeline restart; `None` = a broken hook fails forever but keeps trying). The reference catches webhook exceptions and prints them so the hook itself never counts as failed. + - `max_allowable_consecutive_failures`: keep `None` unless the user prefers auto-disable semantics; explain the tradeoff (finite = a flaky endpoint silently disables the hook until the next pipeline restart, visible as FAILED then DISABLED rows in `hook_progress`; `None` = a broken hook fails forever but keeps trying). The reference catches every exception and prints, so the hook itself never counts as failed. 2. **Python vs SQL pipelines.** Hooks are Python-only. If the user's pipeline is SQL, do not rewrite it: add one small Python file containing just the imports and the hook, and add it to the pipeline's `libraries`. The hook still sees all events of the update. 3. **Event-log publication.** If the user's pipeline resource has no `event_log` block, add one (catalog, schema, table name; pick a name coupled to the pipeline to avoid collisions in a shared schema). If the pipeline already publishes an event log, reuse the existing table and do not rename it: downstream consumers may already query it. 4. **The backstop alert.** Adapt the reference resource: diff --git a/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/self-verify.md b/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/self-verify.md index 29c8a6d..d0017f1 100644 --- a/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/self-verify.md +++ b/assets/sdp-expectation-notifications/template/{{.skill_dir}}/skills/sdp-expectation-notifications/references/self-verify.md @@ -23,7 +23,7 @@ WHERE event_type = 'hook_progress' ORDER BY timestamp DESC; ``` -PASS when the notifier hook appears with state `ENABLED` for the update you ran. Interpret it honestly: `hook_progress` records enable/disable state at registration only. It proves the hook is alive, not that it processed any particular event. +PASS when the notifier hook appears with state `ENABLED` for the update you ran. Interpret it honestly: `hook_progress` records ENABLED, FAILED, and DISABLED transitions (observed live), so a FAILED or DISABLED row is a red flag worth reporting, but the absence of failures does not prove the hook processed any particular event; per-invocation output has no event-log surface. ## 2. Expectation results recorded (event log) @@ -48,7 +48,16 @@ PASS when every expectation the pipeline declares shows up with counts for the u The hook's output (`EXPECTATION VIOLATION | ...` lines, and any `WEBHOOK_DELIVERY_FAILED` lines) exists only in the pipeline compute's driver log. Ask the user to: open the pipeline in the workspace UI, open the update's compute, open Logs, and search for `EXPECTATION VIOLATION`. -PASS when the printed counts match check 2's counts for the same `update_id` exactly. If a webhook is configured, also confirm the message arrived at the endpoint (the user's channel or the endpoint's own log). +PASS when the printed counts match check 2's counts for the same `update_id` (expect ONE `EXPECTATION VIOLATION` line per expectation per throttle window, not one per microbatch: suppressed repeats print nothing). If a webhook is configured, also confirm the message arrived at the endpoint (the user's channel or the endpoint's own log). + +## 3b. Throttle state markers (only when `dq_notify.state_dir` is set) + +```bash +databricks fs ls dbfs:/dq_notify_state/ +databricks fs cat "dbfs:/dq_notify_state//__.json" +``` + +PASS when a folder named after the pipeline exists and each notified expectation has one marker whose `last_notified_at` matches the run and whose counts match check 2. A second run inside the throttle window must NOT advance `last_notified_at` (suppression proof); a run after the window must. Marker files are state, not history: one file per expectation, overwritten in place. If the log shows nothing despite check 2 showing failures: first suspect timing (delivery is best-effort; events emitted just before compute termination can be lost, which is exactly why the backstop exists), then the hook's filter logic. A `mode: development` target keeps compute warm and its log long-lived, which makes this check easier to run and the loss scenario less likely to occur; see the development-mode caveat in `references/adapt-the-pattern.md`. @@ -62,8 +71,8 @@ If the log shows nothing despite check 2 showing failures: first suspect timing databricks alerts-v2 get-alert ``` - PASS when `state` is `TRIGGERED` (given `failed_records > 0` in step 1) and `last_evaluated_at` is fresh. With no violations in the window, `OK` is the correct passing state; do not manufacture violations in real data to force a trigger. -3. Delivery: only the recipient can confirm the email arrived (sender `noreply@databricks.com`, subject containing the alert display name and state). Ask the user; record their answer as the delivery confirmation. + PASS when `state` is `TRIGGERED` (given `failed_records > 0` in step 1) and `last_evaluated_at` is fresh. With no violations in the window, `OK` is the correct passing state; do not manufacture violations in real data to force a trigger. A persistent `ERROR` state means the query itself fails and the alert emails on EVERY evaluation until fixed (observed live); treat it as urgent, not as noise to ignore. +3. Delivery: only the recipient can confirm the email arrived (sender `noreply@databricks.com`, subject containing the alert display name and state). Expected volume, observed live: one email per state transition, plus one recovery email if `notify_on_ok` is true, plus windowed re-sends only if `retrigger_seconds` is set. Ask the user; record their answer as the delivery confirmation. ## Interpreting results honestly diff --git a/assets/sdp-expectation-notifications/template/{{.target_dir}}/expectation_notifications_pipeline.py b/assets/sdp-expectation-notifications/template/{{.target_dir}}/expectation_notifications_pipeline.py index a2efabf..6316994 100644 --- a/assets/sdp-expectation-notifications/template/{{.target_dir}}/expectation_notifications_pipeline.py +++ b/assets/sdp-expectation-notifications/template/{{.target_dir}}/expectation_notifications_pipeline.py @@ -1,25 +1,52 @@ -"""Lakeflow SDP pipeline demonstrating per-expectation notification via an event hook. +"""Lakeflow SDP pipeline: per-expectation notification via a throttled event hook. One streaming table carries a deliberately tight WARN expectation that fails on the healthy public sample, so the hook has something to notify about on the first run. The hook filters `flow_progress` events for expectation results with -failures and notifies via `print` (always) and an optional webhook. +failures, de-duplicates, and notifies via `print` (always) and an optional +webhook (Slack, Teams, or generic JSON). + +Noise control is a two-layer, time-aware throttle (validated live 2026-07-18): + +- In-memory layer (always on): a last-notified timestamp per (pipeline, + dataset, expectation). Without it, one failing expectation notifies once per + microbatch: a multi-file backlog or a continuous pipeline multiplies every + violation (measured: 4 notifications for one expectation in one update). + Time-aware on purpose: continuous pipelines keep one Python process alive + for days, so a notify-once-per-process set would go silent forever. +- Durable marker layer (optional): JSON marker files under + `/dq_notify_state//__.json` in a + UC Volume, carrying the throttle across updates and compute restarts. Point + `dq_notify.state_dir` at a path under an EXISTING volume to enable; leave + empty for in-memory only. One file per expectation, overwritten on notify, + never growing. Fail-open: any state error means notify, never stay silent. The hook is the fast path only. Event hook delivery is best-effort: the -platform waits a fixed, non-configurable period before terminating pipeline -compute and does not guarantee every hook runs on every event. The companion -backstop alert (see the resource in `resources/`) sweeps the published event -log on a schedule as the guaranteed path. +platform waits only seconds after an update completes before terminating +compute (measured: a hook needing 20s per event lost 6 of 6 queued +notifications under normal teardown), and does not guarantee every hook runs +on every event. Keep hook work in single-digit seconds. The companion backstop +alert (see `resources/`) sweeps the published event log on a schedule as the +guaranteed path. Do not write Delta tables from inside a hook via `spark.sql()`. SDP restricts -`spark.sql()` in pipeline Python to a read-oriented command allowlist (DDL such -as CREATE TABLE fails with `UNSUPPORTED_SPARK_SQL_COMMAND`), and the patched -`spark.sql()` does not accept the parameterized-query `args=` kwarg. The -supported notification surface inside a hook is `print` and HTTP calls such as -`requests` (both used by the official event-hooks docs examples). +`spark.sql()` in pipeline Python to a read allowlist (INSERT fails with +`UNSUPPORTED_SPARK_SQL_COMMAND`; allowed: SELECT, DESCRIBE, SHOW variants, +USE; observed live 2026-07-18). Durable state goes through plain-Python UC +Volume file I/O instead (create/overwrite/read work; append does not). + +Webhook URLs are credentials (a Slack incoming webhook grants posting rights). +Resolve them from a secret scope: set `dq_notify.webhook_secret_scope` (and +key) and the module resolves via `dbutils.secrets.get`, which works at module +level in serverless SDP pipeline Python (validated live; `{{secrets/...}}` +interpolation in pipeline configuration does NOT resolve and arrives as the +literal string). The plain `dq_notify.webhook_url` config is the demo-only +fallback. A missing scope is caught and the hook degrades to print-only. """ import json +import os +import time import requests from pyspark import pipelines as dp @@ -28,11 +55,35 @@ spark = SparkSession.getActiveSession() SOURCE_TABLE = spark.conf.get("dq_notify.source", "samples.nyctaxi.trips") -# Optional webhook for real delivery (Slack incoming webhook or similar). -# Empty string disables it and the hook only prints to the driver log. -# For authenticated endpoints, resolve tokens from a Databricks secret scope at -# module level (as the docs' Slack example does), not inside the hook body. + +# Throttle: at most one notification per (pipeline, dataset, expectation) per +# window. 0 disables throttling entirely (v1.11 behavior: one notification per +# failing expectation per microbatch; expect floods on multi-batch pipelines). +THROTTLE_SECONDS = int(spark.conf.get("dq_notify.throttle_seconds", "3600")) +# Durable state root under an EXISTING UC Volume; empty = in-memory only. +STATE_DIR = spark.conf.get("dq_notify.state_dir", "").rstrip("/") + +# Webhook payload format: "slack" (validated live), "teams" (documented +# Workflows Adaptive Card envelope, not live-tested), or "generic" (plain JSON +# with the violation fields, for webhook receivers you control). +CHANNEL_FORMAT = spark.conf.get("dq_notify.channel_format", "slack") + +# Webhook resolution: secret scope first (real setups), plain config second +# (demo), empty means print-only. Never log the resolved URL. +WEBHOOK_SECRET_SCOPE = spark.conf.get("dq_notify.webhook_secret_scope", "") +WEBHOOK_SECRET_KEY = spark.conf.get("dq_notify.webhook_secret_key", "slack_webhook_url") WEBHOOK_URL = spark.conf.get("dq_notify.webhook_url", "") +if WEBHOOK_SECRET_SCOPE: + try: + WEBHOOK_URL = dbutils.secrets.get(WEBHOOK_SECRET_SCOPE, WEBHOOK_SECRET_KEY) # noqa: F821 + except Exception as e: + print( + f"DQ_NOTIFY_SECRET_UNAVAILABLE: scope={WEBHOOK_SECRET_SCOPE} " + f"key={WEBHOOK_SECRET_KEY} ({type(e).__name__}); webhook disabled, print-only." + ) + WEBHOOK_URL = "" + +_MEM_LAST = {} # (pipeline, dataset, expectation) -> epoch of last notification @dp.table(comment="NYC taxi trips with one deliberately tight warn expectation.") @@ -64,38 +115,156 @@ def _event_details(event): return details +def _sanitize(name): + """Make a pipeline/dataset/expectation name safe as a file or folder name.""" + return "".join(c if (c.isalnum() or c in "._-") else "_" for c in str(name)) + + +def _should_notify(pipeline, dataset, expectation, failed, passed, now): + """Two-layer time-aware throttle. Returns True when a notification is due. + + Layer 1 (memory) covers repeats within one Python process: later + microbatches of the same update, and a continuous pipeline's whole run. + Layer 2 (marker files, only when STATE_DIR is set) covers repeats across + updates and compute restarts. Both layers fail open: a state error must + produce a notification, never silence. + """ + if THROTTLE_SECONDS <= 0: + return True + key = (pipeline, dataset, expectation) + last = _MEM_LAST.get(key) + if last is not None and (now - last) < THROTTLE_SECONDS: + return False + marker = None + if STATE_DIR: + marker = ( + f"{STATE_DIR}/dq_notify_state/{_sanitize(pipeline)}/" + f"{_sanitize(dataset)}__{_sanitize(expectation)}.json" + ) + try: + with open(marker) as f: + prior = float((json.load(f) or {}).get("last_notified_epoch")) + if (now - prior) < THROTTLE_SECONDS: + _MEM_LAST[key] = prior + return False + except Exception: + pass # no marker or unreadable: fail open and notify + _MEM_LAST[key] = now + if marker: + try: + os.makedirs(os.path.dirname(marker), exist_ok=True) + with open(marker, "w") as f: + json.dump( + { + "pipeline": pipeline, + "dataset": dataset, + "expectation": expectation, + "last_notified_at": time.strftime( + "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now) + ), + "last_notified_epoch": now, + "last_failed_records": failed, + "last_passed_records": passed, + "throttle_seconds": THROTTLE_SECONDS, + }, + f, + indent=2, + ) + except Exception as e: + print(f"DQ_NOTIFY_MARKER_WRITE_FAILED: {type(e).__name__}: {e}") + return True + + +def _payload(msg, pipeline, dataset, expectation, failed, passed): + """Build the webhook payload for the configured channel format.""" + if CHANNEL_FORMAT == "teams": + # MS Teams Workflows incoming webhook: Adaptive Card envelope + # (documented format; not live-tested by this asset, see README). + return { + "type": "message", + "attachments": [ + { + "contentType": "application/vnd.microsoft.card.adaptive", + "content": { + "type": "AdaptiveCard", + "version": "1.4", + "body": [{"type": "TextBlock", "text": msg, "wrap": True}], + }, + } + ], + } + if CHANNEL_FORMAT == "generic": + return { + "text": msg, + "pipeline": pipeline, + "dataset": dataset, + "expectation": expectation, + "failed_records": failed, + "passed_records": passed, + } + return {"text": msg} # slack (default) + + @dp.on_event_hook(max_allowable_consecutive_failures=None) def notify_on_expectation_violation(event): - """Notify the moment an expectation result with failures is logged. + """Notify, at most once per throttle window, when an expectation fails. Expectation counts arrive in `flow_progress` events under `details.flow_progress.data_quality.expectations`, one entry per - expectation per flow. Hooks run serialized (one at a time), so anything - slow in here delays every other hook and widens the window in which - queued events can be lost at compute termination: keep it fast. + expectation per flow, once per microbatch (so an unthrottled hook repeats + per batch). The pipeline identity comes from the event's own `origin` + (observed live: `origin.pipeline_name` and `origin.pipeline_id` are + present in hook events), so this same code runs unchanged in any number of + pipelines, each throttling in its own state folder. + Hooks run serialized and the post-update grace budget is seconds: keep + this fast (marker I/O ~100ms, webhook timeout 5s, no retries). `max_allowable_consecutive_failures=None` means the hook is never disabled - by its own failures; a finite value would silently disable it until the - next pipeline restart. Either way, `hook_progress` events record only - enable/disable state, not per-invocation execution, so the webhook failure - branch below prints rather than raises. + by its own failures; `hook_progress` records ENABLED, FAILED, and DISABLED + transitions (observed live), so a finite value here is event-log-visible + but silently stops notifications until the next restart. """ - if event.get("event_type") != "flow_progress": - return - details = _event_details(event) - data_quality = (details.get("flow_progress") or {}).get("data_quality") or {} - for exp in data_quality.get("expectations") or []: - if (exp.get("failed_records") or 0) > 0: + try: + if event.get("event_type") != "flow_progress": + return + details = _event_details(event) + data_quality = (details.get("flow_progress") or {}).get("data_quality") or {} + failing = [ + e + for e in (data_quality.get("expectations") or []) + if (e.get("failed_records") or 0) > 0 + ] + if not failing: + return + origin = event.get("origin") or {} + pipeline = origin.get("pipeline_name") or origin.get("pipeline_id") or "unknown_pipeline" + for exp in failing: + now = time.time() + dataset = exp.get("dataset") + expectation = exp.get("name") + failed = exp.get("failed_records") or 0 + passed = exp.get("passed_records") or 0 + if not _should_notify(pipeline, dataset, expectation, failed, passed, now): + continue msg = ( "EXPECTATION VIOLATION" - f" | dataset={exp.get('dataset')}" - f" | expectation={exp.get('name')}" - f" | failed={exp.get('failed_records')}" - f" | passed={exp.get('passed_records')}" + f" | pipeline={pipeline}" + f" | dataset={dataset}" + f" | expectation={expectation}" + f" | failed={failed}" + f" | passed={passed}" ) print(msg) if WEBHOOK_URL: try: - requests.post(WEBHOOK_URL, json={"text": msg}, timeout=10) + requests.post( + WEBHOOK_URL, + json=_payload(msg, pipeline, dataset, expectation, failed, passed), + timeout=5, + ) except Exception as e: - print(f"WEBHOOK_DELIVERY_FAILED: {e}") + print(f"WEBHOOK_DELIVERY_FAILED: {type(e).__name__}: {e}") + except Exception as e: + # A raising hook risks racking up FAILED states; notification code + # must never take the hook down. + print(f"DQ_NOTIFY_HOOK_UNEXPECTED: {type(e).__name__}: {e}") diff --git a/tests/assets/test_sdp_expectation_notifications.py b/tests/assets/test_sdp_expectation_notifications.py index f7b8c7f..e2152ff 100644 --- a/tests/assets/test_sdp_expectation_notifications.py +++ b/tests/assets/test_sdp_expectation_notifications.py @@ -23,6 +23,9 @@ """ import ast +import json +import os +import time from pathlib import Path import pytest @@ -122,6 +125,107 @@ def test_webhook_is_optional_and_guarded(installed: Path): assert 'spark.conf.get("dq_notify.webhook_url", "")' in src assert "if WEBHOOK_URL:" in src assert "WEBHOOK_DELIVERY_FAILED" in src + assert "timeout=5" in src + + +def test_webhook_secret_resolution_is_guarded(installed: Path): + """Secret-scope resolution goes through dbutils.secrets.get (config + interpolation of {{secrets/...}} does not resolve in SDP pipeline + configuration, observed live) and a missing scope degrades to print-only + instead of failing the pipeline.""" + src = (installed / DEFAULT_TARGET_DIR / PIPELINE_SOURCE).read_text(encoding="utf-8") + assert 'spark.conf.get("dq_notify.webhook_secret_scope", "")' in src + assert "dbutils.secrets.get(WEBHOOK_SECRET_SCOPE, WEBHOOK_SECRET_KEY)" in src + assert "DQ_NOTIFY_SECRET_UNAVAILABLE" in src + + +def _throttle_ns(src: str, state_dir: str, throttle_seconds: int) -> dict: + """Exec only the pure throttle helpers from the installed source with + stubbed module globals, so the decision logic is testable offline.""" + tree = ast.parse(src) + wanted = {"_sanitize", "_should_notify"} + subset = ast.Module( + body=[n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in wanted], + type_ignores=[], + ) + ns = { + "os": os, + "json": json, + "time": time, + "THROTTLE_SECONDS": throttle_seconds, + "STATE_DIR": state_dir, + "_MEM_LAST": {}, + "print": lambda *a, **k: None, + } + exec(compile(subset, "", "exec"), ns) # noqa: S102 (test-only) + return ns + + +def test_throttle_suppresses_within_window_in_memory(installed: Path, tmp_path: Path): + src = (installed / DEFAULT_TARGET_DIR / PIPELINE_SOURCE).read_text(encoding="utf-8") + ns = _throttle_ns(src, str(tmp_path), 3600) + t0 = 1_000_000.0 + assert ns["_should_notify"]("pipe", "cat.sch.tbl", "rule", 5, 95, t0) is True + assert ns["_should_notify"]("pipe", "cat.sch.tbl", "rule", 5, 95, t0 + 10) is False + assert ns["_should_notify"]("pipe", "cat.sch.tbl", "rule", 5, 95, t0 + 3601) is True + + +def test_throttle_is_durable_across_processes_and_scoped_per_pipeline( + installed: Path, tmp_path: Path +): + """A fresh namespace simulates the fresh Python process each update gets in + non-development mode (observed live): the marker file must carry the + suppression across, and a different pipeline scope must not be suppressed.""" + src = (installed / DEFAULT_TARGET_DIR / PIPELINE_SOURCE).read_text(encoding="utf-8") + t0 = 1_000_000.0 + ns1 = _throttle_ns(src, str(tmp_path), 3600) + assert ns1["_should_notify"]("pipe_a", "cat.sch.tbl", "rule", 5, 95, t0) is True + ns2 = _throttle_ns(src, str(tmp_path), 3600) # fresh process, same state dir + assert ns2["_should_notify"]("pipe_a", "cat.sch.tbl", "rule", 5, 95, t0 + 10) is False + assert ns2["_should_notify"]("pipe_b", "cat.sch.tbl", "rule", 5, 95, t0 + 10) is True + + +def test_throttle_marker_is_human_readable_and_overwritten(installed: Path, tmp_path: Path): + src = (installed / DEFAULT_TARGET_DIR / PIPELINE_SOURCE).read_text(encoding="utf-8") + ns = _throttle_ns(src, str(tmp_path), 3600) + t0 = 1_000_000.0 + ns["_should_notify"]("My Pipeline", "cat.sch.tbl", "rule", 7, 93, t0) + marker_dir = tmp_path / "dq_notify_state" / "My_Pipeline" + markers = list(marker_dir.glob("*.json")) + assert [m.name for m in markers] == ["cat.sch.tbl__rule.json"] + body = json.loads(markers[0].read_text(encoding="utf-8")) + assert body["pipeline"] == "My Pipeline" + assert body["last_failed_records"] == 7 + assert body["last_notified_epoch"] == t0 + assert "last_notified_at" in body and body["throttle_seconds"] == 3600 + ns["_should_notify"]("My Pipeline", "cat.sch.tbl", "rule", 9, 91, t0 + 4000) + assert len(list(marker_dir.glob("*.json"))) == 1, "marker must be overwritten, not accumulated" + assert json.loads(markers[0].read_text(encoding="utf-8"))["last_failed_records"] == 9 + + +def test_throttle_zero_disables_and_empty_state_dir_stays_in_memory( + installed: Path, tmp_path: Path +): + src = (installed / DEFAULT_TARGET_DIR / PIPELINE_SOURCE).read_text(encoding="utf-8") + ns0 = _throttle_ns(src, str(tmp_path), 0) + t0 = 1_000_000.0 + assert ns0["_should_notify"]("p", "d", "r", 1, 1, t0) is True + assert ns0["_should_notify"]("p", "d", "r", 1, 1, t0 + 1) is True + ns_mem = _throttle_ns(src, "", 3600) + assert ns_mem["_should_notify"]("p", "d", "r", 1, 1, t0) is True + assert ns_mem["_should_notify"]("p", "d", "r", 1, 1, t0 + 10) is False + assert not list(tmp_path.rglob("*.json")), "empty state dir must write no markers" + + +def test_payload_formats_cover_all_channels(installed: Path): + """One payload function, three formats; Teams is the documented Adaptive + Card envelope (doc-confirmed, not live-tested; the README says so).""" + src = (installed / DEFAULT_TARGET_DIR / PIPELINE_SOURCE).read_text(encoding="utf-8") + tree = ast.parse(src) + payload_defs = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == "_payload"] + assert payload_defs, "_payload function missing" + assert '"teams"' in src and '"generic"' in src + assert "application/vnd.microsoft.card.adaptive" in src def test_demo_expectation_is_warn_tripwire(installed: Path): @@ -170,8 +274,16 @@ def test_resource_library_points_at_installed_source(installed: Path): def test_resource_config_and_tags(installed: Path): data = _load_yaml(installed, f"resources/{DEFAULT_RESOURCE_KEY}.pipeline.yml") pipe = data["resources"]["pipelines"][DEFAULT_RESOURCE_KEY] - assert pipe["configuration"]["dq_notify.source"] == "samples.nyctaxi.trips" - assert pipe["configuration"]["dq_notify.webhook_url"] == "" + config = pipe["configuration"] + assert config["dq_notify.source"] == "samples.nyctaxi.trips" + assert config["dq_notify.throttle_seconds"] == "3600" + # The placeholder volume path must template to an empty state_dir + # (in-memory throttle only), never leak the placeholder into the config. + assert config["dq_notify.state_dir"] == "" + assert config["dq_notify.channel_format"] == "slack" + assert config["dq_notify.webhook_secret_scope"] == "" + assert config["dq_notify.webhook_secret_key"] == "slack_webhook_url" + assert config["dq_notify.webhook_url"] == "" assert pipe["tags"]["created_by"] == "dabs-asset" assert pipe["tags"]["asset"] == "sdp-expectation-notifications" @@ -205,7 +317,13 @@ def test_alert_evaluation_and_schedule(installed: Path): assert evaluation["comparison_operator"] == "GREATER_THAN" assert evaluation["source"]["name"] == "failed_records" assert evaluation["threshold"]["value"]["double_value"] == 0 - assert evaluation["notification"]["notify_on_ok"] is False + # notify_on_ok true is the v1.12 default: exactly one recovery email on + # TRIGGERED -> OK (observed live), closing the loop after the masking + # window of state-change-only notification. + assert evaluation["notification"]["notify_on_ok"] is True + assert "retrigger_seconds" not in evaluation["notification"], ( + "retrigger_seconds must ship commented out, not active" + ) assert evaluation["notification"]["subscriptions"] == [{"user_email": "EMAIL_PLACEHOLDER"}] schedule = alert["schedule"] assert schedule["pause_status"] == "UNPAUSED" @@ -239,6 +357,7 @@ def test_custom_values_flow_through(install_asset): "schema": "observability", "warehouse_id": "abc123def456", "notification_email": "someone@example.com", + "state_volume_path": "/Volumes/sandbox/observability/state_vol", }, ) @@ -255,6 +374,7 @@ def test_custom_values_flow_through(install_asset): assert pipe["catalog"] == "sandbox" assert pipe["schema"] == "observability" assert pipe["event_log"]["schema"] == "observability" + assert pipe["configuration"]["dq_notify.state_dir"] == "/Volumes/sandbox/observability/state_vol" file_paths = [lib["file"]["path"] for lib in pipe["libraries"] if "file" in lib] assert file_paths == [f"../pipelines/dq_notify_demo/{PIPELINE_SOURCE}"] diff --git a/tests/configs/assets/sdp_expectation_notifications.json b/tests/configs/assets/sdp_expectation_notifications.json index c9e3daf..ecd89a5 100644 --- a/tests/configs/assets/sdp_expectation_notifications.json +++ b/tests/configs/assets/sdp_expectation_notifications.json @@ -6,5 +6,6 @@ "schema": "dq_notifications", "warehouse_id": "WAREHOUSE_ID_PLACEHOLDER", "notification_email": "EMAIL_PLACEHOLDER", + "state_volume_path": "VOLUME_PATH_PLACEHOLDER", "skill_dir": ".agents" }