Add support for non-preemptible secondary dataproc workers

mmv1/products/dataproc/api.yaml

@@ -326,8 +326,19 @@ objects:
                       Number of attached SSDs, from 0 to 4.
               - !ruby/object:Api::Type::Boolean
                 name: 'isPreemptible'
+                output: true
                 description: |
                   Specifies if this instance group contains preemptible instances.
+              - !ruby/object:Api::Type::Enum
+                name: 'preemptibility'
+                description: |
+                  Specifies the preemptibility of the instance group.
+                  The default value for master and worker groups is NON_PREEMPTIBLE. This default cannot be changed.
+                  The default value for secondary instances is PREEMPTIBLE.
+                values:
+                - :PREEMPTIBILITY_UNSPECIFIED
+                - :NON_PREEMPTIBLE
+                - :PREEMPTIBLE
               - !ruby/object:Api::Type::NestedObject
                 name: 'managedGroupConfig'
                 output: true
@@ -387,8 +398,19 @@ objects:
                       Number of attached SSDs, from 0 to 4.
               - !ruby/object:Api::Type::Boolean
                 name: 'isPreemptible'
+                output: true
                 description: |
                   Specifies if this instance group contains preemptible instances.
+              - !ruby/object:Api::Type::Enum
+                name: 'preemptibility'
+                description: |
+                  Specifies the preemptibility of the instance group.
+                  The default value for master and worker groups is NON_PREEMPTIBLE. This default cannot be changed.
+                  The default value for secondary instances is PREEMPTIBLE.
+                values:
+                - :PREEMPTIBILITY_UNSPECIFIED
+                - :NON_PREEMPTIBLE
+                - :PREEMPTIBLE
               - !ruby/object:Api::Type::NestedObject
                 name: 'managedGroupConfig'
                 output: true
@@ -448,8 +470,19 @@ objects:
                       Number of attached SSDs, from 0 to 4.
               - !ruby/object:Api::Type::Boolean
                 name: 'isPreemptible'
+                output: true
                 description: |
                   Specifies if this instance group contains preemptible instances.
+              - !ruby/object:Api::Type::Enum
+                name: 'preemptibility'
+                description: |
+                  Specifies the preemptibility of the instance group.
+                  The default value for master and worker groups is NON_PREEMPTIBLE. This default cannot be changed.
+                  The default value for secondary instances is PREEMPTIBLE.
+                values:
+                - :PREEMPTIBILITY_UNSPECIFIED
+                - :NON_PREEMPTIBLE
+                - :PREEMPTIBLE
               - !ruby/object:Api::Type::NestedObject
                 name: 'managedGroupConfig'
                 output: true

mmv1/third_party/terraform/resources/resource_dataproc_cluster.go.erb

@@ -336,6 +336,7 @@ func resourceDataprocCluster() *schema.Resource {
 										Description: `Specifies the number of preemptible nodes to create. Defaults to 0.`,
 										AtLeastOneOf: []string{
 											"cluster_config.0.preemptible_worker_config.0.num_instances",
+											"cluster_config.0.preemptible_worker_config.0.preemptibility",
 											"cluster_config.0.preemptible_worker_config.0.disk_config",
 										},
 									},
@@ -344,13 +345,28 @@ func resourceDataprocCluster() *schema.Resource {
 									// It always uses whatever is specified for the worker_config
 									// "machine_type": { ... }
 									// "min_cpu_platform": { ... }
+									"preemptibility": {
+										Type:		 schema.TypeString,
+										Optional: 	 true,
+										Description: `Specifies the preemptibility of the secondary nodes. Defaults to PREEMPTIBLE.`,
+										AtLeastOneOf: []string{
+											"cluster_config.0.preemptible_worker_config.0.num_instances",
+											"cluster_config.0.preemptible_worker_config.0.preemptibility",
+											"cluster_config.0.preemptible_worker_config.0.disk_config",
+										},
+										ForceNew:     true,
+										ValidateFunc: validation.StringInSlice([]string{"PREEMPTIBILITY_UNSPECIFIED", "NON_PREEMPTIBLE", "PREEMPTIBLE"}, false),
+										Default:      "PREEMPTIBLE",
+									},
+
 									"disk_config": {
 										Type:        schema.TypeList,
 										Optional:    true,
 										Computed:    true,
 										Description: `Disk Config`,
 										AtLeastOneOf: []string{
 											"cluster_config.0.preemptible_worker_config.0.num_instances",
+											"cluster_config.0.preemptible_worker_config.0.preemptibility",
 											"cluster_config.0.preemptible_worker_config.0.disk_config",
 										},
 										MaxItems: 1,
@@ -982,9 +998,6 @@ func expandClusterConfig(d *schema.ResourceData, config *Config) (*dataproc.Clus
 	if cfg, ok := configOptions(d, "cluster_config.0.preemptible_worker_config"); ok {
 		log.Println("[INFO] got preemptible worker config")
 		conf.SecondaryWorkerConfig = expandPreemptibleInstanceGroupConfig(cfg)
-		if conf.SecondaryWorkerConfig.NumInstances > 0 {
-			conf.SecondaryWorkerConfig.IsPreemptible = true
-		}
 	}
 	return conf, nil
 }
@@ -1221,6 +1234,9 @@ func expandPreemptibleInstanceGroupConfig(cfg map[string]interface{}) *dataproc.
 			}
 		}
 	}
