add kernel power query (#44298)

### What does this PR do?
Add query for kernel power events. make room for more queries to be added.

### Motivation
https://datadoghq.atlassian.net/browse/WINA-1970

Co-authored-by: branden.clark <branden.clark@datadoghq.com>

Author: clarkb7

comp/notableevents/impl/collector.go

@@ -10,23 +10,48 @@ package notableeventsimpl
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
 	"github.com/cenkalti/backoff"
+	"golang.org/x/sys/windows"
 
+	"github.com/DataDog/datadog-agent/pkg/logs/util/windowsevent"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 	evtapi "github.com/DataDog/datadog-agent/pkg/util/winutil/eventlog/api"
 	winevtapi "github.com/DataDog/datadog-agent/pkg/util/winutil/eventlog/api/windows"
 	evtsubscribe "github.com/DataDog/datadog-agent/pkg/util/winutil/eventlog/subscription"
 )
 
+// eventDefinition describes a notable event type and how to collect/process it
+type eventDefinition struct {
+	// Event identification (for lookup after receiving event)
+	Provider string
+	EventID  uint
+
+	// Event metadata (for payload)
+	EventType string
+	Title     string
+	Message   string
+
+	// Query definition - inner content of <Query> block
+	Channel   string
+	QueryBody string
+}
+
+// eventKey uniquely identifies an event by provider and event ID
+type eventKey struct {
+	Provider string
+	EventID  uint
+}
+
 // collector monitors Windows Event Log for notable events
 type collector struct {
 	// in
 	api         evtapi.API
-	channelPath string
 	query       string
+	eventLookup map[eventKey]*eventDefinition
 	// out
 	outChan chan<- eventPayload
 	// internal
@@ -35,15 +60,63 @@ type collector struct {
 	wg     sync.WaitGroup
 }
 
+// getEventDefinitions returns the list of notable events to collect
+func getEventDefinitions() []eventDefinition {
+	e := []eventDefinition{
+		{
+			Provider:  "Microsoft-Windows-Kernel-Power",
+			EventID:   41,
+			Channel:   "System",
+			QueryBody: `    <Select Path="System">*[System[Provider[@Name='Microsoft-Windows-Kernel-Power'] and EventID=41]]</Select>`,
+			EventType: "Unexpected reboot",
+			Title:     "Unexpected reboot",
+			Message:   "The system has rebooted without cleanly shutting down first",
+		},
+	}
+	return e
+}
+
+// buildEventLookup creates a map for fast event definition lookup.
+// Returns error if duplicate event keys are found.
+func buildEventLookup(events []eventDefinition) (map[eventKey]*eventDefinition, error) {
+	lookup := make(map[eventKey]*eventDefinition)
+	for i := range events {
+		def := &events[i]
+		key := eventKey{Provider: def.Provider, EventID: def.EventID}
+		if _, exists := lookup[key]; exists {
+			return nil, fmt.Errorf("duplicate event definition: %s/%d", def.Provider, def.EventID)
+		}
+		lookup[key] = def
+	}
+	return lookup, nil
+}
+
+// buildQuery generates full XML query from event definitions.
+// Each event gets its own <Query> block with auto-generated ID.
+func buildQuery(events []eventDefinition) string {
+	var queries []string
+	for i, def := range events {
+		query := fmt.Sprintf(`  <Query Id="%d" Path="%s">
+%s
+  </Query>`, i, def.Channel, def.QueryBody)
+		queries = append(queries, query)
+	}
+	return fmt.Sprintf("<QueryList>\n%s\n</QueryList>", strings.Join(queries, "\n"))
+}
+
 // newCollector creates a new collector instance
-func newCollector(outChan chan<- eventPayload) *collector {
-	// TODO(WINA-1970): make real query
+func newCollector(outChan chan<- eventPayload) (*collector, error) {
+	events := getEventDefinitions()
+	lookup, err := buildEventLookup(events)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build event lookup: %w", err)
+	}
 	return &collector{
 		api:         winevtapi.New(),
-		channelPath: "System",
-		query:       "*[System[(Level=1 or Level=2 or Level=3)]]",
+		query:       buildQuery(events),
 		outChan:     outChan,
-	}
+		eventLookup: lookup,
+	}, nil
 }
 
 // start begins monitoring Windows Event Log
