Merge pull request #7 from lovoo/AS-295-bbq-lag

AS-295 bbq lag

bbq/bbq.go

@@ -174,6 +174,8 @@ func (b *Bbq) Consume(ctx goka.Context, msg interface{}) {
 		msgToSave: msg,
 	}
 
+	b.metrics.mxLagSeconds.Observe(time.Since(ctx.Timestamp()).Seconds())
+
 	table := strings.Replace(string(ctx.Topic()), "-", "_", -1)
 
 	b.uploaders[table].Upload(s)

bbq/metrics.go

@@ -1,6 +1,8 @@
 package bbq
 
 import (
+	"time"
+
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -12,16 +14,19 @@ const (
 type metrics struct {
 	mxErrorUpload  prometheus.Counter
 	mxTableInserts *prometheus.CounterVec
+	mxLagSeconds   prometheus.Histogram
 }
 
 func (m *metrics) Describe(ch chan<- *prometheus.Desc) {
 	m.mxErrorUpload.Describe(ch)
 	m.mxTableInserts.Describe(ch)
+	m.mxLagSeconds.Describe(ch)
 }
 
 func (m *metrics) Collect(ch chan<- prometheus.Metric) {
 	m.mxErrorUpload.Collect(ch)
 	m.mxTableInserts.Collect(ch)
+	m.mxLagSeconds.Collect(ch)
 }
 
 func newMetrics(namespace string) *metrics {
@@ -38,5 +43,12 @@ func newMetrics(namespace string) *metrics {
 			Name:      "table_insert",
 			Help:      "BBQ tables insert counter",
 		}, []string{"table"}),
+		mxLagSeconds: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Name:        "lag",
+			Help:        "Message lag in secondsd",
+			Namespace:   namespace,
+			Buckets:     []float64{0.0, 0.5, 1.0, 10.0, 60.0, 300.0, time.Hour.Seconds(), (time.Hour * 24).Seconds()},
+			ConstLabels: prometheus.Labels{"subsystem": subsystem},
+		}),
 	}
 }