feat(workers): Add metrics for worker server-side grpc connections, refactor repeated code into common location (#2367)

* feat(workers): added latencies histogram for server side multihop workers

* refactor(daemon): combine common grpc metric functions into daemon/internal/metric

Co-authored-by: Todd <github@quaddi.com>

internal/daemon/controller/internal/metric/api.go

@@ -7,10 +7,10 @@ import (
 	"net/http"
 	"path"
 	"regexp"
-	"strconv"
 	"strings"
 
 	"github.com/hashicorp/boundary/globals"
+	"github.com/hashicorp/boundary/internal/daemon/internal/metric"
 	"github.com/hashicorp/boundary/internal/gen/controller/api/services"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -26,6 +26,11 @@ var (
 	expectedPathsToMethods map[string][]string
 )
 
+const (
+	invalidPathValue = "invalid"
+	apiSubSystem     = "controller_api"
+)
+
 func init() {
 	expectedPathsToMethods = apiPathsAndMethods()
 	pathRegex = make(map[*regexp.Regexp]string)
@@ -109,15 +114,6 @@ func buildRegexFromPath(p string) *regexp.Regexp {
 	return regexp.MustCompile(fmt.Sprintf("^%s$", escapedPathRegex))
 }
 
-const (
-	invalidPathValue = "invalid"
-
-	labelHttpCode   = "code"
-	labelHttpPath   = "path"
-	labelHttpMethod = "method"
-	apiSubSystem    = "controller_api"
-)
-
 var (
 	// 100 bytes, 1kb, 10kb, 100kb, 1mb, 10mb, 100mb, 1gb
 	msgSizeBuckets = prometheus.ExponentialBuckets(100, 10, 8)
@@ -133,7 +129,7 @@ var (
 			Help:      "Histogram of latencies for HTTP requests.",
 			Buckets:   prometheus.DefBuckets,
 		},
-		[]string{labelHttpCode, labelHttpPath, labelHttpMethod},
+		metric.ListHttpLabels,
 	)
 
 	// httpRequestSize collections measurements of how large each request
@@ -146,7 +142,7 @@ var (
 			Help:      "Histogram of request sizes for HTTP requests.",
 			Buckets:   msgSizeBuckets,
 		},
-		[]string{labelHttpCode, labelHttpPath, labelHttpMethod},
+		metric.ListHttpLabels,
 	)
 
 	// httpRequestSize collections measurements of how large each response
@@ -159,7 +155,7 @@ var (
 			Help:      "Histogram of response sizes for HTTP responses.",
 			Buckets:   msgSizeBuckets,
 		},
-		[]string{labelHttpCode, labelHttpPath, labelHttpMethod},
+		metric.ListHttpLabels,
 	)
 )
 
