[exporter/googlecloudstorage] add compression support (open-telemetry#45384)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
Add compression support for the Google Cloud Storage exporter, allowing
users to compress log data before uploading to GCS to reduce storage
costs and transfer times. Supports gzip and zstd compression algorithms
with automatic file extension handling (.gz, .zst).

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

<!--Describe what testing was performed and which tests were added.-->
#### Testing
- Added comprehensive unit tests for compression functionality including
gzip and zstd compression verification
- Tests verify that compressed data can be properly decompressed back to
original content
- Tests verify correct file extensions (.gz, .zst) are applied to
compressed files
- Tests verify compression reduces data size compared to uncompressed
data
- All existing tests continue to pass, ensuring backward compatibility
- Added configuration validation tests for compression settings

<!--Describe the documentation added.-->
#### Documentation
- Updated README.md with compression configuration options and examples
- Added usage examples showing how to enable gzip/zstd compression
- Documented the `bucket.compression` configuration parameter

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

---------

Signed-off-by: Alex <alex.s@akeyless.io>
Co-authored-by: Constança Manteigas <113898685+constanca-m@users.noreply.github.com>

Author: 2 people

.chloggen/45337-add-gcs-compression-support.yaml

@@ -0,0 +1,25 @@
+# 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. receiver/filelog)
+component: exporter/googlecloudstorage
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add compression support for Google Cloud Storage exporter
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [45337]
+
+# (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: |
+  The Google Cloud Storage exporter now supports compression of log data before uploading to GCS.
+  Supported compression algorithms: `gzip` and `zstd`.
+
+# 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/googlecloudstorageexporter/README.md

@@ -27,6 +27,7 @@ This exporter writes received OpenTelemetry data to a cloud storage bucket.
 | `bucket.partition`       | Configuration for time-based partitioning. See below for details.                                                                                                                                                        | No       |         |
 | `bucket.reuse_if_exists` | If true, use the existing bucket if it already exists; if false, error if bucket exists.                                                                                                                                 | No       | `false` |
 | `bucket.region`          | Region where the bucket will be created or where it exists. If left empty, it will query the metadata endpoint. It requires the collector to be running in a Google Cloud environment.                                   | Yes      |         |
+| `bucket.compression`     | Compression algorithm used to compress data before uploading. Valid values are `gzip`, `zstd`, or no value set for no compression.                                                                                        | No       |         |
 
 ### Partition Configuration
 
@@ -59,3 +60,18 @@ extensions:
   # text encoding to ensure only the body is placed in the bucket
   text_encoding:
 ```
+
+### Compression Example
+
+```yaml
+exporters:
+  googlecloudstorage:
+    bucket:
+      name: compressed-logs-bucket
+      project_id: my-project
+      region: us-central1
+      compression: gzip
+      file_prefix: logs
+      partition:
+        format: "year=%Y/month=%m/day=%d"
+```

exporter/googlecloudstorageexporter/config.go

@@ -9,12 +9,14 @@ import (
 
 	"github.com/lestrrat-go/strftime"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configcompression"
 	"go.opentelemetry.io/collector/confmap/xconfmap"
 )
 
 var (
-	errNameRequired  = errors.New("name is required")
-	errFormatInvalid = errors.New("invalid format")
+	errNameRequired       = errors.New("name is required")
+	errFormatInvalid      = errors.New("invalid format")
+	errUnknownCompression = errors.New("unknown compression type")
 )
 
 type Config struct {
@@ -52,6 +54,11 @@ type bucketConfig struct {
 	// empty, it will query the metadata endpoint. It requires the collector
 	// to be running in a Google Cloud environment.
 	Region string `mapstructure:"region"`
+
+	// Compression sets the algorithm used to process the payload
+	// before uploading to GCS.
+	// Valid values are: `gzip`, `zstd`, or no value set.
+	Compression configcompression.Type `mapstructure:"compression"`
 }
 
 type partitionConfig struct {
@@ -84,6 +91,17 @@ func (c *bucketConfig) Validate() error {
 	if c.Name == "" {
 		return errNameRequired
 	}
+
+	compression := c.Compression
+	if compression.IsCompressed() {
+		if compression != configcompression.TypeGzip && compression != configcompression.TypeZstd {
+			return fmt.Errorf(
+				"%w %q, valid values are %q and %q",
+				errUnknownCompression, compression,
+				configcompression.TypeGzip, configcompression.TypeZstd)
+		}
+	}
+
 	return nil
 }
 

exporter/googlecloudstorageexporter/config.schema.yaml

@@ -3,6 +3,9 @@ properties:
   bucket:
     type: object
     properties:
+      compression:
+        description: 'Compression sets the algorithm used to process the payload before uploading to GCS. Valid values are: `gzip`, `zstd`, or no value set.'
+        $ref: go.opentelemetry.io/collector/config/configcompression.type
       file_prefix:
         description: 'FilePrefix holds the prefix for the created filename. This prefix is applied after the partition path (if any). Example: file_prefix: "logs" Result: ".../logs_UUID"'
         type: string

exporter/googlecloudstorageexporter/config_test.go

@@ -59,6 +59,38 @@ func TestValidate(t *testing.T) {
 				},
 			},
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "with_gzip_compression"),
+			expected: &Config{
+				Encoding: func() *component.ID {
+					id := component.MustNewID("test")
+					return &id
+				}(),
+				Bucket: bucketConfig{
+					Name:        "test-bucket",
+					Region:      "test-region",
+					ProjectID:   "test-project-id",
+					FilePrefix:  "logs",
+					Compression: "gzip",
+				},
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "with_zstd_compression"),
+			expected: &Config{
+				Encoding: func() *component.ID {
+					id := component.MustNewID("test")
+					return &id
+				}(),
+				Bucket: bucketConfig{
+					Name:        "test-bucket",
+					Region:      "test-region",
+					ProjectID:   "test-project-id",
+					FilePrefix:  "logs",
+					Compression: "zstd",
+				},
+			},
+		},
 		{
 			id:          component.NewIDWithName(metadata.Type, "empty_bucket_name"),
 			expectedErr: errNameRequired,
@@ -67,6 +99,10 @@ func TestValidate(t *testing.T) {
 			id:          component.NewIDWithName(metadata.Type, "invalid_partition_format"),
 			expectedErr: errFormatInvalid,
 		},
+		{
+			id:          component.NewIDWithName(metadata.Type, "unsupported_compression"),
+			expectedErr: errUnknownCompression,
+		},
 	}
 
 	for _, tt := range tests {

exporter/googlecloudstorageexporter/exporter.go

@@ -4,18 +4,24 @@
 package googlecloudstorageexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/googlecloudstorageexporter"
 
 import (
+	"bytes"
+	"compress/gzip"
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"cloud.google.com/go/compute/metadata"
 	"cloud.google.com/go/storage"
 	"github.com/google/uuid"
+	"github.com/klauspost/compress/zstd"
 	"github.com/lestrrat-go/strftime"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configcompression"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -24,6 +30,10 @@ import (
 	"google.golang.org/api/googleapi"
 )
 
+type poolItem interface {
+	io.WriteCloser
+	Reset(io.Writer)
+}
 type signalType string
 
 const (
@@ -46,6 +56,8 @@ type storageExporter struct {
 	bucketHandle    *storage.BucketHandle
 	logger          *zap.Logger
 	partitionFormat *strftime.Strftime
+	gzipWriterPool  *sync.Pool
+	zstdWriterPool  *sync.Pool
 	signal          signalType
 }
 
@@ -116,6 +128,28 @@ func newStorageExporter(
 		logger:          logger,
 		partitionFormat: partitionFormat,
 		signal:          signal,
+		gzipWriterPool: &sync.Pool{
+			New: func() any {
+				// Create a new gzip writer that writes to a dummy buffer initially
+				// It will be reset to the actual destination when used
+				writer := gzip.NewWriter(io.Discard)
+				return writer
+			},
+		},
+		zstdWriterPool: &sync.Pool{
+			New: func() any {
+				// Create a new zstd writer that writes to a dummy buffer initially
+				// It will be reset to the actual destination when used
+				writer, err := zstd.NewWriter(io.Discard)
+				if err != nil {
+					logger.Error("failed to create zstd writer for pool, falling back to on-demand creation",
+						zap.Error(err))
+					// Return nil on error - sync.Pool will handle this gracefully
+					return nil
+				}
+				return writer
+			},
+		},
 	}, nil
 }
 
@@ -218,6 +252,7 @@ func (s *storageExporter) ConsumeTraces(ctx context.Context, td ptrace.Traces) e
 // It starts from a unique ID, and prepends the partitionFormat and the prefix to it.
 func generateFilename(
 	uniqueID, filePrefix, partitionPrefix string,
+	compression configcompression.Type,
 	partitionFormat *strftime.Strftime,
 	now time.Time,
 ) string {
@@ -244,6 +279,15 @@ func generateFilename(
 		}
 		filename = partitionPrefix + filename
 	}
+
+	// Add compression extension
+	switch compression {
+	case configcompression.TypeGzip:
+		filename += ".gz"
+	case configcompression.TypeZstd:
+		filename += ".zst"
+	}
+
 	return filename
 }
 
@@ -253,13 +297,20 @@ func (s *storageExporter) uploadFile(ctx context.Context, content []byte) (err e
 		return nil
 	}
 
+	// Compress the content if compression is configured
+	compressedContent, err := s.compressContent(content)
+	if err != nil {
+		return fmt.Errorf("failed to compress content: %w", err)
+	}
+
 	// if we have multiple files coming, we need to make sure the name is unique so they do
 	// not overwrite each other
 	uniqueID := uuid.New().String()
 	filename := generateFilename(
 		uniqueID,
 		s.cfg.Bucket.FilePrefix,
 		s.cfg.Bucket.Partition.Prefix,
+		s.cfg.Bucket.Compression,
 		s.partitionFormat,
 		time.Now().UTC(),
 	)
@@ -276,18 +327,77 @@ func (s *storageExporter) uploadFile(ctx context.Context, content []byte) (err e
 			)
 		}
 	}()
-	if _, err = writer.Write(content); err != nil {
+	if _, err = writer.Write(compressedContent); err != nil {
 		return fmt.Errorf("failed to write to file: %w", err)
 	}
 	s.logger.Info(
 		"New file uploaded",
 		zap.String("filename", filename),
 		zap.String("bucket", s.cfg.Bucket.Name),
-		zap.Int("size", len(content)),
+		zap.Int("size", len(compressedContent)),
 	)
 	return nil
 }
 
+func (s *storageExporter) compressContent(raw []byte) ([]byte, error) {
+	switch s.cfg.Bucket.Compression {
+	case configcompression.TypeGzip:
+		return compress[*gzip.Writer](s.gzipWriterPool, raw, func(w io.Writer) (*gzip.Writer, error) {
+			return gzip.NewWriter(w), nil
+		})
+	case configcompression.TypeZstd:
+		return compress[*zstd.Encoder](s.zstdWriterPool, raw, func(w io.Writer) (*zstd.Encoder, error) {
+			return zstd.NewWriter(w)
+		})
+	default:
+		return raw, nil
+	}
+}
+
+func compress[T poolItem](pool *sync.Pool, raw []byte, newItem func(io.Writer) (T, error)) ([]byte, error) {
+	if pool == nil {
+		return nil, errors.New("unexpected: compress pool is nil")
+	}
+
+	content := bytes.NewBuffer(nil)
+
+	// Get writer from pool or create new one
+	pooled := pool.Get()
+	var zipper T
+	var fromPool bool
+	if pooled != nil {
+		if w, ok := pooled.(T); ok {
+			zipper = w
+			zipper.Reset(content)
+			fromPool = true
+		}
+	}
+	if !fromPool {
+		var err error
+		zipper, err = newItem(content)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// Write the data
+	if _, err := zipper.Write(raw); err != nil {
+		// Always close to release resources, but ignore close error on write failure
+		_ = zipper.Close()
+		return nil, err
+	}
+
+	// Close the writer
+	if err := zipper.Close(); err != nil {
+		return nil, err
+	}
+
+	// Only return the writer to the pool after successful write and close
+	pool.Put(zipper)
+
+	return content.Bytes(), nil
+}
+
 // loadExtension tries to load an available extension for the given id.
 func loadExtension[T any](host component.Host, id component.ID, _ string, errNotMarshaler error) (T, error) {
 	var zero T