Bufferize custom metrics

fd/file.d.go

@@ -63,7 +63,7 @@ func (f *FileD) Start() {
 }
 
 func (f *FileD) initMetrics() {
-	f.metricCtl = metric.NewCtl("file_d", f.registry, 0, 0)
+	f.metricCtl = metric.NewCtl("file_d", f.registry, 0, 0, 0)
 	f.versionMetric = f.metricCtl.RegisterGaugeVec("version", "", "version")
 	f.versionMetric.WithLabelValues(buildinfo.Version).Inc()
 }

fd/util.go

@@ -36,6 +36,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 	var antispamRules antispam.Rules
 
 	metricHoldDuration := pipeline.DefaultMetricHoldDuration
+	metricFlushInterval := pipeline.DefaultMetricFlushInterval
 	metricMaxLabelValueLength := pipeline.DefaultMetricMaxLabelValueLength
 
 	if settings != nil {
@@ -146,6 +147,15 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 			metricHoldDuration = i
 		}
 
+		str = metrics.Get("flush_interval").MustString()
+		if str != "" {
+			i, err := time.ParseDuration(str)
+			if err != nil {
+				logger.Fatalf("can't parse pipeline metric hold duration: %s", err.Error())
+			}
+			metricFlushInterval = i
+		}
+
 		metricMaxLabelValueLength = metrics.Get("max_label_value_length").MustInt()
 		if metricMaxLabelValueLength < 0 {
 			logger.Warn("negative max_label_value_length value, metric label truncation is disabled")
@@ -176,6 +186,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
 		Pool:                pipeline.PoolType(pool),
 		Metric: &pipeline.MetricSettings{
 			HoldDuration:        metricHoldDuration,
+			FlushInterval:       metricFlushInterval,
 			MaxLabelValueLength: metricMaxLabelValueLength,
 		},
 	}

metric/buffer.go

@@ -0,0 +1,87 @@
+package metric
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type counterBuffer struct {
+	mu    sync.Mutex
+	delta float64
+}
+
+func (b *counterBuffer) add(v float64) {
+	b.mu.Lock()
+	b.delta += v
+	b.mu.Unlock()
+}
+
+func (b *counterBuffer) flush(c prometheus.Counter) {
+	b.mu.Lock()
+	d := b.delta
+	b.delta = 0
+	b.mu.Unlock()
+	if d > 0 {
+		c.Add(d)
+	}
+}
+
+type gaugeBuffer struct {
+	mu       sync.Mutex
+	hasSet   bool
+	setValue float64
+	delta    float64
+}
+
+func (b *gaugeBuffer) set(v float64) {
+	b.mu.Lock()
+	b.hasSet, b.setValue, b.delta = true, v, 0
+	b.mu.Unlock()
+}
+
+func (b *gaugeBuffer) add(v float64) {
+	b.mu.Lock()
+	b.delta += v
+	b.mu.Unlock()
+}
+
+func (b *gaugeBuffer) flush(g prometheus.Gauge) {
+	b.mu.Lock()
+	hasSet, setValue, delta := b.hasSet, b.setValue, b.delta
+	b.hasSet, b.setValue, b.delta = false, 0, 0
+	b.mu.Unlock()
+
+	switch {
+	case hasSet:
+		g.Set(setValue + delta)
+	case delta != 0:
+		g.Add(delta)
+	}
+}
+
+type histogramBuffer struct {
+	mu      sync.Mutex
+	pending []float64
+}
+
+func (b *histogramBuffer) observe(v float64) {
+	b.mu.Lock()
+	b.pending = append(b.pending, v)
+	b.mu.Unlock()
+}
+
+func (b *histogramBuffer) flush(h prometheus.Histogram) {
+	b.mu.Lock()
+	if len(b.pending) == 0 {
+		b.mu.Unlock()
+		return
+	}
+	obs := b.pending
+	b.pending = make([]float64, 0, len(obs))
+	b.mu.Unlock()
+
+	for _, v := range obs {
+		h.Observe(v)
+	}
+}

metric/controller.go

@@ -27,35 +27,53 @@ type Ctl struct {
 	mu                        sync.RWMutex
 }
 