@@ -223,7 +219,7 @@ func pathLabel(incomingPath string) string {
 func InstrumentApiHandler(wrapped http.Handler) http.Handler {
 	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
 		l := prometheus.Labels{
-			labelHttpPath: pathLabel(req.URL.Path),
+			metric.LabelHttpPath: pathLabel(req.URL.Path),
 		}
 		promhttp.InstrumentHandlerDuration(
 			httpRequestLatency.MustCurryWith(l),
@@ -242,31 +238,7 @@ func InstrumentApiHandler(wrapped http.Handler) http.Handler {
 // prometheus register and initializes them to 0 for all possible label
 // combinations.
 func InitializeApiCollectors(r prometheus.Registerer) {
-	if r == nil {
-		return
-	}
-	r.MustRegister(httpResponseSize, httpRequestSize, httpRequestLatency)
-
-	for p, methods := range expectedPathsToMethods {
-		for _, m := range methods {
-			for _, sc := range expectedStatusCodesPerMethod[m] {
-				l := prometheus.Labels{labelHttpCode: strconv.Itoa(sc), labelHttpPath: p, labelHttpMethod: strings.ToLower(m)}
-				httpResponseSize.With(l)
-				httpRequestSize.With(l)
-				httpRequestLatency.With(l)
-			}
-		}
-	}
-
-	// When an invalid path is found, any method is possible both we expect
-	// an error response.
-	p := invalidPathValue
-	for m := range expectedStatusCodesPerMethod {
-		for _, sc := range []int{http.StatusNotFound, http.StatusMethodNotAllowed} {
-			l := prometheus.Labels{labelHttpCode: strconv.Itoa(sc), labelHttpPath: p, labelHttpMethod: strings.ToLower(m)}
-			httpResponseSize.With(l)
-			httpRequestSize.With(l)
-			httpRequestLatency.With(l)
-		}
+	for _, v := range []prometheus.ObserverVec{httpRequestLatency, httpRequestSize, httpResponseSize} {
+		metric.InitializeApiCollectors(r, v, expectedPathsToMethods, expectedStatusCodesPerMethod)
 	}
 }

internal/daemon/controller/internal/metric/cluster.go

@@ -2,28 +2,27 @@ package metric
 
 import (
 	"context"
-	"strings"
 
 	"github.com/hashicorp/boundary/globals"
-	"github.com/hashicorp/boundary/internal/errors"
+	metric "github.com/hashicorp/boundary/internal/daemon/internal/metric"
 	"github.com/prometheus/client_golang/prometheus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/stats"
-	"google.golang.org/grpc/status"
 )
 
 const (
-	labelGRpcCode    = "grpc_code"
-	labelGRpcService = "grpc_service"
-	labelGRpcMethod  = "grpc_method"
 	clusterSubSystem = "controller_cluster"
 )
 
-// gRpcRequestLatency collects measurements of how long it takes
+// Because we use grpc's stats.Handler to track close-to-the-wire server-side communication over grpc,
+// there is no easy way to measure request and response size as we are recording latency. Thus we only
+// track the request latency for server-side grpc connections.
+
+// grpcRequestLatency collects measurements of how long it takes
 // the boundary system to reply to a request to the controller cluster
 // from the time that boundary received the request.
-var gRpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
+var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 	prometheus.HistogramOpts{
 		Namespace: globals.MetricNamespace,
 		Subsystem: clusterSubSystem,
@@ -31,69 +30,12 @@ var gRpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 		Help:      "Histogram of latencies for gRPC requests.",
 		Buckets:   prometheus.DefBuckets,
 	},
-	[]string{labelGRpcCode, labelGRpcService, labelGRpcMethod},
+	metric.ListGrpcLabels,
 )
 
-// statusFromError retrieves the *status.Status from the provided error.  It'll
-// attempt to unwrap the *status.Error, which is something status.FromError
-// does not do.
-func statusFromError(err error) *status.Status {
-	if s, ok := status.FromError(err); ok {
-		return s
-	}
-
-	type gRPCStatus interface {
-		GRPCStatus() *status.Status
-	}
-	var unwrappedStatus gRPCStatus
-	if ok := errors.As(err, &unwrappedStatus); ok {
-		return unwrappedStatus.GRPCStatus()
-	}
-
-	return status.New(codes.Unknown, "Unknown Code")
-}
-
-func splitMethodName(fullMethodName string) (string, string) {
-	fullMethodName = strings.TrimPrefix(fullMethodName, "/") // remove leading slash
-	if i := strings.Index(fullMethodName, "/"); i >= 0 {
-		return fullMethodName[:i], fullMethodName[i+1:]
-	}
-	return "unknown", "unknown"
-}
-
-type metricMethodNameContextKey struct{}
-
-type statsHandler struct{}
-
-func (sh statsHandler) TagRPC(ctx context.Context, i *stats.RPCTagInfo) context.Context {
-	return context.WithValue(ctx, metricMethodNameContextKey{}, i.FullMethodName)
-}
-
-func (sh statsHandler) TagConn(ctx context.Context, _ *stats.ConnTagInfo) context.Context {
-	return ctx
-}
-
-func (sh statsHandler) HandleConn(context.Context, stats.ConnStats) {
-}
-
-func (sh statsHandler) HandleRPC(ctx context.Context, s stats.RPCStats) {
-	switch v := s.(type) {
-	case *stats.End:
-		// Accept the ok, but ignore it. This code doesn't need to panic
-		// and if "fullName" is an empty string splitMethodName will
-		// set service and method to "unknown".
-		fullName, _ := ctx.Value(metricMethodNameContextKey{}).(string)
-		service, method := splitMethodName(fullName)
-		l := prometheus.Labels{
-			labelGRpcMethod:  method,
-			labelGRpcService: service,
-			labelGRpcCode:    statusFromError(v.Error).Code().String(),
-		}
-		gRpcRequestLatency.With(l).Observe(v.EndTime.Sub(v.BeginTime).Seconds())
-	}
-}
-
-var allCodes = []codes.Code{
+// All the codes expected to be returned by boundary or the grpc framework to
+// requests to the cluster server.
+var expectedGrpcCodes = []codes.Code{
 	codes.OK, codes.InvalidArgument, codes.PermissionDenied,
 	codes.FailedPrecondition,
 
@@ -105,29 +47,13 @@ var allCodes = []codes.Code{
 
 // InstrumentClusterStatsHandler returns a gRPC stats.Handler which observes
 // cluster specific metrics. Use with the cluster gRPC server.
-func InstrumentClusterStatsHandler() statsHandler {
-	return statsHandler{}
+func InstrumentClusterStatsHandler(ctx context.Context) (stats.Handler, error) {
+	return metric.NewStatsHandler(ctx, grpcRequestLatency)
 }
 
 // InitializeClusterCollectors registers the cluster metrics to the default
 // prometheus register and initializes them to 0 for all possible label
 // combinations.
 func InitializeClusterCollectors(r prometheus.Registerer, server *grpc.Server) {
-	if r == nil {
-		return
-	}
-	r.MustRegister(gRpcRequestLatency)
-
-	for serviceName, info := range server.GetServiceInfo() {
-		for _, mInfo := range info.Methods {
-			for _, c := range allCodes {
-				l := prometheus.Labels{
-					labelGRpcMethod:  mInfo.Name,
-					labelGRpcService: serviceName,
-					labelGRpcCode:    c.String(),
-				}
-				gRpcRequestLatency.With(l)
-			}
-		}
-	}
+	metric.InitializeGrpcCollectorsFromServer(r, grpcRequestLatency, server, expectedGrpcCodes)
 }

internal/daemon/controller/listeners.go

@@ -195,9 +195,13 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 	if err != nil {
 		return nil, fmt.Errorf("error getting request interceptor for worker proto: %w", err)
 	}
+	statsHandler, err := metric.InstrumentClusterStatsHandler(c.baseContext)
+	if err != nil {
+		return nil, errors.Wrap(c.baseContext, err, op)
+	}
 
 	workerServer := grpc.NewServer(
-		grpc.StatsHandler(metric.InstrumentClusterStatsHandler()),
+		grpc.StatsHandler(statsHandler),
 		grpc.MaxRecvMsgSize(math.MaxInt32),
 		grpc.MaxSendMsgSize(math.MaxInt32),
 		grpc.UnaryInterceptor(

internal/daemon/internal/metric/initialize_metrics.go

@@ -0,0 +1,124 @@
+// Package metric provides functions to initialize the controller specific
+// collectors and hooks to measure metrics and update the relevant collectors.
+package metric
+
+import (
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
+)
+
+const (
+	LabelGrpcService = "grpc_service"
+	LabelGrpcMethod  = "grpc_method"
+	LabelGrpcCode    = "grpc_code"
+	LabelHttpPath    = "path"
+	LabelHttpMethod  = "method"
+	LabelHttpCode    = "code"
+
+	invalidPathValue = "invalid"
+)
+
+var (
+	ListGrpcLabels = []string{LabelGrpcService, LabelGrpcMethod, LabelGrpcCode}
+	ListHttpLabels = []string{LabelHttpPath, LabelHttpMethod, LabelHttpCode}
+)
+
+/* The following methods are used to initialize Prometheus histogram vectors for gRPC connections. */
+
+func rangeProtoFiles(m map[string][]string, fd protoreflect.FileDescriptor) bool {
+	if fd.Services().Len() == 0 {
+		return true
+	}
+
+	for i := 0; i < fd.Services().Len(); i++ {
+		s := fd.Services().Get(i)
+		if s.Methods().Len() == 0 {
+			continue
+		}
+
+		methods := []string{}
+		for j := 0; j < s.Methods().Len(); j++ {
+			methods = append(methods, string(s.Methods().Get(j).Name()))
+		}
+		m[string(s.FullName())] = methods
+	}
+
+	return true
+}
+
+// InitializeGrpcCollectorsFromPackage registers and zeroes a Prometheus
+// histogram, populating all service and method labels by ranging through a
+// given protobuf package.
+func InitializeGrpcCollectorsFromPackage(r prometheus.Registerer, v prometheus.ObserverVec, pkg protoreflect.FileDescriptor, codes []codes.Code) {
+	if r == nil {
+		return
+	}
+	r.MustRegister(v)
+
+	serviceNamesToMethodNames := make(map[string][]string, 0)
+	protoregistry.GlobalFiles.RangeFilesByPackage(
+		pkg.Package(),
+		func(fd protoreflect.FileDescriptor) bool { return rangeProtoFiles(serviceNamesToMethodNames, fd) },
+	)
+
+	for serviceName, serviceMethods := range serviceNamesToMethodNames {
+		for _, sm := range serviceMethods {
+			for _, c := range codes {
+				v.With(prometheus.Labels{LabelGrpcService: serviceName, LabelGrpcMethod: sm, LabelGrpcCode: c.String()})
+			}
+		}
+	}
+}
+
+// InitializeGrpcCollectorsFromServer registers and zeroes a Prometheus
+// histogram, finding all service and method labels from the provided gRPC
+// server.
+func InitializeGrpcCollectorsFromServer(r prometheus.Registerer, v prometheus.ObserverVec, server *grpc.Server, codes []codes.Code) {
+	if r == nil {
+		return
+	}
+	r.MustRegister(v)
+
+	for serviceName, info := range server.GetServiceInfo() {
+		for _, mInfo := range info.Methods {
+			for _, c := range codes {
+				v.With(prometheus.Labels{LabelGrpcService: serviceName, LabelGrpcMethod: mInfo.Name, LabelGrpcCode: c.String()})
+			}
+		}
+	}
+}
+
+/* The following methods are used to initialize Prometheus histogram vectors for http requests. */
+
+// InitializeApiCollectors registers and zeroes a Prometheus
+// histogram, populating all path, code, and method labels from the
+// provided maps in its parameters.
+func InitializeApiCollectors(r prometheus.Registerer, v prometheus.ObserverVec, expectedPathsToMethods map[string][]string, expectedStatusCodesPerMethod map[string][]int) {
+	if r == nil {
+		return
+	}
+	r.MustRegister(v)
+
+	for p, methods := range expectedPathsToMethods {
+		for _, m := range methods {
+			for _, sc := range expectedStatusCodesPerMethod[m] {
+				v.With(prometheus.Labels{LabelHttpPath: p, LabelHttpMethod: strings.ToLower(m), LabelHttpCode: strconv.Itoa(sc)})
+			}
+		}
+	}
+
+	// When an invalid path is found, any method is possible, but we expect an error response.
+	p := invalidPathValue
+	for m := range expectedStatusCodesPerMethod {
+		for _, sc := range []int{http.StatusNotFound, http.StatusMethodNotAllowed} {
+			v.With(prometheus.Labels{LabelHttpPath: p, LabelHttpMethod: strings.ToLower(m), LabelHttpCode: strconv.Itoa(sc)})
+		}
+	}
+}

internal/daemon/internal/metric/instrument_handlers.go

@@ -0,0 +1,96 @@
+// Package metric provides functions to initialize the controller specific
+// collectors and hooks to measure metrics and update the relevant collectors.
+package metric
+
+import (
+	"context"
+	"strings"
+
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/util"
+	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/stats"
+	"google.golang.org/grpc/status"
+)
+
+/* The following methods are used to instrument handlers for gRPC server and client connections. */
+
+// statsHandler satisfies grpc's stats.Handler interface. This helps measure the latency of grpc requests as close to the
+// wire as possible, and allows us to capture error codes returned by the grpc go library which our service may
+// never return, or error codes for requests that our service may never even see.
+type statsHandler struct {
+	reqLatency prometheus.ObserverVec
+}
+
+// NewStatsHandler takes a request latency metric (prometheus.ObserverVec) and
+// returns a grpc stats.Handler that updates the provided metric with the
+// request latency.
+func NewStatsHandler(ctx context.Context, o prometheus.ObserverVec) (*statsHandler, error) {
+	const op = "metric.NewStatsHandler"
+	if util.IsNil(o) {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "prometheus.ObserverVec is nil")
+	}
+	return &statsHandler{reqLatency: o}, nil
+}
+
+var _ stats.Handler = (*statsHandler)(nil)
+
+type metricMethodNameContextKey struct{}
+
+func (sh *statsHandler) TagRPC(ctx context.Context, i *stats.RPCTagInfo) context.Context {
+	return context.WithValue(ctx, metricMethodNameContextKey{}, i.FullMethodName)
+}
+
+func (sh *statsHandler) TagConn(ctx context.Context, _ *stats.ConnTagInfo) context.Context {
+	return ctx
+}
+
+func (sh *statsHandler) HandleConn(context.Context, stats.ConnStats) {
+}
+
+func (sh *statsHandler) HandleRPC(ctx context.Context, s stats.RPCStats) {
+	switch v := s.(type) {
+	case *stats.End:
+		// Accept the ok, but ignore it. This code doesn't need to panic
+		// and if "fullName" is an empty string SplitMethodName will
+		// set service and method to "unknown".
+		fullName, _ := ctx.Value(metricMethodNameContextKey{}).(string)
+		service, method := SplitMethodName(fullName)
+		labels := prometheus.Labels{
+			LabelGrpcMethod:  method,
+			LabelGrpcService: service,
+			LabelGrpcCode:    StatusFromError(v.Error).Code().String(),
+		}
+		sh.reqLatency.With(labels).Observe(v.EndTime.Sub(v.BeginTime).Seconds())
+	}
+}
+
+// StatusFromError retrieves the *status.Status from the provided error.  It'll
+// attempt to unwrap the *status.Error, which is something status.FromError
+// does not do.
+func StatusFromError(err error) *status.Status {
+	if s, ok := status.FromError(err); ok {
+		return s
+	}
+
+	type gRPCStatus interface {
+		GRPCStatus() *status.Status
+	}
+	var unwrappedStatus gRPCStatus
+	if ok := errors.As(err, &unwrappedStatus); ok {
+		return unwrappedStatus.GRPCStatus()
+	}
+
+	return status.New(codes.Unknown, "Unknown Code")
+}
+
+// SplitMethodName returns the service and the method name when given the full
+// method name as provided by the grpc request handler.
+func SplitMethodName(fullMethodName string) (string, string) {
+	fullMethodName = strings.TrimPrefix(fullMethodName, "/") // remove leading slash
+	if i := strings.Index(fullMethodName, "/"); i >= 0 {
+		return fullMethodName[:i], fullMethodName[i+1:]
+	}
+	return "unknown", "unknown"
+}

internal/daemon/internal/metric/instrument_handlers_test.go

@@ -8,19 +8,13 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
 )
 
-func TestStatsHandler(t *testing.T) {
-	bkpLatency := gRpcRequestLatency
-	defer func() {
-		gRpcRequestLatency = bkpLatency
-	}()
-
-	handler := InstrumentClusterStatsHandler()
-
+func TestNewStatsHandler(t *testing.T) {
 	cases := []struct {
 		name           string
 		stats          []stats.RPCStats
@@ -38,9 +32,9 @@ func TestStatsHandler(t *testing.T) {
 				},
 			},
 			wantedLabels: map[string]string{
-				labelGRpcCode:    "OK",
-				labelGRpcMethod:  "method",
-				labelGRpcService: "some.service.path",
+				LabelGrpcCode:    "OK",
+				LabelGrpcMethod:  "method",
+				LabelGrpcService: "some.service.path",
 			},
 			wantedLatency: (4 * time.Second).Seconds(),
 		},
@@ -68,9 +62,9 @@ func TestStatsHandler(t *testing.T) {
 				},
 			},
 			wantedLabels: map[string]string{
-				labelGRpcCode:    "OK",
-				labelGRpcMethod:  "method",
-				labelGRpcService: "some.service.path",
+				LabelGrpcCode:    "OK",
+				LabelGrpcMethod:  "method",
+				LabelGrpcService: "some.service.path",
 			},
 			wantedLatency: (4 * time.Second).Seconds(),
 		},
@@ -84,9 +78,9 @@ func TestStatsHandler(t *testing.T) {
 				},
 			},
 			wantedLabels: map[string]string{
-				labelGRpcCode:    "OK",
-				labelGRpcMethod:  "unknown",
-				labelGRpcService: "unknown",
+				LabelGrpcCode:    "OK",
+				LabelGrpcMethod:  "unknown",
+				LabelGrpcService: "unknown",
 			},
 			wantedLatency: (4 * time.Second).Seconds(),
 		},
@@ -101,9 +95,9 @@ func TestStatsHandler(t *testing.T) {
 				},
 			},
 			wantedLabels: map[string]string{
-				labelGRpcCode:    "Canceled",
-				labelGRpcMethod:  "method",
-				labelGRpcService: "some.service.path",
+				LabelGrpcCode:    "Canceled",
+				LabelGrpcMethod:  "method",
+				LabelGrpcService: "some.service.path",
 			},
 			wantedLatency: (4 * time.Second).Seconds(),
 		},
@@ -118,19 +112,20 @@ func TestStatsHandler(t *testing.T) {
 				},
 			},
 			wantedLabels: map[string]string{
-				labelGRpcCode:    "InvalidArgument",
-				labelGRpcMethod:  "method",
-				labelGRpcService: "some.service.path",
+				LabelGrpcCode:    "InvalidArgument",
+				LabelGrpcMethod:  "method",
+				LabelGrpcService: "some.service.path",
 			},
 			wantedLatency: (4 * time.Second).Seconds(),
 		},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			testableLatency := &testableObserverVec{}
-			gRpcRequestLatency = testableLatency
-
 			ctx := context.Background()
+			testableLatency := &TestableObserverVec{}
+			handler, err := NewStatsHandler(ctx, testableLatency)
+			require.NoError(t, err)
+
 			ctx = handler.TagRPC(ctx, &stats.RPCTagInfo{
 				FullMethodName: tc.fullMethodName,
 			})
@@ -139,31 +134,9 @@ func TestStatsHandler(t *testing.T) {
 				handler.HandleRPC(ctx, i)
 			}
 
-			assert.Len(t, testableLatency.observations, 1)
-			assert.Equal(t, testableLatency.observations[0].observation, tc.wantedLatency)
-			assert.Equal(t, testableLatency.observations[0].labels, tc.wantedLabels)
+			assert.Len(t, testableLatency.Observations, 1)
+			assert.Equal(t, testableLatency.Observations[0].Observation, tc.wantedLatency)
+			assert.Equal(t, testableLatency.Observations[0].Labels, tc.wantedLabels)
 		})
 	}
 }
