[pkg/stanza] [receiver/windowseventlog] Fix: Windows Event Max Read (ERRNO 1734) (open-telemetry#38149)

Author: BominRahmani

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

@@ -0,0 +1,29 @@
+# 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: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+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]
+
+# (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: |
+  Whenever large events were read in by the Windows event log receiver, via the stanza input operator,
+  the collector would fail with error RPC_S_INVALID_BOUND (1734). Now the operator tries to workaround
+  this issue by reducing the number of events read on each attempt.
+# 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]

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

@@ -46,6 +46,7 @@ func (c *Config) Build(set component.TelemetrySettings) (operator.Operator, erro
 		buffer:           NewBuffer(),
 		channel:          c.Channel,
 		maxReads:         c.MaxReads,
+		currentMaxReads:  c.MaxReads,
 		startAt:          c.StartAt,
 		pollInterval:     c.PollInterval,
 		raw:              c.Raw,

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

@@ -27,6 +27,7 @@ type Input struct {
 	buffer              *Buffer
 	channel             string
 	maxReads            int
+	currentMaxReads     int
 	startAt             string
 	raw                 bool
 	excludeProviders    map[string]struct{}
@@ -197,7 +198,14 @@ func (i *Input) readOnInterval(ctx context.Context) {
 
 // read will read events from the subscription.
 func (i *Input) read(ctx context.Context) {
-	events, err := i.subscription.Read(i.maxReads)
+	events, actualMaxReads, err := i.subscription.Read(i.currentMaxReads)
+
+	// Update the current max reads if it changed
+	if err == nil && actualMaxReads < i.currentMaxReads {
+		i.currentMaxReads = actualMaxReads
+		i.Logger().Debug("Encountered RPC_S_INVALID_BOUND, reduced batch size", zap.Int("current_batch_size", i.currentMaxReads), zap.Int("original_batch_size", i.maxReads))
+	}
+
 	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)) {
@@ -230,6 +238,9 @@ func (i *Input) read(ctx context.Context) {
 		}
 		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))
+			}
 		}
 		event.Close()
 	}

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

