feat: implement object storage delete validations for warehouse destinations

warehouse/internal/model/validation.go

@@ -4,6 +4,7 @@ import backendconfig "github.com/rudderlabs/rudder-server/backend-config"
 
 const (
 	VerifyingObjectStorage       = "Verifying Object Storage"
+	VerifyingObjectStorageDelete = "Verifying delete permissions in object storage"
 	VerifyingConnections         = "Verifying Connections"
 	VerifyingCreateSchema        = "Verifying Create Schema"
 	VerifyingCreateAndAlterTable = "Verifying Create and Alter Table"

warehouse/validations/steps.go

@@ -24,6 +24,13 @@ func StepsToValidate(dest *backendconfig.DestinationT) *model.StepsResponse {
 		ID:   len(steps) + 1,
 		Name: model.VerifyingObjectStorage,
 	}}
+	cleanupObjectStorageFiles, _ := dest.Config[model.CleanupObjectStorageFilesSetting.String()].(bool)
+	if cleanupObjectStorageFiles {
+		steps = append(steps, &model.Step{
+			ID:   len(steps) + 1,
+			Name: model.VerifyingObjectStorageDelete,
+		})
+	}
 
 	switch destType {
 	case warehouseutils.GCSDatalake, warehouseutils.AzureDatalake:

warehouse/validations/steps_test.go

@@ -27,9 +27,27 @@ func TestValidationSteps(t *testing.T) {
 				DestinationDefinition: backendconfig.DestinationDefinitionT{
 					Name: warehouseutils.GCSDatalake,
 				},
+				Config: map[string]interface{}{
+					model.CleanupObjectStorageFilesSetting.String(): false,
+				},
 			},
 			steps: []string{model.VerifyingObjectStorage},
 		},
+		{
+			name: "GCS with cleanup enabled",
+			dest: backendconfig.DestinationT{
+				DestinationDefinition: backendconfig.DestinationDefinitionT{
+					Name: warehouseutils.GCSDatalake,
+				},
+				Config: map[string]interface{}{
+					model.CleanupObjectStorageFilesSetting.String(): true,
+				},
+			},
+			steps: []string{
+				model.VerifyingObjectStorage,
+				model.VerifyingObjectStorageDelete,
+			},
+		},
 		{
 			name: "Azure",
 			dest: backendconfig.DestinationT{
@@ -83,6 +101,26 @@ func TestValidationSteps(t *testing.T) {
 				model.VerifyingLoadTable,
 			},
 		},
+		{
+			name: "RS with cleanup enabled",
+			dest: backendconfig.DestinationT{
+				DestinationDefinition: backendconfig.DestinationDefinitionT{
+					Name: warehouseutils.RS,
+				},
+				Config: map[string]interface{}{
+					model.CleanupObjectStorageFilesSetting.String(): true,
+				},
+			},
+			steps: []string{
+				model.VerifyingObjectStorage,
+				model.VerifyingObjectStorageDelete,
+				model.VerifyingConnections,
+				model.VerifyingCreateSchema,
+				model.VerifyingCreateAndAlterTable,
+				model.VerifyingFetchSchema,
+				model.VerifyingLoadTable,
+			},
+		},
 	}
 
 	for _, tc := range testCases {

warehouse/validations/validate.go

@@ -56,6 +56,10 @@
 	table       string
 }
 
+type objectStorageDelete struct {
+	destination *backendconfig.DestinationT
+}
+
 type DestinationValidator interface {
 	Validate(ctx context.Context, dest *backendconfig.DestinationT) *model.DestinationValidationResponse
 }
@@ -220,6 +224,10 @@
 			manager:     operations,
 			table:       getTable(dest),
 		}, nil
+	case model.VerifyingObjectStorageDelete:
+		return &objectStorageDelete{
+			destination: dest,
+		}, nil
 	}
 
 	return nil, fmt.Errorf("invalid step: %s", step)
@@ -322,6 +330,24 @@
 	return nil
 }
 
+func (osd *objectStorageDelete) Validate(ctx context.Context) error {
+	var (
+		tempPath     string
+		err          error
+		uploadObject filemanager.UploadedFile
+	)
+	if tempPath, err = CreateTempLoadFile(osd.destination); err != nil {
+		return fmt.Errorf("creating temp load file: %w", err)
+	}
+	if uploadObject, err = uploadFile(ctx, osd.destination, tempPath); err != nil {
+		return fmt.Errorf("upload file: %w", err)
+	}
+	if err = deleteFile(ctx, osd.destination, uploadObject.ObjectName); err != nil {
+		return fmt.Errorf("delete file: %w", err)
+	}
+	return nil
+}
+
 // CreateTempLoadFile creates a temporary load file
 func CreateTempLoadFile(dest *backendconfig.DestinationT) (string, error) {
 	var (
@@ -402,6 +428,20 @@
 	return output, nil
 }
 
+func deleteFile(ctx context.Context, dest *backendconfig.DestinationT, location string) error {
+	var (
+		err error
+		fm  filemanager.FileManager
+	)
+	if fm, err = createFileManager(dest); err != nil {
+		return fmt.Errorf("create file manager: %w", err)
+	}
+	if err = fm.Delete(ctx, []string{location}); err != nil {
+		return fmt.Errorf("delete file: %w", err)
+	}
+	return nil
+}
+
 func downloadFile(ctx context.Context, dest *backendconfig.DestinationT, location string) error {
 	var (
 		err          error

warehouse/validations/validate_test.go

@@ -94,6 +94,23 @@ func TestValidator(t *testing.T) {
 		})
 	})
 
+	t.Run("Object Storage Delete", func(t *testing.T) {
+		v, err := validations.NewValidator(ctx, model.VerifyingObjectStorageDelete, &backendconfig.DestinationT{
+			DestinationDefinition: backendconfig.DestinationDefinitionT{
+				Name: warehouseutils.POSTGRES,
+			},
+			Config: map[string]interface{}{
+				"bucketProvider":  provider,
+				"bucketName":      bucket,
+				"accessKeyID":     minioResource.AccessKeyID,
+				"secretAccessKey": minioResource.AccessKeySecret,
+				"endPoint":        minioResource.Endpoint,
+			},
+		})
+		require.NoError(t, err)
+		require.NoError(t, v.Validate(ctx))
+	})
+
 	t.Run("Connections", func(t *testing.T) {
 		testCases := []struct {
 			name      string