-
-// testableObserverVec allows us to assert which observations are being made
-// with which labels.
-type testableObserverVec struct {
-	observations []*testableObserver
-	prometheus.ObserverVec
-}
-
-func (v *testableObserverVec) With(l prometheus.Labels) prometheus.Observer {
-	ret := &testableObserver{labels: l}
-	v.observations = append(v.observations, ret)
-	return ret
-}
-
-type testableObserver struct {
-	labels      prometheus.Labels
-	observation float64
-}
-
-func (o *testableObserver) Observe(f float64) {
-	o.observation = f
-}

internal/daemon/internal/metric/testing.go

@@ -0,0 +1,47 @@
+package metric
+
+import (
+	"context"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+// testableObserverVec allows us to assert which observations are being made
+// with which labels.
+type TestableObserverVec struct {
+	Observations []*testableObserver
+	prometheus.ObserverVec
+}
+
+func (v *TestableObserverVec) With(l prometheus.Labels) prometheus.Observer {
+	ret := &testableObserver{Labels: l}
+	v.Observations = append(v.Observations, ret)
+	return ret
+}
+
+type testableObserver struct {
+	Labels      prometheus.Labels
+	Observation float64
+}
+
+func (o *testableObserver) Observe(f float64) {
+	o.Observation = f
+}
+
+type TestInvoker struct {
+	T      *testing.T
+	Called bool
+	RetErr error
+}
+
+func (i *TestInvoker) Invoke(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, opts ...grpc.CallOption) error {
+	i.Called = true
+
+	require.NotNil(i.T, ctx)
+	require.NotEmpty(i.T, method)
+	require.NotNil(i.T, req)
+	return i.RetErr
+}

internal/daemon/worker/internal/metric/cluster_client.go

@@ -2,25 +2,17 @@ package metric
 
 import (
 	"context"
-	"strings"
 	"time"
 
 	"github.com/hashicorp/boundary/globals"
-	"github.com/hashicorp/boundary/internal/errors"
+	metric "github.com/hashicorp/boundary/internal/daemon/internal/metric"
 	"github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/prometheus/client_golang/prometheus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-	"google.golang.org/protobuf/reflect/protoreflect"
-	"google.golang.org/protobuf/reflect/protoregistry"
 )
 
 const (
-	labelGrpcCode    = "grpc_code"
-	labelGrpcService = "grpc_service"
-	labelGrpcMethod  = "grpc_method"
-
 	clusterClientSubsystem = "cluster_client"
 )
 
@@ -34,49 +26,26 @@ var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 		Help:      "Histogram of latencies for gRPC requests between the cluster and any of its clients.",
 		Buckets:   prometheus.DefBuckets,
 	},
