Merge branch 'main' into feature/windows-eventId-filtering

Author: Paamicky

cmd/config-translator/translator_test.go

@@ -212,13 +212,7 @@ func TestSampleConfigSchema(t *testing.T) {
 			if re.MatchString(file.Name()) {
 				t.Logf("Validating ../../translator/tocwconfig/sampleConfig/%s\n", file.Name())
 
-				// Special case for windows_eventids.json which has both event_ids and event_levels
-				if file.Name() == "windows_eventids.json" {
-					expectedErrorMap := map[string]int{"number_one_of": 1}
-					checkIfSchemaValidateAsExpected(t, "../../translator/tocwconfig/sampleConfig/"+file.Name(), true, expectedErrorMap)
-				} else {
-					checkIfSchemaValidateAsExpected(t, "../../translator/tocwconfig/sampleConfig/"+file.Name(), true, map[string]int{})
-				}
+				checkIfSchemaValidateAsExpected(t, "../../translator/tocwconfig/sampleConfig/"+file.Name(), true, map[string]int{})
 
 				t.Logf("Validated ../../translator/tocwconfig/sampleConfig/%s\n", file.Name())
 			}

plugins/inputs/logfile/tailersrc.go

@@ -47,6 +47,7 @@ func (le LogEvent) Time() time.Time {
 }
 
 func (le LogEvent) Done() {
+	le.RangeQueue().Enqueue(le.Range())
 }
 
 func (le LogEvent) Range() state.Range {

plugins/inputs/logfile/tailersrc_test.go

@@ -156,7 +156,7 @@ func TestTailerSrc(t *testing.T) {
 	assert.Eventually(t, func() bool { return tail.OpenFileCount.Load() <= beforeCount }, 3*time.Second, time.Second)
 }
 
-func TestStatefulLogEvent(t *testing.T) {
+func TestEventDoneCallback(t *testing.T) {
 	original := multilineWaitPeriod
 	defer resetState(original)
 
@@ -217,7 +217,7 @@ func TestStatefulLogEvent(t *testing.T) {
 		}
 		sle, ok := evt.(logs.StatefulLogEvent)
 		assert.True(t, ok)
-		sle.RangeQueue().Enqueue(sle.Range())
+		sle.Done()
 		i++
 		switch i {
 		case 10:

plugins/inputs/windows_event_log/windows_event_log.go

@@ -31,7 +31,7 @@ var startOnlyOnce sync.Once
 type EventConfig struct {
 	Name          string   `toml:"event_name"`
 	Levels        []string `toml:"event_levels"`
-	EventID       []int    `toml:"event_ids"`
+	EventIDs      []int    `toml:"event_ids"`
 	RenderFormat  string   `toml:"event_format"`
 	BatchReadSize int      `toml:"batch_read_size"`
 	LogGroupName  string   `toml:"log_group_name"`
@@ -107,7 +107,7 @@ func (s *Plugin) Start(acc telegraf.Accumulator) error {
 		eventLog := wineventlog.NewEventLog(
 			eventConfig.Name,
 			eventConfig.Levels,
-			eventConfig.EventID,
+			eventConfig.EventIDs,
 			eventConfig.LogGroupName,
 			eventConfig.LogStreamName,
 			eventConfig.RenderFormat,

plugins/inputs/windows_event_log/wineventlog/utils.go

@@ -36,23 +36,23 @@ func createFilterQuery(levels []string, eventIDs []int) string {
 	}
 
 	//EventID filtering
-	var filterEventID string
+	var filterEventIDs string
 	for i, eventID := range eventIDs {
 		if i == 0 {
-			filterEventID = fmt.Sprintf(eventLogeventIDFilter, eventID)
+			filterEventIDs = fmt.Sprintf(eventLogeventIDFilter, eventID)
 		} else {
-			filterEventID = filterEventID + " or " + fmt.Sprintf(eventLogeventIDFilter, eventID)
+			filterEventIDs = filterEventIDs + " or " + fmt.Sprintf(eventLogeventIDFilter, eventID)
 		}
 	}
 
 	//query results
 	var query string
-	if filterLevels != "" && filterEventID != "" {
-		query = "(" + filterLevels + ") and (" + filterEventID + ")"
-	} else if filterLevels != "" && filterEventID == "" {
+	if filterLevels != "" && filterEventIDs != "" {
+		query = "(" + filterLevels + ") and (" + filterEventIDs + ")"
+	} else if filterLevels != "" && filterEventIDs == "" {
 		query = "(" + filterLevels + ")"
-	} else if filterLevels == "" && filterEventID != "" {
-		query = "(" + filterEventID + ")"
+	} else if filterLevels == "" && filterEventIDs != "" {
+		query = "(" + filterEventIDs + ")"
 	}
 
 	//Ignore events older than 2 weeks

plugins/inputs/windows_event_log/wineventlog/wineventlog.go

@@ -49,7 +49,7 @@ func (e *wevtAPIError) Error() string {
 type windowsEventLog struct {
 	name          string
 	levels        []string
-	eventID       []int
+	eventIDs      []int
 	logGroupName  string
 	logStreamName string
 	logGroupClass string
@@ -67,11 +67,11 @@ type windowsEventLog struct {
 	resubscribeCh chan struct{}
 }
 
-func NewEventLog(name string, levels []string, eventID []int, logGroupName, logStreamName, renderFormat, destination string, stateManager state.FileRangeManager, maximumToRead int, retention int, logGroupClass string) *windowsEventLog {
+func NewEventLog(name string, levels []string, eventIDs []int, logGroupName, logStreamName, renderFormat, destination string, stateManager state.FileRangeManager, maximumToRead int, retention int, logGroupClass string) *windowsEventLog {
 	eventLog := &windowsEventLog{
 		name:          name,
 		levels:        levels,
-		eventID:       eventID,
+		eventIDs:      eventIDs,
 		logGroupName:  logGroupName,
 		logStreamName: logStreamName,
 		logGroupClass: logGroupClass,
@@ -212,7 +212,7 @@ func (w *windowsEventLog) open() error {
 	if err != nil {
 		return err
 	}
-	query, err := CreateQuery(w.name, w.levels, w.eventID)
+	query, err := CreateQuery(w.name, w.levels, w.eventIDs)
 	if err != nil {
 		return err
 	}
@@ -256,10 +256,6 @@ func (w *windowsEventLog) SetEventOffset(eventOffset uint64) {
 	w.eventOffset = eventOffset
 }
 
-func (w *windowsEventLog) Done(offset state.Range) {
-	w.stateManager.Enqueue(offset)
-}
-
 func (w *windowsEventLog) ResubscribeCh() chan struct{} {
 	return w.resubscribeCh
 }
@@ -315,6 +311,7 @@ func (le LogEvent) Time() time.Time {
 }
 
 func (le LogEvent) Done() {
+	le.RangeQueue().Enqueue(le.Range())
 }
 
 func (le LogEvent) Range() state.Range {

plugins/inputs/windows_event_log/wineventlog/wineventlog_test.go

@@ -24,7 +24,7 @@ var (
 	NAME = "Application"
 	// 2 is ERROR
 	LEVELS          = []string{"2"}
-	EVENTID         = []int{777}
+	EVENTIDS        = []int{777}
 	GROUP_NAME      = "fake"
 	STREAM_NAME     = "fake"
 	RENDER_FMT      = FormatPlainText
@@ -37,7 +37,7 @@ var (
 
 // TestNewEventLog verifies constructor's default values.
 func TestNewEventLog(t *testing.T) {
-	elog := newTestEventLog(t, NAME, LEVELS, EVENTID)
+	elog := newTestEventLog(t, NAME, LEVELS, EVENTIDS)
 	assert.Equal(t, NAME, elog.name)
 	assert.Equal(t, uint64(0), elog.eventOffset)
 	assert.Zero(t, elog.eventHandle)
@@ -47,18 +47,18 @@ func TestNewEventLog(t *testing.T) {
 // And fails with invalid inputs.
 func TestOpen(t *testing.T) {
 	// Happy path.
-	elog := newTestEventLog(t, NAME, LEVELS, EVENTID)
+	elog := newTestEventLog(t, NAME, LEVELS, EVENTIDS)
 	assert.NoError(t, elog.Open())
 	assert.NotZero(t, elog.eventHandle)
 	assert.NoError(t, elog.Close())
 	// Bad event log source name does not cause Open() to fail.
 	// But eventHandle will be 0 and Close() will fail because of it.
-	elog = newTestEventLog(t, "FakeBadElogName", LEVELS, EVENTID)
+	elog = newTestEventLog(t, "FakeBadElogName", LEVELS, EVENTIDS)
 	assert.NoError(t, elog.Open())
 	assert.Zero(t, elog.eventHandle)
 	assert.Error(t, elog.Close())
 	// bad LEVELS does not cause Open() to fail.
-	elog = newTestEventLog(t, NAME, []string{"498"}, EVENTID)
+	elog = newTestEventLog(t, NAME, []string{"498"}, EVENTIDS)
 	assert.NoError(t, elog.Open())
 	assert.NotZero(t, elog.eventHandle)
 	assert.NoError(t, elog.Close())
@@ -67,7 +67,7 @@ func TestOpen(t *testing.T) {
 	assert.NotZero(t, elog.eventHandle)
 	assert.NoError(t, elog.Close())
 	// bad wlog.eventOffset does not cause Open() to fail.
-	elog = newTestEventLog(t, NAME, []string{"498"}, EVENTID)
+	elog = newTestEventLog(t, NAME, []string{"498"}, EVENTIDS)
 	elog.eventOffset = 9987
 	assert.NoError(t, elog.Open())
 	assert.NotZero(t, elog.eventHandle)
@@ -77,7 +77,7 @@ func TestOpen(t *testing.T) {
 // TestReadGoodSource will verify we can read events written by a registered
 // event log source.
 func TestReadGoodSource(t *testing.T) {
-	elog := newTestEventLog(t, NAME, LEVELS, EVENTID)
+	elog := newTestEventLog(t, NAME, LEVELS, EVENTIDS)
 	assert.NoError(t, elog.Open())
 	seekToEnd(t, elog)
 	writeEvents(t, 10, true, "CWA_UnitTest111", 777)
@@ -89,7 +89,7 @@ func TestReadGoodSource(t *testing.T) {
 // TestReadBadSource will verify that we cannot read events written by an
 // unregistered event log source.
 func TestReadBadSource(t *testing.T) {
-	elog := newTestEventLog(t, NAME, LEVELS, EVENTID)
+	elog := newTestEventLog(t, NAME, LEVELS, EVENTIDS)
 	assert.NoError(t, elog.Open())
 	seekToEnd(t, elog)
 	writeEvents(t, 10, false, "CWA_UnitTest222", 888)
@@ -102,7 +102,7 @@ func TestReadBadSource(t *testing.T) {
 // registered event log source, even if the batch contains events from an
 // unregistered source too.
 func TestReadWithBothSources(t *testing.T) {
-	elog := newTestEventLog(t, NAME, LEVELS, EVENTID)
+	elog := newTestEventLog(t, NAME, LEVELS, EVENTIDS)
 	assert.NoError(t, elog.Open())
 	seekToEnd(t, elog)
 	writeEvents(t, 10, true, "CWA_UnitTest111", 777)
@@ -134,11 +134,8 @@ func seekToEnd(t *testing.T, elog *windowsEventLog) {
 // Fail the test if an error occurs.
 func writeEvents(t *testing.T, msgCount int, doRegister bool, logSrc string, eventId uint32) {
 	if doRegister {
-		err := eventlog.InstallAsEventCreate(logSrc, eventlog.Info|eventlog.Warning|eventlog.Error)
-		if err != nil {
-			t.Logf("Warning: Failed to install event source %s: %v (may need admin privileges)", logSrc, err)
-			// Continue anyway as it might already be registered
-		}
+		// Expected to fail if unit test previously ran and installed the event src.
+		_ = eventlog.InstallAsEventCreate(logSrc, eventlog.Info|eventlog.Warning|eventlog.Error)
 	}
 	wlog, err := eventlog.Open(logSrc)
 	assert.NoError(t, err)
@@ -180,13 +177,13 @@ func checkEvents(t *testing.T, records []*windowsEventLogRecord, substring strin
 	assert.Equal(t, count, found, "expected %v, %v, actual %v", substring, count, found)
 }
 
-func newTestEventLog(t *testing.T, name string, levels []string, eventid []int) *windowsEventLog {
+func newTestEventLog(t *testing.T, name string, levels []string, eventids []int) *windowsEventLog {
 	t.Helper()
 	manager := state.NewFileRangeManager(state.ManagerConfig{
 		StateFileDir:    t.TempDir(),
 		StateFilePrefix: logscommon.WindowsEventLogPrefix,
 		Name:            GROUP_NAME + "_" + STREAM_NAME + "_" + name,
 	})
-	return NewEventLog(name, levels, eventid, GROUP_NAME, STREAM_NAME, RENDER_FMT, DEST,
+	return NewEventLog(name, levels, eventids, GROUP_NAME, STREAM_NAME, RENDER_FMT, DEST,
 		manager, BATCH_SIZE, RETENTION, LOG_GROUP_CLASS)
 }

plugins/outputs/cloudwatchlogs/internal/pusher/batch_test.go

@@ -33,6 +33,14 @@ func newMockEntityProvider(entity *cloudwatchlogs.Entity) *mockEntityProvider {
 	return ep
 }
 
+type mockDoneCallback struct {
+	mock.Mock
+}
+
+func (m *mockDoneCallback) Done() {
+	m.Called()
+}
+
 func TestLogEvent(t *testing.T) {
 	now := time.Now()
 	e := newLogEvent(now, "test message", nil)
@@ -165,10 +173,8 @@ func TestLogEventBatch(t *testing.T) {
 	t.Run("WithStatefulLogEvents", func(t *testing.T) {
 		batch := newLogEventBatch(Target{Group: "G", Stream: "S"}, nil)
 
-		callbackCalled := false
-		callback := func() {
-			callbackCalled = true
-		}
+		mdc1 := &mockDoneCallback{}
+		mdc1.On("Done").Panic("should not be called")
 
 		mrq1 := &mockRangeQueue{}
 		mrq1.On("ID").Return("test")
@@ -178,25 +184,31 @@ func TestLogEventBatch(t *testing.T) {
 		mrq2.On("ID").Return("test2")
 		mrq2.On("Enqueue", state.NewRange(5, 20)).Once()
 
-		event1 := newStatefulLogEvent(time.Now(), "Test", callback, &logEventState{
+		event1 := newStatefulLogEvent(time.Now(), "Test", mdc1.Done, &logEventState{
 			r:     state.NewRange(20, 40),
 			queue: mrq1,
 		})
-		event2 := newStatefulLogEvent(time.Now(), "Test2", callback, &logEventState{
+		event2 := newStatefulLogEvent(time.Now(), "Test2", mdc1.Done, &logEventState{
 			r:     state.NewRange(5, 20),
 			queue: mrq2,
 		})
-		event3 := newStatefulLogEvent(time.Now(), "Test3", callback, &logEventState{
+		event3 := newStatefulLogEvent(time.Now(), "Test3", mdc1.Done, &logEventState{
 			r:     state.NewRange(40, 50),
 			queue: mrq1,
 		})
+
+		mdc2 := &mockDoneCallback{}
+		mdc2.On("Done").Return().Once()
+		event4 := newLogEvent(time.Now(), "Test2", mdc2.Done)
 		batch.append(event1)
 		batch.append(event2)
 		batch.append(event3)
+		batch.append(event4)
 		batch.done()
 
-		mrq1.AssertNumberOfCalls(t, "Enqueue", 1)
-		mrq2.AssertNumberOfCalls(t, "Enqueue", 1)
-		assert.False(t, callbackCalled, "Done callback should not have been called")
+		mrq1.AssertExpectations(t)
+		mrq2.AssertExpectations(t)
+		mdc1.AssertNotCalled(t, "Done")
+		mdc2.AssertExpectations(t)
 	})
 }

translator/defaultKeyCase.go

@@ -74,6 +74,26 @@ func DefaultStringArrayCase(key string, defaultVal, input interface{}) (returnKe
 	return
 }
 
+func DefaultIntegralArrayCase(key string, defaultVal, input interface{}) (string, interface{}) {
+	returnKey, returnVal := DefaultCase(key, defaultVal, input)
+	if arrayVal, ok := returnVal.([]interface{}); ok {
+		intArrayVal := []int{}
+		for _, v := range arrayVal {
+			if floatVal, ok := v.(float64); ok {
+				intVal := int(floatVal)
+				intArrayVal = append(intArrayVal, intVal)
+				
+			}
+		}
+		returnVal = intArrayVal
+	} else {
+		AddErrorMessages(
+			fmt.Sprintf("int array key: %s", key),
+			fmt.Sprintf("%s value (%v) in json is not valid as an array of integers.", key, returnVal))
+	}
+	return returnKey, returnVal
+}
+
 func DefaultRetentionInDaysCase(key string, defaultVal, input interface{}) (returnKey string, returnVal interface{}) {
 	returnKey, returnVal = DefaultIntegralCase(key, defaultVal, input)
 	if intVal, ok := returnVal.(int); ok && IsValidRetentionDays(intVal) {

translator/isValid.go

@@ -96,3 +96,4 @@ func IsValidRetentionDays(days int) bool {
 func IsValidLogGroupClass(class string) bool {
 	return slices.Contains(ValidLogGroupClasses, class) || class == ""
 }
+