add metadata to gce cluster config in dataproc (#999)

google/resource_dataproc_cluster.go

@@ -177,6 +177,13 @@ func resourceDataprocCluster() *schema.Resource {
 										ForceNew: true,
 										Default:  false,
 									},
+
+									"metadata": &schema.Schema{
+										Type:     schema.TypeMap,
+										Optional: true,
+										Elem:     schema.TypeString,
+										ForceNew: true,
+									},
 								},
 							},
 						},
@@ -507,6 +514,9 @@ func expandGceClusterConfig(d *schema.ResourceData, config *Config) (*dataproc.G
 	if v, ok := cfg["internal_ip_only"]; ok {
 		conf.InternalIpOnly = v.(bool)
 	}
+	if v, ok := cfg["metadata"]; ok {
+		conf.Metadata = convertStringMap(v.(map[string]interface{}))
+	}
 	return conf, nil
 }
 
@@ -752,6 +762,7 @@ func flattenGceClusterConfig(d *schema.ResourceData, gcc *dataproc.GceClusterCon
 		"service_account":  gcc.ServiceAccount,
 		"zone":             GetResourceNameFromSelfLink(gcc.ZoneUri),
 		"internal_ip_only": gcc.InternalIpOnly,
+		"metadata":         gcc.Metadata,
 	}
 
 	if gcc.NetworkUri != "" {

google/resource_dataproc_cluster_test.go

@@ -159,6 +159,29 @@ func TestAccDataprocCluster_basicWithInternalIpOnlyTrue(t *testing.T) {
 	})
 }
 
+func TestAccDataprocCluster_basicWithMetadata(t *testing.T) {
+	t.Parallel()
+
+	var cluster dataproc.Cluster
+	rnd := acctest.RandString(10)
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckDataprocClusterDestroy(false),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDataprocCluster_basicWithMetadata(rnd),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDataprocClusterExists("google_dataproc_cluster.basic", &cluster),
+
+					resource.TestCheckResourceAttr("google_dataproc_cluster.basic", "cluster_config.0.gce_cluster_config.0.metadata.foo", "bar"),
+					resource.TestCheckResourceAttr("google_dataproc_cluster.basic", "cluster_config.0.gce_cluster_config.0.metadata.baz", "qux"),
+				),
+			},
+		},
+	})
+}
+
 func TestAccDataprocCluster_basicWithAutogenDeleteTrue(t *testing.T) {
 	t.Parallel()
 
@@ -740,6 +763,24 @@ resource "google_dataproc_cluster" "basic" {
 `, rnd, rnd, rnd)
 }
 
+func testAccDataprocCluster_basicWithMetadata(rnd string) string {
+	return fmt.Sprintf(`
+resource "google_dataproc_cluster" "basic" {
+	name   = "dproc-cluster-test-%s"
+	region = "us-central1"
+
+	cluster_config {
+		gce_cluster_config {
+			metadata {
+				foo = "bar"
+				baz = "qux"
+			}
+		}
+	}
+}
+`, rnd)
+}
+
 func testAccDataprocCluster_basicWithAutogenDeleteTrue(rnd string) string {
 	return fmt.Sprintf(`
 resource "google_dataproc_cluster" "basic" {