Added - Support for Data Flow CascadeDeleteApplication API

Author: Jay Patel

examples/dataflow/main.tf

@@ -82,8 +82,19 @@ resource "oci_dataflow_pool" "test_pool" {
   freeform_tags = {
     "Department" = "Finance"
   }
-  configurations = [{shape: "VM.Standard2.1", shapeConfig: {ocpus: 1, memoryInGBs: 15}, min: 0, max: 1}]
-  schedules = [{dayOfWeek: "SUNDAY", startTime: 3}]
+  configurations {
+    shape = "VM.Standard2.1"
+    shape_config {
+      ocpus = 1
+      memory_in_gbs = 15
+    }
+    min = 0
+    max = 1
+  }
+  schedules {
+    day_of_week = "SUNDAY"
+    start_time = 3
+  }
 }
 
 resource "oci_dataflow_application" "tf_application" {
@@ -116,6 +127,7 @@ resource "oci_dataflow_application" "tf_application" {
 
   #warehouse_bucket_uri = var.application_warehouse_bucket_uri}"
   metastore_id = var.metastore_id
+  terminate_runs_on_deletion = true
 }
 
 data "oci_dataflow_applications" "tf_applications" {

internal/service/dataflow/dataflow_application_resource.go

@@ -7,6 +7,11 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"strings"
+	"time"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	oci_common "github.com/oracle/oci-go-sdk/v65/common"
 
 	"github.com/oracle/terraform-provider-oci/internal/client"
 	"github.com/oracle/terraform-provider-oci/internal/tfresource"
@@ -273,6 +278,11 @@ func DataflowApplicationResource() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
+			"terminate_runs_on_deletion": {
+				Type:     schema.TypeBool,
+				Optional: true,
+				Default:  false,
+			},
 		},
 	}
 }
