[receiver/awscloudwatch]: fix issue with polling of deleted log group while polling (#40571)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Resolves #38940

If a log group was deleted while a poll was being executed we were
silently failing after the first log message. While examining the code I
don't believe we were failing on future polls but can imagine that this
change is an improvement as it now radiates a tad bit more on the
`pollForLogs` function as it now actually returns up errors.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes #38940


<!--Describe what testing was performed and which tests were added.-->
#### Testing

Unit testing the situation described in the issue

#### Documentation

changelog entry

Author: schmikei

.chloggen/fix_log-group-deletion.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: awscloudwatchreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fixes issue with autodiscovered groups that were deleted preventing logs during that poll
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38940]
+
+# (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/awscloudwatchreceiver/logs.go

@@ -13,6 +13,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -268,7 +269,6 @@ func (l *logsReceiver) pollForLogs(ctx context.Context, pc groupRequest, startTi
 
 	for nextToken != nil {
 		select {
-		// if done, we want to stop processing paginated stream of events
 		case _, ok := <-l.doneChan:
 			if !ok {
 				return nil
@@ -277,17 +277,24 @@ func (l *logsReceiver) pollForLogs(ctx context.Context, pc groupRequest, startTi
 			input := pc.request(l.maxEventsPerRequest, *nextToken, &startTime, &endTime)
 			resp, err := l.client.FilterLogEvents(ctx, input)
 			if err != nil {
-				l.settings.Logger.Error("unable to retrieve logs from cloudatch",
-					zap.String("logGroup", logGroup),
-					zap.Error(err))
-				break
+				var resourceNotFoundException *types.ResourceNotFoundException
+				if errors.As(err, &resourceNotFoundException) {
+					l.settings.Logger.Warn("log group no longer exists, skipping",
+						zap.String("logGroup", logGroup),
+						zap.Error(err))
+					return fmt.Errorf("log group %s no longer exists: %w", logGroup, err)
+				}
+				return fmt.Errorf("failed to retrieve logs from log group %s: %w", logGroup, err)
 			}
+
 			observedTime := pcommon.NewTimestampFromTime(time.Now())
 			logs := l.processEvents(observedTime, logGroup, resp)
 			if logs.LogRecordCount() > 0 {
 				if err = l.consumer.ConsumeLogs(ctx, logs); err != nil {
-					l.settings.Logger.Error("unable to consume logs", zap.Error(err))
-					break
+					l.settings.Logger.Error("unable to consume logs",
+						zap.String("logGroup", logGroup),
+						zap.Error(err))
+					return fmt.Errorf("failed to consume logs from log group %s: %w", logGroup, err)
 				}
 			}
 			nextToken = resp.NextToken
@@ -340,7 +347,6 @@ func (l *logsReceiver) processEvents(now pcommon.Timestamp, logGroupName string,
 			}
 			group[logStreamName] = resourceLogs
 
-			// Ensure one scopeLogs is initialized so we can handle in standardized way going forward.
 			_ = resourceLogs.ScopeLogs().AppendEmpty()
 		}
 

receiver/awscloudwatchreceiver/logs_test.go

@@ -274,6 +274,223 @@ func TestAutodiscoverLimit(t *testing.T) {
 	require.Len(t, grs, cfg.Logs.Groups.AutodiscoverConfig.Limit)
 }
 
+func TestDeletedLogGroupContinuesPolling(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	cfg.Region = "us-west-1"
+	cfg.Logs.PollInterval = 1 * time.Second
+	cfg.Logs.Groups = GroupConfig{
+		NamedConfigs: map[string]StreamConfig{
+			"existing-group": {
+				Names: []*string{aws.String("stream1")},
+			},
+			"deleted-group": {
+				Names: []*string{aws.String("stream2")},
+			},
+		},
+	}
+
+	sink := &consumertest.LogsSink{}
+	logsRcvr := newLogsReceiver(cfg, receiver.Settings{
+		TelemetrySettings: component.TelemetrySettings{
+			Logger: zap.NewNop(),
+		},
+	}, sink)
+	mc := &mockClient{}
+
+	mc.On("FilterLogEvents", mock.Anything, mock.MatchedBy(func(input *cloudwatchlogs.FilterLogEventsInput) bool {
+		return *input.LogGroupName == "existing-group"
+	}), mock.Anything).Return(&cloudwatchlogs.FilterLogEventsOutput{
+		Events: []types.FilteredLogEvent{
+			{
+				EventId:       aws.String("event1"),
+				LogStreamName: aws.String("stream1"),
+				Message:       aws.String("test message"),
+				Timestamp:     aws.Int64(time.Now().UnixMilli()),
+			},
+		},
+		NextToken: nil,
+	}, nil)
+
+	mc.On("FilterLogEvents", mock.Anything, mock.MatchedBy(func(input *cloudwatchlogs.FilterLogEventsInput) bool {
+		return *input.LogGroupName == "deleted-group"
+	}), mock.Anything).Return((*cloudwatchlogs.FilterLogEventsOutput)(nil), &types.ResourceNotFoundException{
+		Message: aws.String("The specified log group does not exist"),
+	})
+
+	logsRcvr.client = mc
+
+	err := logsRcvr.Start(context.Background(), componenttest.NewNopHost())
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return sink.LogRecordCount() > 0
+	}, 2*time.Second, 10*time.Millisecond)
+	logs := sink.AllLogs()
+	require.Len(t, logs, 1)
+	require.Equal(t, 1, logs[0].LogRecordCount())
+
+	logRecord := logs[0].ResourceLogs().At(0)
+	require.Equal(t, "existing-group", logRecord.Resource().Attributes().AsRaw()["cloudwatch.log.group.name"])
+
+	err = logsRcvr.Shutdown(context.Background())
+	require.NoError(t, err)
+
+	// Verify all mock expectations were met
+	mc.AssertExpectations(t)
+}
+
+func TestDeletedLogGroupDuringAutodiscover(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	cfg.Region = "us-west-1"
+	cfg.Logs.PollInterval = 1 * time.Second
+	cfg.Logs.Groups = GroupConfig{
+		AutodiscoverConfig: &AutodiscoverConfig{
+			Limit:  3,
+			Prefix: "/aws/",
+		},
+	}
+
+	firstPollDone := make(chan struct{}, 1)
+
+	sink := &consumertest.LogsSink{}
+	logsRcvr := newLogsReceiver(cfg, receiver.Settings{
+		TelemetrySettings: component.TelemetrySettings{
+			Logger: zap.NewNop(),
+		},
+	}, sink)
+
+	mc := &mockClient{}
+
+	// First DescribeLogGroups call returns two groups
+	mc.On("DescribeLogGroups", mock.Anything, mock.MatchedBy(func(input *cloudwatchlogs.DescribeLogGroupsInput) bool {
+		return input.LogGroupNamePrefix != nil && *input.LogGroupNamePrefix == "/aws/"
+	}), mock.Anything).Return(&cloudwatchlogs.DescribeLogGroupsOutput{
+		LogGroups: []types.LogGroup{
+			{
+				LogGroupName: aws.String("/aws/working-group"),
+			},
+			{
+				LogGroupName: aws.String("/aws/delete-group"),
+			},
+			{
+				LogGroupName: aws.String("/aws/third-group"),
+			},
+		},
+		NextToken: nil,
+	}, nil).Once()
+
+	// Second DescribeLogGroups call returns working group plus a new group
+	mc.On("DescribeLogGroups", mock.Anything, mock.MatchedBy(func(input *cloudwatchlogs.DescribeLogGroupsInput) bool {
+		return input.LogGroupNamePrefix != nil && *input.LogGroupNamePrefix == "/aws/"
+	}), mock.Anything).Return(&cloudwatchlogs.DescribeLogGroupsOutput{
+		LogGroups: []types.LogGroup{
+			{
+				LogGroupName: aws.String("/aws/working-group"),
+			},
+			{
+				LogGroupName: aws.String("/aws/third-group"),
+			},
+		},
+		NextToken: nil,
+	}, nil).Run(func(_ mock.Arguments) {
+		close(firstPollDone)
+	})
+
+	// Setup mock for working-group to return normal logs
+	mc.On("FilterLogEvents", mock.Anything, mock.MatchedBy(func(input *cloudwatchlogs.FilterLogEventsInput) bool {
+		return input.LogGroupName != nil && *input.LogGroupName == "/aws/working-group"
+	}), mock.Anything).Return(&cloudwatchlogs.FilterLogEventsOutput{
+		Events: []types.FilteredLogEvent{
+			{
+				EventId:       aws.String("event1"),
+				LogStreamName: aws.String("stream1"),
+				Message:       aws.String("test message"),
+				Timestamp:     aws.Int64(time.Now().UnixMilli()),
+			},
+		},
+		NextToken: nil,
+	}, nil)
+
+	// Setup mock for /aws/delete-group to return ResourceNotFoundException
+	mc.On("FilterLogEvents", mock.Anything, mock.MatchedBy(func(input *cloudwatchlogs.FilterLogEventsInput) bool {
+		return input.LogGroupName != nil && *input.LogGroupName == "/aws/delete-group"
+	}), mock.Anything).Return(&cloudwatchlogs.FilterLogEventsOutput{
+		Events: []types.FilteredLogEvent{
+			{
+				EventId:       aws.String("event1"),
+				LogStreamName: aws.String("stream1"),
+				Message:       aws.String("test message"),
+				Timestamp:     aws.Int64(time.Now().UnixMilli()),
+			},
+		},
+		NextToken: aws.String("next"),
+	}, nil).Once()
+
+	mc.On("FilterLogEvents", mock.Anything, mock.MatchedBy(func(input *cloudwatchlogs.FilterLogEventsInput) bool {
+		return input.LogGroupName != nil && *input.LogGroupName == "/aws/delete-group" && input.NextToken != nil && *input.NextToken == "next"
+	}), mock.Anything).Return(&cloudwatchlogs.FilterLogEventsOutput{}, &types.ResourceNotFoundException{
+		Message: aws.String("The specified log group does not exist"),
+	}).Once()
+
+	// Setup mock for /aws/working-group or /aws/third-group to return normal logs
+	mc.On("FilterLogEvents",
+		mock.Anything,
+		mock.MatchedBy(func(input *cloudwatchlogs.FilterLogEventsInput) bool {
+			return input.LogGroupName != nil && (*input.LogGroupName == "/aws/working-group" || *input.LogGroupName == "/aws/third-group")
+		}), mock.Anything).Return(&cloudwatchlogs.FilterLogEventsOutput{
+		Events: []types.FilteredLogEvent{
+			{
+				EventId:       aws.String("event2"),
+				LogStreamName: aws.String("stream2"),
+				Message:       aws.String("test message from group"),
+				Timestamp:     aws.Int64(time.Now().UnixMilli()),
+			},
+		},
+		NextToken: nil,
+	}, nil)
+
+	logsRcvr.client = mc
+
+	err := logsRcvr.Start(context.Background(), componenttest.NewNopHost())
+	require.NoError(t, err)
+
+	// Wait for first poll to complete
+	require.Eventually(t, func() bool {
+		select {
+		case <-firstPollDone:
+			return true
+		default:
+			return false
+		}
+	}, 15*time.Second, 10*time.Millisecond)
+
+	err = logsRcvr.Shutdown(context.Background())
+	require.NoError(t, err)
+
+	groupNames := make([]string, 0, len(logsRcvr.groupRequests))
+	for _, gr := range logsRcvr.groupRequests {
+		groupNames = append(groupNames, gr.groupName())
+	}
+	// validate deleted group is not in the groupRequests
+	require.ElementsMatch(t, []string{"/aws/working-group", "/aws/third-group"}, groupNames)
+	logs := sink.AllLogs()
+	require.GreaterOrEqual(t, len(logs), 3)
+
+	firstWorkingGroupLog := logs[0].ResourceLogs().At(0)
+	require.Equal(t, "/aws/working-group", firstWorkingGroupLog.Resource().Attributes().AsRaw()["cloudwatch.log.group.name"])
+	require.Equal(t, 1, firstWorkingGroupLog.ScopeLogs().Len())
+
+	secondWorkingGroupLog := logs[1].ResourceLogs().At(0)
+	require.Equal(t, "/aws/delete-group", secondWorkingGroupLog.Resource().Attributes().AsRaw()["cloudwatch.log.group.name"])
+	require.Equal(t, 1, secondWorkingGroupLog.ScopeLogs().Len())
+
+	thirdWorkingGroupLog := logs[2].ResourceLogs().At(0)
+	require.Equal(t, "/aws/third-group", thirdWorkingGroupLog.Resource().Attributes().AsRaw()["cloudwatch.log.group.name"])
+	require.Equal(t, 1, thirdWorkingGroupLog.ScopeLogs().Len())
+
+	mc.AssertExpectations(t)
+}
+
 func defaultMockClient() client {
 	mc := &mockClient{}
 	mc.On("DescribeLogGroups", mock.Anything, mock.Anything, mock.Anything).Return(