-	[]string{labelGrpcCode, labelGrpcService, labelGrpcMethod},
+	metric.ListGrpcLabels,
 )
 
-// statusFromError retrieves the *status.Status from the provided error. It'll
-// attempt to unwrap the *status.Error, which is something status.FromError
-// does not do.
-func statusFromError(err error) *status.Status {
-	if s, ok := status.FromError(err); ok {
-		return s
-	}
-
-	type gRPCStatus interface {
-		GRPCStatus() *status.Status
-	}
-	var unwrappedStatus gRPCStatus
-	if ok := errors.As(err, &unwrappedStatus); ok {
-		return unwrappedStatus.GRPCStatus()
-	}
-
-	return status.New(codes.Unknown, "Unknown Code")
-}
-
-func splitMethodName(fullMethodName string) (string, string) {
-	fullMethodName = strings.TrimPrefix(fullMethodName, "/") // remove leading slash
-	if i := strings.Index(fullMethodName, "/"); i >= 0 {
-		return fullMethodName[:i], fullMethodName[i+1:]
-	}
-	return "unknown", "unknown"
-}
-
 type requestRecorder struct {
-	labels prometheus.Labels
+	reqLatency prometheus.ObserverVec
+	labels     prometheus.Labels
 
 	// measurements
 	start time.Time
 }
 
