Merge pull request #3 from inloco/feature/endpoint-latency-metrics

Feature: add endpoint latency metrics

Author: racevedoo

README.md

@@ -105,6 +105,7 @@ dashboards as inspiration for your application.
 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).
 
 ## Development
 

VERSION

@@ -1 +1 @@
-0.5.0
+0.5.1

cmd/injector.go

@@ -46,8 +46,8 @@ func main() {
 		MetricsUpdateInterval: os.Getenv("KAFKA_CONSUMER_METRICS_UPDATE_INTERVAL"),
 		RecordType:            os.Getenv("KAFKA_CONSUMER_RECORD_TYPE"),
 	}
-
-	service := injector.NewService(logger)
+	metricsPublisher := metrics.NewMetricsPublisher()
+	service := injector.NewService(logger, metricsPublisher)
 	p.SetReadinessCheck(service.ReadinessCheck)
 
 	endpoints := injector.MakeEndpoints(service)
@@ -57,7 +57,7 @@ func main() {
 		level.Error(logger).Log("err", err, "message", "error creating kafka consumer")
 		panic(err)
 	}
-	k := kafka.NewKafka(os.Getenv("KAFKA_ADDRESS"), consumer)
+	k := kafka.NewKafka(os.Getenv("KAFKA_ADDRESS"), consumer, metricsPublisher)
 
 	signals := make(chan os.Signal, 1)
 	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)

docker-compose.yml

@@ -59,6 +59,8 @@ services:
       - KAFKA_CONSUMER_BATCH_SIZE=10
       - METRICS_PORT=9102
       - LOG_LEVEL=DEBUG
+    ports:
+      - "9102:9102"
   producer:
     image: 'inlocomedia/kafka-elasticsearch-injector:producer-local'
     container_name: producer

src/injector/middleware.go

@@ -0,0 +1,24 @@
+package injector
+
+import (
+	"time"
+
+	"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
+	"github.com/inloco/kafka-elasticsearch-injector/src/models"
+)
+
+type instrumentingMiddleware struct {
+	metricsPublisher metrics.MetricsPublisher
+	next             Service
+}
+
+func (s instrumentingMiddleware) Insert(records []*models.Record) error {
+	begin := time.Now()
+	err := s.next.Insert(records)
+	s.metricsPublisher.RecordEndpointLatency(time.Since(begin).Seconds())
+	return err
+}
+
+func (s instrumentingMiddleware) ReadinessCheck() bool {
+	return s.next.ReadinessCheck()
+}

src/injector/service.go

@@ -1,9 +1,10 @@
 package injector
 
 import (
+	"github.com/go-kit/kit/log"
 	"github.com/inloco/kafka-elasticsearch-injector/src/injector/store"
+	"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
-	"github.com/go-kit/kit/log"
 )
 
 type Service interface {
@@ -23,8 +24,11 @@ func (s basicService) ReadinessCheck() bool {
 	return s.store.ReadinessCheck()
 }
 
-func NewService(logger log.Logger) Service {
-	return basicService{
-		store.NewStore(logger),
+func NewService(logger log.Logger, metrics metrics.MetricsPublisher) Service {
+	return instrumentingMiddleware{
+		metricsPublisher: metrics,
+		next: basicService{
+			store.NewStore(logger),
+		},
 	}
 }

src/kafka/consumer.go

@@ -49,7 +49,7 @@ type topicPartitionOffset struct {
 	offset    int64
 }
 
-func NewKafka(address string, consumer Consumer) kafka {
+func NewKafka(address string, consumer Consumer, metrics metrics.MetricsPublisher) kafka {
 	brokers := []string{address}
 	config := cluster.NewConfig()
 	config.Consumer.Return.Errors = true
@@ -61,7 +61,7 @@ func NewKafka(address string, consumer Consumer) kafka {
 		brokers:          brokers,
 		config:           config,
 		consumer:         consumer,
-		metricsPublisher: metrics.NewMetricsPublisher(),
+		metricsPublisher: metrics,
 		consumerCh:       make(chan *sarama.ConsumerMessage, consumer.BufferSize),
 		offsetCh:         make(chan *topicPartitionOffset),
 	}

src/kafka/consumer_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/inloco/kafka-elasticsearch-injector/src/elasticsearch"
 	"github.com/inloco/kafka-elasticsearch-injector/src/kafka/fixtures"
 	"github.com/inloco/kafka-elasticsearch-injector/src/logger_builder"
+	"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
 	"github.com/inloco/kafka-elasticsearch-injector/src/schema_registry"
 	"github.com/olivere/elastic"
@@ -88,7 +89,7 @@ func TestMain(m *testing.M) {
 		BatchSize:             1,
 		MetricsUpdateInterval: 30 * time.Second,
 	}
-	k = NewKafka("localhost:9092", consumer)
+	k = NewKafka("localhost:9092", consumer, metrics.NewMetricsPublisher())
 	retCode := m.Run()
 	os.Exit(retCode)
 }

src/metrics/metrics.go

@@ -12,17 +12,22 @@ import (
 )
 
 type metrics struct {
-	logger                 log.Logger
-	partitionDelay         *kitprometheus.Gauge
-	recordsConsumed        *kitprometheus.Counter
-	lock                   sync.RWMutex
-	topicPartitionToOffset map[string]map[int32]int64
+	logger                   log.Logger
+	partitionDelay           *kitprometheus.Gauge
+	recordsConsumed          *kitprometheus.Counter
+	endpointLatencyHistogram *kitprometheus.Summary
+	lock                     sync.RWMutex
+	topicPartitionToOffset   map[string]map[int32]int64
 }
 
 func (m *metrics) IncrementRecordsConsumed(count int) {
 	m.recordsConsumed.Add(float64(count))
 }
 
+func (m *metrics) RecordEndpointLatency(latency float64) {
+	m.endpointLatencyHistogram.Observe(latency)
+}
+
 func (m *metrics) UpdateOffset(topic string, partition int32, offset int64) {
 	m.lock.Lock()
 	currentOffset, exists := m.topicPartitionToOffset[topic][partition]
@@ -58,6 +63,7 @@ type MetricsPublisher interface {
 	PublishOffsetMetrics(highWaterMarks map[string]map[int32]int64)
 	UpdateOffset(topic string, partition int32, delay int64)
 	IncrementRecordsConsumed(count int)
+	RecordEndpointLatency(latency float64)
 }
 
 func NewMetricsPublisher() MetricsPublisher {
@@ -70,11 +76,16 @@ func NewMetricsPublisher() MetricsPublisher {
 		Name: "kafka_consumer_partition_delay",
 		Help: "Kafka consumer partition delay",
 	}, []string{"partition", "topic"})
+	endpointLatencySummary := kitprometheus.NewSummaryFrom(stdprometheus.SummaryOpts{
+		Name: "kafka_consumer_endpoint_latency_histogram_seconds",
+		Help: "Kafka consumer endpoint latency histogram in seconds",
+	}, []string{})
 	return &metrics{
-		logger:          logger,
-		partitionDelay:  partitionDelay,
-		recordsConsumed: recordsConsumed,
-		lock:            sync.RWMutex{},
+		logger:                   logger,
+		partitionDelay:           partitionDelay,
+		recordsConsumed:          recordsConsumed,
+		endpointLatencyHistogram: endpointLatencySummary,
+		lock: sync.RWMutex{},
 		topicPartitionToOffset: make(map[string]map[int32]int64),
 	}
 }