fix(inputs.win_eventlog): Do not replay the channel on restart

[skartikey]

Summary

With from_beginning = false and a statefile, the plugin replays a channel's entire history on every start after the first, for as long as the channel stays quiet. The bookmark is only advanced for events we actually receive, so a run that receives none persists a well-formed but positionless bookmark, and Windows cannot resume from that: it starts at the channel's oldest record instead.

The subscription now anchors the bookmark at the newest event matching the query before subscribing, so the state carries a real position from the first start. The current run collects the same events, as starting after the newest event and subscribing to the future ones are the same thing.

A channel that holds no event matching the query has no position to anchor to, but there the empty bookmark is already correct: it means we are at the beginning of the stream, and restoring it starts at the oldest record, so the next run picks up everything that arrived while Telegraf was stopped. The persisted state is unchanged, still the bookmark XML. A state file written by an earlier version replays the channel once on the first start after the upgrade and carries a real position from then on.

from_beginning = true is left alone, and a failed anchoring query warns and keeps the configured start point.

This needs EvtQuery, which was not bound yet.

Checklist

• No AI generated code was used in this PR
• AI generated code used in this PR follows the InfluxData Policy on AI-Generated Code Contributions

Related issues

resolves #19355

[srebhan]

Thanks @skartikey for your PR!

However, I think your approach is flawed.

Imagine you start Telegraf with from_beginning = false on an empty category log (or more general without any match on your query). Through the lifetime of this Telegraf instance no new log arrives.
We now take Telegraf down and for me the "saved state" should be the start of that log as that is where we left. After Telegraf is down many logs matching the query arrive.
Once Telegraf is restarted, your approach will miss all the log arriving in the meantime as you simply continue on the end of the stream, don't you?

As I already tried to say in the issue, I think what should be done is:

• tack if the bookmark was updated, i.e. if at least one event arrived
• when the state should be saved (i.e. in GetState()) check
  • if the bookmark was updated, return the bookmark. That's the normal case where new data arrives.
  • if the bookmark was not updated and it is not empty, return the (previous) bookmark. That's the case where we had a position in the stream before but nothing new arrived.
  • if the bookmark was not updated and it is empty, enforce a bookmark update by
    running EvtQuery with the EvtQueryReverseDirection flag and run EvtNext with EventsSize being once on the result-set to get the latest event.
    • if the set is empty (i.e. Returned is zero), save the empty bookmark as we are on the beginning of the stream.
    • otherwise, call EvtUpdateBookmark with the returned event. Render the updated bookmark XML and return it.

This approach guarantees that we really continue on where we left.

Drawback: The above cannot fix the current issue with existing saved states. The user has to manually delete the saved state for the eventlog plugin. However, I don't think this one-time effort justifies adding future issues with potential data loss when we leave on the start but come back at the end.

[skartikey]

@srebhan You are right, that case loses everything arriving while we are down, thanks. Pushed a fix, though I took a slightly different route than the one you sketched and I would like your read on it.

The plugin now remembers why the bookmark has no position. If the anchoring query ran and the channel had no matching event, we know we are sitting at the beginning of the stream, and the state records that next to the bookmark. Restoring such a state starts the subscription at the oldest record, so the logs that arrived while Telegraf was down are picked up. A positionless bookmark without that marker still means "we have no idea where we are", so it keeps the configured start and does not replay.

That is also what avoids the manual state deletion: the state is now a small struct whose UnmarshalJSON still accepts the plain bookmark string, so a state file written by an older version reads as "no position, no marker" and behaves exactly as it does today.

I kept the anchoring before the subscription rather than moving the query into GetState(). Once the marker exists the query at shutdown does not add anything, and anchoring at shutdown would move the bookmark past events that arrived after the last gather cycle (there is no final gather, the loop stops before Stop()), so a quiet run would lose them. Same EvtQuery call, one place, and it runs at most once per start.

[srebhan]

@skartikey your approach is much more complex than what I had in mind. For me what you need to do is to

1. Add an eventReceived bool flag to WinEventLog which is set to true once fetchEvents passes the error check in Gather (original at line 165).
2. In GetState do the following