@@ -54,12 +127,12 @@ func (c *collector) start() error {
 
 	// Create subscription object (will be started in the event loop)
 	c.sub = evtsubscribe.NewPullSubscription(
-		c.channelPath,
+		"", // empty chennel path when XML query is used
 		c.query,
 		evtsubscribe.WithWindowsEventLogAPI(c.api),
 	)
 
-	log.Infof("Initialized Windows Event Log subscription: channel=%s, query=%s", c.channelPath, c.query)
+	log.Debugf("Initialized Windows Event Log subscription: query=%s", c.query)
 
 	// Start processing events in background
 	c.wg.Add(1)
@@ -109,7 +182,7 @@ func (c *collector) run(ctx context.Context) {
 		// Check if loop should exit
 		select {
 		case <-ctx.Done():
-			log.Info("Notable events collector context cancelled, shutting down")
+			log.Debug("Notable events collector context cancelled, shutting down")
 			return
 		default:
 		}
@@ -123,7 +196,7 @@ func (c *collector) run(ctx context.Context) {
 					return err
 				}
 				// Subscription started successfully
-				log.Infof("Started Windows Event Log subscription: channel=%s, query=%s", c.channelPath, c.query)
+				log.Debugf("Started Windows Event Log subscription: query=%s", c.query)
 				return nil
 			})
 			if err != nil {
@@ -136,7 +209,7 @@ func (c *collector) run(ctx context.Context) {
 		// Subscription is running, wait for events or cancellation
 		select {
 		case <-ctx.Done():
-			log.Info("Notable events collector context cancelled, shutting down")
+			log.Debug("Notable events collector context cancelled, shutting down")
 			return
 		case events, ok := <-c.sub.GetEvents():
 			if !ok {
@@ -169,16 +242,45 @@ func (c *collector) processEvent(renderCtx evtapi.EventRenderContextHandle, even
 	}
 	defer vals.Close()
 
-	// Extract Event ID
+	// Extract provider and event ID for lookup
+	providerName, err := vals.String(evtapi.EvtSystemProviderName)
+	if err != nil {
+		return fmt.Errorf("failed to get provider name: %w", err)
+	}
 	eventID, err := vals.UInt(evtapi.EvtSystemEventID)
 	if err != nil {
 		return fmt.Errorf("failed to get event ID: %w", err)
 	}
-	providerName, err := vals.String(evtapi.EvtSystemProviderName)
+
+	// Lookup event definition
+	def, found := c.eventLookup[eventKey{Provider: providerName, EventID: uint(eventID)}]
+	if !found {
+		return fmt.Errorf("unknown event: %s/%d", providerName, eventID)
+	}
+
+	// Render full event XML
+	xmlUTF16, err := c.api.EvtRenderEventXml(eventRecord.EventRecordHandle)
 	if err != nil {
-		return fmt.Errorf("failed to get provider name: %w", err)
+		return fmt.Errorf("failed to render event XML: %w", err)
+	}
+	xmlString := windows.UTF16ToString(xmlUTF16)
+
+	// Convert XML to JSON map using windowsevent package
+	eventMap, err := windowsevent.NewMapXMLWithOptions([]byte(xmlString), windowsevent.TransformOptions{
+		FormatEventData:  true,
+		FormatBinaryData: false, // Skip buggy binary transform
+		NormalizeEventID: true,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to parse event XML: %w", err)
+	}
+
+	// Build custom data with windows_event_log
+	customData := map[string]interface{}{
+		"windows_event_log": eventMap.Map,
 	}
-	// TODO: get nanoseconds precision from event log
+
+	// Get timestamp
 	var timestamp time.Time
 	unixTimestamp, err := vals.Time(evtapi.EvtSystemTimeCreated)
 	if err != nil {
@@ -188,13 +290,15 @@ func (c *collector) processEvent(renderCtx evtapi.EventRenderContextHandle, even
 		timestamp = time.Unix(unixTimestamp, 0)
 	}
 
-	log.Debugf("Collected notable event: channel=%s, event_id=%d", c.channelPath, eventID)
+	log.Debugf("Collected notable event: provider=%s, event_id=%d, title=%s", providerName, eventID, def.Title)
 
+	// Build and send payload
 	payload := eventPayload{
-		Channel:   c.channelPath,
-		Provider:  providerName,
-		EventID:   uint(eventID),
 		Timestamp: timestamp,
+		EventType: def.EventType,
+		Title:     def.Title,
+		Message:   def.Message,
+		Custom:    customData,
 	}
 	c.outChan <- payload
 

comp/notableevents/impl/collector_test.go

@@ -8,6 +8,7 @@
 package notableeventsimpl
 
 import (
+	"fmt"
 	"slices"
 	"testing"
 	"time"
@@ -18,6 +19,102 @@ import (
 	eventlog_test "github.com/DataDog/datadog-agent/pkg/util/winutil/eventlog/test"
 )
 
+// TestBuildEventLookup asserts that events are correctly indexed by provider and event ID.
+func TestBuildEventLookup(t *testing.T) {
+	events := []eventDefinition{
+		{Provider: "Provider-A", EventID: 1, EventType: "Type1", Title: "Title1", Message: "Msg1"},
+		{Provider: "Provider-A", EventID: 2, EventType: "Type2", Title: "Title2", Message: "Msg2"},
+		{Provider: "Provider-B", EventID: 1, EventType: "Type3", Title: "Title3", Message: "Msg3"},
+	}
+
+	lookup, err := buildEventLookup(events)
+	require.NoError(t, err)
+	require.Len(t, lookup, 3)
+
+	// Verify lookups
+	def, found := lookup[eventKey{Provider: "Provider-A", EventID: 1}]
+	require.True(t, found)
+	assert.Equal(t, "Type1", def.EventType)
+
+	def, found = lookup[eventKey{Provider: "Provider-A", EventID: 2}]
+	require.True(t, found)
+	assert.Equal(t, "Type2", def.EventType)
+
+	def, found = lookup[eventKey{Provider: "Provider-B", EventID: 1}]
+	require.True(t, found)
+	assert.Equal(t, "Type3", def.EventType)
+
+	// Non-existent key
+	_, found = lookup[eventKey{Provider: "Provider-B", EventID: 999}]
+	assert.False(t, found)
+}
+
+// TestBuildEventLookup_DuplicateKey asserts that duplicate event definitions are not allowed.
+func TestBuildEventLookup_DuplicateKey(t *testing.T) {
+	events := []eventDefinition{
+		{Provider: "Provider-A", EventID: 1, EventType: "Type1"},
+		{Provider: "Provider-A", EventID: 1, EventType: "Type2"}, // Duplicate
+	}
+
+	lookup, err := buildEventLookup(events)
+	require.Error(t, err)
+	assert.Nil(t, lookup)
+	assert.Contains(t, err.Error(), "duplicate event definition")
+	assert.Contains(t, err.Error(), "Provider-A/1")
+}
+
+func TestBuildQuery(t *testing.T) {
+	events := []eventDefinition{
+		{
+			Channel:   "System",
+			QueryBody: `    <Select Path="System">*[System[Provider[@Name='Test-Provider'] and EventID=123]]</Select>`,
+		},
+	}
+
+	query := buildQuery(events)
+
+	expected := `<QueryList>
+  <Query Id="0" Path="System">
+    <Select Path="System">*[System[Provider[@Name='Test-Provider'] and EventID=123]]</Select>
+  </Query>
+</QueryList>`
+	assert.Equal(t, expected, query)
+}
+
+func TestBuildQuery_MultipleEvents(t *testing.T) {
+	events := []eventDefinition{
+		{
+			Channel:   "System",
+			QueryBody: `    <Select Path="System">*[System[EventID=1]]</Select>`,
+		},
+		{
+			Channel:   "Application",
+			QueryBody: `    <Select Path="Application">*[System[EventID=2]]</Select>`,
+		},
+		{
+			Channel: "Security",
+			QueryBody: `    <Select Path="Security">*[System[EventID=3]]</Select>
+    <Suppress Path="Security">*[EventData[Data='exclude']]</Suppress>`,
+		},
+	}
+
+	query := buildQuery(events)
+
+	expected := `<QueryList>
+  <Query Id="0" Path="System">
+    <Select Path="System">*[System[EventID=1]]</Select>
+  </Query>
+  <Query Id="1" Path="Application">
+    <Select Path="Application">*[System[EventID=2]]</Select>
+  </Query>
+  <Query Id="2" Path="Security">
+    <Select Path="Security">*[System[EventID=3]]</Select>
+    <Suppress Path="Security">*[EventData[Data='exclude']]</Suppress>
+  </Query>
+</QueryList>`
+	assert.Equal(t, expected, query)
+}
+
 func TestCollector_CollectEvents(t *testing.T) {
 	ctx := t.Context()
 
@@ -48,12 +145,23 @@ func TestCollector_CollectEvents(t *testing.T) {
 	outChan := make(chan eventPayload)
 
 	// Create collector using constructor
-	collector := newCollector(outChan)
+	collector, err := newCollector(outChan)
+	require.NoError(t, err)
 
 	// Customize for testing with test API and test channel
 	collector.api = ti.API()
-	collector.channelPath = channelPath
-	collector.query = "*" // Collect all events from the test channel
+	collector.query = fmt.Sprintf(`<QueryList><Query Id="0"><Select Path="%s">*</Select></Query></QueryList>`, channelPath) // Collect all events from the test channel
+
+	// Add test event source to the lookup so test events can be processed
+	testEventDef := &eventDefinition{
+		Provider:  eventSource,
+		EventID:   1000,
+		EventType: "Test event",
+		Title:     "Test event title",
+		Message:   "Test event message",
+		Channel:   channelPath,
+	}
+	collector.eventLookup[eventKey{Provider: testEventDef.Provider, EventID: testEventDef.EventID}] = testEventDef
 
 	// Start collector
 	err = collector.start()
@@ -95,10 +203,13 @@ func TestCollector_CollectEvents(t *testing.T) {
 	// Verify we received all expected events
 	require.Len(t, receivedEvents, 3, "Should have received 3 events")
 
-	// Verify event payloads have correct channel
+	// Verify event payloads have correct metadata from test event definition
 	for i, event := range receivedEvents {
-		assert.Equal(t, channelPath, event.Channel, "Event %d should have correct channel", i)
-		assert.NotZero(t, event.EventID, "Event %d should have non-zero Event ID", i)
+		assert.Equal(t, "Test event title", event.Title, "Event %d should have correct title", i)
+		assert.Equal(t, "Test event", event.EventType, "Event %d should have correct event type", i)
+		assert.Equal(t, "Test event message", event.Message, "Event %d should have correct message", i)
+		assert.NotNil(t, event.Custom, "Event %d should have custom data", i)
+		assert.Contains(t, event.Custom, "windows_event_log", "Event %d custom data should contain windows_event_log", i)
 	}
 
 	// Verify no more events are in the channel