feat: add support for mapping s3 bucket prefix to OTel resource attributes

.chloggen/feat_s3exp-mapping-attributes.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: awss3exporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add configuration field `resource_attrs_to_s3/s3_prefix` to support mapping s3 bucket prefix to OTel resource attributes
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37858]
+
+# (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 `resource_attrs_to_s3/s3_prefix` is configured, s3 prefix will be determined based on the specified resource attribute and `s3uploader/s3_prefix` will serve as a fallback.
+
+# 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, api]

exporter/awss3exporter/README.md

@@ -38,7 +38,7 @@ The following exporter configuration parameters are supported.
 | `compression`             | should the file be compressed                                                                                                              | none                                        |
 | `sending_queue`           | [exporters common queuing](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)          | disabled                                    |
 | `timeout`                 | [exporters common timeout](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)          | 5s                                          |
-
+| `resource_attrs_to_s3`        | determines the mapping of S3 configuration values to resource attribute values for uploading operations.                                   |                                             |
 
 ### Marshaler
 
@@ -61,6 +61,12 @@ See https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/
 - `none` (default): No compression will be applied
 - `gzip`: Files will be compressed with gzip. **This does not support `sumo_ic`marshaler.**
 
+### resource_attrs_to_s3
+- `s3_prefix`: Defines which resource attribute's value should be used as the S3 prefix.
+  When this option is set, it dynamically overrides `s3uploader/s3_prefix`. 
+  If the specified resource attribute exists in the data,  
+  its value will be used as the prefix; otherwise, `s3uploader/s3_prefix` will serve as the fallback.
+
 # Example Configurations
 
 Following example configuration defines to store output in 'eu-central' region and bucket named 'databucket'.
