Add collector updating interceptor to the controller cluster grpc server (#1964)
This commit is contained in:
3 files changed
+418
No files matched your search
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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})
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in new issue
Block a user