added generic handler metrics with message type as the label (#144)

* added generic handler metrics with message type as the label

* add handler name label to the metrics

* adding new metrics to the read me

Author: danielwitz

docs/METRICS.md

@@ -4,7 +4,9 @@ grabbit exposes and reports the following metrics to Prometheus
 
 |  Namespace    | Subsystem     | Name                              | Description                                                                 |
 | ------------- | ------------- | ----------------------------------| --------------------------------------------------------------------------- |
-| grabbit       | handler       | [name of message handler]_result  | records and counts each succesfull or failed execution of a message handler |
-| grabbit       | handler       | [name of message handler]_latency | records the execution time of each handler                                  |
+| grabbit       | handlers      | [name of message handler]_result  | records and counts each successful or failed execution of a message handler |
+| grabbit       | handlers      | [name of message handler]_latency | records the execution time of each handler                                  |
+| grabbit       | handlers      | result                            | records and counts each run of a handler, having the handler's name, message type and the result as labels|
+| grabbit       | handlers      | latency                           | records the execution time of each run of a handler, having the handler's name, message type as labels|
 | grabbit       | messages      | rejected_messages                 | increments each time a message gets rejected                                |
 | grabbit       | saga          | timedout_sagas                   | counting the number of timedout saga instances                              |

gbus/metrics/handler_metrics.go

@@ -2,31 +2,47 @@ package metrics
 
 import (
 	"fmt"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"sync"
 
 	"github.com/prometheus/client_golang/prometheus"
-	io_prometheus_client "github.com/prometheus/client_model/go"
+	"github.com/prometheus/client_model/go"
 	"github.com/sirupsen/logrus"
 )
 
 var (
 	handlerMetricsByHandlerName = &sync.Map{}
+	handlersResultCounter       = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: grabbitPrefix,
+		Subsystem: handlers,
+		Name:      handlerResult,
+		Help:      "The result of the message handler. The handler's name, message type and result are labeled",
+	}, []string{handler, messageTypeLabel, handlerResult})
+	handlersLatencySummary = promauto.NewSummaryVec(prometheus.SummaryOpts{
+		Namespace: grabbitPrefix,
+		Subsystem: handlers,
+		Name:      handlerLatency,
+		Help:      "The latency of the message handler. The handler's name and message type are labeled",
+	}, []string{handler, messageTypeLabel})
 )
 
 const (
-	failure       = "failure"
-	success       = "success"
-	handlerResult = "result"
-	handlers      = "handlers"
-	grabbitPrefix = "grabbit"
+	failure          = "failure"
+	success          = "success"
+	handlerResult    = "result"
+	handlers         = "handlers"
+	grabbitPrefix    = "grabbit"
+	messageTypeLabel = "message_type"
+	handlerLatency   = "latency"
+	handler          = "handler"
 )
 
 type handlerMetrics struct {
 	result  *prometheus.CounterVec
 	latency prometheus.Summary
 }
 
