Update datastream_stream resource to include kms fields

.changelog/7126.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+datastream: add `customer_managed_encryption_key` and `destination_config.bigquery_destination_config.source_hierarchy_datasets.dataset_template.kms_key_name` fields to `datastream_stream` resource
+```

google/resource_datastream_stream.go

@@ -169,6 +169,15 @@ See https://cloud.google.com/bigquery/docs/locations for supported locations.`,
 																Description: `If supplied, every created dataset will have its name prefixed by the provided value.
 The prefix and name will be separated by an underscore. i.e. _.`,
 															},
+															"kms_key_name": {
+																Type:     schema.TypeString,
+																Optional: true,
+																ForceNew: true,
+																Description: `Describes the Cloud KMS encryption key that will be used to protect destination BigQuery
+table. The BigQuery Service Account associated with your project requires access to this
+encryption key. i.e. projects/{project}/locations/{location}/keyRings/{key_ring}/cryptoKeys/{cryptoKey}.
+See https://cloud.google.com/bigquery/docs/customer-managed-encryption for more information.`,
+															},
 														},
 													},
 												},
@@ -576,6 +585,13 @@ https://dev.mysql.com/doc/refman/8.0/en/data-types.html`,
 				},
 				ExactlyOneOf: []string{"backfill_all", "backfill_none"},
 			},
+			"customer_managed_encryption_key": {
+				Type:     schema.TypeString,
+				Optional: true,
+				ForceNew: true,
+				Description: `A reference to a KMS encryption key. If provided, it will be used to encrypt the data. If left blank, data
+will be encrypted using an internal Stream-specific encryption key provisioned through KMS.`,
+			},
 			"labels": {
 				Type:        schema.TypeMap,
 				Optional:    true,
@@ -653,6 +669,12 @@ func resourceDatastreamStreamCreate(d *schema.ResourceData, meta interface{}) er
 	} else if v, ok := d.GetOkExists("backfill_none"); ok || !reflect.DeepEqual(v, backfillNoneProp) {
 		obj["backfillNone"] = backfillNoneProp
 	}
+	customerManagedEncryptionKeyProp, err := expandDatastreamStreamCustomerManagedEncryptionKey(d.Get("customer_managed_encryption_key"), d, config)
+	if err != nil {
+		return err
+	} else if v, ok := d.GetOkExists("customer_managed_encryption_key"); !isEmptyValue(reflect.ValueOf(customerManagedEncryptionKeyProp)) && (ok || !reflect.DeepEqual(v, customerManagedEncryptionKeyProp)) {
+		obj["customerManagedEncryptionKey"] = customerManagedEncryptionKeyProp
+	}
 
 	obj, err = resourceDatastreamStreamEncoder(d, meta, obj)
 	if err != nil {
@@ -787,6 +809,9 @@ func resourceDatastreamStreamRead(d *schema.ResourceData, meta interface{}) erro
 	if err := d.Set("backfill_none", flattenDatastreamStreamBackfillNone(res["backfillNone"], d, config)); err != nil {
 		return fmt.Errorf("Error reading Stream: %s", err)
 	}
+	if err := d.Set("customer_managed_encryption_key", flattenDatastreamStreamCustomerManagedEncryptionKey(res["customerManagedEncryptionKey"], d, config)); err != nil {
+		return fmt.Errorf("Error reading Stream: %s", err)
+	}
 
 	return nil
 }
@@ -1501,6 +1526,8 @@ func flattenDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHier
 		flattenDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHierarchyDatasetsDatasetTemplateLocation(original["location"], d, config)
 	transformed["dataset_id_prefix"] =
 		flattenDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHierarchyDatasetsDatasetTemplateDatasetIdPrefix(original["datasetIdPrefix"], d, config)
+	transformed["kms_key_name"] =
+		flattenDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHierarchyDatasetsDatasetTemplateKmsKeyName(original["kmsKeyName"], d, config)
 	return []interface{}{transformed}
 }
 func flattenDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHierarchyDatasetsDatasetTemplateLocation(v interface{}, d *schema.ResourceData, config *Config) interface{} {
@@ -1511,6 +1538,10 @@ func flattenDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHier
 	return v
 }
 
