diff --git a/automated_ingestion/iac/main.tf b/automated_ingestion/iac/main.tf index 47bd0dac3..295cf3a35 100644 --- a/automated_ingestion/iac/main.tf +++ b/automated_ingestion/iac/main.tf @@ -107,6 +107,24 @@ resource "google_storage_bucket_iam_member" "ingestion_object_admin" { member = "serviceAccount:${google_service_account.ingestion.email}" } + +# objectAdmin covers objects and says nothing about the bucket itself, so it +# does not include storage.buckets.get. gcsfs checks a bucket exists before +# writing to it, that check is denied, and GCS reports a denial as absence -- +# so the pipeline fails with "Bucket does not exist" for a bucket that plainly +# does. +# +# legacyBucketReader adds buckets.get and objects.list and nothing else. It is +# the narrowest standard role that makes the existence check succeed; the +# alternative, storage.admin, would also grant deletion of the bucket. +resource "google_storage_bucket_iam_member" "ingestion_bucket_reader" { + for_each = google_storage_bucket.ingestion_raw + + bucket = each.value.name + role = "roles/storage.legacyBucketReader" + 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 diff --git a/automated_ingestion/sources/san_acacia/dlt_pipeline.py b/automated_ingestion/sources/san_acacia/dlt_pipeline.py index 25b08a270..4c64088f4 100644 --- a/automated_ingestion/sources/san_acacia/dlt_pipeline.py +++ b/automated_ingestion/sources/san_acacia/dlt_pipeline.py @@ -171,8 +171,16 @@ def _approved_timestamps( return set() -def build_pipeline(environment: str) -> Any: - """A dlt pipeline writing parquet to the raw zone for one environment.""" +def build_pipeline() -> Any: + """A dlt pipeline writing parquet to the raw zone. + + The pipeline is named after the bucket it writes to rather than after a + separately supplied environment. Those were two sources of truth for one + fact, and they disagreed the first time this ran in Dagster+: a pipeline + called ``san_acacia_staging`` writing to the production bucket, because the + name came from a run tag that was absent and the bucket came from the + environment. Deriving one from the other makes that impossible. + """ # gcsfs resolves Application Default Credentials the same way the Cloud SQL # connector does, and Serverless supplies none of its own. from automated_ingestion.shared.credentials import ( @@ -181,10 +189,11 @@ def build_pipeline(environment: str) -> Any: ensure_application_default_credentials() + bucket = raw_zone_bucket() return dlt.pipeline( - pipeline_name=f"san_acacia_{environment}", + pipeline_name=f"{SOURCE.key}_{bucket}", destination=dlt.destinations.filesystem( - bucket_url=f"gs://{raw_zone_bucket()}", + bucket_url=f"gs://{bucket}", layout=RAW_LAYOUT, ), dataset_name=SOURCE.dataset_name, diff --git a/automated_ingestion/sources/san_acacia/ingest.py b/automated_ingestion/sources/san_acacia/ingest.py index 6486f2a2d..4d7d8a021 100644 --- a/automated_ingestion/sources/san_acacia/ingest.py +++ b/automated_ingestion/sources/san_acacia/ingest.py @@ -50,7 +50,7 @@ def raw_san_acacia_locations(context: AssetExecutionContext) -> Output[int]: client = _client() points = list(client.monitoring_points(PROJECT_ID)) - pipeline = build_pipeline(context.run.tags.get("environment", "staging")) + pipeline = build_pipeline() pipeline.run(vanessen_locations(client)) context.log.info("landed %s monitoring points", len(points)) @@ -84,7 +84,7 @@ def raw_san_acacia_readings(context: AssetExecutionContext) -> Output[int]: ] end = int(datetime.now(tz=timezone.utc).timestamp()) - pipeline = build_pipeline(context.run.tags.get("environment", "staging")) + pipeline = build_pipeline() failures: list[dict[str, Any]] = [] info = pipeline.run(vanessen_readings(client, points, end, failures)) rows = _row_count(info) diff --git a/automated_ingestion/tests/test_gcs.py b/automated_ingestion/tests/test_gcs.py index b9875a1d5..ad219715f 100644 --- a/automated_ingestion/tests/test_gcs.py +++ b/automated_ingestion/tests/test_gcs.py @@ -59,4 +59,19 @@ def test_layout_partitions_by_date(): assert "day={DD}" in RAW_LAYOUT +def test_pipeline_name_follows_the_bucket(monkeypatch): + # The name and the destination must not be able to disagree. They did once: + # a pipeline called san_acacia_staging wrote to the production bucket, + # because the name came from an absent run tag and the bucket from the + # environment. + monkeypatch.setenv(BUCKET_ENV_VAR, "ocotillo-ingestion-production") + monkeypatch.delenv("GCS_BUCKET_NAME", raising=False) + monkeypatch.setenv("INGESTION_GCP_CREDENTIALS_JSON", "") + + from automated_ingestion.sources.san_acacia.dlt_pipeline import build_pipeline + + pipeline = build_pipeline() + assert "ocotillo-ingestion-production" in pipeline.pipeline_name + + # ============= EOF =============================================