awsfirehosereceiver: ignore CONTROL_MESSAGE log records

.chloggen/firehose-control-messages.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awsfirehosereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix cwlogs encoding to not consider CONTROL_MESSAGE records invalid
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38433]
+
+# (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]

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/testdata/only_control

@@ -0,0 +1 @@
+{"messageType":"CONTROL_MESSAGE","owner":"CloudwatchLogs","logGroup":"","logStream":"","subscriptionFilters":[],"logEvents":[{"id":"","timestamp":1741312971934,"message":"CWL CONTROL MESSAGE: Checking health of destination Firehose."}]}

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/unmarshaler.go

@@ -29,7 +29,12 @@ const (
 	attributeAWSCloudWatchLogStreamName = "aws.cloudwatch.log_stream_name"
 )
 
-var errInvalidRecords = errors.New("record format invalid")
+var (
+	errInvalidRecords   = errors.New("record format invalid")
+	errMissingOwner     = errors.New("cloudwatch log record is missing owner field")
+	errMissingLogGroup  = errors.New("cloudwatch log record is missing logGroup field")
+	errMissingLogStream = errors.New("cloudwatch log record is missing logStream field")
+)
 
 // Unmarshaler for the CloudWatch Log JSON record format.
 type Unmarshaler struct {
@@ -70,22 +75,28 @@ func (u *Unmarshaler) UnmarshalLogs(compressedRecord []byte) (plog.Logs, error)
 	byResource := make(map[resourceKey]plog.LogRecordSlice)
 
 	// Multiple logs in each record separated by newline character
+	var anyErrors bool
 	scanner := bufio.NewScanner(r)
 	for datumIndex := 0; scanner.Scan(); datumIndex++ {
-		var log cWLog
-		if err := jsoniter.ConfigFastest.Unmarshal(scanner.Bytes(), &log); err != nil {
+		log, control, err := parseLog(scanner.Bytes())
+		if err != nil {
+			anyErrors = true
 			u.logger.Error(
 				"Unable to unmarshal input",
 				zap.Error(err),
 				zap.Int("datum_index", datumIndex),
 			)
 			continue
 		}
-		if !isValid(log) {
-			u.logger.Error(
-				"Invalid log",
-				zap.Int("datum_index", datumIndex),
-			)
+		if control {
+			for _, event := range log.LogEvents {
+				u.logger.Debug(
+					"Skipping CloudWatch control message event",
+					zap.Time("timestamp", time.UnixMilli(event.Timestamp)),
+					zap.String("message", event.Message),
+					zap.Int("datum_index", datumIndex),
+				)
+			}
 			continue
 		}
 
@@ -110,9 +121,14 @@ func (u *Unmarshaler) UnmarshalLogs(compressedRecord []byte) (plog.Logs, error)
 	}
 	if err := scanner.Err(); err != nil {
 		// Treat this as a non-fatal error, and handle the data below.
+		anyErrors = true
 		u.logger.Error("Error scanning for newline-delimited JSON", zap.Error(err))
 	}
-	if len(byResource) == 0 {
+	if anyErrors && len(byResource) == 0 {
+		// Note that there is a valid scenario where there are no errors
+		// but also byResource is empty: when there are only control messages
+		// in the record. In this case we want to return an initialized but
+		// empty plog.Logs below.
 		return plog.Logs{}, errInvalidRecords
 	}
 
@@ -137,9 +153,27 @@ func (u *Unmarshaler) UnmarshalLogs(compressedRecord []byte) (plog.Logs, error)
 	return logs, nil
 }
 
-// isValid validates that the cWLog has been unmarshalled correctly.
-func isValid(log cWLog) bool {
-	return log.Owner != "" && log.LogGroup != "" && log.LogStream != ""
+func parseLog(data []byte) (log cWLog, control bool, _ error) {
+	if err := jsoniter.ConfigFastest.Unmarshal(data, &log); err != nil {
+		return cWLog{}, false, fmt.Errorf("error unmarshalling JSON: %w", err)
+	}
+	switch log.MessageType {
+	case "DATA_MESSAGE":
+		if log.Owner == "" {
+			return cWLog{}, false, errMissingOwner
+		}
+		if log.LogGroup == "" {
+			return cWLog{}, false, errMissingLogGroup
+		}
+		if log.LogStream == "" {
+			return cWLog{}, false, errMissingLogStream
+		}
+		return log, false, nil
+	case "CONTROL_MESSAGE":
+		return log, true, nil
+	default:
+		return cWLog{}, false, fmt.Errorf("invalid message type %q", log.MessageType)
+	}
 }
 
 // Type of the serialized messages.

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/unmarshaler_test.go

@@ -67,6 +67,13 @@ func TestUnmarshal(t *testing.T) {
 			wantResourceLogGroups:  nil, // not checking log group names because logs are unordered
 			wantResourceLogStreams: nil, // not checking log stream names because logs are unordered
 		},
+		"WithOnlyControlMessages": {
+			filename:               "only_control",
+			wantResourceCount:      0,
+			wantLogCount:           0,
+			wantResourceLogGroups:  nil,
+			wantResourceLogStreams: nil,
+		},
 	}
 	for name, testCase := range testCases {
 		t.Run(name, func(t *testing.T) {