Push EMF logs in batch (#2572)

Author: bjrara

exporter/awsemfexporter/emf_exporter.go

@@ -89,7 +89,6 @@ func NewEmfExporter(
 	config configmodels.Exporter,
 	params component.ExporterCreateParams,
 ) (component.MetricsExporter, error) {
-
 	exp, err := New(config, params)
 	if err != nil {
 		return nil, err
@@ -105,12 +104,23 @@ func NewEmfExporter(
 }
 
 func (emf *emfExporter) pushMetricsData(_ context.Context, md pdata.Metrics) (droppedTimeSeries int, err error) {
+	rms := md.ResourceMetrics()
+	labels := map[string]string{}
+	for i := 0; i < rms.Len(); i++ {
+		rm := rms.At(i)
+		am := rm.Resource().Attributes()
+		if am.Len() > 0 {
+			am.ForEach(func(k string, v pdata.AttributeValue) {
+				labels[k] = v.StringVal()
+			})
+		}
+	}
+	emf.logger.Info("Start processing resource metrics", zap.Any("labels", labels))
+
 	groupedMetrics := make(map[interface{}]*GroupedMetric)
 	expConfig := emf.config.(*Config)
 	defaultLogStream := fmt.Sprintf("otel-stream-%s", emf.collectorID)
 
-	rms := md.ResourceMetrics()
-
 	for i := 0; i < rms.Len(); i++ {
 		rm := rms.At(i)
 		emf.metricTranslator.translateOTelToGroupedMetric(&rm, groupedMetrics, expConfig)
@@ -133,14 +143,23 @@ func (emf *emfExporter) pushMetricsData(_ context.Context, md pdata.Metrics) (dr
 				err = wrapErrorIfBadRequest(&returnError)
 				return
 			}
-			returnError = pusher.ForceFlush()
-			if returnError != nil {
-				err = wrapErrorIfBadRequest(&returnError)
-				return
+		}
+	}
+
+	for _, pusher := range emf.listPushers() {
+		returnError := pusher.ForceFlush()
+		if returnError != nil {
+			//TODO now we only have one pusher, so it's ok to return after first error occurred
+			err = wrapErrorIfBadRequest(&returnError)
+			if err != nil {
+				emf.logger.Error("Error force flushing logs. Skipping to next pusher.", zap.Error(err))
 			}
+			return
 		}
 	}
 
+	emf.logger.Info("Finish processing resource metrics", zap.Any("labels", labels))
+
 	return
 }
 
@@ -163,6 +182,19 @@ func (emf *emfExporter) getPusher(logGroup, logStream string) Pusher {
 	return pusher
 }
 
+func (emf *emfExporter) listPushers() []Pusher {
+	emf.pusherMapLock.Lock()
+	defer emf.pusherMapLock.Unlock()
+
+	pushers := []Pusher{}
+	for _, pusherMap := range emf.groupStreamToPusherMap {
+		for _, pusher := range pusherMap {
+			pushers = append(pushers, pusher)
+		}
+	}
+	return pushers
+}
+
 func (emf *emfExporter) ConsumeMetrics(ctx context.Context, md pdata.Metrics) error {
 	exporterCtx := obsreport.ExporterContext(ctx, "emf.exporterFullName")
 
@@ -172,20 +204,12 @@ func (emf *emfExporter) ConsumeMetrics(ctx context.Context, md pdata.Metrics) er
 
 // Shutdown stops the exporter and is invoked during shutdown.
 func (emf *emfExporter) Shutdown(ctx context.Context) error {
-	emf.pusherMapLock.Lock()
-	defer emf.pusherMapLock.Unlock()
-
-	var err error
-	for _, streamToPusherMap := range emf.groupStreamToPusherMap {
-		for _, pusher := range streamToPusherMap {
-			if pusher != nil {
-				returnError := pusher.ForceFlush()
-				if returnError != nil {
-					err = wrapErrorIfBadRequest(&returnError)
-				}
-				if err != nil {
-					emf.logger.Error("Error when gracefully shutting down emf_exporter. Skipping to next pusher.", zap.Error(err))
-				}
+	for _, pusher := range emf.listPushers() {
+		returnError := pusher.ForceFlush()
+		if returnError != nil {
+			err := wrapErrorIfBadRequest(&returnError)
+			if err != nil {
+				emf.logger.Error("Error when gracefully shutting down emf_exporter. Skipping to next pusher.", zap.Error(err))
 			}
 		}
 	}