Use query XML for windows events logs (#2201)

Author: jefchien

go.mod

@@ -191,22 +191,6 @@ processors:
         send_batch_max_size: 10000
         send_batch_size: 10000
         timeout: 30s
-    filter/windows_events_application_1:
-        error_mode: ignore
-        logs:
-            log_record:
-                - not((severity_number == 17 or severity_number == 13))
-        metrics: {}
-        spans: {}
-        traces: {}
-    filter/windows_events_system_0:
-        error_mode: ignore
-        logs:
-            log_record:
-                - not((severity_number == 17 or severity_number == 13))
-        metrics: {}
-        spans: {}
-        traces: {}
     resourcedetection/opentelemetry:
         aks:
             resource_attributes:
@@ -610,12 +594,13 @@ receivers:
                 read_header_timeout: 0s
                 traces_url_path: /v1/traces
                 write_timeout: 0s
-    windowseventlog/application_2122316671:
-        channel: Application
+    windowseventlog/application_3497854245:
+        channel: ""
         id: windows_eventlog_input
         max_reads: 100
         operators: []
         poll_interval: 1s
+        query: <QueryList><Query Id="0"><Select Path="Application">*[System[(Level='2' or Level='3') and TimeCreated[timediff(@SystemTime) &lt;= 1209600000]]]</Select></Query></QueryList>
         resource:
             aws.log.channel: Application
             aws.log.source: windows_events
@@ -627,12 +612,13 @@ receivers:
         start_at: end
         storage: file_storage/opentelemetry
         type: windows_eventlog_input
-    windowseventlog/system_653521908:
-        channel: System
+    windowseventlog/system_396641996:
+        channel: ""
         id: windows_eventlog_input
         max_reads: 100
         operators: []
         poll_interval: 1s
+        query: <QueryList><Query Id="0"><Select Path="System">*[System[(Level='2' or Level='3') and TimeCreated[timediff(@SystemTime) &lt;= 1209600000]]]</Select></Query></QueryList>
         resource:
             aws.log.channel: System
             aws.log.source: windows_events
@@ -681,18 +667,16 @@ service:
             exporters:
                 - forward/opentelemetry
             processors:
-                - filter/windows_events_application_1
                 - transform/windows_events_scope
             receivers:
-                - windowseventlog/application_2122316671
+                - windowseventlog/application_3497854245
         logs/windows_events_system_0:
             exporters:
                 - forward/opentelemetry
             processors:
-                - filter/windows_events_system_0
                 - transform/windows_events_scope
             receivers:
-                - windowseventlog/system_653521908
+                - windowseventlog/system_396641996
         metrics/opentelemetry:
             exporters:
                 - otlphttp/metrics

go.sum

@@ -89,22 +89,6 @@ processors:
         send_batch_max_size: 10000
         send_batch_size: 10000
         timeout: 30s
-    filter/windows_events_application_1:
-        error_mode: ignore
-        logs:
-            log_record:
-                - not((severity_number == 17 or severity_number == 13))
-        metrics: {}
-        spans: {}
-        traces: {}
-    filter/windows_events_system_0:
-        error_mode: ignore
-        logs:
-            log_record:
-                - not((severity_number == 17 or severity_number == 13 or severity_number == 9))
-        metrics: {}
-        spans: {}
-        traces: {}
     resource/windows_events_system_0:
         attributes:
             - action: upsert
@@ -466,12 +450,13 @@ processors:
         metric_statements: []
         trace_statements: []
 receivers:
-    windowseventlog/application_2122316671:
-        channel: Application
+    windowseventlog/application_3497854245:
+        channel: ""
         id: windows_eventlog_input
         max_reads: 100
         operators: []
         poll_interval: 1s
+        query: <QueryList><Query Id="0"><Select Path="Application">*[System[(Level='2' or Level='3') and TimeCreated[timediff(@SystemTime) &lt;= 1209600000]]]</Select></Query></QueryList>
         resource:
             aws.log.channel: Application
             aws.log.source: windows_events
@@ -483,12 +468,13 @@ receivers:
         start_at: end
         storage: file_storage/opentelemetry
         type: windows_eventlog_input
-    windowseventlog/system_653521908:
-        channel: System
+    windowseventlog/system_1384911762:
+        channel: ""
         id: windows_eventlog_input
         max_reads: 100
         operators: []
         poll_interval: 1s
+        query: <QueryList><Query Id="0"><Select Path="System">*[System[(Level='2' or Level='4' or Level='0' or Level='3') and TimeCreated[timediff(@SystemTime) &lt;= 1209600000]]]</Select></Query></QueryList>
         resource:
             aws.log.channel: System
             aws.log.source: windows_events
@@ -524,19 +510,17 @@ service:
             exporters:
                 - forward/opentelemetry
             processors:
-                - filter/windows_events_application_1
                 - transform/windows_events_scope
             receivers:
-                - windowseventlog/application_2122316671
+                - windowseventlog/application_3497854245
         logs/windows_events_system_0:
             exporters:
                 - forward/opentelemetry
             processors:
-                - filter/windows_events_system_0
                 - resource/windows_events_system_0
                 - transform/windows_events_scope
             receivers:
-                - windowseventlog/system_653521908
+                - windowseventlog/system_1384911762
     telemetry:
         logs:
             encoding: console

translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_windows.yaml

@@ -6,48 +6,61 @@ package windowsevents
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/pipeline"
 
+	"github.com/aws/amazon-cloudwatch-agent/internal/util/hash"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/connector/forward"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/filestorage"
-	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/filterprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/resourceprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/transformprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/receiver/windowseventlog"
 )
 