+func flattenDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHierarchyDatasetsDatasetTemplateKmsKeyName(v interface{}, d *schema.ResourceData, config *Config) interface{} {
+	return v
+}
+
 func flattenDatastreamStreamState(v interface{}, d *schema.ResourceData, config *Config) interface{} {
 	return v
 }
@@ -1670,6 +1701,10 @@ func flattenDatastreamStreamBackfillNone(v interface{}, d *schema.ResourceData,
 	return []interface{}{transformed}
 }
 
+func flattenDatastreamStreamCustomerManagedEncryptionKey(v interface{}, d *schema.ResourceData, config *Config) interface{} {
+	return v
+}
+
 func expandDatastreamStreamLabels(v interface{}, d TerraformResourceData, config *Config) (map[string]string, error) {
 	if v == nil {
 		return map[string]string{}, nil
@@ -2358,6 +2393,13 @@ func expandDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHiera
 		transformed["datasetIdPrefix"] = transformedDatasetIdPrefix
 	}
 
+	transformedKmsKeyName, err := expandDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHierarchyDatasetsDatasetTemplateKmsKeyName(original["kms_key_name"], d, config)
+	if err != nil {
+		return nil, err
+	} else if val := reflect.ValueOf(transformedKmsKeyName); val.IsValid() && !isEmptyValue(val) {
+		transformed["kmsKeyName"] = transformedKmsKeyName
+	}
+
 	return transformed, nil
 }
 
@@ -2369,6 +2411,10 @@ func expandDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHiera
 	return v, nil
 }
 
