add new highcardmetrics pkg to grpcmetrics

grpc/grpcmetrics/highcardmetrics/server.go

@@ -0,0 +1,162 @@
+package highcardmetrics
+
+import (
+	"context"
+	"regexp"
+	"strings"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/heroku/x/go-kit/metrics"
+	"github.com/heroku/x/go-kit/metricsregistry"
+	"github.com/heroku/x/grpc/grpcmetrics"
+)
+
+// NewUnaryServerInterceptor returns an interceptor for unary server calls
+// which will report metrics to the given provider.
+func NewUnaryServerInterceptor(p metrics.Provider) grpc.UnaryServerInterceptor {
+	r0 := metricsregistry.New(p)
+	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (_ interface{}, err error) {
+		r1 := metricsregistry.NewPrefixed(r0, "grpc.server")
+
+		defer func(begin time.Time) {
+			service, method := parseFullMethod(info.FullMethod)
+			labels := []string{"service", service, "method", method, "response-status", code(err)}
+
+			instrumentMethod(r1, labels, time.Since(begin))
+		}(time.Now())
+
+		return handler(ctx, req)
+	}
+}
+
+// NewStreamServerInterceptor returns an interceptor for stream server calls
+// which will report metrics to the given provider.
+func NewStreamServerInterceptor(p metrics.Provider) grpc.StreamServerInterceptor {
+	r0 := metricsregistry.New(p)
+	return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) (err error) {
+		r1 := metricsregistry.NewPrefixed(r0, "grpc.server")
+
+		service, method := parseFullMethod(info.FullMethod)
+
+		labels := []string{"service", service, "method", method}
+
+		clients := r1.GetOrRegisterGauge("stream.clients").With(labels...)
+		clients.Add(1)
+
+		defer func(begin time.Time) {
+			clients.Add(-1)
+
+			labels = append(labels, "response-status", code(err))
+
+			instrumentMethod(r1, labels, time.Since(begin))
+		}(time.Now())
+
+		wrapped := &serverStream{r1, ss, labels}
+		return handler(srv, wrapped)
+	}
+}
+
+// serverStream provides a light wrapper over grpc.ServerStream
+// to instrument SendMsg and RecvMsg.
+type serverStream struct {
+	reg metricsregistry.Registry
+	grpc.ServerStream
+	labels []string
+}
+
+//// RecvMsg implements the grpc.Stream interface.
+func (ss *serverStream) SendMsg(m interface{}) (err error) {
+	defer func(begin time.Time) {
+		instrumentStreamSend(ss.reg, ss.labels, time.Since(begin), err)
+	}(time.Now())
+
+	return ss.ServerStream.SendMsg(m)
+}
+
+//// RecvMsg implements the grpc.Stream interface.
+func (ss *serverStream) RecvMsg(m interface{}) (err error) {
+	defer func(begin time.Time) {
+		instrumentStreamRecv(ss.reg, ss.labels, time.Since(begin), err)
+	}(time.Now())
+
+	return ss.ServerStream.RecvMsg(m)
+}
+
+func instrumentMethod(r metricsregistry.Registry, labels []string, duration time.Duration) {
+	r.GetOrRegisterHistogram("request-duration.ms", 50).With(labels...).Observe(ms(duration))
+	r.GetOrRegisterCounter("requests").With(labels...).Add(1)
+}
+
+func instrumentStreamSend(r metricsregistry.Registry, labels []string, duration time.Duration, err error) {
+	r.GetOrRegisterHistogram("stream.send-duration.ms", 50).With(labels...).Observe(ms(duration))
+	r.GetOrRegisterCounter("stream.sends").With(labels...).Add(1)
+
+	if err != nil && !isCanceled(err) {
+		r.GetOrRegisterCounter("stream.sends.errors").Add(1)
+	}
+}
+
+func instrumentStreamRecv(r metricsregistry.Registry, labels []string, duration time.Duration, err error) {
+	r.GetOrRegisterHistogram("stream.recv-duration.ms", 50).With(labels...).Observe(ms(duration))
+	r.GetOrRegisterCounter("stream.recvs").With(labels...).Add(1)
+
+	if err != nil && !isCanceled(err) {
+		r.GetOrRegisterCounter("stream.recvs.errors").With(labels...).Add(1)
+	}
+}
+
+func parseFullMethod(fullMethod string) (string, string) {
+	parts := strings.Split(fullMethod, "/")
+	if len(parts) < 3 {
+		return grpcmetrics.Unknown, grpcmetrics.Unknown
+	}
+
+	fullService := parts[1]
+	method := parts[2]
+
+	sp := strings.Split(fullService, ".")
+	service := sp[len(sp)-1]
+
+	return dasherize(service), dasherize(method)
+}
+
+// code returns the gRPC error code, handling context and unknown errors.
+func code(err error) string {
+	if err == context.Canceled {
+		return grpcmetrics.Canceled
+	}
+
+	st, ok := status.FromError(err)
+	if !ok {
+		return grpcmetrics.Unknown
+	}
+
+	return dasherize(st.Code().String())
+}
+
+var uppers = regexp.MustCompile(`([[:lower:]])([[:upper:]])`)
+
+func dasherize(s string) string {
+	return strings.ToLower(uppers.ReplaceAllString(s, "$1-$2"))
+}
+
+// isCanceled returns true if error is a context or gRPC cancelation error.
+func isCanceled(err error) bool {
+	if err == context.Canceled {
+		return true
+	}
+
+	if st, ok := status.FromError(err); ok {
+		return st.Code() == codes.Canceled
+	}
+
+	return false
+}
+
+func ms(d time.Duration) float64 {
+	return float64(d.Milliseconds())
+}

