[Feature] added support for cloudflare_api_token in `databricks_sto…

…rage_credential` resource (#3835)

## Changes
- Updated the struct, and added necessary secret handling
- Resolves #3834

## Tests
<!-- 
How is this tested? Please see the checklist below and also describe any
other relevant tests
-->

- [x] `make test` run locally
- [x] relevant change in `docs/` folder
- [x] covered with integration tests in `internal/acceptance`
- [x] relevant acceptance tests are passing
- [x] using Go SDK

catalog/resource_metastore_data_access.go

@@ -19,7 +19,7 @@ type GcpServiceAccountKey struct {
 }
 
 var alofCred = []string{"aws_iam_role", "azure_service_principal", "azure_managed_identity",
-	"gcp_service_account_key", "databricks_gcp_service_account"}
+	"gcp_service_account_key", "databricks_gcp_service_account", "cloudflare_api_token"}
 
 func SuppressGcpSAKeyDiff(k, old, new string, d *schema.ResourceData) bool {
 	//ignore changes in private_key
@@ -28,11 +28,9 @@ func SuppressGcpSAKeyDiff(k, old, new string, d *schema.ResourceData) bool {
 
 // it's used by both ResourceMetastoreDataAccess & ResourceStorageCredential
 func adjustDataAccessSchema(m map[string]*schema.Schema) map[string]*schema.Schema {
-	m["aws_iam_role"].AtLeastOneOf = alofCred
-	m["azure_service_principal"].AtLeastOneOf = alofCred
-	m["azure_managed_identity"].AtLeastOneOf = alofCred
-	m["gcp_service_account_key"].AtLeastOneOf = alofCred
-	m["databricks_gcp_service_account"].AtLeastOneOf = alofCred
+	for _, cred := range alofCred {
+		common.CustomizeSchemaPath(m, cred).SetAtLeastOneOf(alofCred)
+	}
 
 	// suppress changes for private_key
 	m["gcp_service_account_key"].DiffSuppressFunc = SuppressGcpSAKeyDiff
@@ -42,6 +40,8 @@ func adjustDataAccessSchema(m map[string]*schema.Schema) map[string]*schema.Sche
 	common.MustSchemaPath(m, "azure_managed_identity", "credential_id").Computed = true
 	common.MustSchemaPath(m, "databricks_gcp_service_account", "email").Computed = true
 	common.MustSchemaPath(m, "databricks_gcp_service_account", "credential_id").Computed = true
+	common.MustSchemaPath(m, "azure_service_principal", "client_secret").Sensitive = true
+	common.MustSchemaPath(m, "cloudflare_api_token", "secret_access_key").Sensitive = true
 
 	m["force_destroy"] = &schema.Schema{
 		Type:     schema.TypeBool,

catalog/resource_storage_credential.go

@@ -19,6 +19,7 @@ type StorageCredentialInfo struct {
 	AzMI                        *catalog.AzureManagedIdentityResponse        `json:"azure_managed_identity,omitempty" tf:"group:access"`
 	GcpSAKey                    *GcpServiceAccountKey                        `json:"gcp_service_account_key,omitempty" tf:"group:access"`
 	DatabricksGcpServiceAccount *catalog.DatabricksGcpServiceAccountResponse `json:"databricks_gcp_service_account,omitempty" tf:"computed"`
+	CloudflareApiToken          *catalog.CloudflareApiToken                  `json:"cloudflare_api_token,omitempty" tf:"group:access"`
 	MetastoreID                 string                                       `json:"metastore_id,omitempty" tf:"computed"`
 	ReadOnly                    bool                                         `json:"read_only,omitempty"`
 	SkipValidation              bool                                         `json:"skip_validation,omitempty"`
@@ -122,14 +123,19 @@ func ResourceStorageCredential() common.Resource {
 				if err != nil {
 					return err
 				}
-				// azure client secret is sensitive, so we need to preserve it
+				// azure client secret, & r2 secret access key are sensitive, so we need to preserve them
 				var scOrig catalog.CreateStorageCredential
 				common.DataToStructPointer(d, storageCredentialSchema, &scOrig)
 				if scOrig.AzureServicePrincipal != nil {
 					if scOrig.AzureServicePrincipal.ClientSecret != "" {
 						storageCredential.CredentialInfo.AzureServicePrincipal.ClientSecret = scOrig.AzureServicePrincipal.ClientSecret
 					}
 				}
+				if scOrig.CloudflareApiToken != nil {
+					if scOrig.CloudflareApiToken.SecretAccessKey != "" {
+						storageCredential.CredentialInfo.CloudflareApiToken.SecretAccessKey = scOrig.CloudflareApiToken.SecretAccessKey
+					}
+				}
 				err = common.StructToData(storageCredential.CredentialInfo, storageCredentialSchema, d)
 				if err != nil {
 					return err
@@ -141,14 +147,19 @@ func ResourceStorageCredential() common.Resource {
 				if err != nil {
 					return err
 				}
-				// azure client secret is sensitive, so we need to preserve it
+				// azure client secret, & r2 secret access key are sensitive, so we need to preserve them
 				var scOrig catalog.CreateStorageCredential
 				common.DataToStructPointer(d, storageCredentialSchema, &scOrig)
 				if scOrig.AzureServicePrincipal != nil {
 					if scOrig.AzureServicePrincipal.ClientSecret != "" {
 						storageCredential.AzureServicePrincipal.ClientSecret = scOrig.AzureServicePrincipal.ClientSecret
 					}
 				}
+				if scOrig.CloudflareApiToken != nil {
+					if scOrig.CloudflareApiToken.SecretAccessKey != "" {
+						storageCredential.CloudflareApiToken.SecretAccessKey = scOrig.CloudflareApiToken.SecretAccessKey
+					}
+				}
 				err = common.StructToData(storageCredential, storageCredentialSchema, d)
 				if err != nil {
 					return err

catalog/resource_storage_credential_test.go

@@ -336,6 +336,59 @@ func TestCreateStorageCredentialsReadOnly(t *testing.T) {
 	}.ApplyNoError(t)
 }
 
+func TestCreateStorageCredentialCloudflare(t *testing.T) {
+	qa.ResourceFixture{
+		Fixtures: []qa.HTTPFixture{
+			{
+				Method:   "POST",
+				Resource: "/api/2.1/unity-catalog/storage-credentials",
+				ExpectedRequest: catalog.CreateStorageCredential{
+					Name: "a",
+					CloudflareApiToken: &catalog.CloudflareApiToken{
+						AccountId:       "1234",
+						AccessKeyId:     "1234",
+						SecretAccessKey: "1234",
+					},
+					Comment: "c",
+				},
+				Response: catalog.StorageCredentialInfo{
+					Name: "a",
+				},
+			},
+			{
+				Method:   "GET",
+				Resource: "/api/2.1/unity-catalog/storage-credentials/a?",
+				Response: catalog.StorageCredentialInfo{
+					Name: "a",
+					CloudflareApiToken: &catalog.CloudflareApiToken{
+						AccountId:   "1234",
+						AccessKeyId: "1234",
+					},
+					MetastoreId: "d",
+					Id:          "1234-5678",
+				},
+			},
+		},
+		Resource: ResourceStorageCredential(),
+		Create:   true,
+		HCL: `
+		name = "a"
+		cloudflare_api_token {
+			account_id = "1234"
+			access_key_id = "1234"
+			secret_access_key = "1234"
+		}
+		comment = "c"
+		`,
+	}.ApplyAndExpectData(t, map[string]any{
+		"cloudflare_api_token.0.secret_access_key": "1234",
+		"cloudflare_api_token.0.access_key_id":     "1234",
+		"cloudflare_api_token.0.account_id":        "1234",
+		"name":                                     "a",
+		"storage_credential_id":                    "1234-5678",
+	})
+}
+
 func TestUpdateStorageCredentials(t *testing.T) {
 	qa.ResourceFixture{
 		Fixtures: []qa.HTTPFixture{

docs/resources/storage_credential.md

@@ -96,6 +96,12 @@ The following arguments are required:
 
 - `email` (output only) - The email of the GCP service account created, to be granted access to relevant buckets.
 
+`cloudflare_api_token` optional configuration block for using a Cloudflare API Token as credential details. This requires account admin access:
+
+- `account_id` - R2 account ID
+- `access_key_id` - R2 API token access key ID
+- `secret_access_key` - R2 API token secret access key
+
 `azure_service_principal` optional configuration block to use service principal as credential details for Azure (Legacy):
 
 - `directory_id` - The directory ID corresponding to the Azure Active Directory (AAD) tenant of the application

internal/acceptance/storage_credential_test.go

@@ -16,6 +16,16 @@ func TestUcAccStorageCredential(t *testing.T) {
 					}
 					skip_validation = true
 					comment = "Managed by TF"
+				}
+				resource "databricks_storage_credential" "r2" {
+					name = "r2-{var.RANDOM}"
+					cloudflare_api_token {
+						account_id = "1234"
+						access_key_id = "1234"
+						secret_access_key = "1234"
+					}
+					skip_validation = true
+					comment = "Managed by TF"
 				}`,
 		})
 	} else if isGcp(t) {