fix(ingestion): grant bucket read, and name the pipeline after the bucket

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

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,

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)

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 =============================================