[exporter/azureblob] support append blob for azure blob exporter (open-telemetry#39076)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Support append blob for azure blob exporter.

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#39075 

<!--Describe what testing was performed and which tests were added.-->
#### Testing

1. unit test
2. e2e test using real azure blob account

<!--Describe the documentation added.-->
#### Documentation

See README.md

<!--Please delete paragraphs that you did not use before submitting.-->

Author: hgaol

.chloggen/39075.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: azureblobexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: support for append blob in azure blob storage exporter
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39075]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/azureblobexporter/README.md

@@ -40,6 +40,9 @@ The following settings can be optionally configured and have default values:
   - logs (default `nil`): encoding component id.
   - metrics (default `nil`): encoding component id.
   - traces (default `nil`): encoding component id.
+- append_blob: configures append blob behavior. When enabled, telemetry data is appended to a single blob instead of creating new blobs. This can be useful for aggregating data or reducing the number of blobs created.
+  - enabled (default `false`): determines whether to use append blob mode.
+  - separator (default `\n`): string to insert between appended data blocks.
 - `retry_on_failure`
   - `enabled` (default = true)
   - `initial_interval` (default = 5s): Time to wait after the first failure before retrying; ignored if `enabled` is `false`
@@ -67,7 +70,15 @@ exporter:
     auth:
       type: "connection_string"
       connection_string: "DefaultEndpointsProtocol=https;AccountName=<your-acount>;AccountKey=<account-key>;EndpointSuffix=core.windows.net"
-          traces: "test"
     encodings:
       logs: text_encoding
