Add translation latency metric (#150)

Author: pglass

interceptor/translation_interceptor.go

@@ -2,8 +2,8 @@ package interceptor
 
 import (
 	"context"
-	"fmt"
 	"strings"
+	"time"
 
 	"go.temporal.io/server/common/api"
 	"go.temporal.io/server/common/log"
@@ -47,17 +47,19 @@ func (i *TranslationInterceptor) Intercept(
 
 		for _, tr := range i.translators {
 			if tr.MatchMethod(info.FullMethod) {
+				start := time.Now()
 				changed, trErr := tr.TranslateRequest(req)
-				logTranslateResult(tr, i.logger, changed, trErr, methodName+"Request", req)
+				logTranslateResult(tr, i.logger, changed, trErr, methodName+"Request", req, time.Since(start))
 			}
 		}
 
 		resp, err := handler(ctx, req)
 
 		for _, tr := range i.translators {
 			if tr.MatchMethod(info.FullMethod) {
+				start := time.Now()
 				changed, trErr := tr.TranslateResponse(resp)
-				logTranslateResult(tr, i.logger, changed, trErr, methodName+"Response", resp)
+				logTranslateResult(tr, i.logger, changed, trErr, methodName+"Response", resp, time.Since(start))
 			}
 		}
 
@@ -84,16 +86,18 @@ type streamTranslator struct {
 
 func (w *streamTranslator) RecvMsg(m any) error {
 	for _, tr := range w.translators {
+		start := time.Now()
 		changed, trErr := tr.TranslateRequest(m)
-		logTranslateResult(tr, w.logger, changed, trErr, "RecvMsg", m)
+		logTranslateResult(tr, w.logger, changed, trErr, "RecvMsg", m, time.Since(start))
 	}
 	return w.ServerStream.RecvMsg(m)
 }
 
 func (w *streamTranslator) SendMsg(m any) error {
 	for _, tr := range w.translators {
+		start := time.Now()
 		changed, trErr := tr.TranslateResponse(m)
-		logTranslateResult(tr, w.logger, changed, trErr, "SendMsg", m)
+		logTranslateResult(tr, w.logger, changed, trErr, "SendMsg", m, time.Since(start))
 	}
 	return w.ServerStream.SendMsg(m)
 }
@@ -110,14 +114,17 @@ func newStreamTranslator(
 	}
 }
 
-func logTranslateResult(tr Translator, logger log.Logger, changed bool, err error, methodName string, obj any) {
+func logTranslateResult(tr Translator, logger log.Logger, changed bool, err error, methodName string, obj any, duration time.Duration) {
+	msgType := metrics.SanitizedTypeName(obj)
+	metrics.TranslationLatency.WithLabelValues(tr.Kind(), msgType).Observe(duration.Seconds())
+
 	methodTag := tag.NewStringTag("method", methodName)
 	if err != nil {
-		logger.Error("translation error", methodTag, tag.Error(err), tag.NewStringTag("type", fmt.Sprintf("%T", obj)))
-		metrics.TranslationErrors.WithLabelValues(tr.Kind(), methodName).Inc()
+		logger.Error("translation error", methodTag, tag.Error(err), tag.NewStringTag("type", msgType))
+		metrics.TranslationErrors.WithLabelValues(tr.Kind(), msgType).Inc()
 	} else if changed {
 		logger.Debug("translation applied", methodTag, tag.NewAnyTag("obj", obj))
-		metrics.TranslationCount.WithLabelValues(tr.Kind(), methodName).Inc()
+		metrics.TranslationCount.WithLabelValues(tr.Kind(), msgType).Inc()
 	} else {
 		logger.Debug("translation not applied", methodTag, tag.NewAnyTag("obj", obj))
 	}

metrics/metrics.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net/http"
 	"regexp"
+	"strings"
 
 	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
 	"github.com/prometheus/client_golang/prometheus"
@@ -33,6 +34,12 @@ func SanitizeForPrometheus(value string) string {
 	return prometheusReplacePattern.ReplaceAllLiteralString(value, "_")
 }
 
+func SanitizedTypeName(m any) string {
+	name := fmt.Sprintf("%T", m)
+	name = strings.ReplaceAll(name, "*", "")
+	return SanitizeForPrometheus(name)
+}
+
 // GetStandardGRPCInterceptor returns a ServerMetrics with our preferred standard config for monitoring gRPC servers.
 // Want to change/add options? Check the docs at https://pkg.go.dev/github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus@v1.1.0#section-documentation
 // Some more handy links: https://prometheus.io/docs/concepts/metric_types/#histogram

metrics/prometheus_defs.go

@@ -56,9 +56,10 @@ var (
 	MuxErrors              = DefaultCounterVec("mux_errors", "Number of errors observed from mux", muxManagerLabels...)
 	MuxConnectionEstablish = DefaultCounterVec("mux_connection_establish", "Number of times mux has established", muxManagerLabels...)
 
-	translationLabels = []string{"kind", "message_type"}
-	TranslationCount  = DefaultCounterVec("translation_success", "Count of message translations", translationLabels...)
-	TranslationErrors = DefaultCounterVec("translation_error", "Count of message translation errors", translationLabels...)
+	translationLabels  = []string{"kind", "message_type"}
+	TranslationCount   = DefaultCounterVec("translation_success", "Count of message translations", translationLabels...)
+	TranslationErrors  = DefaultCounterVec("translation_error", "Count of message translation errors", translationLabels...)
+	TranslationLatency = DefaultHistogramVec("translation_latency", "Latency of message translations", translationLabels...)
 
 	UTF8RepairTranslationKind = "utf8repair"
 	NamespaceTranslationKind  = "namespace"
@@ -102,4 +103,5 @@ func init() {
 
 	prometheus.MustRegister(TranslationCount)
 	prometheus.MustRegister(TranslationErrors)
+	prometheus.MustRegister(TranslationLatency)
 }

proto/compat/codec.go

@@ -1,7 +1,7 @@
 package compat
 
 import (
-	"fmt"
+	"time"
 
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
@@ -60,8 +60,13 @@ func (c *RepairUTF8Codec) Name() string {
 func (c *RepairUTF8Codec) Unmarshal(data mem.BufferSlice, v any) error {
 	err := c.delegate.Unmarshal(data, v)
 	if common.IsInvalidUTF8Error(err) {
-		msgType := fmt.Sprintf("%T", v)
-		if err := convertAndRepairInvalidUTF8(data.Materialize(), v); err != nil {
+		start := time.Now()
+		err := convertAndRepairInvalidUTF8(data.Materialize(), v)
+		duration := time.Since(start)
+
+		msgType := metrics.SanitizedTypeName(v)
+		metrics.TranslationLatency.WithLabelValues(metrics.UTF8RepairTranslationKind, msgType).Observe(duration.Seconds())
+		if err != nil {
 			c.Logger.Error("during UTF-8 repair", tag.Error(err))
 			metrics.TranslationErrors.WithLabelValues(metrics.UTF8RepairTranslationKind, msgType).Inc()
 		} else {