From 976a861e68e4289912e623b1c0f17651dc40b3bc Mon Sep 17 00:00:00 2001 From: jakeross Date: Tue, 18 Aug 2026 15:44:57 -0700 Subject: [PATCH 1/3] chore(ingestion): support dg for code location environment variables The dagster-cloud CLI has no environment-variable command at all; dg does, but refuses to run outside a directory it recognises as a project. Adding the tool.dg blocks satisfies that check. It does not affect what the deployed code location loads, since dagster_cloud.yaml names the entry point explicitly -- verified by loading the definitions and running the suite with the blocks in place. The script sets each phase in one pass. Secrets go through --from-local-env rather than arguments, so they never reach the command line or shell history. Phases are separate deliberately. `database` waits on ingestion_role.sql having been run: pointing CLOUD_SQL_* at a role that does not exist makes database_connectivity fail in a way that looks like the serverless-to-Cloud-SQL problem it exists to test. Note for whoever runs this: it needs a Dagster+ *user* token. An agent token authenticates and returns data for queries, but is unauthorized for these mutations, and dg surfaces that as a KeyError on its own error handler rather than as a permission message. Co-Authored-By: Claude Opus 5 --- .../scripts/set_code_location_env.sh | 70 +++++++++++++++++++ pyproject.toml | 10 +++ 2 files changed, 80 insertions(+) create mode 100755 automated_ingestion/scripts/set_code_location_env.sh diff --git a/automated_ingestion/scripts/set_code_location_env.sh b/automated_ingestion/scripts/set_code_location_env.sh new file mode 100755 index 000000000..5ece6926c --- /dev/null +++ b/automated_ingestion/scripts/set_code_location_env.sh @@ -0,0 +1,70 @@ +#!/usr/bin/env bash +# Set every environment variable the ocotillo-automated-ingestion code location +# needs, in one pass. +# +# Requires a Dagster+ *user* token -- an agent token authenticates but is not +# authorized for these mutations, and dg reports that as an unhelpful KeyError: +# dg plus config set --api-token 'user:...' +# +# Secrets are never passed as arguments. `--from-local-env` reads them from this +# shell, so nothing sensitive reaches the command line, your shell history, or +# the Dagster+ audit log's argument capture. Export them first: +# +# read -rs "DIVERHUB_USERNAME?Diver-HUB username: "; echo +# read -rs "DIVERHUB_PASSWORD?Diver-HUB password: "; echo +# export DIVERHUB_USERNAME DIVERHUB_PASSWORD +# +# Usage: +# ./automated_ingestion/scripts/set_code_location_env.sh storage +# ./automated_ingestion/scripts/set_code_location_env.sh vendor +# ./automated_ingestion/scripts/set_code_location_env.sh database +# +# The phases are separate on purpose. `database` should wait until +# automated_ingestion/sql/ingestion_role.sql has been run: setting CLOUD_SQL_* +# against a role that does not exist yet makes database_connectivity fail in a +# way that looks like the serverless-to-Cloud-SQL problem it is meant to test. +set -euo pipefail + +DG="uv run --with dagster-dg-cli dg" +PHASE="${1:-}" + +set_var() { echo " $1"; $DG plus create env "$@" --global -y >/dev/null; } + +case "$PHASE" in +storage) + echo "Raw-zone buckets (different value per scope):" + set_var INGESTION_GCS_BUCKET ocotillo-ingestion-production --scope full + set_var INGESTION_GCS_BUCKET ocotillo-ingestion-staging --scope branch + ;; +vendor) + : "${DIVERHUB_USERNAME:?export it first, see the header}" + : "${DIVERHUB_PASSWORD:?export it first, see the header}" + echo "Diver-HUB credentials (values read from this shell, not echoed):" + set_var DIVERHUB_USERNAME --from-local-env + set_var DIVERHUB_PASSWORD --from-local-env + ;; +database) + : "${CLOUD_SQL_INSTANCE_NAME:?export it first}" + : "${CLOUD_SQL_DATABASE:?export it first}" + echo "Cloud SQL connection:" + set_var DB_DRIVER cloudsql + set_var CLOUD_SQL_IP_TYPE public + set_var CLOUD_SQL_USER ocotillo_ingestion + set_var CLOUD_SQL_INSTANCE_NAME --from-local-env + set_var CLOUD_SQL_DATABASE --from-local-env + # Prefer IAM auth: it removes the password entirely, and ingestion_role.sql + # documents creating the role as "ocotillo-ingestion@PROJECT.iam" instead. + if [ -n "${CLOUD_SQL_PASSWORD:-}" ]; then + set_var CLOUD_SQL_PASSWORD --from-local-env + else + echo " CLOUD_SQL_PASSWORD unset -- assuming IAM auth" + set_var CLOUD_SQL_IAM_AUTH 1 + fi + ;; +*) + echo "usage: $0 {storage|vendor|database}" >&2 + exit 64 + ;; +esac + +echo "Done. Verify in Dagster+ under Deployment -> Environment variables." diff --git a/pyproject.toml b/pyproject.toml index 0cdc5c86a..b33d0e623 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -131,6 +131,16 @@ packages = [ [tool.dagster] module_name = "automated_ingestion.defs.definitions" +# Required by the `dg` CLI, which refuses to run outside a directory it +# recognises as a project. `dagster_cloud.yaml` names the entry point +# explicitly, so this does not affect what the deployed code location loads -- +# it only lets `dg plus` manage environment variables from this checkout. +[tool.dg] +directory_type = "project" + +[tool.dg.project] +root_module = "automated_ingestion" + # Bare `--cov` measures every imported module, which pulls the whole virtualenv # into the report. Scope it to first-party code instead. Keep this a single # source root -- listing each package separately makes coverage treat every From f246d61107f376a23add2856968eda6c5085c70f Mon Sep 17 00:00:00 2001 From: jakeross Date: Tue, 18 Aug 2026 15:52:13 -0700 Subject: [PATCH 2/3] fix(ingestion): make the IAM database path internally consistent CLOUD_SQL_USER means different things in the two authentication modes and db/engine.py passes it straight to the connector either way. The script set it to the plain role name unconditionally and then chose IAM auth when no password was exported, so the two settings contradicted each other -- a combination that fails as an authentication error looking like a missing grant. It now derives the value from whichever branch it takes. The role DDL leads with the IAM role for the same reason, since that is the configured path, and states the exact string CLOUD_SQL_USER has to match. IAM authentication also needs GCP-side grants that this configuration did not create: cloudsql.client, cloudsql.instanceUser, and the service account registered as a CLOUD_IAM_SERVICE_ACCOUNT database user. Without them the Postgres role exists but cannot be reached. They are gated on a cloud_sql_instance variable so the storage half can still be applied before the database half is decided. Variables are no longer set with --global. This Dagster+ deployment hosts other code locations, and deployment-level scope made the vendor and database credentials readable by all of them. Co-Authored-By: Claude Opus 5 --- automated_ingestion/iac/main.tf | 37 +++++++++++++++++++ automated_ingestion/iac/variables.tf | 6 +++ .../scripts/set_code_location_env.sh | 28 +++++++++++--- automated_ingestion/sql/ingestion_role.sql | 27 ++++++++++---- 4 files changed, 85 insertions(+), 13 deletions(-) diff --git a/automated_ingestion/iac/main.tf b/automated_ingestion/iac/main.tf index 9df05f48a..e432f1808 100644 --- a/automated_ingestion/iac/main.tf +++ b/automated_ingestion/iac/main.tf @@ -101,3 +101,40 @@ resource "google_storage_bucket_iam_member" "ingestion_object_admin" { role = "roles/storage.objectAdmin" member = "serviceAccount:${google_service_account.ingestion.email}" } + +# Database access for the ingestion service account. +# +# Only created when `cloud_sql_instance` is set, so the storage half of this +# configuration can be applied before the database half is decided. +# +# These grants are what make IAM database authentication work. Without them the +# Postgres role in automated_ingestion/sql/ingestion_role.sql exists but cannot +# be reached: the connector fails while acquiring a token, which surfaces as an +# authentication error and reads like a missing GRANT. +resource "google_project_iam_member" "ingestion_cloudsql_client" { + count = var.cloud_sql_instance == null ? 0 : 1 + + project = var.project_id + role = "roles/cloudsql.client" + member = "serviceAccount:${google_service_account.ingestion.email}" +} + +resource "google_project_iam_member" "ingestion_cloudsql_instance_user" { + count = var.cloud_sql_instance == null ? 0 : 1 + + project = var.project_id + role = "roles/cloudsql.instanceUser" + member = "serviceAccount:${google_service_account.ingestion.email}" +} + +# Registers the service account as a database user. The Postgres role itself, +# and its grants, come from ingestion_role.sql -- this only makes the login +# possible. +resource "google_sql_user" "ingestion" { + count = var.cloud_sql_instance == null ? 0 : 1 + + name = trimsuffix(google_service_account.ingestion.email, ".gserviceaccount.com") + instance = var.cloud_sql_instance + project = var.project_id + type = "CLOUD_IAM_SERVICE_ACCOUNT" +} diff --git a/automated_ingestion/iac/variables.tf b/automated_ingestion/iac/variables.tf index c06187f4e..d7440af8c 100644 --- a/automated_ingestion/iac/variables.tf +++ b/automated_ingestion/iac/variables.tf @@ -14,3 +14,9 @@ variable "bucket_location" { description = "Bucket location. US-CENTRAL1 keeps the raw zone in the same region as Cloud SQL, so replay reads do not cross regions." default = "US-CENTRAL1" } + +variable "cloud_sql_instance" { + type = string + description = "Cloud SQL instance name for the IAM database user. Leave null to skip the database grants entirely -- useful before the instance is known, or when using password authentication instead." + default = null +} diff --git a/automated_ingestion/scripts/set_code_location_env.sh b/automated_ingestion/scripts/set_code_location_env.sh index 5ece6926c..35a263b59 100755 --- a/automated_ingestion/scripts/set_code_location_env.sh +++ b/automated_ingestion/scripts/set_code_location_env.sh @@ -14,6 +14,11 @@ # read -rs "DIVERHUB_PASSWORD?Diver-HUB password: "; echo # export DIVERHUB_USERNAME DIVERHUB_PASSWORD # +# Variables are scoped to this code location, not the deployment. If an earlier +# run set them with --global, delete those deployment-level entries in the +# Dagster+ UI afterwards -- otherwise both exist and which one wins is not +# obvious from either place. +# # Usage: # ./automated_ingestion/scripts/set_code_location_env.sh storage # ./automated_ingestion/scripts/set_code_location_env.sh vendor @@ -28,7 +33,10 @@ set -euo pipefail DG="uv run --with dagster-dg-cli dg" PHASE="${1:-}" -set_var() { echo " $1"; $DG plus create env "$@" --global -y >/dev/null; } +# No --global: that sets the variable at deployment level, where every other +# code location in this deployment can read it. This deployment is shared, so +# the vendor and database credentials stay scoped to this location. +set_var() { echo " $1"; $DG plus create env "$@" -y >/dev/null; } case "$PHASE" in storage) @@ -49,16 +57,26 @@ database) echo "Cloud SQL connection:" set_var DB_DRIVER cloudsql set_var CLOUD_SQL_IP_TYPE public - set_var CLOUD_SQL_USER ocotillo_ingestion set_var CLOUD_SQL_INSTANCE_NAME --from-local-env set_var CLOUD_SQL_DATABASE --from-local-env - # Prefer IAM auth: it removes the password entirely, and ingestion_role.sql - # documents creating the role as "ocotillo-ingestion@PROJECT.iam" instead. + + # CLOUD_SQL_USER means different things in the two auth modes, and db/engine.py + # passes it straight to the connector either way. Under IAM auth it must be the + # service account with the .gserviceaccount.com suffix stripped; a plain + # Postgres role name there fails as an authentication error that reads like a + # missing grant. Deriving it here keeps the two settings from contradicting + # each other. if [ -n "${CLOUD_SQL_PASSWORD:-}" ]; then + echo " (password auth)" + set_var CLOUD_SQL_IAM_AUTH 0 + set_var CLOUD_SQL_USER ocotillo_ingestion set_var CLOUD_SQL_PASSWORD --from-local-env else - echo " CLOUD_SQL_PASSWORD unset -- assuming IAM auth" + IAM_SA="${INGESTION_SERVICE_ACCOUNT:-ocotillo-ingestion@waterdatainitiative-271000.iam.gserviceaccount.com}" + IAM_USER="${IAM_SA%.gserviceaccount.com}" + echo " (IAM auth as ${IAM_USER})" set_var CLOUD_SQL_IAM_AUTH 1 + set_var CLOUD_SQL_USER "$IAM_USER" fi ;; *) diff --git a/automated_ingestion/sql/ingestion_role.sql b/automated_ingestion/sql/ingestion_role.sql index 65afb657b..990b9a5e1 100644 --- a/automated_ingestion/sql/ingestion_role.sql +++ b/automated_ingestion/sql/ingestion_role.sql @@ -11,16 +11,27 @@ -- and NMW_* tables, so a bug in an adapter cannot corrupt data no ingestion -- path should ever reach. --- Set the password out of band; do not commit it. It belongs in Secret --- Manager alongside internal-ogc-api-keys. --- CREATE ROLE ocotillo_ingestion LOGIN PASSWORD '...'; +-- IAM authentication is the configured path, and the reason is that it removes +-- the credential rather than rotating it: Cloud SQL mints a short-lived token +-- from the service account, so there is no password to store in Dagster+, in +-- Secret Manager, or here. +-- +-- The role name is the service account with the .gserviceaccount.com suffix +-- stripped. That exact string is also what CLOUD_SQL_USER must be set to -- +-- db/engine.py passes it straight to the connector, and a plain role name there +-- fails as an authentication error that reads like a missing grant. +-- +-- CREATE ROLE "ocotillo-ingestion@waterdatainitiative-271000.iam" WITH LOGIN; +-- GRANT cloudsqliamuser TO "ocotillo-ingestion@waterdatainitiative-271000.iam"; -- --- Or, preferred, use IAM database authentication and create the role for the --- service account instead, so there is no password to rotate: --- CREATE ROLE "ocotillo-ingestion@PROJECT.iam" WITH LOGIN; --- GRANT cloudsqliamuser TO "ocotillo-ingestion@PROJECT.iam"; +-- Password authentication, if IAM is ever unavailable. Set the password out of +-- band and store it in Secret Manager; never commit it, and set +-- CLOUD_SQL_IAM_AUTH=0 so the two settings agree. +-- +-- CREATE ROLE ocotillo_ingestion LOGIN PASSWORD '...'; -\set role_name ocotillo_ingestion +-- Set to match whichever role was created above. +\set role_name "ocotillo-ingestion@waterdatainitiative-271000.iam" GRANT CONNECT ON DATABASE :"db_name" TO :"role_name"; GRANT USAGE ON SCHEMA public TO :"role_name"; From a4259545fd47344f4e8262178198e85463c9b0bb Mon Sep 17 00:00:00 2001 From: jakeross Date: Tue, 18 Aug 2026 15:55:57 -0700 Subject: [PATCH 3/3] fix(ingestion): make db and domain importable in the deployed image database_connectivity failed in Dagster+ with ModuleNotFoundError: No module named 'db', while the code location itself loaded and ingestion_heartbeat ran. The image copies the repository to /opt/dagster/app but never installs it: the generated requirements omit the project, and the build template only runs `pip install .` when a setup.py exists. So db and domain are importable only while that directory is on sys.path -- true when Dagster loads the code location, not guaranteed in the separate process that executes a step, which is exactly where the loader's lazy imports run. Locally an editable install puts the repository on sys.path unconditionally, which is why 43 tests pass and the failure appeared only once deployed. Two tests now cover it: one asserts the path entry exists, the other imports db from a process whose working directory is not the repository. Importing automated_ingestion now adds the repository root itself, so the coupling is satisfied wherever the package is imported from rather than depending on how it was launched. Co-Authored-By: Claude Opus 5 --- automated_ingestion/__init__.py | 17 +++++++++ .../tests/test_connectivity.py | 36 +++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/automated_ingestion/__init__.py b/automated_ingestion/__init__.py index 1fb584ba5..de197f548 100644 --- a/automated_ingestion/__init__.py +++ b/automated_ingestion/__init__.py @@ -28,6 +28,23 @@ the first source; ``shared/`` holds what later sources reuse. See ``docs/automated-ingestion-pipeline-plan.md``. + +Importing this package puts the repository root on ``sys.path``. That is +unusual and deliberate. The Dagster+ image copies the repository to +``/opt/dagster/app`` but never installs it -- the generated requirements omit +the project, and the build template only runs ``pip install .`` when a +``setup.py`` exists -- so ``db`` and ``domain`` are importable only while that +directory happens to be on the path. It is, when Dagster loads the code +location; it is not guaranteed in the separate process that executes a step, +which is where the loader's imports actually run. Locally the editable install +hides the difference entirely, so the failure appears only once deployed. """ +import sys as _sys +from pathlib import Path as _Path + +_REPOSITORY_ROOT = _Path(__file__).resolve().parent.parent +if str(_REPOSITORY_ROOT) not in _sys.path: + _sys.path.insert(0, str(_REPOSITORY_ROOT)) + # ============= EOF ============================================= diff --git a/automated_ingestion/tests/test_connectivity.py b/automated_ingestion/tests/test_connectivity.py index da5453139..35d4d83ae 100644 --- a/automated_ingestion/tests/test_connectivity.py +++ b/automated_ingestion/tests/test_connectivity.py @@ -57,4 +57,40 @@ def test_loading_definitions_does_not_import_db_engine(): assert result.stdout.strip() == "False", result.stdout +def test_importing_the_package_makes_the_repository_importable(): + # The deployed image never installs this project, so `db` and `domain` + # resolve only if the repository root is on sys.path. Locally an editable + # install provides that and hides the difference, which is why this failed + # only once deployed -- the code location loaded fine and the step that + # imported db died. + import sys + + import automated_ingestion + + assert str(automated_ingestion._REPOSITORY_ROOT) in sys.path + + +def test_db_imports_from_an_unrelated_working_directory(): + # Reproduces the deployed condition: a process whose cwd is not the + # repository. The lazy imports in the resource and the connectivity asset + # run at step execution, not at load, so this is the path that broke. + import subprocess + import sys + + result = subprocess.run( + [ + sys.executable, + "-c", + "import automated_ingestion; " + "from db.transducer import TransducerObservation; " + "print(TransducerObservation.__tablename__)", + ], + capture_output=True, + text=True, + cwd="/", + ) + assert result.returncode == 0, result.stderr + assert "transducer_observation" in result.stdout + + # ============= EOF =============================================