newlog: add filter, dedup and counter functions

Newlog will have 3 ways to reduce the amount of logs:
- filter: filter logs out based on the source code line that
produced them
- counter: count the number of logs produced by a specific
source code line. Add that number to the first occurance of the log
and remove the rest
- deduplicator (for errors only): record the last X errors in a sliding
window and remove duplicates

The benchmarking of the newlogd with the new features is in the
dedup_test.go file. It shows that CPU and RAM usage increase by a factor
of 3 when the features are enabled. So they can be disabled by setting
the deduplication window size to 0 and not providing anything to the
filter and counter functions.

Signed-off-by: Paul Gaiduk <paulg@zededa.com>

Author: europaul

pkg/newlog/cmd/counter.go

@@ -0,0 +1,75 @@
+// Copyright (c) 2025 Zededa, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"os"
+	"sync/atomic"
+
+	"github.com/lf-edge/eve-api/go/logs"
+)
+
+var (
+	// logsToCount is a list of Filename fields (src file + line number) of log entries that should be counted.
+	logsToCount atomic.Value
+
+	counterSuppressedLogs = 0
+)
+
+func countLogsInFile(file *os.File) map[string]int {
+	logCounter := make(map[string]int)
+	for _, logSrcLine := range logsToCount.Load().([]string) {
+		logCounter[logSrcLine] = 0
+	}
+	preScanner := bufio.NewScanner(file)
+	for preScanner.Scan() {
+		var logEntry logs.LogEntry
+		// we ignore the errors here, they might be coming from non-json lines like the metadata line
+		_ = json.Unmarshal(preScanner.Bytes(), &logEntry)
+
+		if currentCount, ok := logCounter[logEntry.Filename]; ok {
+			logCounter[logEntry.Filename] = currentCount + 1
+		}
+	}
+	if err := preScanner.Err(); err != nil {
+		log.Errorf("Error scanning file for log occurrence count: %v", err)
+	}
+	if _, err := file.Seek(0, 0); err != nil {
+		log.Errorf("Failed to reset file pointer: %v", err)
+	}
+	return logCounter
+}
+
+// addLogCount updates the log entry with a count tag based on the occurrence count
+// provided in the filterMap, and manages suppression of duplicate log entries.
+// It returns true if the log entry should be included in the output and false if it should be suppressed.
+func addLogCount(logEntry *logs.LogEntry, filterMap map[string]int) bool {
+	if count, ok := filterMap[logEntry.Filename]; !ok {
+		return true
+	} else {
+		if count == 0 {
+			// the count was already included in another entry
+			counterSuppressedLogs++
+			return false
+		}
+
+		if count == 1 {
+			// no need to add additional count field if there is only one occurrence
+			return true
+		}
+
+		if logEntry.Tags == nil {
+			logEntry.Tags = make(map[string]string)
+		}
+		logEntry.Tags["count"] = fmt.Sprint(count)
+
+		// mark the log entry as counted
+		filterMap[logEntry.Filename] = 0
+
+		return true
+	}
+}

pkg/newlog/cmd/dedup.go

@@ -0,0 +1,121 @@
+// Copyright (c) 2025 Zededa, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"container/ring"
+	"encoding/json"
+	"sync/atomic"
+
+	"github.com/lf-edge/eve-api/go/logs"
+)
+
+var dedupWindowSize atomic.Uint32
+
+func init() {
+	dedupWindowSize.Store(100)
+}
+
+type containsMsg struct {
+	Msg string `json:"msg"`
+}
+
+var numDedupedLogs = 0
+
+// deduplicateLogs can be used to deduplicate logs on the fly reading from a channel
+// and writing to another channel
+func deduplicateLogs(in <-chan inputEntry, out chan<- inputEntry) {
+	// 'seen' counts occurrences of each file in the current window.
+	seen := make(map[string]string)
+	// 'queue' holds the file fields of the last bufferSize logs.
+	// TODO changing dedupWindowSize will have no effect
+	// since the ring buffer is created with the initial value of dedupWindowSize
+	// Need to find another solution here
+	queue := ring.New(int(dedupWindowSize.Load()))
+
+	for logEntry := range in {
+		// No queue means no deduplication.
+		if queue == nil {
+			out <- logEntry
+			continue
+		}
+
+		dedupField := ""
+		// If logEntry.content is a valid JSON, extract the field "msg" from it.
+		if logEntry.content != "" {
+			var content containsMsg
+			err := json.Unmarshal([]byte(logEntry.content), &content)
+			if err == nil && content.Msg != "" {
+				dedupField = content.Msg
+			} else {
+				dedupField = logEntry.content
+			}
+		}
+
+		// If the file hasn't appeared in the last bufferSize logs, forward it.
+		if _, ok := seen[dedupField]; ok && logEntry.severity == "error" {
+			log.Tracef("Deduped log at %s because of the log at %s\n", logEntry.timestamp, seen[dedupField])
+			numDedupedLogs++
+		} else {
+			out <- logEntry
+		}
+
+		// Remove the oldest log from the window.
+		if oldest := queue.Value; oldest != nil {
+			delete(seen, oldest.(string))
+		}
+
+		// Add the current log to the window.
+		queue.Value = dedupField
+		seen[dedupField] = logEntry.timestamp
+
+		// Move the window.
+		queue = queue.Next()
+	}
+
+	close(out)
+}
+
+// dedupLogEntry returns a boolean indicating whether the log entry should be used and the updated queue.
+// It can be used to deduplicate logs based on the content of a file
+func dedupLogEntry(logEntry *logs.LogEntry, seen map[string]uint64, queue *ring.Ring) (bool, *ring.Ring) {
+	useEntry := true
+
+	// No queue means no deduplication.
+	if queue == nil {
+		return useEntry, queue
+	}
+
+	dedupField := ""
+	// If logEntry.content is a valid JSON, extract the field "msg" from it.
+	if logEntry.Content != "" {
+		var content containsMsg
+		err := json.Unmarshal([]byte(logEntry.Content), &content)
+		if err == nil && content.Msg != "" {
+			dedupField = content.Msg
+		} else {
+			dedupField = logEntry.Content
+		}
+	}
+
+	// If the file hasn't appeared in the last bufferSize logs, forward it.
+	if _, ok := seen[dedupField]; ok && logEntry.Severity == "error" {
+		useEntry = false
+		log.Tracef("Deduped log id %d because of the log id %d\n", logEntry.Msgid, seen[dedupField])
+		numDedupedLogs++
+	} else {
+		useEntry = true
+	}
+
+	// Remove the oldest log from the window.
+	if oldest := queue.Value; oldest != nil {
+		delete(seen, oldest.(string))
+	}
+
+	// Add the current log to the window.
+	queue.Value = dedupField
+	seen[dedupField] = logEntry.Msgid
+
+	return useEntry, queue.Next()
+}