grpc/grpcmetrics/highcardmetrics/server_test.go

@@ -0,0 +1,117 @@
+package highcardmetrics
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"google.golang.org/grpc"
+
+	"github.com/heroku/x/go-kit/metrics/testmetrics"
+)
+
+func TestUnaryServerInterceptor(t *testing.T) {
+	p := testmetrics.NewProvider(t)
+	usi := NewUnaryServerInterceptor(p)
+	handler := func(resp interface{}, err error) grpc.UnaryHandler {
+		return func(ctx context.Context, req interface{}) (interface{}, error) {
+			return resp, err
+		}
+	}
+	info := &grpc.UnaryServerInfo{
+		FullMethod: "/spec.Hello/Ping",
+	}
+
+	resp, err := usi(context.Background(), "ping", info, handler("pong", nil))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp != "pong" {
+		t.Fatalf("resp = %v, want %q", resp, "pong")
+	}
+
+	_, err = usi(context.Background(), "ping", info, handler(nil, errors.New("test")))
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+
+	_, err = usi(context.Background(), "ping", info, handler(nil, context.Canceled))
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+
+	p.CheckCounter("grpc.server.requests", 1, "service", "hello", "method", "ping", "response-status", "ok")
+	p.CheckCounter("grpc.server.requests", 1, "service", "hello", "method", "ping", "response-status", "canceled")
+	p.CheckCounter("grpc.server.requests", 1, "service", "hello", "method", "ping", "response-status", "unknown")
+
+	p.CheckObservationCount("grpc.server.request-duration.ms", 1, "service", "hello", "method", "ping", "response-status", "ok")
+	p.CheckObservationCount("grpc.server.request-duration.ms", 1, "service", "hello", "method", "ping", "response-status", "canceled")
+	p.CheckObservationCount("grpc.server.request-duration.ms", 1, "service", "hello", "method", "ping", "response-status", "unknown")
+}
+
+func TestStreamServerInterceptor(t *testing.T) {
+	p := testmetrics.NewProvider(t)
+	ssi := NewStreamServerInterceptor(p)
+	handler := func(err error) grpc.StreamHandler {
+		return func(srv interface{}, stream grpc.ServerStream) error {
+			if err == nil {
+				if err := stream.SendMsg("ping"); err != nil {
+					t.Fatal("unexpected error", err)
+				}
+				if err := stream.RecvMsg("pong"); err != nil {
+					t.Fatal("unexpected error", err)
+				}
+				if err := stream.SendMsg("ping"); err != nil {
+					t.Fatal("unexpected error", err)
+				}
+			}
+			return err
+		}
+	}
+	info := &grpc.StreamServerInfo{
+		FullMethod: "/spec.Hello/StreamUpdates",
+	}
+
+	err := ssi(nil, &testServerStream{}, info, handler(nil))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = ssi(nil, &testServerStream{}, info, func(srv interface{}, stream grpc.ServerStream) error {
+		p.CheckGauge("grpc.server.stream.clients", 1, "service", "hello", "method", "stream-updates")
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = ssi(nil, &testServerStream{}, info, handler(errors.New("test")))
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+
+	p.CheckCounter("grpc.server.requests", 2, "service", "hello", "method", "stream-updates", "response-status", "ok")
+	p.CheckCounter("grpc.server.requests", 1, "service", "hello", "method", "stream-updates", "response-status", "unknown")
+	p.CheckObservationCount("grpc.server.request-duration.ms", 2, "service", "hello", "method", "stream-updates", "response-status", "ok")
+	p.CheckObservationCount("grpc.server.request-duration.ms", 1, "service", "hello", "method", "stream-updates", "response-status", "unknown")
+
+	p.CheckGauge("grpc.server.stream.clients", 0, "service", "hello", "method", "stream-updates")
+
+	p.CheckCounter("grpc.server.stream.sends", 2, "service", "hello", "method", "stream-updates")
+	p.CheckObservationCount("grpc.server.stream.send-duration.ms", 2, "service", "hello", "method", "stream-updates")
+
+	p.CheckCounter("grpc.server.stream.recvs", 1, "service", "hello", "method", "stream-updates")
+	p.CheckObservationCount("grpc.server.stream.recv-duration.ms", 1, "service", "hello", "method", "stream-updates")
+}
+
+type testServerStream struct {
+	grpc.ServerStream
+}
+
+func (*testServerStream) SendMsg(m interface{}) error {
+	return nil
+}
+
+func (*testServerStream) RecvMsg(m interface{}) error {
+	return nil
+}

grpc/grpcserver/options.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/heroku/x/go-kit/metrics"
 	"github.com/heroku/x/grpc/grpcmetrics"
+	"github.com/heroku/x/grpc/grpcmetrics/highcardmetrics"
 	"github.com/heroku/x/grpc/panichandler"
 	"github.com/heroku/x/tlsconfig"
 )
@@ -26,11 +27,12 @@ var defaultLogOpts = []grpc_logrus.Option{
 }
 
 type options struct {
-	logEntry               *logrus.Entry
-	metricsProvider        metrics.Provider
-	authUnaryInterceptor   grpc.UnaryServerInterceptor
-	authStreamInterceptor  grpc.StreamServerInterceptor
-	useValidateInterceptor bool
+	logEntry                   *logrus.Entry
+	metricsProvider            metrics.Provider
+	authUnaryInterceptor       grpc.UnaryServerInterceptor
+	authStreamInterceptor      grpc.StreamServerInterceptor
+	useValidateInterceptor     bool
+	highCardinalityInterceptor bool
 
 	grpcOptions []grpc.ServerOption
 }
@@ -86,6 +88,14 @@ func ValidateInterceptor() ServerOption {
 	}
 }
 
