added metric for ooo lag for each datapoint (#534)

* added hist metric for ooo lag for each datapoint

* linter fix, styles, increared number of buckets for ooo writes lag

* fixed naming

* fixed naming

carbon/app.go

@@ -311,7 +311,9 @@ func (app *App) startPersister() {
 		p.SetRemoveEmptyFile(app.Config.Whisper.RemoveEmptyFile)
 		p.SetWorkers(app.Config.Whisper.Workers)
 		p.SetHashFilenames(app.Config.Whisper.HashFilenames)
-
+		if app.Config.Prometheus.Enabled {
+			p.InitPrometheus(app.PromRegisterer)
+		}
 		if app.Tags != nil {
 			p.SetTagsEnabled(true)
 			p.SetTaggedFn(app.Tags.Add)

persister/whisper.go

@@ -13,6 +13,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	whisper "github.com/go-graphite/go-whisper"
 	"go.uber.org/zap"
 
@@ -77,14 +79,20 @@ type Whisper struct {
 			logicalSizeChanges  int64
 		}
 	}
-
+	prometheus whisperPrometheus
 	// blockThrottleNs        uint64 // sum ns counter
 	// blockQueueGetNs        uint64 // sum ns counter
 	// blockAvoidConcurrentNs uint64 // sum ns counter
 	// blockUpdateManyNs      uint64 // sum ns counter
 }
 
 // NewWhisper create instance of Whisper
+type whisperPrometheus struct {
+	enabled             bool
+	outOfOrderWriteLags prometheus.Histogram
+	outOfOrderWriteLag  func(time.Duration)
+}
+
 func NewWhisper(
 	rootPath string,
 	schemas WhisperSchemas,
@@ -108,6 +116,23 @@ func NewWhisper(
 	}
 }
 
+func (p *Whisper) InitPrometheus(reg prometheus.Registerer) {
+	p.prometheus = whisperPrometheus{
+		enabled: true,
+		outOfOrderWriteLags: prometheus.NewHistogram(
+			prometheus.HistogramOpts{
+				Name:    "out_of_order_write_lag_exp",
+				Help:    "Lag for incoming datapoints (exponential buckets)",
+				Buckets: prometheus.ExponentialBuckets(time.Millisecond.Seconds(), 2.0, 30),
+			},
+		),
+	}
+	p.prometheus.outOfOrderWriteLag = func(t time.Duration) {
+		p.prometheus.outOfOrderWriteLags.Observe(t.Seconds())
+	}
+	reg.MustRegister(p.prometheus.outOfOrderWriteLags)
+}
+
 // SetOnlineMigration enable online migration
 func (p *Whisper) EnableOnlineMigration(rate int, scope []string) {
 	p.onlineMigration.enabled = true
@@ -188,7 +213,16 @@ func fnv32(key string) uint32 {
 	}
 	return hash
 }
-
+func (p *Whisper) registerOutOfOrderWriteLags(points []*whisper.TimeSeriesPoint) {
+	if !p.prometheus.enabled {
+		return
+	}
+	now := time.Now()
+	for _, point := range points {
+		lag := now.Sub(time.Unix(int64(point.Time), 0))
+		p.prometheus.outOfOrderWriteLag(lag)
+	}
+}
 func (p *Whisper) updateMany(w *whisper.Whisper, path string, points []*whisper.TimeSeriesPoint) {
 	defer func() {
 		if r := recover(); r != nil {
@@ -200,6 +234,7 @@ func (p *Whisper) updateMany(w *whisper.Whisper, path string, points []*whisper.
 	}()
 
 	// start = time.Now()
+	p.registerOutOfOrderWriteLags(points)
 	if err := w.UpdateMany(points); err != nil {
 		p.logger.Error("fail to update metric",
 			zap.String("path", path),