+    append_blob:
+      enabled: true
+      separator: "\n"
 ```
+
+When `append_blob` is enabled:
+- The exporter will create append blobs instead of block blobs
+- New data will be appended to existing blobs rather than creating new ones
+- The configured separator will be inserted between data blocks
+- If the blob doesn't exist, it will be created automatically

exporter/azureblobexporter/config.go

@@ -34,6 +34,11 @@ type BlobNameFormat struct {
 	Params         map[string]string `mapstructure:"params"`
 }
 
+type AppendBlob struct {
+	Enabled   bool   `mapstructure:"enabled"`
+	Separator string `mapstructure:"separator"`
+}
+
 type Authentication struct {
 	// Type is the authentication type. supported values are connection_string, service_principal, system_managed_identity and user_managed_identity
 	Type AuthType `mapstructure:"type"`
@@ -75,6 +80,9 @@ type Config struct {
 	// FormatType is the format of encoded telemetry data. Supported values are json and proto.
 	FormatType string `mapstructure:"format"`
 
+	// AppendBlob configures append blob behavior
+	AppendBlob *AppendBlob `mapstructure:"append_blob"`
+
 	// Encoding extension to apply for logs/metrics/traces. If present, overrides the marshaler configuration option and format.
 	Encodings *Encodings `mapstructure:"encodings"`
 

exporter/azureblobexporter/config_test.go

@@ -53,6 +53,10 @@ func TestLoadConfig(t *testing.T) {
 				FormatType:    "json",
 				Encodings:     &Encodings{},
 				BackOffConfig: configretry.NewDefaultBackOffConfig(),
+				AppendBlob: &AppendBlob{
+					Enabled:   false,
+					Separator: "\n",
+				},
 			},
 		},
 		{
@@ -77,6 +81,10 @@ func TestLoadConfig(t *testing.T) {
 				FormatType:    "proto",
 				Encodings:     &Encodings{},
 				BackOffConfig: configretry.NewDefaultBackOffConfig(),
+				AppendBlob: &AppendBlob{
+					Enabled:   false,
+					Separator: "\n",
+				},
 			},
 		},
 		{
@@ -102,6 +110,10 @@ func TestLoadConfig(t *testing.T) {
 				FormatType:    "json",
 				Encodings:     &Encodings{},
 				BackOffConfig: configretry.NewDefaultBackOffConfig(),
+				AppendBlob: &AppendBlob{
+					Enabled:   false,
+					Separator: "\n",
+				},
 			},
 		},
 		{
@@ -126,6 +138,10 @@ func TestLoadConfig(t *testing.T) {
 				FormatType:    "json",
 				Encodings:     &Encodings{},
 				BackOffConfig: configretry.NewDefaultBackOffConfig(),
+				AppendBlob: &AppendBlob{
+					Enabled:   false,
+					Separator: "\n",
+				},
 			},
 		},
 		{

exporter/azureblobexporter/exporter.go

@@ -6,13 +6,16 @@ package azureblobexporter // import "github.com/open-telemetry/opentelemetry-col
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math/rand/v2"
 	"time"
 
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/appendblob"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -33,6 +36,45 @@ type azureBlobExporter struct {
 type azblobClient interface {
 	UploadStream(ctx context.Context, containerName string, blobName string, body io.Reader, o *azblob.UploadStreamOptions) (azblob.UploadStreamResponse, error)
 	URL() string
+	AppendBlock(ctx context.Context, containerName string, blobName string, data []byte, o *appendblob.AppendBlockOptions) error
+}
+
+type azblobClientImpl struct {
+	client *azblob.Client
+}
+
+func (c *azblobClientImpl) UploadStream(ctx context.Context, containerName string, blobName string, body io.Reader, o *azblob.UploadStreamOptions) (azblob.UploadStreamResponse, error) {
+	return c.client.UploadStream(ctx, containerName, blobName, body, o)
+}
+
+func (c *azblobClientImpl) URL() string {
+	return c.client.URL()
+}
+
+func (c *azblobClientImpl) AppendBlock(ctx context.Context, containerName string, blobName string, data []byte, o *appendblob.AppendBlockOptions) error {
+	containerClient := c.client.ServiceClient().NewContainerClient(containerName)
+	appendBlobClient := containerClient.NewAppendBlobClient(blobName)
+
+	_, err := appendBlobClient.AppendBlock(ctx, newReadSeekCloserWrapper(data), o)
+	if err == nil {
+		return nil
+	}
+
+	// Handle BlobNotFound error by creating the blob and retrying
+	var cerr *azcore.ResponseError
+	if errors.As(err, &cerr) && cerr.ErrorCode == "BlobNotFound" {
+		if _, err = appendBlobClient.Create(ctx, nil); err != nil {
+			return fmt.Errorf("failed to create append blob: %w", err)
+		}
+
+		_, err = appendBlobClient.AppendBlock(ctx, newReadSeekCloserWrapper(data), o)
+		if err != nil {
+			return fmt.Errorf("failed to append block after creation: %w", err)
+		}
+		return nil
+	}
+
+	return fmt.Errorf("failed to append block: %w", err)
 }
 
 func newAzureBlobExporter(config *Config, logger *zap.Logger, signal pipeline.Signal) *azureBlobExporter {
@@ -58,9 +100,11 @@ func (e *azureBlobExporter) start(_ context.Context, host component.Host) error
 
 	// create client based on auth type
 	authType := e.config.Auth.Type
+	azblobClient := &azblobClientImpl{}
+	e.client = azblobClient
 	switch authType {
 	case ConnectionString:
-		e.client, err = azblob.NewClientFromConnectionString(e.config.Auth.ConnectionString, nil)
+		azblobClient.client, err = azblob.NewClientFromConnectionString(e.config.Auth.ConnectionString, nil)
 		if err != nil {
 			return fmt.Errorf("failed to create client from connection string: %w", err)
 		}
@@ -73,7 +117,7 @@ func (e *azureBlobExporter) start(_ context.Context, host component.Host) error
 		if err != nil {
 			return fmt.Errorf("failed to create service principal credential: %w", err)
 		}
-		e.client, err = azblob.NewClient(e.config.URL, cred, nil)
+		azblobClient.client, err = azblob.NewClient(e.config.URL, cred, nil)
 		if err != nil {
 			return fmt.Errorf("failed to create client with service principal: %w", err)
 		}
@@ -82,7 +126,7 @@ func (e *azureBlobExporter) start(_ context.Context, host component.Host) error
 		if err != nil {
 			return fmt.Errorf("failed to create system managed identity credential: %w", err)
 		}
-		e.client, err = azblob.NewClient(e.config.URL, cred, nil)
+		azblobClient.client, err = azblob.NewClient(e.config.URL, cred, nil)
 		if err != nil {
 			return fmt.Errorf("failed to create client with system managed identity: %w", err)
 		}
@@ -93,7 +137,7 @@ func (e *azureBlobExporter) start(_ context.Context, host component.Host) error
 		if err != nil {
 			return fmt.Errorf("failed to create user managed identity credential: %w", err)
 		}
-		e.client, err = azblob.NewClient(e.config.URL, cred, nil)
+		azblobClient.client, err = azblob.NewClient(e.config.URL, cred, nil)
 		if err != nil {
 			return fmt.Errorf("failed to create client with user managed identity: %w", err)
 		}
@@ -118,7 +162,7 @@ func (e *azureBlobExporter) generateBlobName(signal pipeline.Signal) (string, er
 	default:
 		return "", fmt.Errorf("unsupported signal type: %v", signal)
 	}
-	return fmt.Sprintf("%s_%d", now.Format(format), randomInRange(1, int(e.config.BlobNameFormat.SerialNumRange))), nil
+	return fmt.Sprintf("%s_%d", now.Format(format), randomInRange(0, int(e.config.BlobNameFormat.SerialNumRange))), nil
 }
 
 func (e *azureBlobExporter) Capabilities() consumer.Capabilities {
@@ -162,17 +206,51 @@ func (e *azureBlobExporter) consumeData(ctx context.Context, data []byte, signal
 		return fmt.Errorf("failed to generate blobname: %w", err)
 	}
 
-	blobContentReader := bytes.NewReader(data)
-	_, err = e.client.UploadStream(ctx, e.config.Container.Traces, blobName, blobContentReader, nil)
+	var containerName string
+	switch signal {
+	case pipeline.SignalMetrics:
+		containerName = e.config.Container.Metrics
+	case pipeline.SignalLogs:
+		containerName = e.config.Container.Logs
+	case pipeline.SignalTraces:
+		containerName = e.config.Container.Traces
+	default:
+		return fmt.Errorf("unsupported signal type: %v", signal)
+	}
+
+	if e.config.AppendBlob != nil && e.config.AppendBlob.Enabled {
+		// Add separator if configured
+		if e.config.AppendBlob.Separator != "" {
+			data = append(data, []byte(e.config.AppendBlob.Separator)...)
+		}
+		err = e.client.AppendBlock(ctx, containerName, blobName, data, nil)
+	} else {
+		blobContentReader := bytes.NewReader(data)
+		_, err = e.client.UploadStream(ctx, containerName, blobName, blobContentReader, nil)
+	}
+
 	if err != nil {
-		return fmt.Errorf("failed to upload traces data: %w", err)
+		return fmt.Errorf("failed to upload data: %w", err)
 	}
 
 	e.logger.Debug("Successfully exported data to Azure Blob Storage",
 		zap.String("account", e.client.URL()),
-		zap.String("container", e.config.Container.Traces),
+		zap.String("container", containerName),
 		zap.String("blob", blobName),
 		zap.Int("size", len(data)))
 
 	return nil
 }
+
+func newReadSeekCloserWrapper(data []byte) *readSeekCloserWrapper {
+	return &readSeekCloserWrapper{bytes.NewReader(data)}
+}
+
+// readSeekCloserWrapper wraps a bytes.Reader to implement io.ReadSeekCloser
+type readSeekCloserWrapper struct {
+	*bytes.Reader
+}
+
+func (r readSeekCloserWrapper) Close() error {
+	return nil
+}