added metrics for current messages in db

metrics.go

@@ -55,6 +55,8 @@ type metrics struct {
 	promErrorMessagesReceivedTotal prometheus.Counter
 	// Metrics for sent error messages
 	promErrorMessagesSentTotal prometheus.Counter
+	// Metrics for the amount of messages currently in db.
+	promDBMessagesCurrent prometheus.Gauge
 }
 
 // newMetrics will prepare and return a *metrics.
@@ -155,6 +157,12 @@ func newMetrics(hostAndPort string) *metrics {
 	})
 	m.promRegistry.MustRegister(m.promErrorMessagesSentTotal)
 
+	m.promDBMessagesCurrent = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "steward_db_messages_current",
+		Help: "The current value messages in database",
+	})
+	m.promRegistry.MustRegister(m.promDBMessagesCurrent)
+
 	return &m
 }
 

ringbuffer.go

@@ -106,6 +106,17 @@ func (r *ringBuffer) start(inCh chan subjectAndMessage, outCh chan samDBValueAnd
 
 	// Start the process that will handle messages present in the ringbuffer.
 	go r.processBufferMessages(outCh)
+
+	go func() {
+		ticker := time.NewTicker(time.Second * 5)
+
+		for {
+			select {
+			case <-ticker.C:
+				r.dbUpdateMetrics(r.samValueBucket)
+			}
+		}
+	}()
 }
 
 // fillBuffer will fill the buffer in the ringbuffer  reading from the inchannel.
@@ -364,6 +375,19 @@ func (r *ringBuffer) deleteKeyFromBucket(bucket string, key string) error {
 	return err
 }
 
+// db update metrics.
+func (r *ringBuffer) dbUpdateMetrics(bucket string) error {
+	err := r.db.Update(func(tx *bolt.Tx) error {
+		bu := tx.Bucket([]byte(bucket))
+
+		r.metrics.promDBMessagesCurrent.Set(float64(bu.Stats().KeyN))
+
+		return nil
+	})
+
+	return err
+}
+
 // getIndexValue will get the last index value stored in DB.
 func (r *ringBuffer) getIndexValue() int {
 	const indexKey string = "index"