[pkg/stanza] Windows input operator logs noisy error while collecting…

… successfully (open-telemetry#35520)

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

If `Input::read` returns 0, the operator will attempt to open the
subscription again. This happens after a resubscribe *or* whenever there
are no event logs to read, which is often. As a result, this error gets
logged constantly:
```
"Failed to re-open subscription for remote server"..."error: subscription handle is already open"
```
To fix this, the handle should only attempt to re-open when the
resubscribe happens.

This is a followup to open-telemetry#35175 which fixed that resubscribe issue for the
user that raised it, but they have also been confused by this error
being logged.

**Link to tracking Issue:** <Issue number if applicable> **N/A**

**Testing:** <Describe what testing was performed and which tests were
added.> Manually verified log collection is successful before and after
remote host reconnects, and that the noisy error is silenced except
during resubscribe, where it is relevant.

**Documentation:** <Describe the documentation added.> **N/A**

.chloggen/stanza-windows-remote-fix-noisy-error.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: windowseventlogreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: While collecting from a remote windows host, the stanza operator will no longer log "subscription handle is already open" constantly during successful collection.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35520]
+
+# (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]

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

@@ -172,7 +172,7 @@ func (i *Input) Stop() error {
 	return i.stopRemoteSession()
 }
 
-// readOnInterval will read events with respect to the polling interval.
+// readOnInterval will read events with respect to the polling interval until it reaches the end of the channel.
 func (i *Input) readOnInterval(ctx context.Context) {
 	defer i.wg.Done()
 
@@ -184,36 +184,13 @@ func (i *Input) readOnInterval(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			i.readToEnd(ctx)
-		}
-	}
-}
-
-// readToEnd will read events from the subscription until it reaches the end of the channel.
-func (i *Input) readToEnd(ctx context.Context) {
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			if count := i.read(ctx); count == 0 {
-				if i.isRemote() {
-					if err := i.startRemoteSession(); err != nil {
-						i.Logger().Error("Failed to re-establish remote session", zap.String("server", i.remote.Server), zap.Error(err))
-						return
-					}
-					if err := i.subscription.Open(i.startAt, uintptr(i.remoteSessionHandle), i.channel, i.bookmark); err != nil {
-						i.Logger().Error("Failed to re-open subscription for remote server", zap.String("server", i.remote.Server), zap.Error(err))
-					}
-				}
-				return
-			}
+			i.read(ctx)
 		}
 	}
 }
 
 // read will read events from the subscription.
-func (i *Input) read(ctx context.Context) int {
+func (i *Input) read(ctx context.Context) {
 	events, err := i.subscription.Read(i.maxReads)
 	if err != nil {
 		i.Logger().Error("Failed to read events from subscription", zap.Error(err))
@@ -222,12 +199,20 @@ func (i *Input) read(ctx context.Context) int {
 			closeErr := i.subscription.Close()
 			if closeErr != nil {
 				i.Logger().Error("Failed to close remote subscription", zap.Error(closeErr))
-				return 0
+				return
 			}
 			i.Logger().Info("Resubscribing, creating remote subscription")
 			i.subscription = NewRemoteSubscription(i.remote.Server)
+			if err := i.startRemoteSession(); err != nil {
+				i.Logger().Error("Failed to re-establish remote session", zap.String("server", i.remote.Server), zap.Error(err))
+				return
+			}
+			if err := i.subscription.Open(i.startAt, uintptr(i.remoteSessionHandle), i.channel, i.bookmark); err != nil {
+				i.Logger().Error("Failed to re-open subscription for remote server", zap.String("server", i.remote.Server), zap.Error(err))
+				return
+			}
 		}
-		return 0
+		return
 	}
 
 	for n, event := range events {
@@ -237,8 +222,6 @@ func (i *Input) read(ctx context.Context) int {
 		}
 		event.Close()
 	}
-
-	return len(events)
 }
 
 func (i *Input) getPublisherName(event Event) (name string, excluded bool) {