-//AddHandlerMetrics adds a handlere to be tracked with metrics
+//AddHandlerMetrics adds a handler to be tracked with metrics
 func AddHandlerMetrics(handlerName string) {
 	handlerMetrics := newHandlerMetrics(handlerName)
 	_, exists := handlerMetricsByHandlerName.LoadOrStore(handlerName, handlerMetrics)
@@ -37,29 +53,31 @@ func AddHandlerMetrics(handlerName string) {
 }
 
 //RunHandlerWithMetric runs a specific handler with metrics being collected and reported to prometheus
-func RunHandlerWithMetric(handleMessage func() error, handlerName string, logger logrus.FieldLogger) error {
+func RunHandlerWithMetric(handleMessage func() error, handlerName, messageType string, logger logrus.FieldLogger) error {
 	handlerMetrics := GetHandlerMetrics(handlerName)
 	defer func() {
 		if p := recover(); p != nil {
 			if handlerMetrics != nil {
 				handlerMetrics.result.WithLabelValues(failure).Inc()
 			}
-
+			handlersResultCounter.With(prometheus.Labels{handler: handlerName, messageTypeLabel: messageType, handlerResult: failure}).Inc()
 			panic(p)
 		}
 	}()
 
 	if handlerMetrics == nil {
 		logger.WithField("handler", handlerName).Warn("Running with metrics - couldn't find metrics for the given handler")
-		return handleMessage()
+		return trackTime(handleMessage, handlersLatencySummary.WithLabelValues(handlerName, messageType))
 	}
 
-	err := trackTime(handleMessage, handlerMetrics.latency)
+	err := trackTime(handleMessage, handlerMetrics.latency, handlersLatencySummary.WithLabelValues(handlerName, messageType))
 
 	if err != nil {
 		handlerMetrics.result.WithLabelValues(failure).Inc()
+		handlersResultCounter.With(prometheus.Labels{handler: handlerName, messageTypeLabel: messageType, handlerResult: failure}).Inc()
 	} else {
 		handlerMetrics.result.WithLabelValues(success).Inc()
+		handlersResultCounter.With(prometheus.Labels{handler: handlerName, messageTypeLabel: messageType, handlerResult: success}).Inc()
 	}
 
 	return err
@@ -95,41 +113,75 @@ func newHandlerMetrics(handlerName string) *handlerMetrics {
 	}
 }
 
-func trackTime(functionToTrack func() error, observer prometheus.Observer) error {
-	timer := prometheus.NewTimer(observer)
-	defer timer.ObserveDuration()
+func trackTime(functionToTrack func() error, observers ...prometheus.Observer) error {
+	timers := make([]*prometheus.Timer, 0)
+	for _, observer := range observers {
+		timers = append(timers, prometheus.NewTimer(observer))
+	}
+
+	defer func() {
+		for _, timer := range timers {
+			timer.ObserveDuration()
+		}
+	}()
 
 	return functionToTrack()
 }
 
+//GetSuccessCountByMessageTypeAndHandlerName gets the counter value for the successful handlers' run for a given message type and handler's name
+func GetSuccessCountByMessageTypeAndHandlerName(messageType, handlerName string) (float64, error) {
+	return getCounterValue(handlersResultCounter.With(prometheus.Labels{messageTypeLabel: messageType, handler: handlerName, handlerResult: success}))
+}
+
+//GetFailureCountByMessageTypeAndHandlerName gets the counter value for the failed handlers' run for a given message type and handler's name
+func GetFailureCountByMessageTypeAndHandlerName(messageType, handlerName string) (float64, error) {
+	return getCounterValue(handlersResultCounter.With(prometheus.Labels{messageTypeLabel: messageType, handler: handlerName, handlerResult: failure}))
+}
+
+//GetLatencySampleCountByMessageTypeAndHandlerName gets the summary sample count value for the handlers' run for a given message type and handler's name
+func GetLatencySampleCountByMessageTypeAndHandlerName(messageType, handlerName string) (*uint64, error) {
+	summary, ok := handlersLatencySummary.With(prometheus.Labels{messageTypeLabel: messageType, handler: handlerName}).(prometheus.Summary)
+
+	if !ok {
+		return nil, fmt.Errorf("couldn't find summary for event type: %s", messageType)
+	}
+
+	return getSummarySampleCount(summary)
+}
+
 //GetSuccessCount gets the value of the handlers success value
 func (hm *handlerMetrics) GetSuccessCount() (float64, error) {
-	return hm.getLabeledCounterValue(success)
+	return getCounterValue(hm.result.WithLabelValues(success))
 }
 
 //GetFailureCount gets the value of the handlers failure value
 func (hm *handlerMetrics) GetFailureCount() (float64, error) {
-	return hm.getLabeledCounterValue(failure)
+	return getCounterValue(hm.result.WithLabelValues(failure))
 }
 
 //GetLatencySampleCount gets the value of the handlers latency value
 func (hm *handlerMetrics) GetLatencySampleCount() (*uint64, error) {
+	return getSummarySampleCount(hm.latency)
+}
+
+func getCounterValue(counter prometheus.Counter) (float64, error) {
 	m := &io_prometheus_client.Metric{}
-	err := hm.latency.Write(m)
+	err := counter.Write(m)
+
 	if err != nil {
-		return nil, err
+		return 0, err
 	}
 
-	return m.GetSummary().SampleCount, nil
+	return m.GetCounter().GetValue(), nil
 }
 
-func (hm *handlerMetrics) getLabeledCounterValue(label string) (float64, error) {
+func getSummarySampleCount(summary prometheus.Summary) (*uint64, error) {
 	m := &io_prometheus_client.Metric{}
-	err := hm.result.WithLabelValues(label).Write(m)
+	err := summary.Write(m)
 
 	if err != nil {
-		return 0, err
+		return nil, err
 	}
 
-	return m.GetCounter().GetValue(), nil
+	return m.GetSummary().SampleCount, nil
 }

gbus/saga/instance.go

@@ -73,7 +73,7 @@ func (si *Instance) invoke(exchange, routingKey string, invocation *sagaInvocati
 				return val.Interface().(error)
 			}
 			return nil
-		}, methodName, invocation.Log())
+		}, methodName, message.PayloadFQN, invocation.Log())
 
 		if err != nil {
 			return err

gbus/worker.go

@@ -211,7 +211,7 @@ func (worker *worker) invokeDeadletterHandler(delivery amqp.Delivery) {
 		handlerWrapper := func() error {
 			return worker.deadletterHandler(tx, &delivery)
 		}
-		return metrics.RunHandlerWithMetric(handlerWrapper, worker.deadletterHandler.Name(), worker.log())
+		return metrics.RunHandlerWithMetric(handlerWrapper, worker.deadletterHandler.Name(), fmt.Sprintf("deadletter_%s", delivery.Type), worker.log())
 	}
 
 	err := worker.withTx(txWrapper)
@@ -250,7 +250,7 @@ func (worker *worker) runGlobalHandler(delivery *amqp.Delivery) error {
 				return worker.withTx(txWrapper)
 			}
 			//run the global handler with metrics
-			return metrics.RunHandlerWithMetric(metricsWrapper, handlerName, worker.log())
+			return metrics.RunHandlerWithMetric(metricsWrapper, handlerName, delivery.Type, worker.log())
 		}
 		return worker.SafeWithRetries(retryAction, MaxRetryCount)
 	}
