consistency changes

Author: BominRahmani

.chloggen/windows-events-err-1734.yaml

@@ -7,7 +7,7 @@ change_type: bug_fix
 component: pkg/stanza
 
 # A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
-note: Adds dynamic batching whenever err no 1734 (invalid bounds) has occured.
+note: Add retries when calls to retrieve Windows event via `EvtNext` fail with error RPC_S_INVALID_BOUND (1734).
 
 # Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
 issues: [38149]
@@ -16,8 +16,8 @@ issues: [38149]
 # These lines will be padded with 2 spaces and then inserted directly into the document.
 # Use pipe (|) for multiline entries.
 subtext: |
-  Whever large events were read in by the windows event log receiver / stanza input operator for windows
-  There would be an Invalid array bounds error that popped up (1734). This can be solved by reducing the batch size / maxreads.
+  Whenever large events were read in by the windows event log receiver / stanza input operator for windows,
+  The collector would fail with error RPC_S_INVALID_BOUND (1734). This can be solved by reducing max reads and retrying.
 # 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.

pkg/stanza/operator/input/windows/input.go

@@ -41,7 +41,6 @@ type Input struct {
 	remoteSessionHandle windows.Handle
 	startRemoteSession  func() error
 	processEvent        func(context.Context, Event) error
-	logger              *zap.Logger
 }
 
 // newInput creates a new Input operator.
@@ -202,7 +201,7 @@ func (i *Input) read(ctx context.Context) {
 	events, err := i.subscription.Read(i.currentMaxReads)
 	if errors.Is(err, ErrBatchSizeReduced) {
 		i.currentMaxReads = max(i.currentMaxReads/2, 1)
-		i.Logger().Warn("Encountered RPC_S_INVALID_BOUND, reducing batch size", zap.Int("current batch size", i.currentMaxReads), zap.Int("original batch size", i.maxReads))
+		i.Logger().Debug("Encountered RPC_S_INVALID_BOUND, reducing batch size", zap.Int("current batch size", i.currentMaxReads), zap.Int("original batch size", i.maxReads))
 	} else if err != nil {
 		i.Logger().Error("Failed to read events from subscription", zap.Error(err))
 		if i.isRemote() && (errors.Is(err, windows.ERROR_INVALID_HANDLE) || errors.Is(err, errSubscriptionHandleNotOpen)) {
@@ -231,12 +230,12 @@ func (i *Input) read(ctx context.Context) {
 
 	for n, event := range events {
 		if err := i.processEvent(ctx, event); err != nil {
-			i.logger.Error("process event", zap.Error(err))
+			i.Logger().Error("process event", zap.Error(err))
 		}
 		if len(events) == n+1 {
 			i.updateBookmarkOffset(ctx, event)
 			if err := i.subscription.bookmark.Update(event); err != nil {
-				i.logger.Error("Failed to update bookmark from event", zap.Error(err))
+				i.Logger().Error("Failed to update bookmark from event", zap.Error(err))
 			}
 		}
 		event.Close()

pkg/stanza/operator/input/windows/input_test.go

@@ -175,7 +175,7 @@ func TestInputRead_RPCInvalidBound(t *testing.T) {
 	}()
 
 	// Create a logger with an observer for testing log output
-	core, logs := observer.New(zap.WarnLevel)
+	core, logs := observer.New(zap.DebugLevel)
 	logger := zap.New(core)
 
 	// Create input instance with mocked dependencies
@@ -186,7 +186,6 @@ func TestInputRead_RPCInvalidBound(t *testing.T) {
 	// Set up test values
 	input.maxReads = 100
 	input.currentMaxReads = 100
-	input.logger = logger
 
 	// Set up subscription with valid handle and enough info to reopen
 	input.subscription = Subscription{

pkg/stanza/operator/input/windows/subscription.go

@@ -112,7 +112,7 @@ func (s *Subscription) readWithRetry(maxReads int) ([]Event, error) {
 		return nil, nil
 	}
 
-	if err != nil && errors.Is(err, windows.RPC_S_INVALID_BOUND) {
+	if errors.Is(err, windows.RPC_S_INVALID_BOUND) {
 		// close current subscription
 		if closeErr := s.Close(); closeErr != nil {
 			return nil, fmt.Errorf("failed to close subscription during recovery: %w", closeErr)