-func newRequestRecorder(fullMethodName string) requestRecorder {
-	service, method := splitMethodName(fullMethodName)
+// NewRequestRecorder creates a requestRecorder struct which is used to measure gRPC client request latencies.
+// For testing purposes, this method is exported.
+func newRequestRecorder(fullMethodName string, reqLatency prometheus.ObserverVec) requestRecorder {
+	service, method := metric.SplitMethodName(fullMethodName)
 	r := requestRecorder{
+		reqLatency: reqLatency,
 		labels: prometheus.Labels{
-			labelGrpcMethod:  method,
-			labelGrpcService: service,
+			metric.LabelGrpcMethod:  method,
+			metric.LabelGrpcService: service,
 		},
 		start: time.Now(),
 	}
@@ -84,9 +53,17 @@ func newRequestRecorder(fullMethodName string) requestRecorder {
 	return r
 }
 
-func (r requestRecorder) record(err error) {
-	r.labels[labelGrpcCode] = statusFromError(err).Code().String()
-	grpcRequestLatency.With(r.labels).Observe(time.Since(r.start).Seconds())
+func (r requestRecorder) Record(err error) {
+	r.labels[metric.LabelGrpcCode] = metric.StatusFromError(err).Code().String()
+	r.reqLatency.With(r.labels).Observe(time.Since(r.start).Seconds())
+}
+
+// The expected codes returned by the grpc client calls to cluster services.
+var expectedGrpcClientCodes = []codes.Code{
+	codes.OK, codes.Canceled, codes.Unknown, codes.InvalidArgument, codes.DeadlineExceeded, codes.NotFound,
+	codes.AlreadyExists, codes.PermissionDenied, codes.Unauthenticated, codes.ResourceExhausted,
+	codes.FailedPrecondition, codes.Aborted, codes.OutOfRange, codes.Unimplemented, codes.Internal,
+	codes.Unavailable, codes.DataLoss,
 }
 
 // InstrumentClusterClient wraps a UnaryClientInterceptor and records
@@ -94,62 +71,16 @@ func (r requestRecorder) record(err error) {
 // between the cluster and its clients.
 func InstrumentClusterClient() grpc.UnaryClientInterceptor {
 	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
-		r := newRequestRecorder(method)
+		r := newRequestRecorder(method, grpcRequestLatency)
 		err := invoker(ctx, method, req, reply, cc, opts...)
-		r.record(err)
+		r.Record(err)
 		return err
 	}
 }
 
-var allCodes = []codes.Code{
-	codes.OK, codes.Canceled, codes.Unknown, codes.InvalidArgument, codes.DeadlineExceeded, codes.NotFound,
-	codes.AlreadyExists, codes.PermissionDenied, codes.Unauthenticated, codes.ResourceExhausted,
-	codes.FailedPrecondition, codes.Aborted, codes.OutOfRange, codes.Unimplemented, codes.Internal,
-	codes.Unavailable, codes.DataLoss,
-}
-
+// InitializeClusterClientCollectors registers the cluster client metrics to the
+// prometheus register and initializes them to 0 for all possible label
+// combinations.
 func InitializeClusterClientCollectors(r prometheus.Registerer) {
-	if r == nil {
-		return
-	}
-	r.MustRegister(grpcRequestLatency)
-
-	serviceNamesToMethodNames := make(map[string][]string, 0)
-	protoregistry.GlobalFiles.RangeFilesByPackage(
-		services.File_controller_servers_services_v1_session_service_proto.Package(),
-		func(fd protoreflect.FileDescriptor) bool { return rangeProtofiles(serviceNamesToMethodNames, fd) },
-	)
-
-	for serviceName, serviceMethods := range serviceNamesToMethodNames {
-		for _, sm := range serviceMethods {
-			for _, c := range allCodes {
-				grpcRequestLatency.With(prometheus.Labels{
-					labelGrpcCode:    c.String(),
-					labelGrpcMethod:  sm,
-					labelGrpcService: serviceName,
-				})
-			}
-		}
-	}
-}
-
-func rangeProtofiles(m map[string][]string, fd protoreflect.FileDescriptor) bool {
-	if fd.Services().Len() == 0 {
-		return true
-	}
-
-	for i := 0; i < fd.Services().Len(); i++ {
-		s := fd.Services().Get(i)
-		if s.Methods().Len() == 0 {
-			continue
-		}
-
-		methods := []string{}
-		for j := 0; j < s.Methods().Len(); j++ {
-			methods = append(methods, string(s.Methods().Get(j).Name()))
-		}
-		m[string(s.FullName())] = methods
-	}
-
-	return true
+	metric.InitializeGrpcCollectorsFromPackage(r, grpcRequestLatency, services.File_controller_servers_services_v1_session_service_proto, expectedGrpcClientCodes)
 }

internal/daemon/worker/internal/metric/cluster_client_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	metric "github.com/hashicorp/boundary/internal/daemon/internal/metric"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -32,9 +33,9 @@ func TestRecorder(t *testing.T) {
 			methodName: "/some.service.path/method",
 			err:        nil,
 			wantedLabels: map[string]string{
-				labelGrpcCode:    "OK",
-				labelGrpcMethod:  "method",
-				labelGrpcService: "some.service.path",
+				metric.LabelGrpcCode:    "OK",
+				metric.LabelGrpcMethod:  "method",
+				metric.LabelGrpcService: "some.service.path",
 			},
 		},
 		{
@@ -42,9 +43,9 @@ func TestRecorder(t *testing.T) {
 			methodName: "unrecognized",
 			err:        nil,
 			wantedLabels: map[string]string{
-				labelGrpcCode:    "OK",
-				labelGrpcMethod:  "unknown",
-				labelGrpcService: "unknown",
+				metric.LabelGrpcCode:    "OK",
+				metric.LabelGrpcMethod:  "unknown",
+				metric.LabelGrpcService: "unknown",
 			},
 		},
 		{
@@ -52,9 +53,9 @@ func TestRecorder(t *testing.T) {
 			methodName: "/some.service.path/method",
 			err:        status.Error(codes.Canceled, ""),
 			wantedLabels: map[string]string{
-				labelGrpcCode:    "Canceled",
-				labelGrpcMethod:  "method",
-				labelGrpcService: "some.service.path",
+				metric.LabelGrpcCode:    "Canceled",
+				metric.LabelGrpcMethod:  "method",
+				metric.LabelGrpcService: "some.service.path",
 			},
 		},
 		{
@@ -62,9 +63,9 @@ func TestRecorder(t *testing.T) {
 			methodName: "/some.service.path/method",
 			err:        status.Error(codes.PermissionDenied, ""),
 			wantedLabels: map[string]string{
-				labelGrpcCode:    "PermissionDenied",
-				labelGrpcMethod:  "method",
-				labelGrpcService: "some.service.path",
+				metric.LabelGrpcCode:    "PermissionDenied",
+				metric.LabelGrpcMethod:  "method",
+				metric.LabelGrpcService: "some.service.path",
 			},
 		},
 	}
@@ -73,18 +74,15 @@ func TestRecorder(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			ogReqLatency := grpcRequestLatency
 			defer func() { grpcRequestLatency = ogReqLatency }()
-
-			testableLatency := &testableObserverVec{}
-			grpcRequestLatency = testableLatency
-
+			testableLatency := &metric.TestableObserverVec{}
 			start := time.Now()
-			tested := newRequestRecorder(tc.methodName)
-			tested.record(tc.err)
+			tested := newRequestRecorder(tc.methodName, testableLatency)
+			tested.Record(tc.err)
 
-			require.Len(t, testableLatency.observations, 1)
-			assert.Greater(t, testableLatency.observations[0].observation, float64(0))
-			assert.LessOrEqual(t, testableLatency.observations[0].observation, time.Since(start).Seconds())
-			assert.Equal(t, testableLatency.observations[0].labels, tc.wantedLabels)
+			require.Len(t, testableLatency.Observations, 1)
+			assert.Greater(t, testableLatency.Observations[0].Observation, float64(0))
+			assert.LessOrEqual(t, testableLatency.Observations[0].Observation, time.Since(start).Seconds())
+			assert.Equal(t, testableLatency.Observations[0].Labels, tc.wantedLabels)
 		})
 	}
 }
@@ -93,76 +91,39 @@ func TestInstrumentClusterClient(t *testing.T) {
 	ogReqLatency := grpcRequestLatency
 	defer func() { grpcRequestLatency = ogReqLatency }()
 
-	testableLatency := &testableObserverVec{}
+	testableLatency := &metric.TestableObserverVec{}
 	grpcRequestLatency = testableLatency
 
 	interceptor := InstrumentClusterClient()
-	i := &testInvoker{t: t, retErr: nil}
+	i := &metric.TestInvoker{T: t, RetErr: nil}
 
 	start := time.Now()
-	err := interceptor(context.Background(), "/some.service.path/method", wrapperspb.Bytes([]byte{1}), nil, nil, i.invoke, []grpc.CallOption{}...)
+	err := interceptor(context.Background(), "/some.service.path/method", wrapperspb.Bytes([]byte{1}), nil, nil, i.Invoke, []grpc.CallOption{}...)
 	require.NoError(t, err)
-	require.True(t, i.called)
+	require.True(t, i.Called)
 
-	require.Len(t, testableLatency.observations, 1)
-	assert.Greater(t, testableLatency.observations[0].observation, float64(0))
-	assert.LessOrEqual(t, testableLatency.observations[0].observation, time.Since(start).Seconds())
+	require.Len(t, testableLatency.Observations, 1)
+	assert.Greater(t, testableLatency.Observations[0].Observation, float64(0))
+	assert.LessOrEqual(t, testableLatency.Observations[0].Observation, time.Since(start).Seconds())
 }
 
 func TestInstrumentClusterClient_InvokerError(t *testing.T) {
 	ogReqLatency := grpcRequestLatency
 	defer func() { grpcRequestLatency = ogReqLatency }()
 
-	testableLatency := &testableObserverVec{}
+	testableLatency := &metric.TestableObserverVec{}
 	grpcRequestLatency = testableLatency
 
 	interceptor := InstrumentClusterClient()
-	i := &testInvoker{t: t, retErr: fmt.Errorf("oops!")}
+	i := &metric.TestInvoker{T: t, RetErr: fmt.Errorf("oops!")}
 
 	start := time.Now()
-	err := interceptor(context.Background(), "/some.service.path/method", wrapperspb.Bytes([]byte{1}), nil, nil, i.invoke, []grpc.CallOption{}...)
+	err := interceptor(context.Background(), "/some.service.path/method", wrapperspb.Bytes([]byte{1}), nil, nil, i.Invoke, []grpc.CallOption{}...)
 	require.EqualError(t, err, "oops!")
-	require.True(t, i.called)
+	require.True(t, i.Called)
 
 	// We still assert request latency in error states.
-	require.Len(t, testableLatency.observations, 1)
-	assert.Greater(t, testableLatency.observations[0].observation, float64(0))
-	assert.LessOrEqual(t, testableLatency.observations[0].observation, time.Since(start).Seconds())
-}
-
-type testInvoker struct {
-	t      *testing.T
-	called bool
-	retErr error
-}
-
-func (i *testInvoker) invoke(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, opts ...grpc.CallOption) error {
-	i.called = true
-
-	require.NotNil(i.t, ctx)
-	require.NotEmpty(i.t, method)
-	require.NotNil(i.t, req)
-	return i.retErr
-}
-
-// testableObserverVec allows us to assert which observations are being made
-// with which labels.
-type testableObserverVec struct {
-	observations []*testableObserver
-	prometheus.ObserverVec
-}
-
-func (v *testableObserverVec) With(l prometheus.Labels) prometheus.Observer {
-	ret := &testableObserver{labels: l}
-	v.observations = append(v.observations, ret)
-	return ret
-}
-
-type testableObserver struct {
-	labels      prometheus.Labels
-	observation float64
-}
-
-func (o *testableObserver) Observe(f float64) {
-	o.observation = f
+	require.Len(t, testableLatency.Observations, 1)
+	assert.Greater(t, testableLatency.Observations[0].Observation, float64(0))
+	assert.LessOrEqual(t, testableLatency.Observations[0].Observation, time.Since(start).Seconds())
 }

internal/daemon/worker/internal/metric/cluster_server.go

@@ -0,0 +1,56 @@
+package metric
+
+import (
+	"context"
+
+	"github.com/hashicorp/boundary/globals"
+	metric "github.com/hashicorp/boundary/internal/daemon/internal/metric"
+	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/stats"
+)
+
+const (
+	workerClusterSubsystem = "worker_cluster"
+)
+
+// Because we use grpc's stats.Handler to track close-to-the-wire server-side communication over grpc,
+// there is no easy way to measure request and response size as we are recording latency. Thus we only
+// track the request latency for server-side grpc connections.
+
+var grpcServerRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
+	prometheus.HistogramOpts{
+		Namespace: globals.MetricNamespace,
+		Subsystem: workerClusterSubsystem,
+		Name:      "grpc_request_duration_seconds",
+		Help:      "Histogram of latencies for gRPC requests between the a worker server and a worker client.",
+		Buckets:   prometheus.DefBuckets,
+	},
+	metric.ListGrpcLabels,
+)
+
+// All the codes expected to be returned by boundary or the grpc framework to
+// requests to the cluster server.
+var expectedGrpcCodes = []codes.Code{
+	codes.OK, codes.InvalidArgument, codes.PermissionDenied,
+	codes.FailedPrecondition,
+
+	// Codes which can be generated by the gRPC framework
+	codes.Canceled, codes.Unknown, codes.DeadlineExceeded,
+	codes.ResourceExhausted, codes.Unimplemented, codes.Internal,
+	codes.Unavailable, codes.Unauthenticated,
+}
+
+// InstrumentClusterStatsHandler returns a gRPC stats.Handler which observes
+// cluster-specific metrics for a gRPC server.
+func InstrumentClusterStatsHandler(ctx context.Context) (stats.Handler, error) {
+	return metric.NewStatsHandler(ctx, grpcServerRequestLatency)
+}
+
+// InitializeClusterServerCollectors registers the cluster server metrics to the
+// prometheus register and initializes them to 0 for all possible label
+// combinations.
+func InitializeClusterServerCollectors(r prometheus.Registerer, server *grpc.Server) {
+	metric.InitializeGrpcCollectorsFromServer(r, grpcServerRequestLatency, server, expectedGrpcCodes)
+}

internal/daemon/worker/internal/metric/proxy_http.go

@@ -6,22 +6,37 @@ import (
 	"fmt"
 	"net/http"
 	"path"
-	"strconv"
-	"strings"
 
 	"github.com/hashicorp/boundary/globals"
+	"github.com/hashicorp/boundary/internal/daemon/internal/metric"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 const (
-	invalidPathValue = "invalid"
+	proxySubSystem   = "worker_proxy"
 	proxyPathValue   = "/v1/proxy"
+	invalidPathValue = "invalid"
+)
 
-	labelHttpCode   = "code"
-	labelHttpPath   = "path"
-	labelHttpMethod = "method"
-	proxySubSystem  = "worker_proxy"
+var (
+	expectedPathsToMethods = map[string][]string{
+		proxyPathValue: {http.MethodGet},
+	}
+
+	expectedHttpErrCodes = []int{
+		http.StatusUpgradeRequired,
+		http.StatusMethodNotAllowed,
+		http.StatusBadRequest,
+		http.StatusForbidden,
+		http.StatusNotImplemented,
+		http.StatusSwitchingProtocols,
+		http.StatusInternalServerError,
+	}
+
+	expectedCodesPerMethod = map[string][]int{
+		http.MethodGet: expectedHttpErrCodes,
+	}
 )
 
 // httpTimeUntilHeader collects measurements of how long it takes
@@ -34,19 +49,9 @@ var httpTimeUntilHeader prometheus.ObserverVec = prometheus.NewHistogramVec(
 		Help:      "Histogram of time elapsed after the TLS connection is established to when the first http header is written back from the server.",
 		Buckets:   prometheus.DefBuckets,
 	},
-	[]string{labelHttpCode, labelHttpPath, labelHttpMethod},
+	metric.ListHttpLabels,
 )
 
-var expectedHttpErrCodes = []int{
-	http.StatusUpgradeRequired,
-	http.StatusMethodNotAllowed,
-	http.StatusBadRequest,
-	http.StatusForbidden,
-	http.StatusNotImplemented,
-	http.StatusSwitchingProtocols,
-	http.StatusInternalServerError,
-}
-
 // pathLabel maps the requested path to the label value recorded for metric
 func pathLabel(incomingPath string) string {
 	if incomingPath == "" || incomingPath[0] != '/' {
@@ -66,7 +71,7 @@ func pathLabel(incomingPath string) string {
 func InstrumentHttpHandler(wrapped http.Handler) http.Handler {
 	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
 		l := prometheus.Labels{
-			labelHttpPath: pathLabel(req.URL.Path),
+			metric.LabelHttpPath: pathLabel(req.URL.Path),
 		}
 		promhttp.InstrumentHandlerTimeToWriteHeader(
 			httpTimeUntilHeader.MustCurryWith(l),
@@ -79,23 +84,5 @@ func InstrumentHttpHandler(wrapped http.Handler) http.Handler {
 // prometheus register and initializes them to 0 for the most likely label
 // combinations.
 func InitializeHttpCollectors(r prometheus.Registerer) {
-	if r == nil {
-		return
-	}
-	r.MustRegister(httpTimeUntilHeader)
-
-	p := proxyPathValue
-	method := http.MethodGet
-	for _, sc := range expectedHttpErrCodes {
-		l := prometheus.Labels{labelHttpCode: strconv.Itoa(sc), labelHttpPath: p, labelHttpMethod: strings.ToLower(method)}
-		httpTimeUntilHeader.With(l)
-	}
-
-	// When an invalid path is found, any method is possible, but we expect
-	// an error response.
-	p = invalidPathValue
-	for _, sc := range []int{http.StatusNotFound, http.StatusMethodNotAllowed} {
-		l := prometheus.Labels{labelHttpCode: strconv.Itoa(sc), labelHttpPath: p, labelHttpMethod: strings.ToLower(method)}
-		httpTimeUntilHeader.With(l)
-	}
+	metric.InitializeApiCollectors(r, httpTimeUntilHeader, expectedPathsToMethods, expectedCodesPerMethod)
 }

internal/daemon/worker/listeners.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/daemon/common"
+	"github.com/hashicorp/boundary/internal/daemon/worker/internal/metric"
 	"github.com/hashicorp/boundary/internal/daemon/worker/session"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
@@ -171,7 +172,12 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 		return nil, fmt.Errorf("error instantiating non-worker split listener: %w", err)
 	}
 
+	statsHandler, err := metric.InstrumentClusterStatsHandler(w.baseContext)
+	if err != nil {
+		return nil, errors.Wrap(w.baseContext, err, op)
+	}
 	downstreamServer := grpc.NewServer(
+		grpc.StatsHandler(statsHandler),
 		grpc.MaxRecvMsgSize(math.MaxInt32),
 		grpc.MaxSendMsgSize(math.MaxInt32),
 	)
@@ -182,6 +188,8 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 		}
 	}
 
+	metric.InitializeClusterServerCollectors(w.conf.PrometheusRegisterer, downstreamServer)
+
 	ln.GrpcServer = downstreamServer
 
 	eventingListener, err := common.NewEventingListener(cancelCtx, nodeeAuthListener)