feat(pkg/stanza/operator/input/windows): add support for XML queries

Signed-off-by: Szilard Parrag <[email protected]>

Author: OverOrion

.chloggen/eventlogreceiver-query.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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: eventlogreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add raw XML query filtering option
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38517]
+
+# (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, api]

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

@@ -37,6 +37,7 @@ type Config struct {
 	SuppressRenderingInfo bool          `mapstructure:"suppress_rendering_info,omitempty"`
 	ExcludeProviders      []string      `mapstructure:"exclude_providers,omitempty"`
 	Remote                RemoteConfig  `mapstructure:"remote,omitempty"`
+	Query                 *string       `mapstructure:"query,omitempty"`
 }
 
 // RemoteConfig is the configuration for a remote server.

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

@@ -24,8 +24,12 @@ func (c *Config) Build(set component.TelemetrySettings) (operator.Operator, erro
 		return nil, err
 	}
 
-	if c.Channel == "" {
-		return nil, fmt.Errorf("missing required `channel` field")
+	if c.Channel == "" && c.Query == nil {
+		return nil, fmt.Errorf("either `channel` or `query` must be set")
+	}
+
+	if c.Channel != "" && c.Query != nil {
+		return nil, fmt.Errorf("either `channel` or `query` must be set, but not both")
 	}
 
 	if c.MaxReads < 1 {
@@ -52,6 +56,7 @@ func (c *Config) Build(set component.TelemetrySettings) (operator.Operator, erro
 		raw:              c.Raw,
 		excludeProviders: excludeProvidersSet(c.ExcludeProviders),
 		remote:           c.Remote,
+		query:            c.Query,
 	}
 	input.startRemoteSession = input.defaultStartRemoteSession
 

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

@@ -26,6 +26,7 @@ type Input struct {
 	bookmark            Bookmark
 	buffer              *Buffer
 	channel             string
+	query               *string
 	maxReads            int
 	currentMaxReads     int
 	startAt             string
@@ -132,7 +133,7 @@ func (i *Input) Start(persister operator.Persister) error {
 		subscription = NewRemoteSubscription(i.remote.Server)
 	}
 
-	if err := subscription.Open(i.startAt, uintptr(i.remoteSessionHandle), i.channel, i.bookmark); err != nil {
+	if err := subscription.Open(i.startAt, uintptr(i.remoteSessionHandle), i.channel, i.query, i.bookmark); err != nil {
 		if isNonTransientError(err) {
 			if i.isRemote() {
 				return fmt.Errorf("failed to open subscription for remote server %s: %w", i.remote.Server, err)
@@ -224,7 +225,7 @@ func (i *Input) read(ctx context.Context) {
 				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 {
+			if err := i.subscription.Open(i.startAt, uintptr(i.remoteSessionHandle), i.channel, i.query, 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
 			}

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

@@ -20,13 +20,14 @@ type Subscription struct {
 	startAt       string
 	sessionHandle uintptr
 	channel       string
+	query         *string
 	bookmark      Bookmark
 }
 
 // Open will open the subscription handle.
 // It returns an error if the subscription handle is already open or if any step in the process fails.
 // If the remote server is not reachable, it returns an error indicating the failure.
-func (s *Subscription) Open(startAt string, sessionHandle uintptr, channel string, bookmark Bookmark) error {
+func (s *Subscription) Open(startAt string, sessionHandle uintptr, channel string, query *string, bookmark Bookmark) error {
 	if s.handle != 0 {
 		return fmt.Errorf("subscription handle is already open")
 	}
@@ -39,13 +40,24 @@ func (s *Subscription) Open(startAt string, sessionHandle uintptr, channel strin
 		_ = windows.CloseHandle(signalEvent)
 	}()
 
+	if channel != "" && query != nil {
+		return fmt.Errorf("can not supply both query and channel")
+	}
+
 	channelPtr, err := syscall.UTF16PtrFromString(channel)
 	if err != nil {
 		return fmt.Errorf("failed to convert channel to utf16: %w", err)
 	}
 
+	var queryPtr *uint16
+	if query != nil {
+		if queryPtr, err = syscall.UTF16PtrFromString(*query); err != nil {
+			return fmt.Errorf("failed to convert channel to utf16: %w", err)
+		}
+	}
+
 	flags := s.createFlags(startAt, bookmark)
-	subscriptionHandle, err := evtSubscribeFunc(sessionHandle, signalEvent, channelPtr, nil, bookmark.handle, 0, 0, flags)
+	subscriptionHandle, err := evtSubscribeFunc(sessionHandle, signalEvent, channelPtr, queryPtr, bookmark.handle, 0, 0, flags)
 	if err != nil {
 		return fmt.Errorf("failed to subscribe to %s channel: %w", channel, err)
 	}
@@ -55,6 +67,7 @@ func (s *Subscription) Open(startAt string, sessionHandle uintptr, channel strin
 	s.sessionHandle = sessionHandle
 	s.channel = channel
 	s.bookmark = bookmark
+	s.query = query
 	return nil
 }
 
@@ -109,7 +122,7 @@ func (s *Subscription) readWithRetry(maxReads int) ([]Event, int, error) {
 		}
 
 		// reopen subscription with the same parameters
-		if openErr := s.Open(s.startAt, s.sessionHandle, s.channel, s.bookmark); openErr != nil {
+		if openErr := s.Open(s.startAt, s.sessionHandle, s.channel, s.query, s.bookmark); openErr != nil {
 			return nil, maxReads, fmt.Errorf("failed to reopen subscription during recovery: %w", openErr)
 		}