-func NewCtl(subsystem string, registry *prometheus.Registry, metricHoldDuration time.Duration, metricMaxLabelValueLength int) *Ctl {
+func NewCtl(
+	subsystem string,
+	registry *prometheus.Registry,
+	metricHoldDuration time.Duration,
+	metricFlushInterval time.Duration,
+	metricMaxLabelValueLength int,
+) *Ctl {
 	ctl := &Ctl{
 		subsystem:                 subsystem,
 		register:                  registry,
 		metrics:                   make(map[string]prometheus.Collector),
 		metricMaxLabelValueLength: metricMaxLabelValueLength,
 	}
-
 	if metricHoldDuration != 0 {
-		ctl.holder = NewHolder(metricHoldDuration)
+		ctl.holder = NewHolder(metricHoldDuration, metricFlushInterval)
 	}
-
 	return ctl
 }
 
-func (mc *Ctl) Maintenance() {
-	if mc.holder == nil {
-		return
+func (mc *Ctl) Start() {
+	if mc.holder != nil {
+		mc.holder.start()
 	}
+}
 
-	mc.holder.maintenance()
+func (mc *Ctl) Stop() {
+	if mc.holder != nil {
+		mc.holder.stop()
+	}
 }
 
-func (mc *Ctl) AddToHolder(mv heldMetricVec) {
-	if mc.holder == nil {
-		return
+func (mc *Ctl) Flush() {
+	if mc.holder != nil {
+		mc.holder.flush()
+	}
+}
+
+func (mc *Ctl) Maintenance() {
+	if mc.holder != nil {
+		mc.holder.maintenance()
 	}
+}
 
-	mc.holder.addMetricVec(mv)
+func (mc *Ctl) AddToHolder(mv heldMetricVec) {
+	if mc.holder != nil {
+		mc.holder.addMetricVec(mv)
+	}
 }
 
 func (mc *Ctl) RegisterCounter(name, help string) *Counter {
@@ -142,6 +160,5 @@ func (mc *Ctl) registerMetric(name string, newMetric prometheus.Collector) prome
 		mc.metrics[name] = metric
 		mc.register.MustRegister(metric)
 	}
-
 	return metric
 }

metric/counter.go

@@ -18,12 +18,20 @@ func newCounter(c prometheus.Counter) *Counter {
 }
 
 func (c *Counter) Inc() {
-	c.metric.Inc()
+	if buf := c.buffer.Load(); buf != nil {
+		buf.(*counterBuffer).add(1)
+	} else {
+		c.metric.Inc()
+	}
 	c.updateUsage()
 }
 
 func (c *Counter) Add(v float64) {
-	c.metric.Add(v)
+	if buf := c.buffer.Load(); buf != nil {
+		buf.(*counterBuffer).add(v)
+	} else {
+		c.metric.Add(v)
+	}
 	c.updateUsage()
 }
 
@@ -44,6 +52,12 @@ func newCounterVec(cv *prometheus.CounterVec, maxLabelValueLength int) *CounterV
 	}
 }
 
+func (cv *CounterVec) enableBuffering() {
+	cv.store.enableBuffering(func() any {
+		return &counterBuffer{}
+	})
+}
+
 func (cv *CounterVec) WithLabelValues(lvs ...string) *Counter {
 	return &Counter{
 		heldMetric: cv.store.GetOrCreate(lvs, cv.vec.WithLabelValues),
@@ -57,3 +71,9 @@ func (cv *CounterVec) DeleteLabelValues(lvs ...string) bool {
 func (cv *CounterVec) DeleteOldMetrics(holdDuration time.Duration) {
 	cv.store.DeleteOldMetrics(holdDuration, cv.vec)
 }
+
+func (cv *CounterVec) flush() {
+	cv.store.flushAll(func(hm *heldMetric[prometheus.Counter]) {
+		hm.buffer.Load().(*counterBuffer).flush(hm.metric)
+	})
+}

metric/gauge.go

@@ -11,40 +11,60 @@ type Gauge struct {
 	*heldMetric[prometheus.Gauge]
 }
 
-func newGauge(c prometheus.Gauge) *Gauge {
+func newGauge(g prometheus.Gauge) *Gauge {
 	return &Gauge{
-		heldMetric: newHeldMetric(nil, c),
+		heldMetric: newHeldMetric(nil, g),
 	}
 }
 
 func (g *Gauge) Set(v float64) {
-	g.metric.Set(v)
+	if buf := g.buffer.Load(); buf != nil {
+		buf.(*gaugeBuffer).set(v)
+	} else {
+		g.metric.Set(v)
+	}
 	g.updateUsage()
 }
 
 func (g *Gauge) Inc() {
-	g.metric.Inc()
+	if buf := g.buffer.Load(); buf != nil {
+		buf.(*gaugeBuffer).add(1)
+	} else {
+		g.metric.Inc()
+	}
 	g.updateUsage()
 }
 
 func (g *Gauge) Dec() {
-	g.metric.Dec()
+	if buf := g.buffer.Load(); buf != nil {
+		buf.(*gaugeBuffer).add(-1)
+	} else {
+		g.metric.Dec()
+	}
 	g.updateUsage()
 }
 
 func (g *Gauge) Add(v float64) {
-	g.metric.Add(v)
+	if buf := g.buffer.Load(); buf != nil {
+		buf.(*gaugeBuffer).add(v)
+	} else {
+		g.metric.Add(v)
+	}
 	g.updateUsage()
 }
 
 func (g *Gauge) Sub(v float64) {
-	g.metric.Sub(v)
+	if buf := g.buffer.Load(); buf != nil {
+		buf.(*gaugeBuffer).add(-v)
+	} else {
+		g.metric.Sub(v)
+	}
 	g.updateUsage()
 }
 
 // should only be used in tests
-func (c *Gauge) ToFloat64() float64 {
-	return testutil.ToFloat64(c.metric)
+func (g *Gauge) ToFloat64() float64 {
+	return testutil.ToFloat64(g.metric)
 }
 
 type GaugeVec struct {
@@ -59,16 +79,26 @@ func newGaugeVec(gv *prometheus.GaugeVec, maxLabelValueLength int) *GaugeVec {
 	}
 }
 
+func (gv *GaugeVec) enableBuffering() {
+	gv.store.enableBuffering(func() any { return &gaugeBuffer{} })
+}
+
 func (gv *GaugeVec) WithLabelValues(lvs ...string) *Gauge {
 	return &Gauge{
 		heldMetric: gv.store.GetOrCreate(lvs, gv.vec.WithLabelValues),
 	}
 }
 
-func (cv *GaugeVec) DeleteLabelValues(lvs ...string) bool {
-	return cv.store.Delete(lvs, cv.vec)
+func (gv *GaugeVec) DeleteLabelValues(lvs ...string) bool {
+	return gv.store.Delete(lvs, gv.vec)
 }
 
 func (gv *GaugeVec) DeleteOldMetrics(holdDuration time.Duration) {
 	gv.store.DeleteOldMetrics(holdDuration, gv.vec)
 }
+
+func (gv *GaugeVec) flush() {
+	gv.store.flushAll(func(hm *heldMetric[prometheus.Gauge]) {
+		hm.buffer.Load().(*gaugeBuffer).flush(hm.metric)
+	})
+}

metric/histogram.go

@@ -10,14 +10,18 @@ type Histogram struct {
 	*heldMetric[prometheus.Histogram]
 }
 
-func newHistogram(c prometheus.Histogram) *Histogram {
+func newHistogram(h prometheus.Histogram) *Histogram {
 	return &Histogram{
-		heldMetric: newHeldMetric(nil, c),
+		heldMetric: newHeldMetric(nil, h),
 	}
 }
 
 func (h *Histogram) Observe(v float64) {
-	h.metric.Observe(v)
+	if buf := h.buffer.Load(); buf != nil {
+		buf.(*histogramBuffer).observe(v)
+	} else {
+		h.metric.Observe(v)
+	}
 	h.updateUsage()
 }
 
@@ -33,6 +37,10 @@ func newHistogramVec(hv *prometheus.HistogramVec, maxLabelValueLength int) *Hist
 	}
 }
 
+func (hv *HistogramVec) enableBuffering() {
+	hv.store.enableBuffering(func() any { return &histogramBuffer{} })
+}
+
 func (hv *HistogramVec) WithLabelValues(lvs ...string) *Histogram {
 	return &Histogram{
 		heldMetric: hv.store.GetOrCreate(lvs, func(s ...string) prometheus.Histogram {
@@ -41,10 +49,16 @@ func (hv *HistogramVec) WithLabelValues(lvs ...string) *Histogram {
 	}
 }
 
-func (cv *HistogramVec) DeleteLabelValues(lvs ...string) bool {
-	return cv.store.Delete(lvs, cv.vec)
+func (hv *HistogramVec) DeleteLabelValues(lvs ...string) bool {
+	return hv.store.Delete(lvs, hv.vec)
 }
 
 func (hv *HistogramVec) DeleteOldMetrics(holdDuration time.Duration) {
 	hv.store.DeleteOldMetrics(holdDuration, hv.vec)
 }
+
+func (hv *HistogramVec) flush() {
+	hv.store.flushAll(func(hm *heldMetric[prometheus.Histogram]) {
+		hm.buffer.Load().(*histogramBuffer).flush(hm.metric)
+	})
+}