[exporter/awscloudwatchlogsexporter] Improve performance of the awscloudwatchlogs exporter

exporter/awscloudwatchlogsexporter/exporter.go

@@ -92,7 +92,7 @@
 		params,
 		config,
 		logsPusher.consumeLogs,
 		exporterhelper.WithQueue(expConfig.QueueSettings),
 		exporterhelper.WithRetry(expConfig.RetrySettings),
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		exporterhelper.WithShutdown(logsPusher.shutdown),
@@ -101,25 +101,21 @@
 
 func (e *cwlExporter) consumeLogs(_ context.Context, ld plog.Logs) error {
 	pusher := e.pusherFactory.CreateMultiStreamPusher()
-	var errs []error
+	var errs error
 
 	err := pushLogsToCWLogs(e.logger, ld, e.Config, pusher)
 
 	if err != nil {
-		errs = append(errs, fmt.Errorf("Error pushing logs: %w", err))
+		errs = errors.Join(errs, fmt.Errorf("Error pushing logs: %w", err))
 	}
 
 	err = pusher.ForceFlush()
 
 	if err != nil {
-		errs = append(errs, fmt.Errorf("Error flushing logs: %w", err))
+		errs = errors.Join(errs, fmt.Errorf("Error flushing logs: %w", err))
 	}
 
-	if len(errs) != 0 {
-		return errors.Join(errs...)
-	}
-
-	return nil
+	return errs
 }
 
 func (e *cwlExporter) shutdown(_ context.Context) error {
@@ -130,10 +126,10 @@
 	n := ld.ResourceLogs().Len()
 
 	if n == 0 {
 		return nil
 	}
 
-	var errs []error
+	var errs error
 
 	rls := ld.ResourceLogs()
 	for i := 0; i < rls.Len(); i++ {
@@ -152,18 +148,14 @@
 				} else {
 					err := pusher.AddLogEntry(event)
 					if err != nil {
-						errs = append(errs, err)
+						errs = errors.Join(errs, err)
 					}
 				}
 			}
 		}
 	}
 
-	if len(errs) != 0 {
-		return errors.Join(errs...)
-	}
-
-	return nil
+	return errs
 }
 
 type cwLogBody struct {

internal/aws/cwlogs/pusher.go

@@ -328,16 +328,15 @@
 
 func (m *multiStreamPusher) AddLogEntry(event *Event) error {
 	if err := m.logStreamManager.InitStream(event.StreamKey); err != nil {
 		return err
 	}
 
 	var pusher Pusher
+	var ok bool
 
-	if _, ok := m.pusherMap[event.StreamKey]; !ok {
+	if pusher, ok = m.pusherMap[event.StreamKey]; !ok {
 		pusher = NewPusher(event.StreamKey, 1, m.client, m.logger)
 		m.pusherMap[event.StreamKey] = pusher
-	} else {
-		pusher = m.pusherMap[event.StreamKey]
 	}
 
 	return pusher.AddLogEntry(event)
@@ -349,13 +348,13 @@
 	for _, val := range m.pusherMap {
 		err := val.ForceFlush()
 		if err != nil {
 			errs = append(errs, err)
 		}
 	}
 
 	if len(errs) != 0 {
 		return errors.Join(errs...)
 	}
 
 	return nil
 }