diff --git a/automated_ingestion/defs/definitions.py b/automated_ingestion/defs/definitions.py index aabacc42a..3d4bd6c72 100644 --- a/automated_ingestion/defs/definitions.py +++ b/automated_ingestion/defs/definitions.py @@ -24,10 +24,16 @@ from dagster import Definitions from automated_ingestion.defs.assets import all_assets +from automated_ingestion.defs.jobs.san_acacia import ( + san_acacia_job, + san_acacia_weekly_schedule, +) from automated_ingestion.defs.resources import OcotilloDatabase defs = Definitions( assets=all_assets(), + jobs=[san_acacia_job], + schedules=[san_acacia_weekly_schedule], resources={"database": OcotilloDatabase()}, ) diff --git a/automated_ingestion/defs/jobs/san_acacia.py b/automated_ingestion/defs/jobs/san_acacia.py new file mode 100644 index 000000000..d38c1c35c --- /dev/null +++ b/automated_ingestion/defs/jobs/san_acacia.py @@ -0,0 +1,80 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +The scheduled run for San Acacia Reach. + +One job over the whole `san_acacia` asset group, so the three steps stay in +order: land the point roster, land the readings, then map and load them. Dagster +resolves that from the asset dependencies rather than from anything declared +here, which is why the selection is by group -- a fourth asset added to the +group joins the schedule without this file changing. + +Weekly rather than daily. These are five-minute diver readings and nobody is +watching them in real time; the vendor's endpoint answers 500 when pushed, and a +weekly cadence keeps each run's windows comfortably inside what it serves. The +watermark makes the interval a matter of freshness rather than correctness: a +run fetches from wherever the last one finished, so a missed week is picked up +by the next run rather than lost. +""" + +from dagster import ( + AssetSelection, + DefaultScheduleStatus, + RetryPolicy, + ScheduleDefinition, + define_asset_job, +) + +SAN_ACACIA_GROUP = "san_acacia" + +san_acacia_job = define_asset_job( + name="san_acacia_ingest", + selection=AssetSelection.groups(SAN_ACACIA_GROUP), + description=( + "Land the San Acacia point roster and readings in the raw zone, then " + "map and load them into Ocotillo." + ), + # A retry covers the vendor dropping a request or a token expiring mid-run. + # Two attempts, not more: a persistent 500 means the window is wrong or the + # endpoint is unwell, and hammering it makes both worse. + op_retry_policy=RetryPolicy(max_retries=2, delay=60), +) + +san_acacia_weekly_schedule = ScheduleDefinition( + name="san_acacia_weekly", + job=san_acacia_job, + # Mondays at 05:00 America/Denver -- after midnight so a run covers whole + # days, and early enough that a failure is visible at the start of the week + # rather than discovered the following Monday. + cron_schedule="0 5 * * 1", + execution_timezone="America/Denver", + # Local time rather than UTC deliberately: the wells, the people who read + # the data, and the working day are all in one timezone, so a schedule that + # shifts by an hour twice a year would be the surprising choice. + # + # Stopped by default. Turning it on starts writing to Ocotillo, and the + # first run for the 24 wells without history fetches back to the + # `INITIAL_START` floor. That should be somebody's decision, taken once, + # rather than a consequence of a merge. + default_status=DefaultScheduleStatus.STOPPED, + description=( + "Weekly San Acacia ingest. Each run resumes from each series' " + "watermark, so a missed week is caught up rather than lost." + ), +) + + +# ============= EOF ============================================= diff --git a/automated_ingestion/tests/test_schedule.py b/automated_ingestion/tests/test_schedule.py new file mode 100644 index 000000000..4b2000e87 --- /dev/null +++ b/automated_ingestion/tests/test_schedule.py @@ -0,0 +1,74 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +The weekly schedule selects what it claims to. + +A group name is a string, so a typo yields a schedule that runs successfully and +ingests nothing -- which looks like everything is fine. +""" + +from dagster import DefaultScheduleStatus + +from automated_ingestion.defs.definitions import defs + +EXPECTED = { + "raw_san_acacia_locations", + "raw_san_acacia_readings", + "san_acacia_observations", +} + + +def _schedule(): + return next(s for s in defs.schedules if s.name == "san_acacia_weekly") + + +def test_the_schedule_is_registered(): + assert _schedule().job.name == "san_acacia_ingest" + + +def test_it_selects_every_san_acacia_asset_and_nothing_else(): + selected = { + key.to_user_string() + for key in _schedule().job.selection.resolve(list(defs.assets)) + } + assert selected == EXPECTED + + +def test_operations_assets_are_excluded(): + # ingestion_heartbeat and database_connectivity are diagnostics. Running + # them weekly would add noise and, for connectivity, a pointless query. + selected = { + key.to_user_string() + for key in _schedule().job.selection.resolve(list(defs.assets)) + } + assert "ingestion_heartbeat" not in selected + assert "database_connectivity" not in selected + + +def test_it_runs_weekly_in_local_time(): + schedule = _schedule() + assert schedule.cron_schedule == "0 5 * * 1" + assert schedule.execution_timezone == "America/Denver" + + +def test_it_is_stopped_until_somebody_starts_it(): + # Turning it on begins writing to Ocotillo, and the first run for the wells + # without history fetches back to the floor. That is a decision, not a + # consequence of a merge. + assert _schedule().default_status is DefaultScheduleStatus.STOPPED + + +# ============= EOF ============================================= diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index 87a1349ab..5eaffdab5 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -419,11 +419,14 @@ Covers raw already in GCS with only the mapping wrong: adapter or unit bug, newl ### 4.4 — Schedule, observability, alerting -- `san_acacia_schedule` runs the daily pipeline; cron avoids contention with existing Dagster+ jobs in the org, recorded in the source registry. -- Dagster logs bridge into the repo's existing logging setup, so ingestion failures surface where the team already looks. Confirm which error-tracking destination is current before wiring this — do not assume the repo's existing integrations are live. -- A failed run notifies someone — not discovered via a stale hydrograph. -- Every run emits rows ingested, rows upserted, entities processed, entities failed, adapter failures, resulting watermark per series. -- A zero-new-rows run succeeds and is distinguishable in the logs from a failure. +Schedule built. `defs/jobs/san_acacia.py` — `san_acacia_ingest` over the whole `san_acacia` asset group, on `san_acacia_weekly`. + +- ✅ **Weekly, not daily.** These are five-minute diver readings nobody watches in real time, the vendor's endpoint answers 500 when pushed, and the watermark makes the interval a question of freshness rather than correctness — a missed week is caught up by the next run, not lost. +- ✅ Mondays 05:00 **America/Denver**, not UTC. The wells, the people reading the data and the working day are in one timezone; a schedule drifting an hour twice a year would be the surprising choice. After midnight so a run covers whole days, early enough that a failure is visible at the start of the week. +- ✅ Selected **by group**, so an asset added to `san_acacia` joins the schedule without touching the job. A test asserts the selection resolves to exactly the three ingest assets and excludes `ingestion_heartbeat` and `database_connectivity` — a group-name typo would otherwise produce a schedule that runs happily and ingests nothing. +- ✅ `RetryPolicy(max_retries=2, delay=60)` covers a dropped request or a token expiring mid-run. Two attempts, not more: a persistent 500 means the window is wrong or the endpoint is unwell, and hammering it makes both worse. +- ✅ **`DefaultScheduleStatus.STOPPED`.** Turning it on starts writing to Ocotillo, and the first run for the 24 wells without history fetches back to `INITIAL_START`. That should be a decision taken once, not a consequence of a merge. +- ⬜ Observability and alerting — log bridge, failure notification, run metadata. ### 4.5 — Documentation