@@ -110,6 +116,31 @@ In this case, logs and traces would be stored in the following path format.
 metric/YYYY/MM/DD/HH/mm
 ```
 
+## Data routing based on resource attributes
+When `resource_attrs_to_s3/s3_prefix` is configured, the S3 prefix is dynamically derived from a specified resource attribute in your data.
+If the attribute value is unavailable, the prefix will fall back to the value defined in `s3uploader/s3_prefix`.
+```yaml
+exporters:
+  awss3:
+    s3uploader:
+      region: 'eu-central-1'
+      s3_bucket: 'databucket'
+      s3_prefix: 'metric'
+      s3_partition_format: '%Y/%m/%d/%H/%M'
+    resource_attrs_to_s3:
+      s3_prefix: "com.awss3.prefix"
+```
+In this case, metrics, logs and traces would be stored in the following path format examples:
+
+```console
+prefix1/YYYY/MM/DD/HH/mm
+foo-prefix/YYYY/MM/DD/HH/mm
+prefix-bar/YYYY/MM/DD/HH/mm
+metric/YYYY/MM/DD/HH/mm
+...
+```
+
+
 ## AWS Credential Configuration
 
 This exporter follows default credential resolution for the

exporter/awss3exporter/config.go

@@ -51,6 +51,12 @@ const (
 	Body         MarshalerType = "body"
 )
 
+// ResourceAttrsToS3 defines the mapping of S3 uploading configuration values to resource attribute values.
+type ResourceAttrsToS3 struct {
+	// S3Prefix indicates the mapping of the key (directory) prefix used for writing into the bucket to a specific resource attribute value.
+	S3Prefix string `mapstructure:"s3_prefix"`
+}
+
 // Config contains the main configuration options for the s3 exporter
 type Config struct {
 	QueueSettings   exporterhelper.QueueBatchConfig `mapstructure:"sending_queue"`
@@ -59,8 +65,9 @@ type Config struct {
 	MarshalerName   MarshalerType                   `mapstructure:"marshaler"`
 
 	// Encoding to apply. If present, overrides the marshaler configuration option.
-	Encoding              *component.ID `mapstructure:"encoding"`
-	EncodingFileExtension string        `mapstructure:"encoding_file_extension"`
+	Encoding              *component.ID     `mapstructure:"encoding"`
+	EncodingFileExtension string            `mapstructure:"encoding_file_extension"`
+	ResourceAttrsToS3     ResourceAttrsToS3 `mapstructure:"resource_attrs_to_s3"`
 }
 
 func (c *Config) Validate() error {

exporter/awss3exporter/config_test.go

@@ -135,7 +135,7 @@ func TestConfigS3ACL(t *testing.T) {
 	factories.Exporters[factory.Type()] = factory
 	// https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/33594
 	cfg, err := otelcoltest.LoadConfigAndValidate(
-		filepath.Join("testdata", "config-s3_storage_class.yaml"), factories)
+		filepath.Join("testdata", "config-s3_acl.yaml"), factories)
 
 	require.NoError(t, err)
 	require.NotNil(t, cfg)
@@ -152,7 +152,8 @@ func TestConfigS3ACL(t *testing.T) {
 			S3Prefix:          "bar",
 			S3PartitionFormat: "year=%Y/month=%m/day=%d/hour=%H/minute=%M",
 			Endpoint:          "http://endpoint.com",
-			StorageClass:      "STANDARD_IA",
+			StorageClass:      "STANDARD",
+			ACL:               "bucket-owner-read",
 		},
 		QueueSettings:   queueCfg,
 		TimeoutSettings: timeoutCfg,
@@ -408,3 +409,40 @@ func TestCompressionName(t *testing.T) {
 	}, e,
 	)
 }
+
+func TestResourceAttrsToS3(t *testing.T) {
+	factories, err := otelcoltest.NopFactories()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[factory.Type()] = factory
+	cfg, err := otelcoltest.LoadConfigAndValidate(
+		filepath.Join("testdata", "config-s3_resource-attrs-to-s3.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	queueCfg := exporterhelper.NewDefaultQueueConfig()
+	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
+
+	e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
+
+	assert.Equal(t, &Config{
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
+		S3Uploader: S3UploaderConfig{
+			Region:            "us-east-1",
+			S3Bucket:          "foo",
+			S3Prefix:          "bar",
+			S3PartitionFormat: "year=%Y/month=%m/day=%d/hour=%H/minute=%M",
+			Endpoint:          "http://endpoint.com",
+			StorageClass:      "STANDARD",
+		},
+		MarshalerName: "otlp_json",
+		ResourceAttrsToS3: ResourceAttrsToS3{
+			S3Prefix: "com.awss3.prefix",
+		},
+	}, e,
+	)
+}

exporter/awss3exporter/exporter.go

@@ -7,6 +7,8 @@
 	"context"
 	"fmt"
 
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
@@ -39,6 +41,19 @@
 	return s3Exporter
 }
 
+func (e *s3Exporter) getUploadOpts(res pcommon.Resource) *upload.UploadOptions {
+	s3Prefix := ""
+	if s3PrefixKey := e.config.ResourceAttrsToS3.S3Prefix; s3PrefixKey != "" {
+		if value, ok := res.Attributes().Get(s3PrefixKey); ok {
+			s3Prefix = value.AsString()
+		}
+	}
+	uploadOpts := &upload.UploadOptions{
+		OverridePrefix: s3Prefix,
+	}
+	return uploadOpts
+}
+
 func (e *s3Exporter) start(ctx context.Context, host component.Host) error {
 	var m marshaler
 	var err error
@@ -72,7 +87,8 @@
 		return err
 	}
 
-	return e.uploader.Upload(ctx, buf)
+	uploadOpts := e.getUploadOpts(md.ResourceMetrics().At(0).Resource())
+	return e.uploader.Upload(ctx, buf, uploadOpts)
 }
 
 func (e *s3Exporter) ConsumeLogs(ctx context.Context, logs plog.Logs) error {
@@ -81,7 +97,9 @@
 		return err
 	}
 
-	return e.uploader.Upload(ctx, buf)
+	uploadOpts := e.getUploadOpts(logs.ResourceLogs().At(0).Resource())
+
+	return e.uploader.Upload(ctx, buf, uploadOpts)
 }
 
 func (e *s3Exporter) ConsumeTraces(ctx context.Context, traces ptrace.Traces) error {
@@ -90,5 +108,7 @@
 		return err
 	}
 
-	return e.uploader.Upload(ctx, buf)
+	uploadOpts := e.getUploadOpts(traces.ResourceSpans().At(0).Resource())
+
+	return e.uploader.Upload(ctx, buf, uploadOpts)
 }

exporter/awss3exporter/exporter_test.go

@@ -5,21 +5,29 @@
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awss3exporter/internal/upload"
+
 	"github.com/stretchr/testify/assert"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.uber.org/zap"
 )
 
-var testLogs = []byte(`{"resourceLogs":[{"resource":{"attributes":[{"key":"_sourceCategory","value":{"stringValue":"logfile"}},{"key":"_sourceHost","value":{"stringValue":"host"}}]},"scopeLogs":[{"scope":{},"logRecords":[{"observedTimeUnixNano":"1654257420681895000","body":{"stringValue":"2022-06-03 13:57:00.62739 +0200 CEST m=+14.018296742 log entry14"},"attributes":[{"key":"log.file.path_resolved","value":{"stringValue":"logwriter/data.log"}}],"traceId":"","spanId":""}]}],"schemaUrl":"https://opentelemetry.io/schemas/1.6.1"}]}`)
+var (
+	s3PrefixKey    = "_sourceHost"
+	overridePrefix = "host"
+	testLogs       = []byte(fmt.Sprintf(`{"resourceLogs":[{"resource":{"attributes":[{"key":"_sourceCategory","value":{"stringValue":"logfile"}},{"key":"%s","value":{"stringValue":"%s"}}]},"scopeLogs":[{"scope":{},"logRecords":[{"observedTimeUnixNano":"1654257420681895000","body":{"stringValue":"2022-06-03 13:57:00.62739 +0200 CEST m=+14.018296742 log entry14"},"attributes":[{"key":"log.file.path_resolved","value":{"stringValue":"logwriter/data.log"}}],"traceId":"","spanId":""}]}],"schemaUrl":"https://opentelemetry.io/schemas/1.6.1"}]}`, s3PrefixKey, overridePrefix))
+)
 
 type TestWriter struct {
 	t *testing.T
 }
 
-func (testWriter *TestWriter) Upload(_ context.Context, buf []byte) error {
+func (testWriter *TestWriter) Upload(_ context.Context, buf []byte, uploadOpts *upload.UploadOptions) error {
 	assert.Equal(testWriter.t, testLogs, buf)
+	assert.Equal(testWriter.t, &upload.UploadOptions{OverridePrefix: ""}, uploadOpts)
 	return nil
 }
 
@@ -46,3 +54,32 @@
 	exporter := getLogExporter(t)
 	assert.NoError(t, exporter.ConsumeLogs(context.Background(), logs))
 }
+
+type TestWriterWithResourceAttrs struct {
+	t *testing.T
+}
+
+func (testWriterWO *TestWriterWithResourceAttrs) Upload(_ context.Context, buf []byte, uploadOpts *upload.UploadOptions) error {
+	assert.Equal(testWriterWO.t, testLogs, buf)
+	assert.Equal(testWriterWO.t, &upload.UploadOptions{OverridePrefix: overridePrefix}, uploadOpts)
+	return nil
+}
+
+func getLogExporterWithResourceAttrs(t *testing.T) *s3Exporter {
+	marshaler, _ := newMarshaler("otlp_json", zap.NewNop())
+	config := createDefaultConfig().(*Config)
+	config.ResourceAttrsToS3.S3Prefix = s3PrefixKey
+	exporter := &s3Exporter{
+		config:    config,
+		uploader:  &TestWriterWithResourceAttrs{t},
+		logger:    zap.NewNop(),
+		marshaler: marshaler,
+	}
+	return exporter
+}
+
+func TestLogWithResourceAttrs(t *testing.T) {
+	logs := getTestLogs(t)
+	exporter := getLogExporterWithResourceAttrs(t)
+	assert.NoError(t, exporter.ConsumeLogs(context.Background(), logs))
+}