Re-create purged cluster for databricks_mount for AWS S3

storage/aws_s3_mount.go

@@ -116,11 +116,7 @@ func preprocessS3Mount(ctx context.Context, d *schema.ResourceData, m interface{
 			return fmt.Errorf("cluster %s must have EC2 instance profile attached", clusterID)
 		}
 	} else if instanceProfile != "" {
-		cluster, err := GetOrCreateMountingClusterWithInstanceProfile(clustersAPI, instanceProfile)
-		if err != nil {
-			return err
-		}
-		return d.Set("cluster_id", cluster.ClusterID)
+		return mountS3ViaProfileAndSetClusterID(clustersAPI, instanceProfile, d)
 	}
 	return nil
 }

storage/aws_s3_mount_test.go

@@ -5,8 +5,8 @@ import (
 	"testing"
 
 	"github.com/databrickslabs/terraform-provider-databricks/clusters"
-	"github.com/databrickslabs/terraform-provider-databricks/common"
 	"github.com/databrickslabs/terraform-provider-databricks/commands"
+	"github.com/databrickslabs/terraform-provider-databricks/common"
 
 	"github.com/databrickslabs/terraform-provider-databricks/qa"
 	"github.com/stretchr/testify/assert"
@@ -82,7 +82,7 @@ func TestResourceAwsS3MountCreate_invalid_arn(t *testing.T) {
 		},
 		Create: true,
 	}.Apply(t)
-	require.EqualError(t, err, "invalid arn: this_mount")
+	require.EqualError(t, err, "mount via profile: invalid arn: this_mount")
 }
 
 func TestResourceAwsS3MountRead(t *testing.T) {

storage/resource_mount_test.go

@@ -288,7 +288,7 @@ func TestResourceAwsS3MountGenericCreate_invalid_arn(t *testing.T) {
 		},
 		Create: true,
 	}.Apply(t)
