RSDK-9345 Deduplicate noisy logs

config/config.go

@@ -71,6 +71,10 @@ type Config struct {
 	// Revision contains the current revision of the config.
 	Revision string
 
+	// DisableLogDeduplication controls whether deduplication of noisy logs
+	// should be turned off. Defaults to false.
+	DisableLogDeduplication bool
+
 	// toCache stores the JSON marshalled version of the config to be cached. It should be a copy of
 	// the config pulled from cloud with minor changes.
 	// This version is kept because the config is changed as it moves through the system.
@@ -86,21 +90,22 @@ type MaintenanceConfig struct {
 
 // NOTE: This data must be maintained with what is in Config.
 type configData struct {
-	Cloud               *Cloud                        `json:"cloud,omitempty"`
-	Modules             []Module                      `json:"modules,omitempty"`
-	Remotes             []Remote                      `json:"remotes,omitempty"`
-	Components          []resource.Config             `json:"components,omitempty"`
-	Processes           []pexec.ProcessConfig         `json:"processes,omitempty"`
-	Services            []resource.Config             `json:"services,omitempty"`
-	Packages            []PackageConfig               `json:"packages,omitempty"`
-	Network             NetworkConfig                 `json:"network"`
-	Auth                AuthConfig                    `json:"auth"`
-	Debug               bool                          `json:"debug,omitempty"`
-	DisablePartialStart bool                          `json:"disable_partial_start"`
-	EnableWebProfile    bool                          `json:"enable_web_profile"`
-	LogConfig           []logging.LoggerPatternConfig `json:"log,omitempty"`
-	Revision            string                        `json:"revision,omitempty"`
-	MaintenanceConfig   *MaintenanceConfig            `json:"maintenance,omitempty"`
+	Cloud                   *Cloud                        `json:"cloud,omitempty"`
+	Modules                 []Module                      `json:"modules,omitempty"`
+	Remotes                 []Remote                      `json:"remotes,omitempty"`
+	Components              []resource.Config             `json:"components,omitempty"`
+	Processes               []pexec.ProcessConfig         `json:"processes,omitempty"`
+	Services                []resource.Config             `json:"services,omitempty"`
+	Packages                []PackageConfig               `json:"packages,omitempty"`
+	Network                 NetworkConfig                 `json:"network"`
+	Auth                    AuthConfig                    `json:"auth"`
+	Debug                   bool                          `json:"debug,omitempty"`
+	DisablePartialStart     bool                          `json:"disable_partial_start"`
+	EnableWebProfile        bool                          `json:"enable_web_profile"`
+	LogConfig               []logging.LoggerPatternConfig `json:"log,omitempty"`
+	Revision                string                        `json:"revision,omitempty"`
+	MaintenanceConfig       *MaintenanceConfig            `json:"maintenance,omitempty"`
+	DisableLogDeduplication bool                          `json:"disable_log_deduplication"`
 }
 
 // AppValidationStatus refers to the.

config/proto_conversions.go

@@ -106,6 +106,10 @@ func FromProto(proto *pb.RobotConfig, logger logging.Logger) (*Config, error) {
 		cfg.MaintenanceConfig = maintenanceConfig
 	}
 
+	if proto.DisableLogDeduplication {
+		cfg.DisableLogDeduplication = true
+	}
+
 	return &cfg, nil
 }
 

go.mod

@@ -75,7 +75,7 @@ require (
 	go.uber.org/atomic v1.11.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
-	go.viam.com/api v0.1.372
+	go.viam.com/api v0.1.373
 	go.viam.com/test v1.2.4
 	go.viam.com/utils v0.1.118
 	goji.io v2.0.2+incompatible

go.sum

@@ -1513,8 +1513,8 @@ go.uber.org/zap v1.18.1/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
 go.uber.org/zap v1.23.0/go.mod h1:D+nX8jyLsMHMYrln8A0rJjFt/T/9/bGgIhAqxv5URuY=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
-go.viam.com/api v0.1.372 h1:Al9P7yojBDdNVAF7nrr5BAbzCvb+vrSp8N7BitbV0mQ=
-go.viam.com/api v0.1.372/go.mod h1:g5eipXHNm0rQmW7DWya6avKcmzoypLmxnMlAaIsE5Ls=
+go.viam.com/api v0.1.373 h1:1vpxO9thQFeEOJhvGWcjHAHyBH5EKHWKTaJahAo5b/A=
+go.viam.com/api v0.1.373/go.mod h1:g5eipXHNm0rQmW7DWya6avKcmzoypLmxnMlAaIsE5Ls=
 go.viam.com/test v1.2.4 h1:JYgZhsuGAQ8sL9jWkziAXN9VJJiKbjoi9BsO33TW3ug=
 go.viam.com/test v1.2.4/go.mod h1:zI2xzosHdqXAJ/kFqcN+OIF78kQuTV2nIhGZ8EzvaJI=
 go.viam.com/utils v0.1.118 h1:Kp6ebrCBiYReeSC1XnWPTjtBJoTUsQ6YWAomQkQF/mE=

logging/impl.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"os"
 	"runtime"
+	"sync"
 	"testing"
 	"time"
 
@@ -15,6 +16,14 @@ import (
 	"go.uber.org/zap/zaptest"
 )
 
+var (
+	// Window duration over which to consider log messages "noisy.".
+	noisyMessageWindowDuration = 10 * time.Second
+	// Count threshold within `noisyMessageWindowDuration` after which to
+	// consider log messages "noisy.".
+	noisyMessageCountThreshold = 3
+)
+
 type (
 	impl struct {
 		name  string
@@ -26,6 +35,15 @@ type (
 		// avoid that. This function is a no-op for non-test loggers. See `NewTestAppender`
 		// documentation for more details.
 		testHelper func()
+
+		// recentMessageMu guards the recentMessage fields below.
+		recentMessageMu sync.Mutex
+		// Map of messages to counts of that message being `Write`ten within window.
+		recentMessageCounts map[string]int
+		// Map of messages to last `LogEntry` with that message within window.
+		recentMessageEntries map[string]LogEntry
+		// Start of current window.
+		recentMessageWindowStart time.Time
 	}
 
 	// LogEntry embeds a zapcore Entry and slice of Fields.
@@ -41,6 +59,32 @@ type (
 	}
 )
 
+// String stringifies a `LogEntry`. Should be used to emplace a log entry in
+// `recentMessageEntries`, i.e. `LogEntry`s that `String`ify identically should
+// be treated as identical with respect to noisiness and deduplication. Should
+// not be used for actual writing of the entry to an appender.
+func (le *LogEntry) String() string {
+	ret := le.Message
+	for _, field := range le.Fields {
+		ret += " " + field.Key + " "
+
+		// Assume field's value is held in one of `Integer`, `Interface`, or
+		// `String`. Otherwise (field has no value or is equivalent to 0 or "") use
+		// the string "undefined".
+		switch {
+		case field.Integer != 0:
+			ret += fmt.Sprintf("%d", field.Integer)
+		case field.Interface != nil:
+			ret += fmt.Sprintf("%v", field.Interface)
+		case field.String != "":
+			ret += field.String
+		default:
+			ret += "undefined"
+		}
+	}
+	return ret
+}
+
 func (imp *impl) NewLogEntry() *LogEntry {
 	ret := &LogEntry{}
 	ret.Time = time.Now()
@@ -84,6 +128,10 @@ func (imp *impl) Sublogger(subname string) Logger {
 		imp.appenders,
 		imp.registry,
 		imp.testHelper,
+		sync.Mutex{},
+		make(map[string]int),
+		make(map[string]LogEntry),
+		time.Now(),
 	}
 
 	// If there are multiple callers racing to create the same logger name (e.g: `viam.networking`),
@@ -198,6 +246,46 @@ func (imp *impl) shouldLog(logLevel Level) bool {
 }
 
 func (imp *impl) Write(entry *LogEntry) {
+	if !DisableLogDeduplication.Load() {
+		// If we have entered a new recentMessage window, output noisy logs from
+		// the last window.
+		imp.recentMessageMu.Lock()
+		if time.Since(imp.recentMessageWindowStart) > noisyMessageWindowDuration {
+			for stringifiedEntry, count := range imp.recentMessageCounts {
+				if count > noisyMessageCountThreshold {
+					collapsedEntry := imp.recentMessageEntries[stringifiedEntry]
+					collapsedEntry.Message = fmt.Sprintf("Message logged %d times in past %v: %s",
+						count, noisyMessageWindowDuration, collapsedEntry.Message)
+
+					imp.testHelper()
+					for _, appender := range imp.appenders {
+						err := appender.Write(collapsedEntry.Entry, collapsedEntry.Fields)
+						if err != nil {
+							fmt.Fprint(os.Stderr, err)
+						}
+					}
+				}
+			}
+
+			// Clear maps and reset window.
+			clear(imp.recentMessageCounts)
+			clear(imp.recentMessageEntries)
+			imp.recentMessageWindowStart = time.Now()
+		}
+
+		// Track entry in recentMessage maps.
+		stringifiedEntry := entry.String()
+		imp.recentMessageCounts[stringifiedEntry]++
+		imp.recentMessageEntries[stringifiedEntry] = *entry
+
+		if imp.recentMessageCounts[stringifiedEntry] > noisyMessageCountThreshold {
+			// If entry's message is reportedly "noisy," return early.
+			imp.recentMessageMu.Unlock()
+			return
+		}
+		imp.recentMessageMu.Unlock()
+	}
+
 	imp.testHelper()
 	for _, appender := range imp.appenders {
 		err := appender.Write(entry.Entry, entry.Fields)