Merge pull request #6 from inloco/feature/buffer-full-metrics

feat(metrics): add buffer full gauge

Author: racevedoo

README.md

@@ -106,6 +106,7 @@ The exported metrics are:
 - `kafka_consumer_partition_delay`: number of records betweeen last record consumed successfully and the last record on kafka, by partition and topic.
 - `kafka_consumer_records_consumed_successfully`: number of records consumed successfully by this instance.
 - `kafka_consumer_endpoint_latency_histogram_seconds`: endpoint latency in seconds (insertion to elasticsearch).
+- `kafka_consumer_buffer_full`: indicates whether the app buffer is full(meaning that elasticsearch is not being able to keep up with the topic volume).
 
 ## Development
 

VERSION

@@ -1 +1 @@
-0.5.1
+0.5.2

src/kafka/consumer.go

@@ -103,8 +103,10 @@ func (k *kafka) Start(signals chan os.Signal, notifications chan<- Notification)
 						"message", "Buffer is full ",
 						"channelSize", cap(k.consumerCh),
 					)
+					k.metricsPublisher.BufferFull(true)
 				}
 				k.consumerCh <- msg
+				k.metricsPublisher.BufferFull(false)
 			}
 		case err, more := <-consumer.Errors():
 			if more {

src/metrics/metrics.go

@@ -3,19 +3,21 @@ package metrics
 import (
 	"strconv"
 
+	"sync"
+
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	"github.com/inloco/kafka-elasticsearch-injector/src/logger_builder"
 	stdprometheus "github.com/prometheus/client_golang/prometheus"
-	"sync"
 )
 
 type metrics struct {
 	logger                   log.Logger
 	partitionDelay           *kitprometheus.Gauge
 	recordsConsumed          *kitprometheus.Counter
 	endpointLatencyHistogram *kitprometheus.Summary
+	bufferFullGauge          *kitprometheus.Gauge
 	lock                     sync.RWMutex
 	topicPartitionToOffset   map[string]map[int32]int64
 }
@@ -59,11 +61,20 @@ func (m *metrics) PublishOffsetMetrics(highWaterMarks map[string]map[int32]int64
 	}
 }
 
+func (m *metrics) BufferFull(full bool) {
+	val := 0.0
+	if full {
+		val = 1.0
+	}
+	m.bufferFullGauge.Set(val)
+}
+
 type MetricsPublisher interface {
 	PublishOffsetMetrics(highWaterMarks map[string]map[int32]int64)
 	UpdateOffset(topic string, partition int32, delay int64)
 	IncrementRecordsConsumed(count int)
 	RecordEndpointLatency(latency float64)
+	BufferFull(full bool)
 }
 
 func NewMetricsPublisher() MetricsPublisher {
@@ -80,12 +91,17 @@ func NewMetricsPublisher() MetricsPublisher {
 		Name: "kafka_consumer_endpoint_latency_histogram_seconds",
 		Help: "Kafka consumer endpoint latency histogram in seconds",
 	}, []string{})
+	bufferFullGauge := kitprometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+		Name: "kafka_consumer_buffer_full",
+		Help: "Kafka consumer boolean indicating if app buffer is full",
+	}, []string{})
 	return &metrics{
 		logger:                   logger,
 		partitionDelay:           partitionDelay,
 		recordsConsumed:          recordsConsumed,
 		endpointLatencyHistogram: endpointLatencySummary,
-		lock: sync.RWMutex{},
+		bufferFullGauge:          bufferFullGauge,
+		lock:                     sync.RWMutex{},
 		topicPartitionToOffset: make(map[string]map[int32]int64),
 	}
 }