From 5c715478e428760b0ab5d3dc26e943d38d4a9d1f Mon Sep 17 00:00:00 2001 From: Haotian Date: Tue, 6 Sep 2022 13:40:52 -0700 Subject: [PATCH] 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 --- .../daemon/controller/internal/metric/api.go | 52 ++------ .../controller/internal/metric/cluster.go | 102 ++------------ internal/daemon/controller/listeners.go | 6 +- .../internal/metric/initialize_metrics.go | 124 +++++++++++++++++ .../internal/metric/instrument_handlers.go | 96 ++++++++++++++ .../metric/instrument_handlers_test.go} | 75 ++++------- internal/daemon/internal/metric/testing.go | 47 +++++++ .../worker/internal/metric/cluster_client.go | 125 ++++-------------- .../internal/metric/cluster_client_test.go | 107 +++++---------- .../worker/internal/metric/cluster_server.go | 56 ++++++++ .../worker/internal/metric/proxy_http.go | 63 ++++----- internal/daemon/worker/listeners.go | 8 ++ 12 files changed, 473 insertions(+), 388 deletions(-) create mode 100644 internal/daemon/internal/metric/initialize_metrics.go create mode 100644 internal/daemon/internal/metric/instrument_handlers.go rename internal/daemon/{controller/internal/metric/cluster_test.go => internal/metric/instrument_handlers_test.go} (65%) create mode 100644 internal/daemon/internal/metric/testing.go create mode 100644 internal/daemon/worker/internal/metric/cluster_server.go diff --git a/internal/daemon/controller/internal/metric/api.go b/internal/daemon/controller/internal/metric/api.go index 06f542b52..987f5200f 100644 --- a/internal/daemon/controller/internal/metric/api.go +++ b/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) } } diff --git a/internal/daemon/controller/internal/metric/cluster.go b/internal/daemon/controller/internal/metric/cluster.go index a66cc466f..6449e2750 100644 --- a/internal/daemon/controller/internal/metric/cluster.go +++ b/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) } diff --git a/internal/daemon/controller/listeners.go b/internal/daemon/controller/listeners.go index 8b87e71a3..35a7d926e 100644 --- a/internal/daemon/controller/listeners.go +++ b/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( diff --git a/internal/daemon/internal/metric/initialize_metrics.go b/internal/daemon/internal/metric/initialize_metrics.go new file mode 100644 index 000000000..0ed80d225 --- /dev/null +++ b/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)}) + } + } +} diff --git a/internal/daemon/internal/metric/instrument_handlers.go b/internal/daemon/internal/metric/instrument_handlers.go new file mode 100644 index 000000000..37f5c8bca --- /dev/null +++ b/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" +} diff --git a/internal/daemon/controller/internal/metric/cluster_test.go b/internal/daemon/internal/metric/instrument_handlers_test.go similarity index 65% rename from internal/daemon/controller/internal/metric/cluster_test.go rename to internal/daemon/internal/metric/instrument_handlers_test.go index 292a3fcb3..327d5c145 100644 --- a/internal/daemon/controller/internal/metric/cluster_test.go +++ b/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 -} diff --git a/internal/daemon/internal/metric/testing.go b/internal/daemon/internal/metric/testing.go new file mode 100644 index 000000000..d6df76cdf --- /dev/null +++ b/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 +} diff --git a/internal/daemon/worker/internal/metric/cluster_client.go b/internal/daemon/worker/internal/metric/cluster_client.go index a6f381020..19e720096 100644 --- a/internal/daemon/worker/internal/metric/cluster_client.go +++ b/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) } diff --git a/internal/daemon/worker/internal/metric/cluster_client_test.go b/internal/daemon/worker/internal/metric/cluster_client_test.go index 9edfeb3ba..f3367f0ad 100644 --- a/internal/daemon/worker/internal/metric/cluster_client_test.go +++ b/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()) } diff --git a/internal/daemon/worker/internal/metric/cluster_server.go b/internal/daemon/worker/internal/metric/cluster_server.go new file mode 100644 index 000000000..394354d53 --- /dev/null +++ b/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) +} diff --git a/internal/daemon/worker/internal/metric/proxy_http.go b/internal/daemon/worker/internal/metric/proxy_http.go index 58888681a..1e35e1257 100644 --- a/internal/daemon/worker/internal/metric/proxy_http.go +++ b/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) } diff --git a/internal/daemon/worker/listeners.go b/internal/daemon/worker/listeners.go index 3f332b4a0..5db68d6ec 100644 --- a/internal/daemon/worker/listeners.go +++ b/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)