-	require.EqualError(t, err, "invalid arn: this_mount")
+	require.EqualError(t, err, "mount via profile: invalid arn: this_mount")
 }
 
 func TestResourceAwsS3MountGenericRead(t *testing.T) {

storage/s3.go

@@ -62,18 +62,29 @@ func preprocessS3MountGeneric(ctx context.Context, s map[string]*schema.Schema,
 	clustersAPI := clusters.NewClustersAPI(ctx, m)
 	if clusterID != "" {
 		clusterInfo, err := clustersAPI.Get(clusterID)
+		if common.IsMissing(err) {
+			if instanceProfile == "" {
+				return fmt.Errorf("instance profile is required to re-create mounting cluster")
+			}
+			return mountS3ViaProfileAndSetClusterID(clustersAPI, instanceProfile, d)
+		}
 		if err != nil {
 			return err
 		}
 		if clusterInfo.AwsAttributes == nil || len(clusterInfo.AwsAttributes.InstanceProfileArn) == 0 {
 			return fmt.Errorf("cluster %s must have EC2 instance profile attached", clusterID)
 		}
 	} else if instanceProfile != "" {
-		cluster, err := GetOrCreateMountingClusterWithInstanceProfile(clustersAPI, instanceProfile)
-		if err != nil {
-			return err
-		}
-		return d.Set("cluster_id", cluster.ClusterID)
+		return mountS3ViaProfileAndSetClusterID(clustersAPI, instanceProfile, d)
 	}
 	return nil
 }
+
+func mountS3ViaProfileAndSetClusterID(clustersAPI clusters.ClustersAPI, 
+	instanceProfile string, d *schema.ResourceData) error {
+	cluster, err := GetOrCreateMountingClusterWithInstanceProfile(clustersAPI, instanceProfile)
+	if err != nil {
+		return fmt.Errorf("mount via profile: %w", err)
+	}
+	return d.Set("cluster_id", cluster.ClusterID)
+}

storage/s3_test.go

@@ -0,0 +1,106 @@
+package storage
+
+import (
+	"context"
+	"testing"
+
+	"github.com/databrickslabs/terraform-provider-databricks/clusters"
+	"github.com/databrickslabs/terraform-provider-databricks/common"
+	"github.com/databrickslabs/terraform-provider-databricks/qa"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPreprocessS3MountOnDeletedClusterNoInstanceProfileSpecifiedError(t *testing.T) {
+	qa.HTTPFixturesApply(t, []qa.HTTPFixture{
+		{
+			Method:   "GET",
+			Resource: "/api/2.0/clusters/get?cluster_id=removed-cluster",
+			Status:   404,
+			Response: common.NotFound("cluster deleted"),
+		},
+	}, func(ctx context.Context, client *common.DatabricksClient) {
+		r := ResourceMount()
+		d := r.TestResourceData()
+		d.Set("uri", "s3://bucket")
+		d.Set("cluster_id", "removed-cluster")
+		err := preprocessS3MountGeneric(ctx, r.Schema, d, client)
+		assert.EqualError(t, err, "instance profile is required to re-create mounting cluster")
+	})
+}
+
+func TestPreprocessS3MountOnDeletedClusterWorks(t *testing.T) {
+	qa.HTTPFixturesApply(t, []qa.HTTPFixture{
+		{
+			Method:   "GET",
+			Resource: "/api/2.0/clusters/get?cluster_id=removed-cluster",
+			Status:   404,
+			Response: common.NotFound("cluster deleted"),
+		},
+		{
+			Method:   "GET",
+			Resource: "/api/2.0/clusters/list",
+			Response: clusters.ClusterList{
+				Clusters: []clusters.ClusterInfo{},
+			},
+		},
+		{
+			ReuseRequest: true,
+			Method:       "GET",
+			Resource:     "/api/2.0/clusters/spark-versions",
+		},
+		{
+			ReuseRequest: true,
+			Method:       "GET",
+			Resource:     "/api/2.0/clusters/list-node-types",
+		},
+		{
+			Method:   "POST",
+			Resource: "/api/2.0/clusters/create",
+			ExpectedRequest: clusters.Cluster{
+				CustomTags: map[string]string{
+					"ResourceClass": "SingleNode",
+				},
+				ClusterName:  "terraform-mount-s3-access",
+				SparkVersion: "7.3.x-scala2.12",
+				NumWorkers:   0,
+				NodeTypeID:   "i3.xlarge",
+				AwsAttributes: &clusters.AwsAttributes{
+					Availability:       "SPOT",
+					InstanceProfileArn: "arn:aws:iam::1234567:instance-profile/s3-access",
+				},
+				AutoterminationMinutes: 10,
+				SparkConf: map[string]string{
+					"spark.databricks.cluster.profile": "singleNode",
+					"spark.master":                     "local[*]",
+					"spark.scheduler.mode":             "FIFO",
+				},
+			},
+			Response: clusters.ClusterID{
+				ClusterID: "new-cluster",
+			},
+		},
+		{
+			Method:   "GET",
+			Resource: "/api/2.0/clusters/get?cluster_id=new-cluster",
+			Response: clusters.ClusterInfo{
+				ClusterID:    "new-cluster",
+				State:        "RUNNING",
+				StateMessage: "created",
+			},
+		},
+	}, func(ctx context.Context, client *common.DatabricksClient) {
+		r := ResourceMount()
+		d := r.TestResourceData()
+		d.MarkNewResource()
+		common.StructToData(GenericMount{
+			URI:       "s3://bucket",
+			ClusterID: "removed-cluster",
+			S3: &S3IamMount{
+				InstanceProfile: "arn:aws:iam::1234567:instance-profile/s3-access",
+			},
+		}, r.Schema, d)
+		err := preprocessS3MountGeneric(ctx, r.Schema, d, client)
+		assert.NoError(t, err)
+		assert.Equal(t, "new-cluster", d.Get("cluster_id"))
+	})
+}