Add iceberg header on table calls (#17912)

Author: rambleraptor

mmv1/products/biglakeiceberg/IcebergTable.yaml

@@ -43,6 +43,10 @@ custom_code:
   update_encoder: templates/terraform/encoders/biglake_iceberg_table.go.tmpl
   constants: templates/terraform/constants/biglake_iceberg_table.go.tmpl
   post_read: templates/terraform/post_read/biglake_iceberg_table.go.tmpl
+  pre_create: templates/terraform/pre_create/biglake_iceberg_table.go.tmpl
+  pre_read: templates/terraform/pre_read/biglake_iceberg_table.go.tmpl
+  pre_update: templates/terraform/pre_update/biglake_iceberg_table.go.tmpl
+  pre_delete: templates/terraform/pre_delete/biglake_iceberg_table.go.tmpl
 samples:
   - name: biglake_iceberg_table_basic
     primary_resource_id: my_iceberg_table

mmv1/templates/terraform/constants/biglake_iceberg_table.go.tmpl

@@ -51,3 +51,34 @@ func expandIcebergTableSortOrderForCommit(v interface{}) map[string]interface{}
 		"fields":   fields,
 	}
 }
+
+{{- if not (contains $.ProductMetadata.Compiler "terraformgoogleconversion") }}
+// addIcebergTableAccessDelegationHeader sets the X-Iceberg-Access-Delegation
+// header that the Iceberg REST catalog requires for every table operation when
+// the parent catalog uses credential_mode = CREDENTIAL_MODE_VENDED_CREDENTIALS.
+// The credential mode is a property of the catalog, not the table, so it is
+// looked up at request time. Catalogs in the default end-user mode reject the
+// header, so it is only added when vending is actually enabled.
+func addIcebergTableAccessDelegationHeader(d *schema.ResourceData, config *transport_tpg.Config, billingProject, userAgent string, headers http.Header) error {
+	url, err := tpgresource.ReplaceVars(d, config, "{{"{{"}}BiglakeIcebergBasePath{{"}}"}}iceberg/v1/restcatalog/extensions/projects/{{"{{"}}project{{"}}"}}/catalogs/{{"{{"}}catalog{{"}}"}}")
+	if err != nil {
+		return err
+	}
+
+	catalog, err := transport_tpg.SendRequest(transport_tpg.SendRequestOptions{
+		Config:    config,
+		Method:    "GET",
+		Project:   billingProject,
+		RawURL:    url,
+		UserAgent: userAgent,
+	})
+	if err != nil {
+		return fmt.Errorf("Error reading IcebergCatalog to determine credential mode: %s", err)
+	}
+
+	if mode, ok := catalog["credential-mode"].(string); ok && mode == "CREDENTIAL_MODE_VENDED_CREDENTIALS" {
+		headers.Add("X-Iceberg-Access-Delegation", "vended-credentials")
+	}
+	return nil
+}
+{{- end }}

mmv1/templates/terraform/pre_create/biglake_iceberg_table.go.tmpl

@@ -0,0 +1,3 @@
+if err := addIcebergTableAccessDelegationHeader(d, config, billingProject, userAgent, headers); err != nil {
+	return err
+}

mmv1/templates/terraform/pre_delete/biglake_iceberg_table.go.tmpl

@@ -0,0 +1,3 @@
+if err := addIcebergTableAccessDelegationHeader(d, config, billingProject, userAgent, headers); err != nil {
+	return err
+}

mmv1/templates/terraform/pre_read/biglake_iceberg_table.go.tmpl

@@ -0,0 +1,3 @@
+if err := addIcebergTableAccessDelegationHeader(d, config, billingProject, userAgent, headers); err != nil {
+	return err
+}

mmv1/templates/terraform/pre_update/biglake_iceberg_table.go.tmpl

@@ -0,0 +1,3 @@
+if err := addIcebergTableAccessDelegationHeader(d, config, billingProject, userAgent, headers); err != nil {
+	return err
+}

mmv1/third_party/terraform/services/biglakeiceberg/resource_biglake_iceberg_table_test.go

@@ -50,6 +50,87 @@ func TestAccBiglakeIcebergIcebergTable_update(t *testing.T) {
 	})
 }
 
+// TestAccBiglakeIcebergIcebergTable_vendedCredentials verifies that a table can
+// be created in a catalog configured with
+// credential_mode = CREDENTIAL_MODE_VENDED_CREDENTIALS. In this mode every table
+// operation must send the X-Iceberg-Access-Delegation: vended-credentials header;
+// without it the REST catalog rejects the create call.
+func TestAccBiglakeIcebergIcebergTable_vendedCredentials(t *testing.T) {
+	t.Parallel()
+
+	context := map[string]interface{}{
+		"random_suffix": acctest.RandString(t, 10),
+	}
+
+	acctest.VcrTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.AccTestPreCheck(t) },
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories(t),
+		CheckDestroy:             testAccCheckBiglakeIcebergIcebergTableDestroyProducer(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccBiglakeIcebergIcebergTable_vendedCredentials(context),
+			},
+			{
+				ResourceName:            "google_biglake_iceberg_table.my_iceberg_table",
+				ImportState:             true,
+				ImportStateVerify:       true,
+				ImportStateVerifyIgnore: []string{"catalog", "namespace"},
+			},
+		},
+	})
+}
+
+func testAccBiglakeIcebergIcebergTable_vendedCredentials(context map[string]interface{}) string {
+	return acctest.Nprintf(`
+resource "google_storage_bucket" "bucket" {
+  name          = "tf-test-my-bucket-%{random_suffix}"
+  location      = "us-central1"
+  force_destroy = true
+  uniform_bucket_level_access = true
+}
+
+resource "google_biglake_iceberg_catalog" "catalog" {
+  name = google_storage_bucket.bucket.name
+  catalog_type = "CATALOG_TYPE_GCS_BUCKET"
+  credential_mode = "CREDENTIAL_MODE_VENDED_CREDENTIALS"
+}
+
+# Credential vending downscopes a generated service account to access the GCS
+# bucket, so it must be granted access before any table operation runs.
+resource "google_storage_bucket_iam_member" "cv_sa_storage_admin" {
+  bucket = google_storage_bucket.bucket.name
+  role   = "roles/storage.admin"
+  member = "serviceAccount:${google_biglake_iceberg_catalog.catalog.biglake_service_account}"
+}
+
+resource "google_biglake_iceberg_namespace" "namespace" {
+  catalog = google_biglake_iceberg_catalog.catalog.name
+  namespace_id = "my_namespace_%{random_suffix}"
+}
+
+resource "google_biglake_iceberg_table" "my_iceberg_table" {
+  catalog   = google_biglake_iceberg_catalog.catalog.name
+  namespace = google_biglake_iceberg_namespace.namespace.namespace_id
+  name      = "tf-test-my_table_%{random_suffix}"
+  schema {
+    type = "struct"
+    fields {
+      id       = 1
+      name     = "id"
+      type     = "long"
+      required = true
+    }
+  }
+
+  properties = {
+    key = "initial"
+  }
+
+  depends_on = [google_storage_bucket_iam_member.cv_sa_storage_admin]
+}
+`, context)
+}
+
 func testAccBiglakeIcebergIcebergTable_updateInitial(context map[string]interface{}) string {
 	return acctest.Nprintf(`
 resource "google_storage_bucket" "bucket" {