+// HighCardinalityInterceptor sets interceptors that use
+// Attributes/Labels on the instrumentation.
+func HighCardinalityInterceptor() ServerOption {
+	return func(o *options) {
+		o.highCardinalityInterceptor = true
+	}
+}
+
 func (o *options) unaryInterceptors() []grpc.UnaryServerInterceptor {
 	l := o.logEntry
 	if l == nil {
@@ -100,7 +110,11 @@ func (o *options) unaryInterceptors() []grpc.UnaryServerInterceptor {
 		unaryPeerNameTagger,
 	}
 	if o.metricsProvider != nil {
-		i = append(i, grpcmetrics.NewUnaryServerInterceptor(o.metricsProvider)) // report metrics on unwrapped errors
+		if o.highCardinalityInterceptor {
+			i = append(i, highcardmetrics.NewUnaryServerInterceptor(o.metricsProvider))
+		} else {
+			i = append(i, grpcmetrics.NewUnaryServerInterceptor(o.metricsProvider)) // report metrics on unwrapped errors
+		}
 	}
 	i = append(i,
 		unaryServerErrorUnwrapper, // unwrap after we've logged
@@ -129,7 +143,11 @@ func (o *options) streamInterceptors() []grpc.StreamServerInterceptor {
 		streamPeerNameTagger,
 	}
 	if o.metricsProvider != nil {
-		i = append(i, grpcmetrics.NewStreamServerInterceptor(o.metricsProvider)) // report metrics on unwrapped errors
+		if o.highCardinalityInterceptor {
+			i = append(i, highcardmetrics.NewStreamServerInterceptor(o.metricsProvider))
+		} else {
+			i = append(i, grpcmetrics.NewStreamServerInterceptor(o.metricsProvider)) // report metrics on unwrapped errors
+		}
 	}
 	i = append(i,
 		streamServerErrorUnwrapper, // unwrap after we've logged