Add Azure Blob Storage support to ClickPipe object storage (#332)

Author: EvandroLG

examples/clickpipe/object_storage_azure_blob/README.md

@@ -0,0 +1,31 @@
+## ClickPipe Object Storage with Azure Blob Storage example
+
+This example demonstrates how to deploy a ClickPipe with an Azure Blob Storage container as the input source,
+authenticated with a connection string.
+
+## Prerequisites
+
+- Azure Storage Account with a blob container
+- Files in the container (e.g., JSON files in `data/` folder)
+- Azure Storage Account connection string
+
+## How to run
+
+- Rename `variables.sample.tfvars` to `variables.tfvars` and fill in all needed data.
+- Run `terraform init`
+- Run `terraform <plan|apply> -var-file=variables.tfvars`
+
+## Azure Blob Storage Configuration
+
+The example uses:
+- `type = "azureblobstorage"` for Azure Blob Storage
+- `authentication = "CONNECTION_STRING"` with Azure connection string
+- `azure_container_name` to specify the container
+- `path` to specify file path within the container (supports wildcards)
+
+## Connection String Format
+
+The Azure connection string should be in the format:
+```
+DefaultEndpointsProtocol=https;AccountName=<account_name>;AccountKey=<account_key>;EndpointSuffix=core.windows.net
+```

examples/clickpipe/object_storage_azure_blob/main.tf

@@ -0,0 +1,80 @@
+variable "organization_id" {}
+variable "token_key" {}
+variable "token_secret" {}
+
+variable "service_id" {
+  description = "ClickHouse service ID"
+}
+
+variable "azure_connection_string" {
+  description = "Azure Blob Storage connection string"
+  sensitive   = true
+}
+
+variable "azure_container_name" {
+  description = "Azure Blob Storage container name"
+}
+
+variable "azure_path" {
+  description = "Path to the file(s) within the Azure container"
+  default     = "data/*.json"
+}
+
+resource "clickhouse_clickpipe" "azure_blob" {
+  name        = "Azure Blob Storage 🚀 ClickPipe"
+  service_id = var.service_id
+  state = "Running"
+  source = {
+    object_storage = {
+      type    = "azureblobstorage"
+      format  = "JSONEachRow"
+
+      path                  = var.azure_path
+      azure_container_name  = var.azure_container_name
+      connection_string     = var.azure_connection_string
+
+      authentication = "CONNECTION_STRING"
+    }
+  }
+
+  destination = {
+    table         = "my_azure_data_table"
+    managed_table = true
+
+    table_definition = {
+      engine = {
+        type = "MergeTree"
+      }
+    }
+
+    columns = [
+      {
+        name = "id"
+        type = "UInt64"
+      }, {
+        name = "name"
+        type = "String"
+      }, {
+        name = "timestamp"
+        type = "DateTime"
+      }
+    ]
+  }
+
+  field_mappings = [
+    {
+      source_field      = "user_id"
+      destination_field = "id"
+    }, {
+      source_field      = "user_name"
+      destination_field = "name"
+    }, {
+      source_field      = "created_at"
+      destination_field = "timestamp"
+    }
+  ]
+}
+
+output "clickpipe_id" {
+  value = clickhouse_clickpipe.azure_blob.id
+}

examples/clickpipe/object_storage_azure_blob/provider.tf

@@ -0,0 +1,15 @@
+# This file is generated automatically please do not edit
+terraform {
+  required_providers {
+    clickhouse = {
+      version = "3.3.3-alpha3"
+      source  = "ClickHouse/clickhouse"
+    }
+  }
+}
+
+provider "clickhouse" {
+  organization_id = var.organization_id
+  token_key       = var.token_key
+  token_secret    = var.token_secret
+}

examples/clickpipe/object_storage_azure_blob/provider.tf.template.alpha

@@ -0,0 +1,14 @@
+terraform {
+  required_providers {
+    clickhouse = {
+      version = "${CLICKHOUSE_TERRAFORM_PROVIDER_VERSION}"
+      source  = "ClickHouse/clickhouse"
+    }
+  }
+}
+
+provider "clickhouse" {
+  organization_id = var.organization_id
+  token_key       = var.token_key
+  token_secret    = var.token_secret
+}

examples/clickpipe/object_storage_azure_blob/variables.sample.tfvars

@@ -0,0 +1,10 @@
+# these keys are for example only and won't work when pointed to a deployed ClickHouse OpenAPI server
+organization_id = "aee076c1-3f83-4637-95b1-ad5a0a825b71"
+token_key       = "avhj1U5QCdWAE9CA9"
+token_secret    = "4b1dROiHQEuSXJHlV8zHFd0S7WQj7CGxz5kGJeJnca"
+service_id      = "aee076c1-3f83-4637-95b1-ad5a0a825b71"
+
+# Azure Blob Storage configuration
+azure_connection_string = "DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey=mykey123;EndpointSuffix=core.windows.net"
+azure_container_name     = "mycontainer"
+azure_path              = "data/*.json"

pkg/internal/api/clickpipe.go

@@ -46,8 +46,9 @@ var (
 )
 
 const (
-	ClickPipeAuthenticationIAMRole = "IAM_ROLE"
-	ClickPipeAuthenticationIAMUser = "IAM_USER"
+	ClickPipeAuthenticationIAMRole          = "IAM_ROLE"
+	ClickPipeAuthenticationIAMUser          = "IAM_USER"
+	ClickPipeAuthenticationConnectionString = "CONNECTION_STRING"
 
 	ClickPipeKafkaAuthenticationPlain       = "PLAIN"
 	ClickPipeKafkaAuthenticationScramSha256 = "SCRAM-SHA-256"
@@ -83,6 +84,7 @@ var ClickPipeKafkaSourceTypes = []string{
 var ClickPipeObjectStorageAuthenticationMethods = []string{
 	ClickPipeAuthenticationIAMRole,
 	ClickPipeAuthenticationIAMUser,
+	ClickPipeAuthenticationConnectionString,
 }
 
 var ClickPipeKinesisAuthenticationMethods = []string{
@@ -110,13 +112,15 @@ var ClickPipeObjectStorageFormats = []string{
 }
 
 const (
-	ClickPipeObjectStorageS3Type  = "s3"
-	ClickPipeObjectStorageGCSType = "gcs"
+	ClickPipeObjectStorageS3Type        = "s3"
+	ClickPipeObjectStorageGCSType       = "gcs"
+	ClickPipeObjectStorageAzureBlobType = "azureblobstorage"
 )
 
 var ClickPipeObjectStorageTypes = []string{
 	ClickPipeObjectStorageS3Type,
 	ClickPipeObjectStorageGCSType,
+	ClickPipeObjectStorageAzureBlobType,
 }
 
 const (

pkg/internal/api/clickpipe_models.go

@@ -61,7 +61,7 @@ type ClickPipeObjectStorageSource struct {
 	Type   string `json:"type"`
 	Format string `json:"format"`
 
-	URL         string  `json:"url"`
+	URL         string  `json:"url,omitempty"`
 	Delimiter   *string `json:"delimiter,omitempty"`
 	Compression *string `json:"compression,omitempty"`
 
@@ -70,6 +70,11 @@ type ClickPipeObjectStorageSource struct {
 	Authentication *string                   `json:"authentication,omitempty"`
 	AccessKey      *ClickPipeSourceAccessKey `json:"accessKey,omitempty"`
 	IAMRole        *string                   `json:"iamRole,omitempty"`
+
+	// Azure Blob Storage specific fields
+	ConnectionString   *string `json:"connectionString,omitempty"`
+	Path               *string `json:"path,omitempty"`
+	AzureContainerName *string `json:"azureContainerName,omitempty"`
 }
 
 type ClickPipeKinesisSource struct {

pkg/resource/clickpipe.go

@@ -322,7 +322,7 @@ func (c *ClickPipeResource) Schema(_ context.Context, _ resource.SchemaRequest,
 						},
 					},
 					"object_storage": schema.SingleNestedAttribute{
-						MarkdownDescription: "The Kafka source configuration for the ClickPipe.",
+						MarkdownDescription: "The compatible object storage source configuration for the ClickPipe.",
 						Optional:            true,
 						Attributes: map[string]schema.Attribute{
 							"type": schema.StringAttribute{
@@ -355,8 +355,8 @@ func (c *ClickPipeResource) Schema(_ context.Context, _ resource.SchemaRequest,
 								},
 							},
 							"url": schema.StringAttribute{
-								MarkdownDescription: "The URL of the S3 bucket. Provide a path to the file(s) you want to ingest. You can specify multiple files using bash-like wildcards. For more information, see the documentation on using wildcards in path: https://clickhouse.com/docs/en/integrations/clickpipes/object-storage#limitations",
-								Required:            true,
+								MarkdownDescription: "The URL of the S3/GCS bucket. Required for S3 and GCS types. Not used for Azure Blob Storage (use path and azure_container_name instead). You can specify multiple files using bash-like wildcards. For more information, see the documentation on using wildcards in path: https://clickhouse.com/docs/en/integrations/clickpipes/object-storage#limitations",
+								Optional:            true,
 								PlanModifiers: []planmodifier.String{
 									stringplanmodifier.RequiresReplace(),
 								},
@@ -391,11 +391,8 @@ func (c *ClickPipeResource) Schema(_ context.Context, _ resource.SchemaRequest,
 								},
 							},
 							"authentication": schema.StringAttribute{
-								MarkdownDescription: fmt.Sprintf(
-									"Authentication method. If not provided, no authentication is used. It can be used to access public buckets.. (%s).",
-									wrapStringsWithBackticksAndJoinCommaSeparated(api.ClickPipeObjectStorageAuthenticationMethods),
-								),
-								Optional: true,
+								MarkdownDescription: "CONNECTION_STRING is for Azure Blob Storage. IAM_ROLE and IAM_USER are for AWS S3/GCS/DigitalOcean. If not provided, no authentication is used",
+								Optional:            true,
 								Validators: []validator.String{
 									stringvalidator.OneOf(api.ClickPipeObjectStorageAuthenticationMethods...),
 								},
@@ -420,6 +417,25 @@ func (c *ClickPipeResource) Schema(_ context.Context, _ resource.SchemaRequest,
 								MarkdownDescription: "The IAM role for the S3 source. Use with `IAM_ROLE` authentication. It can be used with AWS ClickHouse service only. Read more in [ClickPipes documentation page](https://clickhouse.com/docs/en/integrations/clickpipes/object-storage#authentication)",
 								Optional:            true,
 							},
+							"connection_string": schema.StringAttribute{
+								MarkdownDescription: "Connection string for Azure Blob Storage authentication. Required when authentication is CONNECTION_STRING. Example: `DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey=mykey;EndpointSuffix=core.windows.net`",
+								Optional:            true,
+								Sensitive:           true,
+							},
+							"path": schema.StringAttribute{
+								MarkdownDescription: "Path to the file(s) within the Azure container. Used for Azure Blob Storage sources. You can specify multiple files using bash-like wildcards. For more information, see the documentation on using wildcards in path: https://clickhouse.com/docs/en/integrations/clickpipes/object-storage#limitations. Example: `data/logs/*.json`",
+								Optional:            true,
+								PlanModifiers: []planmodifier.String{
+									stringplanmodifier.RequiresReplace(),
+								},
+							},
+							"azure_container_name": schema.StringAttribute{
+								MarkdownDescription: "Container name for Azure Blob Storage. Required when type is azureblobstorage. Example: `mycontainer`",
+								Optional:            true,
+								PlanModifiers: []planmodifier.String{
+									stringplanmodifier.RequiresReplace(),
+								},
+							},
 						},
 						PlanModifiers: []planmodifier.Object{
 							objectplanmodifier.RequiresReplace(),
@@ -957,17 +973,61 @@ func (c *ClickPipeResource) extractSourceFromPlan(ctx context.Context, diagnosti
 			}
 		}
 
-		source.ObjectStorage = &api.ClickPipeObjectStorageSource{
+		storageType := objectStorageModel.Type.ValueString()
+		if storageType == api.ClickPipeObjectStorageAzureBlobType {
+			if !objectStorageModel.URL.IsNull() && !objectStorageModel.URL.IsUnknown() && objectStorageModel.URL.ValueString() != "" {
+				diagnostics.AddError(
+					"Error Creating ClickPipe",
+					"URL field should not be used with Azure Blob Storage. Use 'path' and 'azure_container_name' fields instead",
+				)
+				return nil
+			}
+
+			if objectStorageModel.AzureContainerName.IsNull() || objectStorageModel.AzureContainerName.IsUnknown() || objectStorageModel.AzureContainerName.ValueString() == "" {
+				diagnostics.AddError(
+					"Error Creating ClickPipe",
+					"azure_container_name is required when using Azure Blob Storage",
+				)
+				return nil
+			}
+
+			if objectStorageModel.ConnectionString.IsNull() || objectStorageModel.ConnectionString.IsUnknown() || objectStorageModel.ConnectionString.ValueString() == "" {
+				diagnostics.AddError(
+					"Error Creating ClickPipe",
+					"connection_string is required when using Azure Blob Storage",
+				)
+				return nil
+			}
+		} else {
+			if objectStorageModel.URL.IsNull() || objectStorageModel.URL.IsUnknown() || objectStorageModel.URL.ValueString() == "" {
+				diagnostics.AddError(
+					"Error Creating ClickPipe",
+					fmt.Sprintf("URL is required when using %s storage type", storageType),
+				)
+				return nil
+			}
+		}
+
+		objectStorage := &api.ClickPipeObjectStorageSource{
 			Type:           objectStorageModel.Type.ValueString(),
 			Format:         objectStorageModel.Format.ValueString(),
-			URL:            objectStorageModel.URL.ValueString(),
 			Delimiter:      objectStorageModel.Delimiter.ValueStringPointer(),
 			Compression:    objectStorageModel.Compression.ValueStringPointer(),
 			IsContinuous:   objectStorageModel.IsContinuous.ValueBool(),
 			Authentication: objectStorageModel.Authentication.ValueStringPointer(),
 			AccessKey:      accessKey,
 			IAMRole:        objectStorageModel.IAMRole.ValueStringPointer(),
 		}
+
+		if storageType == api.ClickPipeObjectStorageAzureBlobType {
+			objectStorage.ConnectionString = objectStorageModel.ConnectionString.ValueStringPointer()
+			objectStorage.Path = objectStorageModel.Path.ValueStringPointer()
+			objectStorage.AzureContainerName = objectStorageModel.AzureContainerName.ValueStringPointer()
+		} else {
+			objectStorage.URL = objectStorageModel.URL.ValueString()
+		}
+
+		source.ObjectStorage = objectStorage
 	} else if !sourceModel.Kinesis.IsNull() {
 		kinesisModel := models.ClickPipeKinesisSourceModel{}
 		diagnostics.Append(sourceModel.Kinesis.As(ctx, &kinesisModel, basetypes.ObjectAsOptions{})...)
@@ -1161,14 +1221,25 @@ func (c *ClickPipeResource) syncClickPipeState(ctx context.Context, state *model
 		}
 
 		objectStorageModel := models.ClickPipeObjectStorageSourceModel{
-			Type:           types.StringValue(clickPipe.Source.ObjectStorage.Type),
-			Format:         types.StringValue(clickPipe.Source.ObjectStorage.Format),
-			URL:            types.StringValue(clickPipe.Source.ObjectStorage.URL),
-			Delimiter:      types.StringPointerValue(clickPipe.Source.ObjectStorage.Delimiter),
-			Compression:    types.StringPointerValue(clickPipe.Source.ObjectStorage.Compression),
-			IsContinuous:   types.BoolValue(clickPipe.Source.ObjectStorage.IsContinuous),
-			Authentication: types.StringPointerValue(clickPipe.Source.ObjectStorage.Authentication),
-			IAMRole:        types.StringPointerValue(clickPipe.Source.ObjectStorage.IAMRole),
+			Type:         types.StringValue(clickPipe.Source.ObjectStorage.Type),
+			Format:       types.StringValue(clickPipe.Source.ObjectStorage.Format),
+			Delimiter:    types.StringPointerValue(clickPipe.Source.ObjectStorage.Delimiter),
+			Compression:  types.StringPointerValue(clickPipe.Source.ObjectStorage.Compression),
+			IsContinuous: types.BoolValue(clickPipe.Source.ObjectStorage.IsContinuous),
+			IAMRole:      types.StringPointerValue(clickPipe.Source.ObjectStorage.IAMRole),
+		}
+
+		// Set storage-type-specific fields
+		if clickPipe.Source.ObjectStorage.Type == api.ClickPipeObjectStorageAzureBlobType {
+			// For Azure Blob Storage, preserve all fields from state as API doesn't return them
+			objectStorageModel.Authentication = stateObjectStorageModel.Authentication
+			objectStorageModel.ConnectionString = stateObjectStorageModel.ConnectionString
+			objectStorageModel.Path = stateObjectStorageModel.Path
+			objectStorageModel.AzureContainerName = stateObjectStorageModel.AzureContainerName
+		} else {
+			// For S3-compatible storage, use API response values
+			objectStorageModel.Authentication = types.StringPointerValue(clickPipe.Source.ObjectStorage.Authentication)
+			objectStorageModel.URL = types.StringValue(clickPipe.Source.ObjectStorage.URL)
 		}
 
 		if !stateObjectStorageModel.AccessKey.IsNull() {