@@ -332,7 +342,9 @@ func (s *DataflowApplicationResourceCrud) CreatedTarget() []string {
 }
 
 func (s *DataflowApplicationResourceCrud) DeletedPending() []string {
-	return []string{}
+	return []string{
+		string(oci_dataflow.ApplicationLifecycleStateDeleting),
+	}
 }
 
 func (s *DataflowApplicationResourceCrud) DeletedTarget() []string {
@@ -760,17 +772,146 @@ func (s *DataflowApplicationResourceCrud) Update() error {
 }
 
 func (s *DataflowApplicationResourceCrud) Delete() error {
-	request := oci_dataflow.DeleteApplicationRequest{}
-
 	tmp := s.D.Id()
-	request.ApplicationId = &tmp
 
+	// Execute cascading deletion of application and related runs if terminate_runs_on_deletion flag is set to true
+	if s.D.Get("terminate_runs_on_deletion").(bool) {
+
+		request := oci_dataflow.CascadingDeleteApplicationRequest{}
+		request.ApplicationId = &tmp
+		request.RequestMetadata.RetryPolicy = tfresource.GetRetryPolicy(s.DisableNotFoundRetries, "dataflow")
+
+		response, err := s.Client.CascadingDeleteApplication(context.Background(), request)
+
+		if err != nil {
+			return err
+		}
+
+		workRequestId := response.OpcWorkRequestId
+
+		// Wait until the work-request completes
+		_, delWorkRequestErr := cascadingDeleteApplicationWaitForWorkRequest(workRequestId, "application",
+			oci_dataflow.WorkRequestResourceActionTypeDeleted, s.D.Timeout(schema.TimeoutDelete),
+			s.DisableNotFoundRetries, s.Client)
+
+		return delWorkRequestErr
+	}
+
+	// Normal deletion
+	request := oci_dataflow.DeleteApplicationRequest{}
+	request.ApplicationId = &tmp
 	request.RequestMetadata.RetryPolicy = tfresource.GetRetryPolicy(s.DisableNotFoundRetries, "dataflow")
 
 	_, err := s.Client.DeleteApplication(context.Background(), request)
 	return err
 }
 
+func cascadingDeleteApplicationWaitForWorkRequest(wId *string, entityType string,
+	action oci_dataflow.WorkRequestResourceActionTypeEnum,
+	timeout time.Duration, disableFoundRetries bool, client *oci_dataflow.DataFlowClient) (*string, error) {
+
+	retryPolicy := tfresource.GetRetryPolicy(disableFoundRetries, "dataflow")
+	retryPolicy.ShouldRetryOperation = applicationWorkRequestShouldRetryFunc(timeout)
+
+	response := oci_dataflow.GetWorkRequestResponse{}
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{
+			string(oci_dataflow.WorkRequestStatusInprogress),
+			string(oci_dataflow.WorkRequestStatusAccepted),
+			string(oci_dataflow.WorkRequestStatusCancelling),
+		},
+		Target: []string{
+			string(oci_dataflow.WorkRequestStatusSucceeded),
+			string(oci_dataflow.WorkRequestStatusFailed),
+			string(oci_dataflow.WorkRequestStatusCancelled),
+		},
+		Refresh: func() (interface{}, string, error) {
+			var err error
+			response, err = client.GetWorkRequest(context.Background(),
+				oci_dataflow.GetWorkRequestRequest{
+					WorkRequestId: wId,
+					RequestMetadata: oci_common.RequestMetadata{
+						RetryPolicy: retryPolicy,
+					},
+				})
+			wr := &response.WorkRequest
+			return wr, string(wr.Status), err
+		},
+		Timeout: timeout,
+	}
+	if _, e := stateConf.WaitForState(); e != nil {
+		return nil, e
+	}
+
+	var identifier *string
+	// The work request response contains an array of objects that finished the operation
+	for _, res := range response.Resources {
+		if strings.Contains(strings.ToLower(*res.ResourceType), entityType) {
+			if res.ActionType == action {
+				identifier = res.ResourceId
+				break
+			}
+		}
+	}
+
+	// The workrequest may have failed, check for errors if identifier is not found or work failed or got cancelled
+	if identifier == nil ||
+		response.Status == oci_dataflow.WorkRequestStatusFailed ||
+		response.Status == oci_dataflow.WorkRequestStatusCancelled {
+		return nil, getErrorFromDataflowApplicationWorkRequest(client, wId, retryPolicy, entityType, action)
+	}
+
+	return identifier, nil
+}
+
+func applicationWorkRequestShouldRetryFunc(timeout time.Duration) func(response oci_common.OCIOperationResponse) bool {
+	startTime := time.Now()
+	stopTime := startTime.Add(timeout)
+	return func(response oci_common.OCIOperationResponse) bool {
+
+		// Stop after timeout has elapsed
+		if time.Now().After(stopTime) {
+			return false
+		}
+
+		// Make sure we stop on default rules
+		if tfresource.ShouldRetry(response, false, "dataflow", startTime) {
+			return true
+		}
+
+		// Only stop if the time Finished is set
+		if workRequestResponse, ok := response.Response.(oci_dataflow.GetWorkRequestResponse); ok {
+			return workRequestResponse.TimeFinished == nil
+		}
+		return false
+	}
+}
+
+func getErrorFromDataflowApplicationWorkRequest(client *oci_dataflow.DataFlowClient, workId *string,
+	retryPolicy *oci_common.RetryPolicy, entityType string, action oci_dataflow.WorkRequestResourceActionTypeEnum) error {
+	response, err := client.ListWorkRequestErrors(context.Background(),
+		oci_dataflow.ListWorkRequestErrorsRequest{
+			WorkRequestId: workId,
+			RequestMetadata: oci_common.RequestMetadata{
+				RetryPolicy: retryPolicy,
+			},
+		})
+	if err != nil {
+		return err
+	}
+
+	allErrs := make([]string, 0)
+	for _, wrkErr := range response.Items {
+		allErrs = append(allErrs, *wrkErr.Message)
+	}
+	errorMessage := strings.Join(allErrs, "\n")
+
+	workRequestErr := fmt.Errorf("work request did not succeed, workId: %s, entity: %s, action: %s. Message: %s",
+		*workId, entityType, action, errorMessage)
+
+	return workRequestErr
+}
+
 func (s *DataflowApplicationResourceCrud) SetData() error {
 	if s.Res.ApplicationLogConfig != nil {
 		s.D.Set("application_log_config", []interface{}{ApplicationLogConfigToMap(s.Res.ApplicationLogConfig)})

website/docs/d/dataflow_application.html.markdown

@@ -75,4 +75,5 @@ The following attributes are exported:
 * `time_updated` - The date and time the resource was updated, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `type` - The Spark application processing type. 
 * `warehouse_bucket_uri` - An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
+* `terminate_runs_on_deletion` - A boolean flag which indicates whether related non-terminal Run(s) for the Application should be terminated along with Application deletion or not.
 

website/docs/r/dataflow_application.html.markdown

@@ -66,6 +66,7 @@ resource "oci_dataflow_application" "test_application" {
 	private_endpoint_id = oci_dataflow_private_endpoint.test_private_endpoint.id
 	type = var.application_type
 	warehouse_bucket_uri = var.application_warehouse_bucket_uri
+	terminate_runs_on_deletion = true
 }
 ```
 
@@ -109,7 +110,7 @@ The following arguments are supported:
 * `spark_version` - (Required) (Updatable) The Spark version utilized to run the application. 
 * `type` - (Optional) The Spark application processing type. 
 * `warehouse_bucket_uri` - (Optional) (Updatable) An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
-
+* `terminate_runs_on_deletion` - (Optional) A boolean flag which indicates whether related non-terminal Run(s) for the Application should be terminated along with Application deletion or not.
 
 ** IMPORTANT **
 Any change to a property that does not support update will force the destruction and recreation of the resource with the new property values