MM-22786 enterprise metrics for logging (#15313)

Adds metrics for logging engine which are collected by Prometheus.
- current queue level(s)
- rate of logging records emitted
- rate of logging errors

vendor/github.com/mattermost/logr/go.mod

@@ -4,6 +4,7 @@ go 1.12
 
 require (
 	github.com/francoispqt/gojay v1.2.13
+	github.com/stretchr/testify v1.2.2
 	github.com/wiggin77/cfg v1.0.2
 	github.com/wiggin77/merror v1.0.2
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0

vendor/github.com/mattermost/logr/logr.go

@@ -27,6 +27,13 @@ type Logr struct {
 	shutdown           bool
 	lvlCache           levelCache
 
+	metricsOnce    sync.Once
+	metricsDone    chan struct{}
+	metrics        MetricsCollector
+	queueSizeGauge Gauge
+	loggedCounter  Counter
+	errorCounter   Counter
+
 	bufferPool sync.Pool
 
 	// MaxQueueSize is the maximum number of log records that can be queued.
@@ -95,6 +102,10 @@ type Logr struct {
 
 	// DisableBufferPool when true disables the buffer pool. See MaxPooledBuffer.
 	DisableBufferPool bool
+
+	// MetricsUpdateFreqMillis determines how often polled metrics are updated
+	// when metrics are enabled.
+	MetricsUpdateFreqMillis int64
 }
 
 // Configure adds/removes targets via the supplied `Config`.
@@ -117,6 +128,13 @@ func (logr *Logr) AddTarget(target Target) error {
 	defer logr.tmux.Unlock()
 	logr.targets = append(logr.targets, target)
 
+	var err error
+	if logr.metrics != nil {
+		if tm, ok := target.(TargetWithMetrics); ok {
+			err = tm.EnableMetrics(logr.metrics, logr.MetricsUpdateFreqMillis)
+		}
+	}
+
 	logr.once.Do(func() {
 		logr.maxQueueSizeActual = logr.MaxQueueSize
 		if logr.maxQueueSizeActual == 0 {
@@ -144,7 +162,7 @@ func (logr *Logr) AddTarget(target Target) error {
 		go logr.start()
 	})
 	logr.resetLevelCache()
-	return nil
+	return err
 }
 
 // NewLogger creates a Logger using defaults. A `Logger` is light-weight
@@ -201,6 +219,13 @@ func (logr *Logr) IsLevelEnabled(lvl Level) LevelStatus {
 	return status
 }
 
+// HasTargets returns true only if at least one target exists within the Logr.
+func (logr *Logr) HasTargets() bool {
+	logr.tmux.RLock()
+	defer logr.tmux.RUnlock()
+	return len(logr.targets) > 0
+}
+
 // ResetLevelCache resets the cached results of `IsLevelEnabled`. This is
 // called any time a Target is added or a target's level is changed.
 func (logr *Logr) ResetLevelCache() {
@@ -279,6 +304,10 @@ func (logr *Logr) panic(err interface{}) {
 // timing out. Use `IsTimeoutError` to determine if the returned error is
 // due to a timeout.
 func (logr *Logr) Flush() error {
+	if !logr.HasTargets() {
+		return nil
+	}
+
 	logr.mux.Lock()
 	defer logr.mux.Unlock()
 
@@ -310,6 +339,10 @@ func (logr *Logr) Shutdown() error {
 	}
 	logr.shutdown = true
 	logr.resetLevelCache()
+	if logr.metricsDone != nil {
+		close(logr.metricsDone)
+		logr.metricsDone = nil
+	}
 	logr.mux.Unlock()
 
 	errs := merror.New()
@@ -344,6 +377,9 @@ func (logr *Logr) Shutdown() error {
 // If `OnLoggerError` is not nil, it is called with the error, otherwise the error is
 // output to `os.Stderr`.
 func (logr *Logr) ReportError(err interface{}) {
+	if logr.errorCounter != nil {
+		logr.errorCounter.Inc()
+	}
 	if logr.OnLoggerError == nil {
 		fmt.Fprintln(os.Stderr, err)
 		return
@@ -415,6 +451,29 @@ func (logr *Logr) start() {
 	close(logr.done)
 }
 
+// startMetricsUpdater updates the metrics for any polled values every `MetricsUpdateFreqSecs` seconds until
+// logr is closed.
+func (logr *Logr) startMetricsUpdater() {
+	for {
+		updateFreq := logr.MetricsUpdateFreqMillis
+		if updateFreq == 0 {
+			updateFreq = DefMetricsUpdateFreqMillis
+		}
+		if updateFreq < 250 {
+			updateFreq = 250 // don't peg the CPU
+		}
+
+		select {
+		case <-logr.metricsDone:
+			return
+		case <-time.After(time.Duration(updateFreq) * time.Millisecond):
+			if logr.queueSizeGauge != nil {
+				logr.queueSizeGauge.Set(float64(len(logr.in)))
+			}
+		}
+	}
+}
+
 // fanout pushes a LogRec to all targets.
 func (logr *Logr) fanout(rec *LogRec) {
 	var target Target
@@ -424,13 +483,20 @@ func (logr *Logr) fanout(rec *LogRec) {
 		}
 	}()
 
+	var logged bool
+
 	logr.tmux.RLock()
 	defer logr.tmux.RUnlock()
 	for _, target = range logr.targets {
 		if enabled, _ := target.IsLevelEnabled(rec.Level()); enabled {
 			target.Log(rec)
+			logged = true
 		}
 	}
+
+	if logged && logr.loggedCounter != nil {
+		logr.loggedCounter.Inc()
+	}
 }
 
 // flush drains the queue and notifies when done.

vendor/github.com/mattermost/logr/metrics.go

@@ -0,0 +1,85 @@
+package logr
+
+import (
+	"errors"
+
+	"github.com/wiggin77/merror"
+)
+
+const (
+	DefMetricsUpdateFreqMillis = 15000 // 15 seconds
+)
+
+// Counter is a simple metrics sink that can only increment a value.
+// Implementations are external to Logr and provided via `MetricsCollector`.
+type Counter interface {
+	// Inc increments the counter by 1. Use Add to increment it by arbitrary non-negative values.
+	Inc()
+	// Add adds the given value to the counter. It panics if the value is < 0.
+	Add(float64)
+}
+
+// Gauge is a simple metrics sink that can receive values and increase or decrease.
+// Implementations are external to Logr and provided via `MetricsCollector`.
+type Gauge interface {
+	// Set sets the Gauge to an arbitrary value.
+	Set(float64)
+	// Add adds the given value to the Gauge. (The value can be negative, resulting in a decrease of the Gauge.)
+	Add(float64)
+	// Sub subtracts the given value from the Gauge. (The value can be negative, resulting in an increase of the Gauge.)
+	Sub(float64)
+}
+
+// MetricsCollector provides a way for users of this Logr package to have metrics pushed
+// in an efficient way to any backend, e.g. Prometheus.
+// For each target added to Logr, the supplied MetricsCollector will provide a Gauge
+// and Counters that will be called frequently as logging occurs.
+type MetricsCollector interface {
+	// QueueSizeGauge returns a Gauge that will be updated by the named target.
+	QueueSizeGauge(target string) (Gauge, error)
+	// LoggedCounter returns a Counter that will be incremented by the named target.
+	LoggedCounter(target string) (Counter, error)
+	// ErrorCounter returns a Counter that will be incremented by the named target.
+	ErrorCounter(target string) (Counter, error)
+	// DroppedCounter returns a Counter that will be incremented by the named target.
+	DroppedCounter(target string) (Counter, error)
+	// BlockedCounter returns a Counter that will be incremented by the named target.
+	BlockedCounter(target string) (Counter, error)
+}
+
+// TargetWithMetrics is a target that provides metrics.
+type TargetWithMetrics interface {
+	EnableMetrics(collector MetricsCollector, updateFreqMillis int64) error
+}
+
+// SetMetricsCollector enables metrics collection by supplying a MetricsCollector.
+// The MetricsCollector provides counters and gauges that are updated by log targets.
+func (logr *Logr) SetMetricsCollector(collector MetricsCollector) error {
+	if collector == nil {
+		return errors.New("collector cannot be nil")
+	}
+
+	logr.metrics = collector
+	logr.queueSizeGauge, _ = collector.QueueSizeGauge("_logr")
+	logr.loggedCounter, _ = collector.LoggedCounter("_logr")
+	logr.errorCounter, _ = collector.ErrorCounter("_logr")
+
+	logr.metricsOnce.Do(func() {
+		logr.metricsDone = make(chan struct{})
+		go logr.startMetricsUpdater()
+	})
+
+	merr := merror.New()
+
+	logr.tmux.RLock()
+	defer logr.tmux.RUnlock()
+	for _, target := range logr.targets {
+		if tm, ok := target.(TargetWithMetrics); ok {
+			if err := tm.EnableMetrics(logr.metrics, logr.MetricsUpdateFreqMillis); err != nil {
+				merr.Append(err)
+			}
+		}
+
+	}
+	return merr.ErrorOrNil()
+}

vendor/github.com/mattermost/logr/target.go

@@ -10,6 +10,9 @@ import (
 // Target represents a destination for log records such as file,
 // database, TCP socket, etc.
 type Target interface {
+	// SetName provides an option name for the target.
+	SetName(name string)
+
 	// IsLevelEnabled returns true if this target should emit
 	// logs for the specified level. Also determines if
 	// a stack trace is required.
@@ -33,9 +36,10 @@ type RecordWriter interface {
 
 // Basic provides the basic functionality of a Target that can be used
 // to more easily compose your own Targets. To use, just embed Basic
-// in your target type, implement `RecordWriter`, and call `Start`.
+// in your target type, implement `RecordWriter`, and call `(*Basic).Start`.
 type Basic struct {
 	target Target
+	name   string
 
 	filter    Filter
 	formatter Formatter
@@ -43,6 +47,14 @@ type Basic struct {
 	in   chan *LogRec
 	done chan struct{}
 	w    RecordWriter
+
+	queueSizeGauge Gauge
+	loggedCounter  Counter
+	errorCounter   Counter
+	droppedCounter Counter
+	blockedCounter Counter
+
+	metricsUpdateFreqMillis int64
 }
 
 // Start initializes this target helper and starts accepting log records for processing.
@@ -61,6 +73,14 @@ func (b *Basic) Start(target Target, rw RecordWriter, filter Filter, formatter F
 	b.done = make(chan struct{}, 1)
 	b.w = rw
 	go b.start()
+
+	if b.queueSizeGauge != nil {
+		go b.startMetricsUpdater()
+	}
+}
+
+func (b *Basic) SetName(name string) {
+	b.name = name
 }
 
 // IsLevelEnabled returns true if this target should emit
@@ -97,8 +117,15 @@ func (b *Basic) Log(rec *LogRec) {
 	default:
 		handler := lgr.OnTargetQueueFull
 		if handler != nil && handler(b.target, rec, cap(b.in)) {
+			if b.droppedCounter != nil {
+				b.droppedCounter.Inc()
+			}
 			return // drop the record
 		}
+		if b.blockedCounter != nil {
+			b.blockedCounter.Inc()
+		}
+
 		select {
 		case <-time.After(lgr.enqueueTimeout()):
 			lgr.ReportError(fmt.Errorf("target enqueue timeout for log rec [%v]", rec))
@@ -107,6 +134,39 @@ func (b *Basic) Log(rec *LogRec) {
 	}
 }
 
+// Metrics enables metrics collection using the provided MetricsCollector.
+func (b *Basic) EnableMetrics(collector MetricsCollector, updateFreqMillis int64) error {
+	b.metricsUpdateFreqMillis = updateFreqMillis
+
+	name := fmt.Sprintf("%v", b)
+	var err error
+
+	if b.queueSizeGauge, err = collector.QueueSizeGauge(name); err != nil {
+		return err
+	}
+	if b.loggedCounter, err = collector.LoggedCounter(name); err != nil {
+		return err
+	}
+	if b.errorCounter, err = collector.ErrorCounter(name); err != nil {
+		return err
+	}
+	if b.droppedCounter, err = collector.DroppedCounter(name); err != nil {
+		return err
+	}
+	if b.blockedCounter, err = collector.BlockedCounter(name); err != nil {
+		return err
+	}
+	return nil
+}
+
+// String returns a name for this target. Use `SetName` to specify a name.
+func (b *Basic) String() string {
+	if b.name != "" {
+		return b.name
+	}
+	return fmt.Sprintf("%T", b.target)
+}
+
 // Start accepts log records via In channel and writes to the
 // supplied writer, until Done channel signaled.
 func (b *Basic) start() {
@@ -123,13 +183,41 @@ func (b *Basic) start() {
 		} else {
 			err := b.w.Write(rec)
 			if err != nil {
+				if b.errorCounter != nil {
+					b.errorCounter.Inc()
+				}
 				rec.Logger().Logr().ReportError(err)
+			} else if b.loggedCounter != nil {
+				b.loggedCounter.Inc()
 			}
 		}
 	}
 	close(b.done)
 }
 
+// startMetricsUpdater updates the metrics for any polled values every `MetricsUpdateFreqSecs` seconds until
+// target is closed.
+func (b *Basic) startMetricsUpdater() {
+	for {
+		updateFreq := b.metricsUpdateFreqMillis
+		if updateFreq == 0 {
+			updateFreq = DefMetricsUpdateFreqMillis
+		}
+		if updateFreq < 250 {
+			updateFreq = 250 // don't peg the CPU
+		}
+
+		select {
+		case <-b.done:
+			return
+		case <-time.After(time.Duration(updateFreq) * time.Millisecond):
+			if b.queueSizeGauge != nil {
+				b.queueSizeGauge.Set(float64(len(b.in)))
+			}
+		}
+	}
+}
+
 // flush drains the queue and notifies when done.
 func (b *Basic) flush(done chan<- struct{}) {
 	for {
@@ -141,6 +229,9 @@ func (b *Basic) flush(done chan<- struct{}) {
 			if rec.flush == nil {
 				err = b.w.Write(rec)
 				if err != nil {
+					if b.errorCounter != nil {
+						b.errorCounter.Inc()
+					}
 					rec.Logger().Logr().ReportError(err)
 				}
 			}

vendor/github.com/mattermost/logr/target/file.go

@@ -85,8 +85,3 @@ func (f *File) Shutdown(ctx context.Context) error {
 
 	return errs.ErrorOrNil()
 }
-
-// String returns a string representation of this target.
-func (f *File) String() string {
-	return "FileTarget"
-}

vendor/github.com/mattermost/logr/target/syslog.go

@@ -87,8 +87,3 @@ func (s *Syslog) Write(rec *logr.LogRec) error {
 	}
 	return err
 }
-
-// String returns a string representation of this target.
-func (s *Syslog) String() string {
-	return "SyslogTarget"
-}

vendor/github.com/mattermost/logr/target/writer.go

@@ -38,8 +38,3 @@ func (w *Writer) Write(rec *logr.LogRec) error {
 	_, err = w.out.Write(buf.Bytes())
 	return err
 }
-
-// String returns a string representation of this target.
-func (w *Writer) String() string {
-	return "WriterTarget"
-}