[exporter/awss3] Implement timeout

Author: nimishamehta5

.chloggen/awss3exporter-timeout.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: Implement timeout for S3 exporter
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36264]
+
+# (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/awss3exporter/README.md

@@ -37,6 +37,8 @@ The following exporter configuration parameters are supported.
 | `disable_ssl`             | set this to `true` to disable SSL when sending requests                                                                                    | false       |
 | `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          |
+
 
 ### Marshaler
 
@@ -76,6 +78,9 @@ exporters:
       enabled: true
       num_consumers: 10
       queue_size: 100
+
+    # Optional (5s by default)
+    timeout: 20s      
 ```
 
 Logs and traces will be stored inside 'databucket' in the following path format.

exporter/awss3exporter/config.go

@@ -53,10 +53,10 @@ const (
 
 // Config contains the main configuration options for the s3 exporter
 type Config struct {
-	QueueSettings exporterhelper.QueueConfig `mapstructure:"sending_queue"`
-
-	S3Uploader    S3UploaderConfig `mapstructure:"s3uploader"`
-	MarshalerName MarshalerType    `mapstructure:"marshaler"`
+	QueueSettings   exporterhelper.QueueConfig   `mapstructure:"sending_queue"`
+	TimeoutSettings exporterhelper.TimeoutConfig `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
+	S3Uploader      S3UploaderConfig             `mapstructure:"s3uploader"`
+	MarshalerName   MarshalerType                `mapstructure:"marshaler"`
 
 	// Encoding to apply. If present, overrides the marshaler configuration option.
 	Encoding              *component.ID `mapstructure:"encoding"`

exporter/awss3exporter/config_test.go

@@ -34,9 +34,11 @@ func TestLoadConfig(t *testing.T) {
 
 	queueCfg := exporterhelper.NewDefaultQueueConfig()
 	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
 
 	assert.Equal(t, &Config{
 		QueueSettings:         queueCfg,
+		TimeoutSettings:       timeoutCfg,
 		Encoding:              &encoding,
 		EncodingFileExtension: "baz",
 		S3Uploader: S3UploaderConfig{
@@ -69,10 +71,15 @@ func TestConfig(t *testing.T) {
 		QueueSize:    42,
 	}
 
+	timeoutCfg := exporterhelper.TimeoutConfig{
+		Timeout: 8,
+	}
+
 	e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
 
 	assert.Equal(t, &Config{
-		QueueSettings: queueCfg,
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
 		S3Uploader: S3UploaderConfig{
 			Region:            "us-east-1",
 			S3Bucket:          "foo",
@@ -103,6 +110,7 @@ func TestConfigS3StorageClass(t *testing.T) {
 	e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
 	queueCfg := exporterhelper.NewDefaultQueueConfig()
 	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
 
 	assert.Equal(t, &Config{
 		S3Uploader: S3UploaderConfig{
@@ -114,8 +122,9 @@ func TestConfigS3StorageClass(t *testing.T) {
 			StorageClass:      "STANDARD_IA",
 			ACL:               "private",
 		},
-		QueueSettings: queueCfg,
-		MarshalerName: "otlp_json",
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
+		MarshalerName:   "otlp_json",
 	}, e,
 	)
 }
@@ -136,6 +145,7 @@ func TestConfigS3ACL(t *testing.T) {
 	e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
 	queueCfg := exporterhelper.NewDefaultQueueConfig()
 	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
 
 	assert.Equal(t, &Config{
 		S3Uploader: S3UploaderConfig{
@@ -147,8 +157,9 @@ func TestConfigS3ACL(t *testing.T) {
 			StorageClass:      "STANDARD_IA",
 			ACL:               "private",
 		},
-		QueueSettings: queueCfg,
-		MarshalerName: "otlp_json",
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
+		MarshalerName:   "otlp_json",
 	}, e,
 	)
 }
@@ -167,11 +178,13 @@ func TestConfigForS3CompatibleSystems(t *testing.T) {
 
 	queueCfg := exporterhelper.NewDefaultQueueConfig()
 	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
 
 	e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
 
 	assert.Equal(t, &Config{
-		QueueSettings: queueCfg,
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
 		S3Uploader: S3UploaderConfig{
 			Region:            "us-east-1",
 			S3Bucket:          "foo",
@@ -283,11 +296,13 @@ func TestMarshallerName(t *testing.T) {
 
 	queueCfg := exporterhelper.NewDefaultQueueConfig()
 	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
 
 	e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
 
 	assert.Equal(t, &Config{
-		QueueSettings: queueCfg,
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
 		S3Uploader: S3UploaderConfig{
 			Region:            "us-east-1",
 			S3Bucket:          "foo",
@@ -302,7 +317,8 @@ func TestMarshallerName(t *testing.T) {
 	e = cfg.Exporters[component.MustNewIDWithName("awss3", "proto")].(*Config)
 
 	assert.Equal(t, &Config{
-		QueueSettings: queueCfg,
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
 		S3Uploader: S3UploaderConfig{
 			Region:            "us-east-1",
 			S3Bucket:          "bar",
@@ -329,11 +345,13 @@ func TestCompressionName(t *testing.T) {
 
 	queueCfg := exporterhelper.NewDefaultQueueConfig()
 	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
 
 	e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
 
 	assert.Equal(t, &Config{
-		QueueSettings: queueCfg,
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
 		S3Uploader: S3UploaderConfig{
 			Region:            "us-east-1",
 			S3Bucket:          "foo",
@@ -349,7 +367,8 @@ func TestCompressionName(t *testing.T) {
 	e = cfg.Exporters[component.MustNewIDWithName("awss3", "proto")].(*Config)
 
 	assert.Equal(t, &Config{
-		QueueSettings: queueCfg,
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
 		S3Uploader: S3UploaderConfig{
 			Region:            "us-east-1",
 			S3Bucket:          "bar",

exporter/awss3exporter/factory.go

@@ -28,9 +28,11 @@ func NewFactory() exporter.Factory {
 func createDefaultConfig() component.Config {
 	queueCfg := exporterhelper.NewDefaultQueueConfig()
 	queueCfg.Enabled = false
+	timeoutCfg := exporterhelper.NewDefaultTimeoutConfig()
 
 	return &Config{
-		QueueSettings: queueCfg,
+		QueueSettings:   queueCfg,
+		TimeoutSettings: timeoutCfg,
 		S3Uploader: S3UploaderConfig{
 			Region:            "us-east-1",
 			S3PartitionFormat: "year=%Y/month=%m/day=%d/hour=%H/minute=%M",
@@ -57,6 +59,7 @@ func createLogsExporter(ctx context.Context,
 		s3Exporter.ConsumeLogs,
 		exporterhelper.WithStart(s3Exporter.start),
 		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithTimeout(cfg.TimeoutSettings),
 	)
 }
 
@@ -80,6 +83,7 @@ func createMetricsExporter(ctx context.Context,
 		s3Exporter.ConsumeMetrics,
 		exporterhelper.WithStart(s3Exporter.start),
 		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithTimeout(cfg.TimeoutSettings),
 	)
 }
 
@@ -104,6 +108,7 @@ func createTracesExporter(ctx context.Context,
 		s3Exporter.ConsumeTraces,
 		exporterhelper.WithStart(s3Exporter.start),
 		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithTimeout(cfg.TimeoutSettings),
 	)
 }
 

exporter/awss3exporter/testdata/config.yaml

@@ -7,6 +7,7 @@ exporters:
       enabled: true
       num_consumers: 23
       queue_size: 42
+    timeout: 8
 
     s3uploader:
         region: 'us-east-1'