[exporter/awscloudwatchlogsexporter] Improve performance of the awscloudwatchlogs exporter

.chloggen/parallel-awscloudwatchlogsexporter.yaml

@@ -0,0 +1,22 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# 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: awscloudwatchlogsexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Improve the performance of the awscloudwatchlogsexporter"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26692]
+
+# (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: |
+  Improve the performance by adding support to multiple consumers and removing locks and limiters that are no longer
+  necessary.

exporter/awscloudwatchlogsexporter/README.md

@@ -31,7 +31,10 @@ The following settings can be optionally configured:
 - `endpoint`: The CloudWatch Logs service endpoint which the requests are forwarded to. [See the CloudWatch Logs endpoints](https://docs.aws.amazon.com/general/latest/gr/cwl_region.html) for a list.
 - `log_retention`: LogRetention is the option to set the log retention policy for only newly created CloudWatch Log Groups. Defaults to Never Expire if not specified or set to 0.  Possible values for retention in days are 1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1827, 2192, 2557, 2922, 3288, or 3653. 
 - `tags`: Tags is the option to set tags for the CloudWatch Log Group. If specified, please add at most 50 tags. Input is a string to string map like so: { 'key': 'value' }. Keys must be between 1-128 characters and follow the regex pattern: `^([\p{L}\p{Z}\p{N}_.:/=+\-@]+)$`(alphanumerics, whitespace, and _.:/=+-!). Values must be between 1-256 characters and follow the regex pattern: `^([\p{L}\p{Z}\p{N}_.:/=+\-@]*)$`(alphanumerics, whitespace, and _.:/=+-!).  [Link to tagging restrictions](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_CreateLogGroup.html#:~:text=Required%3A%20Yes-,tags,-The%20key%2Dvalue)
-- `raw_log`: Boolean default false. If set to true, only the log message will be exported to CloudWatch Logs. This needs to be set to true for [EMF logs](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html). 
+- `raw_log`: Boolean default false. If set to true, only the log message will be exported to CloudWatch Logs. This needs to be set to true for [EMF logs](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html).
+- `sending_queue`: [Parameters for the sending queue](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md), where you can control parallelism and the size of the sending buffer. Obs.: this component will always have a sending queue enabled. 
+  - `num_consumers`: Number of consumers that will consume from the sending queue. This parameter controls how many consumers will consume from the sending queue in parallel.
+  - `queue_size`: Maximum number of batches kept in memory before dropping; ignored if enabled is false
 
 ### Examples
 
@@ -63,7 +66,7 @@ exporters:
 - If the log group and/or log stream are specified in an EMF log, that EMF log will be exported to that log group and/or log stream (i.e. ignores the log group and log stream defined in the configuration)
   - The log group and log stream will also be created automatically if they do not already exist.
   - Example of an EMF log with log group and log stream: 
-      ```json
+```json
   {"_aws":{"Timestamp":1574109732004,"LogGroupName":"Foo", "LogStreamName": "Bar", "CloudWatchMetrics":[{"Namespace":"MyApp","Dimensions":[["Operation"]],"Metrics":[{"Name":"ProcessingLatency","Unit":"Milliseconds","StorageResolution":60}]}]},"Operation":"Aggregator","ProcessingLatency":100}
   ```
 - Resource ARNs (Amazon Resource Name (ARN) of the AWS resource running the collector) are currently not supported with the CloudWatch Logs Exporter.

exporter/awscloudwatchlogsexporter/config.go

@@ -42,7 +42,7 @@ type Config struct {
 	Tags map[string]*string `mapstructure:"tags"`
 
 	// QueueSettings is a subset of exporterhelper.QueueSettings,
-	// because only QueueSize is user-settable due to how AWS CloudWatch API works
+	// because we want to force the queue to always be enabled
 	QueueSettings QueueSettings `mapstructure:"sending_queue"`
 
 	logger *zap.Logger
@@ -57,6 +57,8 @@ type Config struct {
 type QueueSettings struct {
 	// QueueSize set the length of the sending queue
 	QueueSize int `mapstructure:"queue_size"`
+	// NumConsumers set the number of consumers consuming from the queue
+	NumConsumers int `mapstructure:"num_consumers"`
 }
 
 var _ component.Config = (*Config)(nil)
@@ -72,6 +74,9 @@ func (config *Config) Validate() error {
 	if config.QueueSettings.QueueSize < 1 {
 		return errors.New("'sending_queue.queue_size' must be 1 or greater")
 	}
+	if config.QueueSettings.NumConsumers < 1 {
+		return errors.New("'sending_queue.num_consumers' must be 1 or greater")
+	}
 	if retErr := cwlogs.ValidateRetentionValue(config.LogRetention); retErr != nil {
 		return retErr
 	}
@@ -81,9 +86,8 @@ func (config *Config) Validate() error {
 
 func (config *Config) enforcedQueueSettings() exporterhelper.QueueSettings {
 	return exporterhelper.QueueSettings{
-		Enabled: true,
-		// due to the sequence token, there can be only one request in flight
-		NumConsumers: 1,
+		Enabled:      true,
+		NumConsumers: config.QueueSettings.NumConsumers,
 		QueueSize:    config.QueueSettings.QueueSize,
 	}
 }

exporter/awscloudwatchlogsexporter/config_test.go

@@ -43,7 +43,8 @@ func TestLoadConfig(t *testing.T) {
 				Endpoint:           "",
 				AWSSessionSettings: awsutil.CreateDefaultSessionConfig(),
 				QueueSettings: QueueSettings{
-					QueueSize: exporterhelper.NewDefaultQueueSettings().QueueSize,
+					NumConsumers: 1,
+					QueueSize:    exporterhelper.NewDefaultQueueSettings().QueueSize,
 				},
 			},
 		},
@@ -62,7 +63,8 @@ func TestLoadConfig(t *testing.T) {
 				LogGroupName:       "test-2",
 				LogStreamName:      "testing",
 				QueueSettings: QueueSettings{
-					QueueSize: 2,
+					NumConsumers: 1,
+					QueueSize:    2,
 				},
 			},
 		},
@@ -80,7 +82,7 @@ func TestLoadConfig(t *testing.T) {
 		},
 		{
 			id:           component.NewIDWithName(metadata.Type, "invalid_queue_setting"),
-			errorMessage: `'sending_queue' has invalid keys: enabled, num_consumers`,
+			errorMessage: `'sending_queue' has invalid keys: enabled`,
 		},
 	}
 
@@ -114,7 +116,8 @@ func TestRetentionValidateCorrect(t *testing.T) {
 		LogRetention:       365,
 		AWSSessionSettings: awsutil.CreateDefaultSessionConfig(),
 		QueueSettings: QueueSettings{
-			QueueSize: exporterhelper.NewDefaultQueueSettings().QueueSize,
+			NumConsumers: 1,
+			QueueSize:    exporterhelper.NewDefaultQueueSettings().QueueSize,
 		},
 	}
 	assert.NoError(t, component.ValidateConfig(cfg))
@@ -214,7 +217,8 @@ func TestValidateTags(t *testing.T) {
 				Tags:               tt.tags,
 				AWSSessionSettings: awsutil.CreateDefaultSessionConfig(),
 				QueueSettings: QueueSettings{
-					QueueSize: exporterhelper.NewDefaultQueueSettings().QueueSize,
+					NumConsumers: 1,
+					QueueSize:    exporterhelper.NewDefaultQueueSettings().QueueSize,
 				},
 			}
 			if tt.errorMessage != "" {

exporter/awscloudwatchlogsexporter/exporter.go

@@ -8,7 +8,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"errors"
-	"sync"
+	"fmt"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -32,8 +32,7 @@ type cwlExporter struct {
 	retryCount       int
 	collectorID      string
 	svcStructuredLog *cwlogs.Client
-	pusherMap        map[cwlogs.PusherKey]cwlogs.Pusher
-	pusherMapLock    sync.RWMutex
+	pusherFactory    cwlogs.MultiStreamPusherFactory
 }
 
 type awsMetadata struct {
@@ -68,24 +67,16 @@ func newCwLogsPusher(expConfig *Config, params exp.CreateSettings) (*cwlExporter
 		return nil, err
 	}
 
-	pusherKey := cwlogs.PusherKey{
-		LogGroupName:  expConfig.LogGroupName,
-		LogStreamName: expConfig.LogStreamName,
-	}
-
-	pusher := cwlogs.NewPusher(pusherKey, *awsConfig.MaxRetries, *svcStructuredLog, params.Logger)
-
-	pusherMap := make(map[cwlogs.PusherKey]cwlogs.Pusher)
-
-	pusherMap[pusherKey] = pusher
+	logStreamManager := cwlogs.NewLogStreamManager(*svcStructuredLog)
+	multiStreamPusherFactory := cwlogs.NewMultiStreamPusherFactory(logStreamManager, *svcStructuredLog, params.Logger)
 
 	logsExporter := &cwlExporter{
 		svcStructuredLog: svcStructuredLog,
 		Config:           expConfig,
 		logger:           params.Logger,
 		retryCount:       *awsConfig.MaxRetries,
 		collectorID:      collectorIdentifier.String(),
-		pusherMap:        pusherMap,
+		pusherFactory:    multiStreamPusherFactory,
 	}
 	return logsExporter, nil
 }
@@ -109,74 +100,34 @@ func newCwLogsExporter(config component.Config, params exp.CreateSettings) (exp.
 }
 
 func (e *cwlExporter) consumeLogs(_ context.Context, ld plog.Logs) error {
-	logEvents, _ := logsToCWLogs(e.logger, ld, e.Config)
-	if len(logEvents) == 0 {
-		return nil
-	}
+	pusher := e.pusherFactory.CreateMultiStreamPusher()
 
-	logPushersUsed := make(map[cwlogs.PusherKey]cwlogs.Pusher)
-	for _, logEvent := range logEvents {
-		pusherKey := cwlogs.PusherKey{
-			LogGroupName:  logEvent.LogGroupName,
-			LogStreamName: logEvent.LogStreamName,
-		}
-		cwLogsPusher := e.getLogPusher(logEvent)
-		e.logger.Debug("Adding log event", zap.Any("event", logEvent))
-		err := cwLogsPusher.AddLogEntry(logEvent)
-		if err != nil {
-			e.logger.Error("Failed ", zap.Int("num_of_events", len(logEvents)))
-		}
-		logPushersUsed[pusherKey] = cwLogsPusher
-	}
-	var flushErrArray []error
-	for _, pusher := range logPushersUsed {
-		flushErr := pusher.ForceFlush()
-		if flushErr != nil {
-			e.logger.Error("Error force flushing logs. Skipping to next logPusher.", zap.Error(flushErr))
-			flushErrArray = append(flushErrArray, flushErr)
-		}
-	}
-	if len(flushErrArray) != 0 {
-		errorString := ""
-		for _, err := range flushErrArray {
-			errorString += err.Error()
-		}
-		return errors.New(errorString)
-	}
-	return nil
-}
+	_, err := logsToCWLogs(e.logger, ld, e.Config, pusher)
 
-func (e *cwlExporter) getLogPusher(logEvent *cwlogs.Event) cwlogs.Pusher {
-	e.pusherMapLock.Lock()
-	defer e.pusherMapLock.Unlock()
-	pusherKey := cwlogs.PusherKey{
-		LogGroupName:  logEvent.LogGroupName,
-		LogStreamName: logEvent.LogStreamName,
+	if err != nil {
+		return fmt.Errorf("Error pushing logs:  %w", err)
 	}
-	if e.pusherMap[pusherKey] == nil {
-		pusher := cwlogs.NewPusher(pusherKey, e.retryCount, *e.svcStructuredLog, e.logger)
-		e.pusherMap[pusherKey] = pusher
+
+	err = pusher.ForceFlush()
+
+	if err != nil {
+		return fmt.Errorf("Error flushing logs:  %w", err)
 	}
-	return e.pusherMap[pusherKey]
+
+	return nil
 }
 
 func (e *cwlExporter) shutdown(_ context.Context) error {
-	if e.pusherMap != nil {
-		for _, pusher := range e.pusherMap {
-			pusher.ForceFlush()
-		}
-	}
 	return nil
 }
 
-func logsToCWLogs(logger *zap.Logger, ld plog.Logs, config *Config) ([]*cwlogs.Event, int) {
+func logsToCWLogs(logger *zap.Logger, ld plog.Logs, config *Config, pusher cwlogs.Pusher) (int, error) {
 	n := ld.ResourceLogs().Len()
 	if n == 0 {
-		return []*cwlogs.Event{}, 0
+		return 0, nil
 	}
 
 	var dropped int
-	var out []*cwlogs.Event
 
 	rls := ld.ResourceLogs()
 	for i := 0; i < rls.Len(); i++ {
@@ -194,12 +145,16 @@ func logsToCWLogs(logger *zap.Logger, ld plog.Logs, config *Config) ([]*cwlogs.E
 					logger.Debug("Failed to convert to CloudWatch Log", zap.Error(err))
 					dropped++
 				} else {
-					out = append(out, event)
+					logger.Debug("Adding log event", zap.Any("event", event))
+					err := pusher.AddLogEntry(event)
+					if err != nil {
+						return dropped, err
+					}
 				}
 			}
 		}
 	}
-	return out, dropped
+	return dropped, nil
 }
 
 type cwLogBody struct {
@@ -268,8 +223,10 @@ func logToCWLog(resourceAttrs map[string]interface{}, log plog.LogRecord, config
 			Timestamp: aws.Int64(int64(log.Timestamp()) / int64(time.Millisecond)), // in milliseconds
 			Message:   aws.String(string(bodyJSON)),
 		},
-		LogGroupName:  logGroupName,
-		LogStreamName: logStreamName,
+		StreamKey: cwlogs.StreamKey{
+			LogGroupName:  logGroupName,
+			LogStreamName: logStreamName,
+		},
 		GeneratedTime: time.Now(),
 	}, nil
 }