diff --git a/internal/servers/controller/internal/metric/cluster.go b/internal/servers/controller/internal/metric/cluster.go new file mode 100644 index 000000000..4d82bcb0e --- /dev/null +++ b/internal/servers/controller/internal/metric/cluster.go @@ -0,0 +1,187 @@ +package metric + +import ( + "context" + "strings" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/prometheus/client_golang/prometheus" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +const ( + labelGRpcCode = "grpc_code" + labelGRpcService = "grpc_service" + labelGRpcMethod = "grpc_method" + clusterSubSystem = "controller_cluster" +) + +var ( + // 100 bytes, 1kb, 10kb, 100kb, 1mb, 10mb + gRpcMsgSizeBuckets = prometheus.ExponentialBuckets(100, 10, 6) + + // 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. + gRpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: globals.MetricNamespace, + Subsystem: clusterSubSystem, + Name: "grpc_request_duration_seconds", + Help: "Histogram of latencies for gRPC requests.", + Buckets: prometheus.DefBuckets, + }, + []string{labelGRpcCode, labelGRpcService, labelGRpcMethod}, + ) + + // gRpcRequestSize collections measurements of how large each request + // to the boundary controller cluster is. + gRpcRequestSize prometheus.ObserverVec = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: globals.MetricNamespace, + Subsystem: clusterSubSystem, + Name: "grpc_request_size_bytes", + Help: "Histogram of request sizes for gRPC requests.", + Buckets: gRpcMsgSizeBuckets, + }, + []string{labelGRpcCode, labelGRpcService, labelGRpcMethod}, + ) + + // gRpcResponseSize collections measurements of how large each response + // from the boundary controller cluster is. + gRpcResponseSize prometheus.ObserverVec = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: globals.MetricNamespace, + Subsystem: clusterSubSystem, + Name: "grpc_response_size_bytes", + Help: "Histogram of response sizes for gRPC responses.", + Buckets: gRpcMsgSizeBuckets, + }, + []string{labelGRpcCode, labelGRpcService, labelGRpcMethod}, + ) +) + +// 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 + + // measurements + reqSize *int + start time.Time +} + +func newRequestRecorder(ctx context.Context, req interface{}, fullMethodName string) requestRecorder { + const op = "metric.newRequestRecorder" + service, method := splitMethodName(fullMethodName) + r := requestRecorder{ + labels: prometheus.Labels{ + labelGRpcMethod: method, + labelGRpcService: service, + }, + start: time.Now(), + } + + reqProto, ok := req.(proto.Message) + switch { + case ok: + reqSize := proto.Size(reqProto) + r.reqSize = &reqSize + default: + event.WriteError(ctx, op, errors.New(ctx, errors.Internal, op, "unable to cast to proto.Message")) + } + return r +} + +func (r requestRecorder) record(ctx context.Context, resp interface{}, err error) { + const op = "metric.(requestRecorder).record" + st := statusFromError(err) + r.labels[labelGRpcCode] = st.Code().String() + + gRpcRequestLatency.With(r.labels).Observe(time.Since(r.start).Seconds()) + + if r.reqSize != nil { + gRpcRequestSize.With(r.labels).Observe(float64(*r.reqSize)) + } + + if respProto, ok := resp.(proto.Message); ok { + respSize := proto.Size(respProto) + gRpcResponseSize.With(r.labels).Observe(float64(respSize)) + } else { + event.WriteError(ctx, op, errors.New(ctx, errors.Internal, op, "unable to cast to proto.Message")) + } +} + +// InstrumentClusterInterceptor wraps a UnaryServerInterceptor and records +// observations for the collectors associated with the cluster's grpc service. +func InstrumentClusterInterceptor() grpc.UnaryServerInterceptor { + return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + recorder := newRequestRecorder(ctx, req, info.FullMethod) + resp, err := handler(ctx, req) + recorder.record(ctx, resp, err) + return resp, 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, +} + +// 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, gRpcRequestSize, gRpcResponseSize) + + 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) + gRpcRequestSize.With(l) + gRpcResponseSize.With(l) + } + } + } +} diff --git a/internal/servers/controller/internal/metric/cluster_test.go b/internal/servers/controller/internal/metric/cluster_test.go new file mode 100644 index 000000000..853ea6bdf --- /dev/null +++ b/internal/servers/controller/internal/metric/cluster_test.go @@ -0,0 +1,228 @@ +package metric + +import ( + "context" + "testing" + "time" + + "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/status" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +// 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 +} + +func floatPtr(i int) *float64 { + f := float64(i) + return &f +} + +func TestRecorder(t *testing.T) { + bkpLatency := gRpcRequestLatency + bkpRespSize := gRpcResponseSize + bkpReqSize := gRpcRequestSize + defer func() { + gRpcRequestLatency = bkpLatency + gRpcResponseSize = bkpRespSize + gRpcRequestSize = bkpReqSize + }() + ctx := context.Background() + + cases := []struct { + name string + methodName string + req interface{} + resp interface{} + err error + wantedLabels prometheus.Labels + wantedReqSize *float64 + wantedRespSize *float64 + }{ + { + name: "basic", + methodName: "/some.service.path/method", + req: wrapperspb.Bytes([]byte{1, 2, 3}), + resp: wrapperspb.Bytes([]byte{1, 2, 3, 4}), + err: nil, + wantedLabels: map[string]string{ + labelGRpcCode: "OK", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3}))), + wantedRespSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3, 4}))), + }, + { + name: "empty request", + methodName: "/some.service.path/method", + req: wrapperspb.Bytes(nil), + resp: wrapperspb.Bytes([]byte{1, 2, 3, 4}), + err: nil, + wantedLabels: map[string]string{ + labelGRpcCode: "OK", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: floatPtr(0), + wantedRespSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3, 4}))), + }, + { + name: "empty response", + methodName: "/some.service.path/method", + req: wrapperspb.Bytes([]byte{1, 2, 3}), + resp: wrapperspb.Bytes(nil), + err: nil, + wantedLabels: map[string]string{ + labelGRpcCode: "OK", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3}))), + wantedRespSize: floatPtr(0), + }, + { + name: "unrecognized method path format", + methodName: "unrecognized", + req: wrapperspb.Bytes([]byte{1, 2, 3}), + resp: wrapperspb.Bytes([]byte{1, 2, 3, 4}), + err: nil, + wantedLabels: map[string]string{ + labelGRpcCode: "OK", + labelGRpcMethod: "unknown", + labelGRpcService: "unknown", + }, + wantedReqSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3}))), + wantedRespSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3, 4}))), + }, + { + name: "cancel error", + methodName: "/some.service.path/method", + req: wrapperspb.Bytes([]byte{1, 2, 3}), + resp: (*wrapperspb.BytesValue)(nil), + err: status.Error(codes.Canceled, ""), + wantedLabels: map[string]string{ + labelGRpcCode: "Canceled", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3}))), + wantedRespSize: floatPtr(0), + }, + { + name: "permission error", + methodName: "/some.service.path/method", + req: wrapperspb.Bytes([]byte{1, 2, 3}), + resp: (*wrapperspb.BytesValue)(nil), + err: status.Error(codes.PermissionDenied, ""), + wantedLabels: map[string]string{ + labelGRpcCode: "PermissionDenied", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3}))), + wantedRespSize: floatPtr(0), + }, + { + name: "error and response", + methodName: "/some.service.path/method", + req: wrapperspb.Bytes([]byte{1, 2, 3}), + resp: wrapperspb.Bytes([]byte{1, 2, 3, 4}), + err: status.Error(codes.PermissionDenied, ""), + wantedLabels: map[string]string{ + labelGRpcCode: "PermissionDenied", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3}))), + wantedRespSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3, 4}))), + }, + { + name: "bad request", + methodName: "/some.service.path/method", + req: "foo", + resp: wrapperspb.Bytes([]byte{1, 2, 3, 4}), + err: status.Error(codes.PermissionDenied, ""), + wantedLabels: map[string]string{ + labelGRpcCode: "PermissionDenied", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: nil, + wantedRespSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3, 4}))), + }, + { + name: "bad response", + methodName: "/some.service.path/method", + req: wrapperspb.Bytes([]byte{1, 2, 3, 4}), + resp: "foo", + err: status.Error(codes.PermissionDenied, ""), + wantedLabels: map[string]string{ + labelGRpcCode: "PermissionDenied", + labelGRpcMethod: "method", + labelGRpcService: "some.service.path", + }, + wantedReqSize: floatPtr(proto.Size(wrapperspb.Bytes([]byte{1, 2, 3, 4}))), + wantedRespSize: nil, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + testableLatency := &testableObserverVec{} + testableReqSize := &testableObserverVec{} + testableRespSize := &testableObserverVec{} + gRpcRequestLatency = testableLatency + gRpcResponseSize = testableRespSize + gRpcRequestSize = testableReqSize + + // record something + start := time.Now() + tested := newRequestRecorder(ctx, tc.req, tc.methodName) + tested.record(ctx, tc.resp, tc.err) + + require.Len(t, testableLatency.observations, 1) + assert.LessOrEqual(t, testableLatency.observations[0].observation, time.Since(start).Seconds()) + assert.Greater(t, testableLatency.observations[0].observation, float64(0)) + assert.Equal(t, testableLatency.observations[0].labels, tc.wantedLabels) + + if tc.wantedReqSize == nil { + require.Len(t, testableReqSize.observations, 0) + } else { + require.Len(t, testableReqSize.observations, 1) + assert.Equal(t, testableReqSize.observations[0], + &testableObserver{observation: *tc.wantedReqSize, labels: tc.wantedLabels}) + } + + if tc.wantedRespSize == nil { + require.Len(t, testableRespSize.observations, 0) + } else { + require.Len(t, testableRespSize.observations, 1) + assert.Equal(t, testableRespSize.observations[0], + &testableObserver{observation: *tc.wantedRespSize, labels: tc.wantedLabels}) + } + }) + } +} diff --git a/internal/servers/controller/listeners.go b/internal/servers/controller/listeners.go index 16e5bca17..cc9543c3f 100644 --- a/internal/servers/controller/listeners.go +++ b/internal/servers/controller/listeners.go @@ -15,6 +15,7 @@ import ( "github.com/hashicorp/boundary/internal/cmd/base" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/servers/controller/handlers/workers" + "github.com/hashicorp/boundary/internal/servers/controller/internal/metric" "github.com/hashicorp/go-multierror" "google.golang.org/grpc" ) @@ -116,6 +117,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error grpc.MaxSendMsgSize(math.MaxInt32), grpc.UnaryInterceptor( grpc_middleware.ChainUnaryServer( + metric.InstrumentClusterInterceptor(), workerReqInterceptor, auditRequestInterceptor(c.baseContext), // before we get started, audit the request auditResponseInterceptor(c.baseContext), // as we finish, audit the response @@ -127,6 +129,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error c.workerStatusUpdateTimes, c.kms) pbs.RegisterServerCoordinationServiceServer(workerServer, workerService) pbs.RegisterSessionServiceServer(workerServer, workerService) + metric.InitializeClusterCollectors(c.conf.PrometheusRegisterer, workerServer) ln.GrpcServer = workerServer