@@ -6,13 +6,16 @@
 package windows // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/windows"
 
 import (
+	"context"
 	"errors"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 	"golang.org/x/sys/windows"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
@@ -128,3 +131,83 @@ func TestInputStart_BadChannelName(t *testing.T) {
 	assert.ErrorContains(t, err, "failed to open subscription for remote server")
 	assert.ErrorContains(t, err, "The specified channel could not be found")
 }
+
+// TestInputRead_RPCInvalidBound tests that the Input handles RPC_S_INVALID_BOUND errors properly
+func TestInputRead_RPCInvalidBound(t *testing.T) {
+	// Save original procs and restore after test
+	originalNextProc := nextProc
+	originalCloseProc := closeProc
+	originalSubscribeProc := subscribeProc
+
+	// Track calls to our mocked functions
+	var nextCalls, closeCalls, subscribeCalls int
+
+	// Mock the procs
+	closeProc = MockProc{
+		call: func(_ ...uintptr) (uintptr, uintptr, error) {
+			closeCalls++
+			return 1, 0, nil
+		},
+	}
+
+	subscribeProc = MockProc{
+		call: func(_ ...uintptr) (uintptr, uintptr, error) {
+			subscribeCalls++
+			return 42, 0, nil
+		},
+	}
+
+	nextProc = MockProc{
+		call: func(_ ...uintptr) (uintptr, uintptr, error) {
+			nextCalls++
+			if nextCalls == 1 {
+				return 0, 0, windows.RPC_S_INVALID_BOUND
+			}
+
+			return 1, 0, nil
+		},
+	}
+
+	defer func() {
+		nextProc = originalNextProc
+		closeProc = originalCloseProc
+		subscribeProc = originalSubscribeProc
+	}()
+
+	// Create a logger with an observer for testing log output
+	core, logs := observer.New(zap.DebugLevel)
+	logger := zap.New(core)
+
+	// Create input instance with mocked dependencies
+	input := newInput(component.TelemetrySettings{
+		Logger: logger,
+	})
+
+	// Set up test values
+	input.maxReads = 100
+	input.currentMaxReads = 100
+
+	// Set up subscription with valid handle and enough info to reopen
+	input.subscription = Subscription{
+		handle:        42, // Dummy handle
+		startAt:       "beginning",
+		sessionHandle: 0,
+		channel:       "test-channel",
+	}
+
+	// Call the method under test
+	ctx := context.Background()
+	input.read(ctx)
+
+	// Verify the correct number of calls to each mock
+	assert.Equal(t, 2, nextCalls, "nextProc should be called twice (initial failure and retry)")
+	assert.Equal(t, 1, closeCalls, "closeProc should be called once to close subscription")
+	assert.Equal(t, 1, subscribeCalls, "subscribeProc should be called once to reopen subscription")
+
+	// Verify that batch size was reduced
+	assert.Equal(t, 50, input.currentMaxReads)
+
+	// Verify that a warning log was generated
+	require.Equal(t, 1, logs.Len())
+	assert.Contains(t, logs.All()[0].Message, "Encountered RPC_S_INVALID_BOUND")
+}

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

@@ -15,8 +15,12 @@ import (
 
 // Subscription is a subscription to a windows eventlog channel.
 type Subscription struct {
-	handle uintptr
-	Server string
+	handle        uintptr
+	Server        string
+	startAt       string
+	sessionHandle uintptr
+	channel       string
+	bookmark      Bookmark
 }
 
 // Open will open the subscription handle.
@@ -47,6 +51,10 @@ func (s *Subscription) Open(startAt string, sessionHandle uintptr, channel strin
 	}
 
 	s.handle = subscriptionHandle
+	s.startAt = startAt
+	s.sessionHandle = sessionHandle
+	s.channel = channel
+	s.bookmark = bookmark
 	return nil
 }
 
@@ -66,26 +74,52 @@ func (s *Subscription) Close() error {
 
 var errSubscriptionHandleNotOpen = errors.New("subscription handle is not open")
 
-// Read will read events from the subscription.
-func (s *Subscription) Read(maxReads int) ([]Event, error) {
+func (s *Subscription) Read(maxReads int) ([]Event, int, error) {
 	if s.handle == 0 {
-		return nil, errSubscriptionHandleNotOpen
+		return nil, 0, errSubscriptionHandleNotOpen
 	}
 
 	if maxReads < 1 {
-		return nil, fmt.Errorf("max reads must be greater than 0")
+		return nil, 0, fmt.Errorf("max reads must be greater than 0")
 	}
 
+	events, actualMaxReads, err := s.readWithRetry(maxReads)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return events, actualMaxReads, nil
+}
+
+// readWithRetry will read events from the subscription with dynamic batch sizing if the RPC_S_INVALID_BOUND error occurs.
+func (s *Subscription) readWithRetry(maxReads int) ([]Event, int, error) {
 	eventHandles := make([]uintptr, maxReads)
 	var eventsRead uint32
+
 	err := evtNext(s.handle, uint32(maxReads), &eventHandles[0], 0, 0, &eventsRead)
 
 	if errors.Is(err, ErrorInvalidOperation) && eventsRead == 0 {
-		return nil, nil
+		return nil, maxReads, nil
+	}
+
+	if errors.Is(err, windows.RPC_S_INVALID_BOUND) {
+		// close current subscription
+		if closeErr := s.Close(); closeErr != nil {
+			return nil, maxReads, fmt.Errorf("failed to close subscription during recovery: %w", closeErr)
+		}
+
+		// reopen subscription with the same parameters
+		if openErr := s.Open(s.startAt, s.sessionHandle, s.channel, s.bookmark); openErr != nil {
+			return nil, maxReads, fmt.Errorf("failed to reopen subscription during recovery: %w", openErr)
+		}
+
+		// retry with half the batch size
+		newMaxReads := max(maxReads/2, 1)
+		return s.readWithRetry(newMaxReads)
 	}
 
 	if err != nil && !errors.Is(err, windows.ERROR_NO_MORE_ITEMS) {
-		return nil, err
+		return nil, maxReads, err
 	}
 
 	events := make([]Event, 0, eventsRead)
@@ -94,7 +128,7 @@ func (s *Subscription) Read(maxReads int) ([]Event, error) {
 		events = append(events, event)
 	}
 
-	return events, nil
+	return events, maxReads, nil
 }
 
 // createFlags will create the necessary subscription flags from the supplied arguments.