+func expandDatastreamStreamDestinationConfigBigqueryDestinationConfigSourceHierarchyDatasetsDatasetTemplateKmsKeyName(v interface{}, d TerraformResourceData, config *Config) (interface{}, error) {
+	return v, nil
+}
+
 func expandDatastreamStreamBackfillAll(v interface{}, d TerraformResourceData, config *Config) (interface{}, error) {
 	l := v.([]interface{})
 	if len(l) == 0 {
@@ -2585,6 +2631,10 @@ func expandDatastreamStreamBackfillNone(v interface{}, d TerraformResourceData,
 	return transformed, nil
 }
 
+func expandDatastreamStreamCustomerManagedEncryptionKey(v interface{}, d TerraformResourceData, config *Config) (interface{}, error) {
+	return v, nil
+}
+
 func resourceDatastreamStreamEncoder(d *schema.ResourceData, meta interface{}, obj map[string]interface{}) (map[string]interface{}, error) {
 	if d.HasChange("desired_state") {
 		obj["state"] = d.Get("desired_state")

google/resource_datastream_stream_generated_test.go

@@ -189,6 +189,7 @@ func TestAccDatastreamStream_datastreamStreamFullExample(t *testing.T) {
 
 	context := map[string]interface{}{
 		"deletion_protection": false,
+		"stream_cmek":         BootstrapKMSKeyInLocation(t, "us-central1").CryptoKey.Name,
 		"random_suffix":       randString(t, 10),
 	}
 
@@ -311,6 +312,12 @@ resource "google_storage_bucket_iam_member" "reader" {
     member = "serviceAccount:service-${data.google_project.project.number}@gcp-sa-datastream.iam.gserviceaccount.com"
 }
 
+resource "google_kms_crypto_key_iam_member" "key_user" {
+    crypto_key_id = "%{stream_cmek}"
+    role          = "roles/cloudkms.cryptoKeyEncrypterDecrypter"
+    member        = "serviceAccount:service-${data.google_project.project.number}@gcp-sa-datastream.iam.gserviceaccount.com"
+}
+
 resource "google_datastream_connection_profile" "destination_connection_profile" {
     display_name          = "Connection profile"
     location              = "us-central1"
@@ -323,6 +330,9 @@ resource "google_datastream_connection_profile" "destination_connection_profile"
 }
 
 resource "google_datastream_stream" "default" {
+    depends_on = [
+        google_kms_crypto_key_iam_member.key_user
+    ]
     stream_id = "tf-test-my-stream%{random_suffix}"
     desired_state = "NOT_STARTED"
     location = "us-central1"
@@ -399,6 +409,159 @@ resource "google_datastream_stream" "default" {
             }
         }
     }
+
+    customer_managed_encryption_key = "%{stream_cmek}"
+}
+`, context)
+}
+
+func TestAccDatastreamStream_datastreamStreamBigqueryExample(t *testing.T) {
+	skipIfVcr(t)
+	t.Parallel()
+
+	context := map[string]interface{}{
+		"deletion_protection":                     false,
+		"bigquery_destination_table_kms_key_name": BootstrapKMSKeyInLocation(t, "us-central1").CryptoKey.Name,
+		"random_suffix":                           randString(t, 10),
+	}
+
+	vcrTest(t, resource.TestCase{
+		PreCheck:  func() { testAccPreCheck(t) },
+		Providers: testAccProviders,
+		ExternalProviders: map[string]resource.ExternalProvider{
+			"random": {},
+			"time":   {},
+		},
+		CheckDestroy: testAccCheckDatastreamStreamDestroyProducer(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDatastreamStream_datastreamStreamBigqueryExample(context),
+			},
+			{
+				ResourceName:            "google_datastream_stream.default",
+				ImportState:             true,
+				ImportStateVerify:       true,
+				ImportStateVerifyIgnore: []string{"stream_id", "location"},
+			},
+		},
+	})
+}
+
+func testAccDatastreamStream_datastreamStreamBigqueryExample(context map[string]interface{}) string {
+	return Nprintf(`
+data "google_project" "project" {
+}
+
+resource "google_sql_database_instance" "instance" {
+    name             = "tf-test-my-instance%{random_suffix}"
+    database_version = "MYSQL_8_0"
+    region           = "us-central1"
+    settings {
+        tier = "db-f1-micro"
+        backup_configuration {
+            enabled            = true
+            binary_log_enabled = true
+        }
+
+        ip_configuration {
+
+            // Datastream IPs will vary by region.
+            authorized_networks {
+                value = "34.71.242.81"
+            }
+
+            authorized_networks {
+                value = "34.72.28.29"
+            }
+
+            authorized_networks {
+                value = "34.67.6.157"
+            }
+
+            authorized_networks {
+                value = "34.67.234.134"
+            }
+
+            authorized_networks {
+                value = "34.72.239.218"
+            }
+        }
+    }
+
+    deletion_protection  = %{deletion_protection}
+}
+
+resource "google_sql_database" "db" {
+    instance = google_sql_database_instance.instance.name
+    name     = "db"
+}
+
+resource "random_password" "pwd" {
+    length = 16
+    special = false
+}
+
+resource "google_sql_user" "user" {
+    name     = "user"
+    instance = google_sql_database_instance.instance.name
+    host     = "%"
+    password = random_password.pwd.result
+}
+
+resource "google_datastream_connection_profile" "source_connection_profile" {
+    display_name          = "Source connection profile"
+    location              = "us-central1"
+    connection_profile_id = "tf-test-source-profile%{random_suffix}"
+
+    mysql_profile {
+        hostname = google_sql_database_instance.instance.public_ip_address
+        username = google_sql_user.user.name
+        password = google_sql_user.user.password
+    }
+}
+
+data "google_bigquery_default_service_account" "bq_sa" {
+}
+
+resource "google_kms_crypto_key_iam_member" "bigquery_key_user" {
+  crypto_key_id = "%{bigquery_destination_table_kms_key_name}"
+  role          = "roles/cloudkms.cryptoKeyEncrypterDecrypter"
+  member        = "serviceAccount:${data.google_bigquery_default_service_account.bq_sa.email}"
+}
+
+resource "google_datastream_connection_profile" "destination_connection_profile" {
+    display_name          = "Connection profile"
+    location              = "us-central1"
+    connection_profile_id = "tf-test-destination-profile%{random_suffix}"
+
+    bigquery_profile {}
+}
+
+resource "google_datastream_stream" "default" {
+    depends_on = [
+        google_kms_crypto_key_iam_member.bigquery_key_user
+    ]
+    stream_id = "tf-test-my-stream%{random_suffix}"
+    location = "us-central1"
+    display_name = "my stream"
+    source_config {
+        source_connection_profile = google_datastream_connection_profile.source_connection_profile.id
+        mysql_source_config {}
+    }
+    destination_config {
+        destination_connection_profile = google_datastream_connection_profile.destination_connection_profile.id
+        bigquery_destination_config {
+            source_hierarchy_datasets {
+                dataset_template {
+                    location = "us-central1"
+                    kms_key_name = "%{bigquery_destination_table_kms_key_name}"
+                }
+            }
+        }
+    }
+
+    backfill_none {
+    }
 }
 `, context)
 }