fix(awscloudwatchreceiver): emit logs from one stream in one resource (…

…#22976)

The receiver treated every event as a separate log. However, it makes a
lot sense to group logs into common resource by their group and stream.
Logs without a stream are grouped into the same resource.
I'm not sure how should `scope` be treated here, but my intuition tells
me that it makes sense to group them also into the same scope.

.chloggen/awscloudwatch-emit-stream-in-one-resource.yaml

@@ -0,0 +1,19 @@
+# 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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awscloudwatchreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: emit logs from one log stream in the same resource
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [22145]
+
+# (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.

receiver/awscloudwatchreceiver/logs.go

@@ -21,6 +21,10 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	noStreamName = "THIS IS INVALID STREAM"
+)
+
 type logsReceiver struct {
 	region              string
 	profile             string
@@ -222,6 +226,9 @@ func (l *logsReceiver) pollForLogs(ctx context.Context, pc groupRequest, startTi
 
 func (l *logsReceiver) processEvents(now pcommon.Timestamp, logGroupName string, output *cloudwatchlogs.FilterLogEventsOutput) plog.Logs {
 	logs := plog.NewLogs()
+
+	resourceMap := map[string](map[string]*plog.ResourceLogs){}
+
 	for _, e := range output.Events {
 		if e.Timestamp == nil {
 			l.logger.Error("unable to determine timestamp of event as the timestamp is nil")
@@ -238,15 +245,35 @@ func (l *logsReceiver) processEvents(now pcommon.Timestamp, logGroupName string,
 			continue
 		}
 
-		rl := logs.ResourceLogs().AppendEmpty()
-		resourceAttributes := rl.Resource().Attributes()
-		resourceAttributes.PutStr("aws.region", l.region)
-		resourceAttributes.PutStr("cloudwatch.log.group.name", logGroupName)
+		group, ok := resourceMap[logGroupName]
+		if !ok {
+			group = map[string]*plog.ResourceLogs{}
+			resourceMap[logGroupName] = group
+		}
+
+		logStreamName := noStreamName
 		if e.LogStreamName != nil {
-			resourceAttributes.PutStr("cloudwatch.log.stream", *e.LogStreamName)
+			logStreamName = *e.LogStreamName
+		}
+
+		resourceLogs, ok := group[logStreamName]
+		if !ok {
+			rl := logs.ResourceLogs().AppendEmpty()
+			resourceLogs = &rl
+			resourceAttributes := resourceLogs.Resource().Attributes()
+			resourceAttributes.PutStr("aws.region", l.region)
+			resourceAttributes.PutStr("cloudwatch.log.group.name", logGroupName)
+			resourceAttributes.PutStr("cloudwatch.log.stream", logStreamName)
+			group[logStreamName] = resourceLogs
+
+			// Ensure one scopeLogs is initialized so we can handle in standardized way going forward.
+			_ = resourceLogs.ScopeLogs().AppendEmpty()
 		}
 
-		logRecord := rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+		// Now we know resourceLogs is initialized and has one scopeLogs so we don't have to handle any special cases.
+
+		logRecord := resourceLogs.ScopeLogs().At(0).LogRecords().AppendEmpty()
+
 		logRecord.SetObservedTimestamp(now)
 		ts := time.UnixMilli(*e.Timestamp)
 		logRecord.SetTimestamp(pcommon.NewTimestampFromTime(ts))

receiver/awscloudwatchreceiver/logs_test.go

@@ -225,23 +225,50 @@ func defaultMockClient() client {
 		&cloudwatchlogs.FilterLogEventsOutput{
 			Events: []*cloudwatchlogs.FilteredLogEvent{
 				{
-					EventId:       &testEventID,
+					EventId:       &testEventIDs[0],
 					IngestionTime: aws.Int64(testIngestionTime),
 					LogStreamName: aws.String(testLogStreamName),
 					Message:       aws.String(testLogStreamMessage),
 					Timestamp:     aws.Int64(testTimeStamp),
 				},
+				{
+					EventId:       &testEventIDs[1],
+					IngestionTime: aws.Int64(testIngestionTime),
+					LogStreamName: aws.String(testLogStreamName),
+					Message:       aws.String(testLogStreamMessage),
+					Timestamp:     aws.Int64(testTimeStamp),
+				},
+				{
+					EventId:       &testEventIDs[2],
+					IngestionTime: aws.Int64(testIngestionTime),
+					LogStreamName: aws.String(testLogStreamName2),
+					Message:       aws.String(testLogStreamMessage),
+					Timestamp:     aws.Int64(testTimeStamp),
+				},
+				{
+					EventId:       &testEventIDs[3],
+					IngestionTime: aws.Int64(testIngestionTime),
+					LogStreamName: aws.String(testLogStreamName2),
+					Message:       aws.String(testLogStreamMessage),
+					Timestamp:     aws.Int64(testTimeStamp),
+				},
 			},
 			NextToken: nil,
 		}, nil)
 	return mc
 }
 
 var (
-	testLogGroupName     = "test-log-group-name"
-	testLogStreamName    = "test-log-stream-name"
-	testLogStreamPrefix  = "test-log-stream"
-	testEventID          = "37134448277055698880077365577645869800162629528367333379"
+	testLogGroupName    = "test-log-group-name"
+	testLogStreamName   = "test-log-stream-name"
+	testLogStreamName2  = "test-log-stream-name-2"
+	testLogStreamPrefix = "test-log-stream"
+	testEventIDs        = []string{
+		"37134448277055698880077365577645869800162629528367333379",
+		"37134448277055698880077365577645869800162629528367333380",
+		"37134448277055698880077365577645869800162629528367333381",
+		"37134448277055698880077365577645869800162629528367333382",
+	}
 	testIngestionTime    = int64(1665166252124)
 	testTimeStamp        = int64(1665166251014)
 	testLogStreamMessage = `"time=\"2022-10-07T18:10:46Z\" level=info msg=\"access granted\" arn=\"arn:aws:iam::892146088969:role/AWSWesleyClusterManagerLambda-NodeManagerRole-16UPVDKA1KBGI\" client=\"127.0.0.1:50252\" groups=\"[]\" method=POST path=/authenticate uid=\"aws-iam-authenticator:892146088969:AROA47OAM7QE2NWPDFDCW\" username=\"eks:node-manager\""`

receiver/awscloudwatchreceiver/testdata/processed/prefixed.yaml

@@ -22,4 +22,48 @@ resourceLogs:
             spanId: ""
             timeUnixNano: "1665166251014000000"
             traceId: ""
+          - attributes:
+              - key: id
+                value:
+                  stringValue: "37134448277055698880077365577645869800162629528367333380"
+            body:
+              stringValue: '"time=\"2022-10-07T18:10:46Z\" level=info msg=\"access granted\" arn=\"arn:aws:iam::892146088969:role/AWSWesleyClusterManagerLambda-NodeManagerRole-16UPVDKA1KBGI\" client=\"127.0.0.1:50252\" groups=\"[]\" method=POST path=/authenticate uid=\"aws-iam-authenticator:892146088969:AROA47OAM7QE2NWPDFDCW\" username=\"eks:node-manager\""'
+            observedTimeUnixNano: "1665166998995360000"
+            spanId: ""
+            timeUnixNano: "1665166251014000000"
+            traceId: ""
+        scope: {}
+  - resource:
+      attributes:
+        - key: aws.region
+          value:
+            stringValue: us-west-1
+        - key: cloudwatch.log.group.name
+          value:
+            stringValue: test-log-group-name
+        - key: cloudwatch.log.stream
+          value:
+            stringValue: test-log-stream-name-2
+    scopeLogs:
+      - logRecords:
+          - attributes:
+              - key: id
+                value:
+                  stringValue: "37134448277055698880077365577645869800162629528367333381"
+            body:
+              stringValue: '"time=\"2022-10-07T18:10:46Z\" level=info msg=\"access granted\" arn=\"arn:aws:iam::892146088969:role/AWSWesleyClusterManagerLambda-NodeManagerRole-16UPVDKA1KBGI\" client=\"127.0.0.1:50252\" groups=\"[]\" method=POST path=/authenticate uid=\"aws-iam-authenticator:892146088969:AROA47OAM7QE2NWPDFDCW\" username=\"eks:node-manager\""'
+            observedTimeUnixNano: "1665166998995360000"
+            spanId: ""
+            timeUnixNano: "1665166251014000000"
+            traceId: ""
+          - attributes:
+              - key: id
+                value:
+                  stringValue: "37134448277055698880077365577645869800162629528367333382"
+            body:
+              stringValue: '"time=\"2022-10-07T18:10:46Z\" level=info msg=\"access granted\" arn=\"arn:aws:iam::892146088969:role/AWSWesleyClusterManagerLambda-NodeManagerRole-16UPVDKA1KBGI\" client=\"127.0.0.1:50252\" groups=\"[]\" method=POST path=/authenticate uid=\"aws-iam-authenticator:892146088969:AROA47OAM7QE2NWPDFDCW\" username=\"eks:node-manager\""'
+            observedTimeUnixNano: "1665166998995360000"
+            spanId: ""
+            timeUnixNano: "1665166251014000000"
+            traceId: ""
         scope: {}