-// severityNumbers maps config level names to OTel severity numbers.
-// Derived from pkg/stanza/operator/input/windows/xml.go parseSeverity():
-//
-//	Level "1"/Critical → Fatal(21), "2"/Error → Error(17),
-//	"3"/Warning → Warn(13), "4"/Information → Info(9), default → Default(0)
-//
-// TODO: Replace with upstream Query XML filtering when collector is bumped past v0.124.
-var severityNumbers = map[string]int{
-	"CRITICAL":    21,
-	"ERROR":       17,
-	"WARNING":     13,
-	"INFORMATION": 9,
-	"VERBOSE":     0,
+// eventLevelNumbers maps config level names to Windows Event Log numeric levels.
+var eventLevelNumbers = map[string][]string{
+	"CRITICAL":    {"1"},
+	"ERROR":       {"2"},
+	"WARNING":     {"3"},
+	"INFORMATION": {"4", "0"},
+	"VERBOSE":     {"5"},
 }
 
 type eventEntry struct {
-	name          string
-	receiverName  string
+	index         int
 	channel       string
-	raw           bool
+	format        string
 	resource      map[string]string
 	logGroupName  string
 	logStreamName string
 	eventLevels   []string
 	eventIDs      []int
 }
 
+func (e eventEntry) raw() bool {
+	return e.format == "xml"
+}
+
+func (e eventEntry) name() string {
+	return fmt.Sprintf("%s_%d", sanitizeName(e.channel), e.index)
+}
+
+func (e eventEntry) receiverName() string {
+	return fmt.Sprintf("%s_%s", sanitizeName(e.channel), e.receiverHash())
+}
+
+func (e eventEntry) receiverHash() string {
+	return hash.HashName(fmt.Sprintf("%s\x00%s\x00%v\x00%v", e.channel, e.format, e.eventLevels, e.eventIDs))
+}
+
 func (e eventEntry) routingAttributes() map[string]string {
+	if e.logGroupName == "" && e.logStreamName == "" {
+		return nil
+	}
 	attrs := make(map[string]string)
 	if e.logGroupName != "" {
 		attrs["aws.log.group.name"] = e.logGroupName
@@ -58,38 +71,41 @@ func (e eventEntry) routingAttributes() map[string]string {
 	return attrs
 }
 
-// filterCondition builds an OTTL drop condition. When both levels and IDs are present, they are ANDed.
-func (e eventEntry) filterCondition() string {
-	var parts []string
+const ignoreOlderThanTwoWeeksMs = int64(14*24*time.Hour) / int64(time.Millisecond)
+
+// queryXML builds a Windows Event Log XPath query XML for native OS-level filtering.
+// Always includes a 2-week time cutoff to prevent replaying stale events on restart.
+func (e eventEntry) queryXML() string {
+	var filters []string
 
 	if len(e.eventLevels) > 0 {
-		var severityChecks []string
+		var levelChecks []string
 		for _, level := range e.eventLevels {
-			if sev, ok := severityNumbers[level]; ok {
-				severityChecks = append(severityChecks, fmt.Sprintf("severity_number == %d", sev))
+			if nums, ok := eventLevelNumbers[level]; ok {
+				for _, n := range nums {
+					levelChecks = append(levelChecks, fmt.Sprintf("Level='%s'", n))
+				}
 			}
 		}
-		if len(severityChecks) > 0 {
-			parts = append(parts, "("+strings.Join(severityChecks, " or ")+")")
+		if len(levelChecks) > 0 {
+			filters = append(filters, "("+strings.Join(levelChecks, " or ")+")")
 		}
 	}
 
 	if len(e.eventIDs) > 0 {
 		var idChecks []string
 		for _, id := range e.eventIDs {
-			idChecks = append(idChecks, fmt.Sprintf("body[\"event_id\"][\"id\"] == %d", id))
+			idChecks = append(idChecks, fmt.Sprintf("EventID='%d'", id))
 		}
 		if len(idChecks) > 0 {
-			parts = append(parts, "("+strings.Join(idChecks, " or ")+")")
+			filters = append(filters, "("+strings.Join(idChecks, " or ")+")")
 		}
 	}
 
-	if len(parts) == 0 {
-		return ""
-	}
+	filters = append(filters, fmt.Sprintf("TimeCreated[timediff(@SystemTime) <= %d]", ignoreOlderThanTwoWeeksMs))
 
-	// Drop records that DON'T match the filter (filter processor drops when condition is true)
-	return "not(" + strings.Join(parts, " and ") + ")"
+	return fmt.Sprintf(`<QueryList><Query Id="0"><Select Path="%s">*[System[%s]]</Select></Query></QueryList>`,
+		e.channel, strings.Join(filters, " and "))
 }
 
 type windowsEventsPipelineTranslator struct {
@@ -99,30 +115,21 @@ type windowsEventsPipelineTranslator struct {
 var _ common.PipelineTranslator = (*windowsEventsPipelineTranslator)(nil)
 
 func (t *windowsEventsPipelineTranslator) ID() pipeline.ID {
-	return pipeline.NewIDWithName(pipeline.SignalLogs, "windows_events_"+t.entry.name)
+	return pipeline.NewIDWithName(pipeline.SignalLogs, "windows_events_"+t.entry.name())
 }
 
 func (t *windowsEventsPipelineTranslator) Translate(_ *confmap.Conf) (*common.ComponentTranslators, error) {
-	if t.entry.raw && len(t.entry.eventIDs) > 0 {
-		return nil, fmt.Errorf("event_ids filtering is not supported with event_format \"xml\" for channel %q", t.entry.channel)
-	}
-
 	fwdConnector := forward.NewTranslator(common.OpenTelemetryKey)
 
+	query := t.entry.queryXML()
 	receivers := common.NewTranslatorMap[component.Config, component.ID]()
-	receivers.Set(windowseventlog.NewTranslator(t.entry.receiverName, t.entry.channel, t.entry.raw, t.entry.resource))
+	receivers.Set(windowseventlog.NewTranslator(t.entry.receiverName(), t.entry.channel, t.entry.raw(), query, t.entry.resource))
 
 	processors := common.NewTranslatorMap[component.Config, component.ID]()
 
-	// TODO: Replace with upstream Query XML filtering when collector is bumped past v0.124.
-	condition := t.entry.filterCondition()
-	if condition != "" {
-		processors.Set(filterprocessor.NewTranslatorWithLogCondition("windows_events_"+t.entry.name, condition, common.OTTLErrorModeIgnore))
-	}
-
 	if attrs := t.entry.routingAttributes(); len(attrs) > 0 {
 		processors.Set(resourceprocessor.NewTranslator(
-			common.WithName("windows_events_"+t.entry.name),
+			common.WithName("windows_events_"+t.entry.name()),
 			resourceprocessor.WithAttributes(attrs),
 		))
 	}