From da3bb0c03d46d0920d01a89b935ee6f13b6ed502 Mon Sep 17 00:00:00 2001 From: Chase Martin Date: Fri, 25 Sep 2026 14:26:38 -0700 Subject: [PATCH 1/2] feat(orchestration): rebuild pygeoapi after product runs finish Add the pygeoapi_rebuild sensor. Once a product job has succeeded and none are queued or running, it starts the die-pygeoapi-rebuild Cloud Build trigger, so the image bakes the new data. It launches no Dagster run and is stopped by default. Co-Authored-By: Claude Opus 5.5 (1M context) --- orchestration/definitions.py | 2 + orchestration/pygeoapi_rebuild.py | 77 +++++++++++++++++++++++++++++++ tests/test_pygeoapi_rebuild.py | 65 ++++++++++++++++++++++++++ 3 files changed, 144 insertions(+) create mode 100644 orchestration/pygeoapi_rebuild.py create mode 100644 tests/test_pygeoapi_rebuild.py diff --git a/orchestration/definitions.py b/orchestration/definitions.py index 9a27a8c..2204f1c 100644 --- a/orchestration/definitions.py +++ b/orchestration/definitions.py @@ -5,6 +5,7 @@ import dagster as dg import yaml +from orchestration.pygeoapi_rebuild import build_rebuild_sensor from orchestration.resources.die_config import DIEConfigResource from orchestration.resources.gcs import GCSResource, AuthedGCSResource from orchestration.resources.geoserver import GeoServerResource @@ -244,6 +245,7 @@ def _build_standalone_jobs_and_schedules( assets=_assets, jobs=list(_cohort_jobs.values()), schedules=_schedules, + sensors=[build_rebuild_sensor(list(_cohort_jobs))], resources={ # USGS_API_KEY is a Dagster+ secret; EnvVar resolves it at run time and # the resource exports it for the NWIS connector. Resolves to None when diff --git a/orchestration/pygeoapi_rebuild.py b/orchestration/pygeoapi_rebuild.py new file mode 100644 index 0000000..307ddab --- /dev/null +++ b/orchestration/pygeoapi_rebuild.py @@ -0,0 +1,77 @@ +"""Rebuild pygeoapi once product runs finish, so the image bakes the new data (SPEC §6.6).""" +import json +import os +from datetime import datetime, timezone + +import dagster as dg + +TRIGGER = "die-pygeoapi-rebuild" +REGION = "us-central1" + +_IN_PROGRESS = [ + dg.DagsterRunStatus.QUEUED, + dg.DagsterRunStatus.NOT_STARTED, + dg.DagsterRunStatus.STARTING, + dg.DagsterRunStatus.STARTED, +] + + +def run_trigger() -> None: + """Start the Cloud Build trigger, using the same credentials as GCS.""" + import google.auth + from google.auth.transport.requests import AuthorizedSession + from google.oauth2 import service_account + + scopes = ["https://www.googleapis.com/auth/cloud-platform"] + key = os.environ.get("GCP_SERVICE_ACCOUNT_KEY") + if key: + info = json.loads(key) + creds = service_account.Credentials.from_service_account_info(info, scopes=scopes) + project = info["project_id"] + else: + creds, project = google.auth.default(scopes=scopes) + url = ( + f"https://cloudbuild.googleapis.com/v1/projects/{project}" + f"/locations/{REGION}/triggers/{TRIGGER}:run" + ) + AuthorizedSession(creds).post(url, json={}).raise_for_status() + + +def build_rebuild_sensor(job_names: list[str]) -> dg.SensorDefinition: + """Sensor that starts one rebuild after a product job succeeds, once no + product job is queued or running. The cursor is the last check time.""" + jobs = set(job_names) + + # Stopped by default so local and branch deployments never rebuild prod; + # turn it on in the prod deployment. + @dg.sensor( + name="pygeoapi_rebuild", + minimum_interval_seconds=600, + default_status=dg.DefaultSensorStatus.STOPPED, + ) + def pygeoapi_rebuild(context: dg.SensorEvaluationContext): + now = datetime.now(timezone.utc) + if context.cursor is None: + context.update_cursor(now.isoformat()) + return dg.SkipReason("First tick; watching for product runs from now.") + + def product_runs(**filters): + runs = context.instance.get_runs(filters=dg.RunsFilter(**filters)) + return [r for r in runs if r.job_name in jobs] + + if product_runs(statuses=_IN_PROGRESS): + return dg.SkipReason("Product runs still in progress.") + succeeded = product_runs( + statuses=[dg.DagsterRunStatus.SUCCESS], + updated_after=datetime.fromisoformat(context.cursor), + ) + if not succeeded: + context.update_cursor(now.isoformat()) + return dg.SkipReason("No product runs succeeded since the last check.") + + run_trigger() + context.update_cursor(now.isoformat()) + context.log.info(f"Started {TRIGGER} after {len(succeeded)} product run(s).") + return dg.SkipReason(f"Started Cloud Build trigger {TRIGGER}.") + + return pygeoapi_rebuild diff --git a/tests/test_pygeoapi_rebuild.py b/tests/test_pygeoapi_rebuild.py new file mode 100644 index 0000000..c9b7bc9 --- /dev/null +++ b/tests/test_pygeoapi_rebuild.py @@ -0,0 +1,65 @@ +from datetime import datetime, timedelta, timezone +from unittest import mock + +import dagster as dg +import pytest + +from orchestration import pygeoapi_rebuild +from orchestration.pygeoapi_rebuild import build_rebuild_sensor + +SUCCESS = dg.DagsterRunStatus.SUCCESS +STARTED = dg.DagsterRunStatus.STARTED + + +@pytest.fixture +def trigger(): + with mock.patch.object(pygeoapi_rebuild, "run_trigger") as run: + yield run + + +def _evaluate(runs, cursor): + """Run the sensor against fake (job_name, status) runs.""" + def get_runs(filters): + return [ + mock.Mock(job_name=job) + for job, status in runs + if status in filters.statuses + ] + + instance = mock.Mock(spec=dg.DagsterInstance) + instance.get_runs.side_effect = get_runs + context = dg.build_sensor_context(instance=instance, cursor=cursor) + result = build_rebuild_sensor(["wl_job", "chem_job"])(context) + return result, context.cursor + + +def _earlier(): + return (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat() + + +def test_first_tick_sets_cursor(trigger): + _, cursor = _evaluate([("wl_job", SUCCESS)], cursor=None) + assert cursor is not None + trigger.assert_not_called() + + +def test_rebuilds_once_product_runs_finish(trigger): + _evaluate([("wl_job", SUCCESS), ("chem_job", SUCCESS)], cursor=_earlier()) + trigger.assert_called_once() + + +def test_waits_while_a_product_run_is_in_progress(trigger): + earlier = _earlier() + _, cursor = _evaluate([("wl_job", SUCCESS), ("chem_job", STARTED)], cursor=earlier) + trigger.assert_not_called() + assert cursor == earlier # keep the success for the next tick + + +def test_ignores_other_jobs(trigger): + _evaluate([("other_job", SUCCESS)], cursor=_earlier()) + trigger.assert_not_called() + + +def test_no_rebuild_without_new_success(trigger): + _evaluate([], cursor=_earlier()) + trigger.assert_not_called() From 3a9eeea3fc09f718816b072aff93e9e0133e67d7 Mon Sep 17 00:00:00 2001 From: Chase Martin Date: Fri, 25 Sep 2026 14:26:38 -0700 Subject: [PATCH 2/2] docs: document the pygeoapi rebuild sensor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update SPEC §6.6, T5 in the performance plan, and orchestration/AGENTS.md. Co-Authored-By: Claude Opus 5.5 (1M context) --- SPEC.md | 15 ++++++--- docs/pygeoapi-performance-plan.md | 52 +++++++++++++------------------ orchestration/AGENTS.md | 6 ++-- 3 files changed, 36 insertions(+), 37 deletions(-) diff --git a/SPEC.md b/SPEC.md index 1343f87..0fdb9c2 100644 --- a/SPEC.md +++ b/SPEC.md @@ -554,11 +554,16 @@ Cloud Run Service (`orchestration/pygeoapi/cloudbuild.yaml`): Baked products (§6.3) are only as fresh as the image: a pipeline run that publishes new data (a combine asset with `skipped_unchanged: false` or `parquet_status: written`) is not served until pygeoapi is rebuilt and redeployed with -`orchestration/pygeoapi/cloudbuild.yaml`. **For now this rebuild is manual**: run the -build after the scheduled product runs finish, and not while any are still running, so -the image doesn't copy a half-updated set of products. Automating it (a Cloud Build -trigger started from Dagster after product runs publish new data) is planned; see -`docs/pygeoapi-performance-plan.md` T5. +`orchestration/pygeoapi/cloudbuild.yaml`. + +The `pygeoapi_rebuild` sensor (`orchestration/pygeoapi_rebuild.py`) does this. Every +10 minutes it checks for product jobs (cohort and standalone) that succeeded since its +last check. Once none are queued or running, it starts the `die-pygeoapi-rebuild` Cloud +Build trigger (`us-central1`) once. Waiting until no runs are active means the image never +bakes a half-updated set of products. Sensor ticks use no Dagster+ credits, and the +sensor launches no Dagster run. It is stopped by default and is turned on only in the +prod deployment. It authenticates with `GCP_SERVICE_ACCOUNT_KEY`, which needs +`roles/cloudbuild.builds.editor`. --- diff --git a/docs/pygeoapi-performance-plan.md b/docs/pygeoapi-performance-plan.md index 36fec66..0c853d3 100644 --- a/docs/pygeoapi-performance-plan.md +++ b/docs/pygeoapi-performance-plan.md @@ -1,7 +1,7 @@ # Plan: pygeoapi performance and correctness (post-standup) -Status: T1–T4 done on DIE-363 and deployed (2026-09-24); T5 deferred (rebuild is manual -for now); T6 partly (format version); T7 partly (per-feature bbox). See "Progress" below. +Status: T1–T4 done on DIE-363 and deployed (2026-09-24); T5 done on DIE-364 +(`pygeoapi_rebuild` sensor); T6 partly (format version); T7 partly (per-feature bbox). See "Progress" below. Picks up from the pygeoapi standup (PR #137, branch `DIE-359-deploy-pygeo-api-on-cloud-run`) and the per-product GeoParquet work (branch `DIE-363-update-the-layer-generation-code-to-land-in-pygeo-api-instead-of-geo-server`, @@ -18,7 +18,7 @@ Measured locally: patched image, files on local disk, 4 workers, 2 GiB. | T2 pin + patch | done | pages are exactly 5,000 rows, no overlap, 48,809 distinct ids; date-only `datetime` returns 200 | | T3 sort | done, changed | sorted by **`id`, `source`**, then `datetime` (not `source` first: with source first the `id` ranges of row groups overlap and well 9033 hit 4 of 9 groups; id first, 1–2). Whole series 1.3–1.4 s vs 1.6–2.5 s time-sorted | | T4 bake | done | `--bake-to`; 31/31 collections pass items/bbox/by-id from baked files; 88 MB baked; `limit=10` in 13–35 ms | -| T5 rebuild | deferred | Not automated yet. Rebuild manually after product runs publish new data (SPEC §6.6) | +| T5 rebuild | done (DIE-364) | `pygeoapi_rebuild` sensor starts the Cloud Build trigger once product runs finish (SPEC §6.6) | | T6 | partial | `PARQUET_FORMAT_VERSION` done (now `"2"`, so every existing `latest.parquet` is rewritten on the next run). Others open | | T7 | partial | per-feature `bbox` removed (with the empty-result fix below). The `bbox` covering column is still a property | @@ -36,10 +36,10 @@ memory stayed at 26–29% of 2 GiB, and the latency breakdown is almost all user total, so clients can fetch pages in parallel. - Not fixed here (separate ticket): WQP `source` labels differ between products. See T6. -Until T5 is done, **each pipeline run that changes data needs a manual pygeoapi build** -(`orchestration/pygeoapi/cloudbuild.yaml`), started after the product runs finish. -Whatever account runs it needs `storage.objectViewer` on the products bucket for the -bake, as well as what builds need today. +The rebuild (`orchestration/pygeoapi/cloudbuild.yaml`) is started by the +`pygeoapi_rebuild` sensor after product runs finish (T5). The build's account needs +`storage.objectViewer` on the products bucket for the bake, as well as what builds need +today. New finding — **memory at 2 GiB with 4 workers.** Under sustained 10,000-row page load the local container's cgroup peak reached its 2 GiB cap (anon 1.88 GiB; workers 470–650 MB @@ -208,29 +208,21 @@ Do them in order. Each is independently shippable. Re-run the measurement protoc - The build log lists each product's baked file and the reason. - All 31 collections return 200. -### T5 — Rebuild after a pipeline run (deferred; manual for now) - -Not being done yet. Until it is, rebuild by hand after product runs publish new data -(SPEC §6.6). Design notes for when it's picked up: - -- Create a Cloud Build **trigger** on the GitHub repo, `main` branch, manual invocation, - using `orchestration/pygeoapi/cloudbuild.yaml`. Triggers fill `COMMIT_SHA` - automatically. -- Start it when a pipeline run has published new data. Recommended: a Dagster - job/sensor that runs after the scheduled product runs complete. If any product's - materialization has `parquet_status: written` or `skipped_unchanged: false`, it calls - the Cloud Build `projects.triggers.run` API. - - One rebuild per run, not per product. Debounce, and never build mid-run, so the image - never bakes a half-updated set. - - The Dagster+ credentials (`GCP_SERVICE_ACCOUNT_KEY`) need permission to run the - trigger (`roles/cloudbuild.builds.editor`, or a narrower custom role). - - Simpler fallback: Cloud Scheduler runs the trigger on a fixed cadence after the - pipeline window. -- Schedules today are **monthly and staggered** (`products.yaml`: days 1, 15, 22, 23, 25 - at 06:00 UTC). Moving to weekly means updating those `schedule:` values. With the - current staggering, the trigger fires up to 5 times a month. -- Accept: a pipeline run that changes a product results in exactly one build, and the new - revision serves the new data with no manual step. +### T5 — Rebuild after a pipeline run (done, DIE-364) + +The `pygeoapi_rebuild` sensor (`orchestration/pygeoapi_rebuild.py`, SPEC §6.6) polls +every 10 minutes. After a product job succeeds, and once no product job is queued or +running, it calls `projects.locations.triggers.run` on the `die-pygeoapi-rebuild` Cloud +Build trigger. Several runs finishing on the same day give one build. + +- It launches no Dagster run. Sensor ticks cost no Dagster+ credits. +- It rebuilds after every successful product run, even if the data didn't change. + Skipping unchanged runs (`parquet_status`) was left out for simplicity; builds fall + within the Cloud Build free tier. +- Schedules are **monthly and staggered** (`products.yaml`: days 1, 15, 22, 23, 25 at + 06:00 America/Denver), so expect up to 5 builds a month. +- It is stopped by default (so local and branch deployments never rebuild prod). Turn it + on in the prod deployment. ### T6 — Pipeline correctness fixes diff --git a/orchestration/AGENTS.md b/orchestration/AGENTS.md index 0e9c92f..383c2da 100644 --- a/orchestration/AGENTS.md +++ b/orchestration/AGENTS.md @@ -55,7 +55,8 @@ uv run python -c "import orchestration.definitions; print('ok')" `assets/products.py:build_product_assets` — into a per-source asset graph: `sources/` → combine (``) → `/geoserver`. - `definitions.py` wires assets, one asset job per product (`_job`), - schedules, and resources (`die_config`, `gcs`, `geoserver`, `io_manager`). + schedules, the `pygeoapi_rebuild` sensor (`pygeoapi_rebuild.py`), and resources + (`die_config`, `gcs`, `geoserver`, `io_manager`). - Resources live in `resources/`. The backend DIE engine is the `nmuwd` package (repo root `backend/`); orchestration only drives it. - Per-source and geoserver assets soft-fail: errors surface as red asset checks @@ -67,7 +68,8 @@ Dagster+ serverless builds from the repo root per `dagster_cloud.yaml` (`module_name: orchestration.definitions`). Runtime deps come from the root `requirements.txt`. Secrets are set as Dagster+ env vars, not in code: -- `GCP_SERVICE_ACCOUNT_KEY` — GCS upload/IO-manager auth (JSON key). +- `GCP_SERVICE_ACCOUNT_KEY` — JSON key for GCS upload/IO-manager and the + `pygeoapi_rebuild` Cloud Build trigger. - `GEOSERVER_URL` / `GEOSERVER_USER` / `GEOSERVER_PASSWORD` / `GEOSERVER_WORKSPACE`. - `USGS_API_KEY` — USGS/NWIS API key. Without it the USGS water data API is heavily rate-limited. Resolved via `dg.EnvVar` into `DIEConfigResource`, which