func (w *WinEventLog) GetState() interface{} {
	// Render the Bookmark XML from the handle
	bookmarkXML, err := w.renderBookmark()
	if err != nil {
		w.Log.Errorf("State-persistence failed, cannot render bookmark: %v", err)
		return nil
	}
	
	// The bookmark was either updated or was filled in a previous run.
	// In both cases the bookmark represents a valid position.
	if w.eventReceived || !w.emptyBookmark(){
		return bookmarkXML
	}

	// Force a bookmark update by finding the last event matching the query
	// and use this position.
	eventHandle, err := w.seekLastEvent()
	if  err != nil {
		// If not a single event matches, we should start at the beginning
		if err == errNoMoreItems {
			return bookmarkXML
		}
		w.Log.Errorf("Seeking latest even failed: %v", err)
		return nil
	}
	if err := evtUpdateBookmark(w.bookmark, eventHandle); err != nil {
		w.Log.Errorf("Updating bookmark failed: %v", err)
	}
	
	// Return the updated bookmark
	// Render the Bookmark XML from the handle
	bookmarkXML, err == w.renderBookmark()
	if err != nil {
		w.Log.Errorf("State-persistence failed, cannot render updated bookmark: %v", err)
		return nil
	}
	return bookmarkXML
}

3. With emptyBookmark either checking the XML string itself (if that's save) or by unmarshal the XML...
4. With seekLastEvent being

func (w *WinEventLog) seekLastEvent() (evtHandle, error) {
	// Query the log in reverse order to find matching events
	logNamePtr, err := syscall.UTF16PtrFromString(w.EventlogName)
	if err != nil {
		return 0, err
	}
	xqueryPtr, err := syscall.UTF16PtrFromString(w.Query)
	if err != nil {
		return 0, err
	}
	resultSet, err := evtQuery(0, logNamePtr, xqueryPtr, evtQueryChannelPath|evtQueryReverseDirection)
	if err != nil {
		return 0, fmt.Errorf("querying failed: %w", err)
	}
	
	// Get the last event
	var evtReturned uint32
	eventHandles := make([]evtHandle, 1)
	if err := evtNext(subsHandle, 1, &eventHandles[0], 0, 0, &evtReturned); err != nil {
		if errors.Is(err, errInvalidOperation) && evtReturned == 0 {
			return 0, errNoMoreItems
		}
		return 0, err
	}
	if evtReturned == 0 {
		return 0, errNoMoreItems
	}
	return eventHandles[0], nil
}

This means we only need to map evtQuery from the Windows API and all magic happens in GetState(). We might need to guard the new nil return value signalling "no state" but that would be a good general addition.

[john-behm-bertelsmann]

any progress in here :?

[skartikey]

@srebhan You are right, that was more complex than needed. I dropped the state struct, the persisted state is the plain bookmark XML again, unchanged from today. The marker was redundant: a positionless bookmark restored with StartAfterBookmark already starts at the oldest record, which is the "continue at the beginning of the stream" case.

I kept the query before the subscription rather than in GetState() though. The persister calls GetState() at load just to derive the state type (reflect.New(reflect.TypeOf(plugin.GetState()))), so anchoring there runs on every start and SetState() discards the result. And there is no final gather at shutdown, so on a quiet run seekLastEvent would bookmark an event that arrived after the last gather and was never collected. Anchored before subscribing instead, such an event is simply redelivered on the next start.

The manual state deletion is avoided too: an old positionless state replays the channel once after the upgrade and carries a real position from then on. Side note on returning nil from GetState(): that panics at load today (reflect.New of a nil type), so the guard you mention would be a good general addition either way.

[telegraf-tiger]

Download PR build artifacts for linux_amd64.tar.gz, darwin_arm64.tar.gz, and windows_amd64.zip.
Downloads for additional architectures and packages are available below.

⚠️ This pull request increases the Telegraf binary size by 6.01 % for linux amd64 (new size: 326.8 MB, nightly size 308.2 MB)

📦 Click here to get additional PR build artifacts

Artifact URLs

. DEB	. RPM	. TAR . GZ	. ZIP
amd64.deb	aarch64.rpm	darwin_amd64.tar.gz	windows_amd64.zip
arm64.deb	armel.rpm	darwin_arm64.tar.gz	windows_arm64.zip
armel.deb	armv6hl.rpm	freebsd_amd64.tar.gz	windows_i386.zip
armhf.deb	i386.rpm	freebsd_armv7.tar.gz
i386.deb	ppc64le.rpm	freebsd_i386.tar.gz
mips.deb	riscv64.rpm	linux_amd64.tar.gz
mipsel.deb	s390x.rpm	linux_arm64.tar.gz
ppc64el.deb	x86_64.rpm	linux_armel.tar.gz
riscv64.deb		linux_armhf.tar.gz
s390x.deb		linux_i386.tar.gz
		linux_mips.tar.gz
		linux_mipsel.tar.gz
		linux_ppc64le.tar.gz
		linux_riscv64.tar.gz
		linux_s390x.tar.gz