@@ -417,7 +417,7 @@ func (worker *worker) invokeHandlers(sctx context.Context, handlers []MessageHan
 				//execute the handler with metrics
 				handlerErr := metrics.RunHandlerWithMetric(func() error {
 					return pinedHandler(invocation, message)
-				}, handlerName, worker.log())
+				}, handlerName, message.PayloadFQN, worker.log())
 
 				if handlerErr != nil {
 					hspan.LogFields(slog.Error(handlerErr))

tests/bus_test.go

@@ -174,14 +174,22 @@ func TestHandlerRetry(t *testing.T) {
 		t.Error("Metrics for handleRetry should be initiated")
 	}
 	f, _ := hm.GetFailureCount()
+	mtf, _ := metrics.GetFailureCountByMessageTypeAndHandlerName(reply.PayloadFQN, "handleRetry")
 	s, _ := hm.GetSuccessCount()
+	mts, _ := metrics.GetSuccessCountByMessageTypeAndHandlerName(reply.PayloadFQN, "handleRetry")
 
 	if f != 2 {
 		t.Errorf("Failure count should be 2 but was %f", f)
 	}
+	if mtf != 2 {
+		t.Errorf("Failure count should be 2 but was %f", mtf)
+	}
 	if s != 1 {
 		t.Errorf("Success count should be 1 but was %f", s)
 	}
+	if mts != 1 {
+		t.Errorf("Success count should be 1 but was %f", mts)
+	}
 }
 
 func handleRetry(invocation gbus.Invocation, message *gbus.BusMessage) error {
@@ -249,8 +257,9 @@ func TestDeadlettering(t *testing.T) {
 	service1.Start()
 	defer assertBusShutdown(service1, t)
 
+	cmd := gbus.NewBusMessage(Command1{})
 	service1.Send(context.Background(), testSvc1, poison)
-	service1.Send(context.Background(), testSvc1, gbus.NewBusMessage(Command1{}))
+	service1.Send(context.Background(), testSvc1, cmd)
 
 	proceedOrTimeout(2, proceed, nil, t)
 
@@ -268,6 +277,10 @@ func TestDeadlettering(t *testing.T) {
 	if failureCount != 0 {
 		t.Errorf("DeadLetterHandler should not have failed, but it failed %f times", failureCount)
 	}
+	poisonF, _ := metrics.GetFailureCountByMessageTypeAndHandlerName(poison.PayloadFQN, "func1")
+	if poisonF != 0 {
+		t.Errorf("DeadLetterHandler should not have failed, but it failed %f times", poisonF)
+	}
 	handlerMetrics = metrics.GetHandlerMetrics("func2")
 	if handlerMetrics == nil {
 		t.Fatal("faulty should be registered for metrics")
@@ -276,6 +289,10 @@ func TestDeadlettering(t *testing.T) {
 	if failureCount == 1 {
 		t.Errorf("faulty should have failed once, but it failed %f times", failureCount)
 	}
+	cmdF, _ := metrics.GetFailureCountByMessageTypeAndHandlerName(cmd.PayloadFQN, "func2")
+	if cmdF == 1 {
+		t.Errorf("faulty should have failed once, but it failed %f times", cmdF)
+	}
 }
 
 func TestRawMessageHandling(t *testing.T) {

tests/metrics_test.go

@@ -59,7 +59,7 @@ func TestRunHandlerWithMetric_FailureCounter(t *testing.T) {
 	}
 
 	for i := 1; i < runningTries; i++ {
-		err := metrics.RunHandlerWithMetric(failure, name, logger)
+		err := metrics.RunHandlerWithMetric(failure, name, name, logger)
 
 		if err == nil {
 			t.Error("Failed handler run should return an error")
@@ -90,7 +90,7 @@ func TestRunHandlerWithMetric_SuccessCounter(t *testing.T) {
 	}
 
 	for i := 1; i < runningTries; i++ {
-		err := metrics.RunHandlerWithMetric(success, name, logger)
+		err := metrics.RunHandlerWithMetric(success, name, name, logger)
 
 		if err != nil {
 			t.Error("Successful handler run shouldn't return an error")
@@ -121,17 +121,23 @@ func TestRunHandlerWithMetric_Latency(t *testing.T) {
 	}
 
 	for i := 1; i < runningTries; i++ {
-		_ = metrics.RunHandlerWithMetric(success, name, logger)
+		_ = metrics.RunHandlerWithMetric(success, name, name, logger)
 		sc, err := hm.GetLatencySampleCount()
 
-		if err != nil {
-			t.Errorf("Failed to get latency value: %e", err)
-		}
-		if sc == nil {
-			t.Errorf("Expected latency sample count not be nil")
-		}
-		if *sc != uint64(i) {
-			t.Errorf("Expected to get %d as the value of the latency sample count, but got %d", uint64(i), *sc)
-		}
+		checkLatency(t, sc, uint64(i), err)
+		mtsc, err := metrics.GetLatencySampleCountByMessageTypeAndHandlerName(name, name)
+		checkLatency(t, mtsc, uint64(i), err)
+	}
+}
+
+func checkLatency(t *testing.T, sc  *uint64, expected uint64, err error) {
+	if err != nil {
+		t.Errorf("Failed to get latency value: %e", err)
+	}
+	if sc == nil {
+		t.Errorf("Expected latency sample count not be nil")
+	}
+	if *sc != expected {
+		t.Errorf("Expected to get %d as the value of the latency sample count, but got %d", expected, *sc)
 	}
 }