+	if p, ok := cfg["preemptibility"]; ok {
+		icg.Preemptibility = p.(string)
+	}
 	return icg
 }
 
@@ -1662,6 +1678,7 @@ func flattenPreemptibleInstanceGroupConfig(d *schema.ResourceData, icg *dataproc
 	if icg != nil {
 		data["num_instances"] = icg.NumInstances
 		data["instance_names"] = icg.InstanceNames
+		data["preemptibility"] = icg.Preemptibility
 		if icg.DiskConfig != nil {
 			disk["boot_disk_size_gb"] = icg.DiskConfig.BootDiskSizeGb
 			disk["num_local_ssds"] = icg.DiskConfig.NumLocalSsds

mmv1/third_party/terraform/tests/resource_dataproc_cluster_test.go.erb

@@ -399,6 +399,27 @@ func TestAccDataprocCluster_updatable(t *testing.T) {
 	})
 }
 
+func TestAccDataprocCluster_nonPreemptibleSecondary(t *testing.T) {
+	t.Parallel()
+
+	rnd := randString(t, 10)
+	var cluster dataproc.Cluster
+	vcrTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckDataprocClusterDestroy(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDataprocCluster_nonPreemptibleSecondary(rnd),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDataprocClusterExists(t, "google_dataproc_cluster.non_preemptible_secondary", &cluster),
+					resource.TestCheckResourceAttr("google_dataproc_cluster.non_preemptible_secondary", "cluster_config.0.preemptible_worker_config.0.preemptibility", "NON_PREEMPTIBLE"),
+				),
+			},
+		},
+	})
+}
+
 func TestAccDataprocCluster_withStagingBucket(t *testing.T) {
 	t.Parallel()
 
@@ -1352,6 +1373,41 @@ resource "google_dataproc_cluster" "updatable" {
 `, rnd, w, p)
 }
 
+func testAccDataprocCluster_nonPreemptibleSecondary(rnd string) string {
+	return fmt.Sprintf(`
+resource "google_dataproc_cluster" "non_preemptible_secondary" {
+  name   = "tf-test-dproc-%s"
+  region = "us-central1"
+
+  cluster_config {
+    master_config {
+	  num_instances = "1"
+	  machine_type  = "e2-medium"
+	  disk_config {
+		boot_disk_size_gb = 35
+	  }
+	}
+
+	worker_config {
+	  num_instances = "2"
+	  machine_type  = "e2-medium"
+	  disk_config {
+		boot_disk_size_gb = 35
+	  }
+	}
+
+	preemptible_worker_config {
+	  num_instances = "1"
+	  preemptibility = "NON_PREEMPTIBLE"
+	  disk_config {
+		boot_disk_size_gb = 35
+	  }
+	}
+  }
+}
+	`, rnd)
+}
+
 func testAccDataprocCluster_withStagingBucketOnly(bucketName string) string {
 	return fmt.Sprintf(`
 resource "google_storage_bucket" "bucket" {

mmv1/third_party/terraform/website/docs/r/dataproc_cluster.html.markdown

@@ -443,6 +443,12 @@ will be set for you based on whatever was set for the `worker_config.machine_typ
 * `num_instances`- (Optional) Specifies the number of preemptible nodes to create.
    Defaults to 0.
 
+* `preemptibility`- (Optional) Specifies the preemptibility of the secondary workers. The default value is `PREEMPTIBLE`
+  Accepted values are:
+  * PREEMPTIBILITY_UNSPECIFIED
+  * NON_PREEMPTIBLE
+  * PREEMPTIBLE
+
 * `disk_config` (Optional) Disk Config
 
     * `boot_disk_type` - (Optional) The disk type of the primary disk attached to each preemptible worker node.