Make dataflow on_delete a virtual field (#3567) (#2110)

* Make dataflow on_delete a virtual field

* Clarify test

* Add attribute check for good measure

* remove comments

* only iterate on "labels" map

Signed-off-by: Modular Magician <[email protected]>

Author: modular-magician

.changelog/3567.txt

@@ -0,0 +1,6 @@
+```release-note:bug
+`dataflow`: fixed an issue where `google_dataflow_job` would try to update `max_workers`
+```
+```release-note:bug
+`dataflow`: fixed an issue where updating `on_delete` in `google_dataflow_job` would cause the job to be replaced
+```

google-beta/resource_dataflow_job.go

@@ -85,6 +85,8 @@ func resourceDataflowJob() *schema.Resource {
 			"max_workers": {
 				Type:     schema.TypeInt,
 				Optional: true,
+				// ForceNew applies to both stream and batch jobs
+				ForceNew: true,
 			},
 
 			"parameters": {
@@ -166,12 +168,15 @@ func resourceDataflowJob() *schema.Resource {
 }
 
 func resourceDataflowJobTypeCustomizeDiff(d *schema.ResourceDiff, meta interface{}) error {
-	// All changes are ForceNew for batch jobs
+	// All non-virtual fields are ForceNew for batch jobs
 	if d.Get("type") == "JOB_TYPE_BATCH" {
 		resourceSchema := resourceDataflowJob().Schema
-		for field, fieldSchema := range resourceSchema {
-			// Each key within a map must be checked for a change
-			if fieldSchema.Type == schema.TypeMap {
+		for field := range resourceSchema {
+			if field == "on_delete" {
+				continue
+			}
+			// Labels map will likely have suppressed changes, so we check each key instead of the parent field
+			if field == "labels" {
 				resourceDataflowJobIterateMapForceNew(field, d)
 			} else if d.HasChange(field) {
 				d.ForceNew(field)
@@ -265,6 +270,11 @@ func resourceDataflowJobRead(d *schema.ResourceData, meta interface{}) error {
 
 // Stream update method. Batch job changes should have been set to ForceNew via custom diff
 func resourceDataflowJobUpdateByReplacement(d *schema.ResourceData, meta interface{}) error {
+	// Don't send an update request if only virtual fields have changes
+	if resourceDataflowJobIsVirtualUpdate(d) {
+		return nil
+	}
+
 	config := meta.(*Config)
 
 	project, err := getProject(d, config)
@@ -456,3 +466,36 @@ func resourceDataflowJobIterateMapForceNew(mapKey string, d *schema.ResourceDiff
 		}
 	}
 }
+
+func resourceDataflowJobIterateMapHasChange(mapKey string, d *schema.ResourceData) bool {
+	obj := d.Get(mapKey).(map[string]interface{})
+	for k := range obj {
+		entrySchemaKey := mapKey + "." + k
+		if d.HasChange(entrySchemaKey) {
+			return true
+		}
+	}
+	return false
+}
+
+func resourceDataflowJobIsVirtualUpdate(d *schema.ResourceData) bool {
+	// on_delete is the only virtual field
+	if d.HasChange("on_delete") {
+		// Check if other fields have changes, which would require an actual update request
+		resourceSchema := resourceDataflowJob().Schema
+		for field := range resourceSchema {
+			if field == "on_delete" {
+				continue
+			}
+			// Labels map will likely have suppressed changes, so we check each key instead of the parent field
+			if (field == "labels" && resourceDataflowJobIterateMapHasChange(field, d)) ||
+				(field != "labels" && d.HasChange(field)) {
+				return false
+			}
+		}
+		// on_delete is changing, but nothing else
+		return true
+	}
+
+	return false
+}

google-beta/resource_dataflow_job_test.go

@@ -259,6 +259,39 @@ func TestAccDataflowJob_streamUpdate(t *testing.T) {
 	})
 }
 
+func TestAccDataflowJob_virtualUpdate(t *testing.T) {
+	// Dataflow responses include serialized java classes and bash commands
+	// This makes body comparison infeasible
+	skipIfVcr(t)
+	t.Parallel()
+
+	suffix := randString(t, 10)
+
+	// If the update is virtual-only, the ID should remain the same after updating.
+	var id string
+	vcrTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckDataflowJobDestroyProducer(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDataflowJob_virtualUpdate(suffix, "drain"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccDataflowJobExists(t, "google_dataflow_job.pubsub_stream"),
+					testAccDataflowSetId(t, "google_dataflow_job.pubsub_stream", &id),
+				),
+			},
+			{
+				Config: testAccDataflowJob_virtualUpdate(suffix, "cancel"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccDataflowCheckId(t, "google_dataflow_job.pubsub_stream", &id),
+					resource.TestCheckResourceAttr("google_dataflow_job.pubsub_stream", "on_delete", "cancel"),
+				),
+			},
+		},
+	})
+}
+
 func testAccCheckDataflowJobDestroyProducer(t *testing.T) func(s *terraform.State) error {
 	return func(s *terraform.State) error {
 		for _, rs := range s.RootModule().Resources {
@@ -323,6 +356,32 @@ func testAccDataflowJobExists(t *testing.T, resource string) resource.TestCheckF
 	}
 }
 
+func testAccDataflowSetId(t *testing.T, resource string, id *string) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		rs, ok := s.RootModule().Resources[resource]
+		if !ok {
+			return fmt.Errorf("resource %q not in state", resource)
+		}
+
+		*id = rs.Primary.ID
+		return nil
+	}
+}
+
+func testAccDataflowCheckId(t *testing.T, resource string, id *string) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		rs, ok := s.RootModule().Resources[resource]
+		if !ok {
+			return fmt.Errorf("resource %q not in state", resource)
+		}
+
+		if rs.Primary.ID != *id {
+			return fmt.Errorf("ID did not match. Expected %s, received %s", *id, rs.Primary.ID)
+		}
+		return nil
+	}
+}
+
 func testAccDataflowJobHasNetwork(t *testing.T, res, expected string) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		instanceTmpl, err := testAccDataflowJobGetGeneratedInstanceTemplate(t, s, res)
@@ -772,3 +831,25 @@ resource "google_dataflow_job" "pubsub_stream" {
 }
   `, suffix, suffix, suffix, suffix, testDataflowJobTemplateTextToPubsub, tempLocation)
 }
+
+func testAccDataflowJob_virtualUpdate(suffix, onDelete string) string {
+	return fmt.Sprintf(`
+resource "google_pubsub_topic" "topic" {
+	name     = "tf-test-dataflow-job-%s"
+}
+resource "google_storage_bucket" "bucket" {
+	name = "tf-test-bucket-%s"
+	force_destroy = true
+}
+resource "google_dataflow_job" "pubsub_stream" {
+	name = "tf-test-dataflow-job-%s"
+	template_gcs_path = "%s"
+	temp_gcs_location = google_storage_bucket.bucket.url
+	parameters = {
+	  inputFilePattern = "${google_storage_bucket.bucket.url}/*.json"
+	  outputTopic    = google_pubsub_topic.topic.id
+	}
+	on_delete = "%s"
+}
+  `, suffix, suffix, suffix, testDataflowJobTemplateTextToPubsub, onDelete)
+}