[exporter/awss3exporter] Added IsArchiveFormat for Sumo IC (open-telemetry#43574)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
The sumo_ic marshaler was incorrectly blocked from using compression.
Sumo Logic's Installed Collector compresses archive files before S3
upload, but the OpenTelemetry exporter prevented this, causing format
incompatibility.

Root cause: The exporter used ContentEncoding: "gzip" for all compressed
uploads, which signals HTTP transfer compression (auto-decompressed by
AWS). Archive formats like sumo_ic need files to stay compressed
permanently.

Solution: Added isArchiveFormat flag to distinguish between archive
formats (files stay compressed) and transfer formats (HTTP
optimization). The flag controls whether ContentEncoding header is set
during upload.


<!--Describe what testing was performed and which tests were added.-->
#### Testing
- Manual testing: Downloaded files from S3, verified gzip compression
with file and gunzip commands
- Unit test passed:
```
➜  awss3exporter git:(sumoCompression) ✗ make test
/Users/timothy.chan/Desktop/opentelemetry-collector-contrib/.tools/gotestsum --rerun-fails=1 --packages="./..." -- -race -timeout 900s -parallel 4 --tags=""
∅  internal/metadata
✓  internal/upload (cached)
✓  . (2.662s)

DONE 98 tests in 2.662s
```

<!--Describe the documentation added.-->
#### Documentation
- Removed outdated README that gzip doesn't support sumo_ic marshaler

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

Author: chan-tim-sumo

.chloggen/sumoCompression.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: exporter/awss3
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Support compression with the sumo_ic marshaller
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [43574]
+
+# (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: []

exporter/awss3exporter/README.md

@@ -82,7 +82,7 @@ See https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/
 
 ### Compression
 - `none` (default): No compression will be applied
-- `gzip`: Files will be compressed with gzip. **This does not support `sumo_ic`marshaler.**
+- `gzip`: Files will be compressed with gzip.
 
 ### resource_attrs_to_s3
 - `s3_bucket`: Defines which resource attribute's value should be used as the S3 bucket.

exporter/awss3exporter/config.go

@@ -146,10 +146,6 @@ func (c *Config) Validate() error {
 		if compression != configcompression.TypeGzip {
 			errs = multierr.Append(errs, errors.New("unknown compression type"))
 		}
-
-		if c.MarshalerName == SumoIC {
-			errs = multierr.Append(errs, errors.New("marshaler does not support compression"))
-		}
 	}
 
 	if c.S3Uploader.RetryMode != "nop" && c.S3Uploader.RetryMode != "standard" && c.S3Uploader.RetryMode != "adaptive" {

exporter/awss3exporter/exporter.go

@@ -75,7 +75,7 @@ func (e *s3Exporter) start(ctx context.Context, host component.Host) error {
 
 	e.marshaler = m
 
-	up, err := newUploadManager(ctx, e.config, e.signalType, m.format())
+	up, err := newUploadManager(ctx, e.config, e.signalType, m.format(), m.compressed())
 	if err != nil {
 		return err
 	}

exporter/awss3exporter/internal/upload/partition.go

@@ -49,6 +49,10 @@ type PartitionKeyBuilder struct {
 	// generating a new unique string to avoid collisions on file upload
 	// across many different instances.
 	UniqueKeyFunc func() string
+	// IsCompressed when true keeps files compressed in S3
+	// by omitting ContentEncoding headers. When false, ContentEncoding
+	// is set for HTTP transfer compression (AWS auto-decompresses).
+	IsCompressed bool
 }
 
 func (pki *PartitionKeyBuilder) Build(ts time.Time, overridePrefix string) string {

exporter/awss3exporter/internal/upload/writer.go

@@ -64,7 +64,10 @@ func (sw *s3manager) Upload(ctx context.Context, data []byte, opts *UploadOption
 	}
 
 	encoding := ""
-	if sw.builder.Compression.IsCompressed() {
+	// Only use ContentEncoding for non-archive formats
+	// Archive formats store files compressed permanently (like .tar.gz)
+	// while ContentEncoding is for HTTP transfer compression
+	if sw.builder.Compression.IsCompressed() && !sw.builder.IsCompressed {
 		encoding = string(sw.builder.Compression)
 	}
 
@@ -79,15 +82,20 @@ func (sw *s3manager) Upload(ctx context.Context, data []byte, opts *UploadOption
 		}
 	}
 
-	_, err = sw.uploader.Upload(ctx, &s3.PutObjectInput{
-		Bucket:          aws.String(overrideBucket),
-		Key:             aws.String(sw.builder.Build(now, overridePrefix)),
-		Body:            content,
-		ContentEncoding: aws.String(encoding),
-		StorageClass:    sw.storageClass,
-		ACL:             sw.acl,
-	})
+	uploadInput := &s3.PutObjectInput{
+		Bucket:       aws.String(overrideBucket),
+		Key:          aws.String(sw.builder.Build(now, overridePrefix)),
+		Body:         content,
+		StorageClass: sw.storageClass,
+		ACL:          sw.acl,
+	}
+
+	// Only set ContentEncoding if we have a non-empty encoding value
+	if encoding != "" {
+		uploadInput.ContentEncoding = aws.String(encoding)
+	}
 
+	_, err = sw.uploader.Upload(ctx, uploadInput)
 	return err
 }
 

exporter/awss3exporter/marshaler.go

@@ -19,6 +19,7 @@ type marshaler interface {
 	MarshalLogs(ld plog.Logs) ([]byte, error)
 	MarshalMetrics(md pmetric.Metrics) ([]byte, error)
 	format() string
+	compressed() bool
 }
 
 var ErrUnknownMarshaler = errors.New("unknown marshaler")
@@ -34,6 +35,7 @@ func newMarshalerFromEncoding(encoding *component.ID, fileFormat string, host co
 	marshaler.metricsMarshaler, _ = e.(pmetric.Marshaler)
 	marshaler.tracesMarshaler, _ = e.(ptrace.Marshaler)
 	marshaler.fileFormat = fileFormat
+	marshaler.IsCompressed = false
 	return marshaler, nil
 }
 
@@ -45,19 +47,23 @@ func newMarshaler(mType MarshalerType, logger *zap.Logger) (marshaler, error) {
 		marshaler.tracesMarshaler = &ptrace.ProtoMarshaler{}
 		marshaler.metricsMarshaler = &pmetric.ProtoMarshaler{}
 		marshaler.fileFormat = "binpb"
+		marshaler.IsCompressed = false
 	case OtlpJSON:
 		marshaler.logsMarshaler = &plog.JSONMarshaler{}
 		marshaler.tracesMarshaler = &ptrace.JSONMarshaler{}
 		marshaler.metricsMarshaler = &pmetric.JSONMarshaler{}
 		marshaler.fileFormat = "json"
+		marshaler.IsCompressed = false
 	case SumoIC:
 		sumomarshaler := newSumoICMarshaler()
 		marshaler.logsMarshaler = &sumomarshaler
-		marshaler.fileFormat = "json.gz"
+		marshaler.fileFormat = "json"
+		marshaler.IsCompressed = true
 	case Body:
 		exportbodyMarshaler := newbodyMarshaler()
 		marshaler.logsMarshaler = &exportbodyMarshaler
 		marshaler.fileFormat = exportbodyMarshaler.format()
+		marshaler.IsCompressed = false
 	default:
 		return nil, ErrUnknownMarshaler
 	}

exporter/awss3exporter/marshaler_test.go

@@ -31,7 +31,7 @@ func TestMarshaler(t *testing.T) {
 		m, err := newMarshaler("sumo_ic", zap.NewNop())
 		assert.NoError(t, err)
 		require.NotNil(t, m)
-		assert.Equal(t, "json.gz", m.format())
+		assert.Equal(t, "json", m.format())
 	}
 	{
 		m, err := newMarshaler("unknown", zap.NewNop())

exporter/awss3exporter/s3_marshaler.go

@@ -16,6 +16,7 @@ type s3Marshaler struct {
 	metricsMarshaler pmetric.Marshaler
 	logger           *zap.Logger
 	fileFormat       string
+	IsCompressed     bool
 }
 
 func (marshaler *s3Marshaler) MarshalTraces(td ptrace.Traces) ([]byte, error) {
@@ -33,3 +34,7 @@ func (marshaler *s3Marshaler) MarshalMetrics(md pmetric.Metrics) ([]byte, error)
 func (marshaler *s3Marshaler) format() string {
 	return marshaler.fileFormat
 }
+
+func (marshaler *s3Marshaler) compressed() bool {
+	return marshaler.IsCompressed
+}

exporter/awss3exporter/s3_writer.go

@@ -24,6 +24,7 @@ func newUploadManager(
 	conf *Config,
 	metadata string,
 	format string,
+	isCompressed bool,
 ) (upload.Manager, error) {
 	configOpts := []func(*config.LoadOptions) error{}
 
@@ -110,6 +111,7 @@ func newUploadManager(
 			Metadata:              metadata,
 			Compression:           conf.S3Uploader.Compression,
 			UniqueKeyFunc:         uniqueKeyFunc,
+			IsCompressed:          isCompressed,
 		},
 		s3.NewFromConfig(cfg, s3Opts...),
 		s3types.StorageClass(conf.S3Uploader.StorageClass),