m/n/c/rpc: implement Span/Trace
This is a first pass at implementing basic support for
Dapper/OpenTracing/OpenTelemetry-style tracing within Metropolis RPCs.
More precisely, this implements an API to expose an RPC-local Span to
RPC handlers (unary and streaming). These Spans are currently backed by
a logtree logger, and aren't processed further (ie. there's no support
for child spans and carrying span information over the wire when
performing remote calls from an active Span). However, this allows us to
at least start emitting Span Events and use them for debugging purposes.
Since we don't yet have OpenTelemetry in our GOPATH, we reimplement a
minimum subset of the Span type that should still be compatible with
real OpenTelemetry types. Once OpenTelemetry lands in our GOPATH (by way
of it landing in k8s, for example), we'll move over to using the real
type instead. Then, we can also begin integrating with OpenTelemetry
proper, ie. start sending traces over to collectors, start
injecting/extracing span information over gRPC, etc.
Another change on top of this one actually uses the Trace(ctx)
functionality within the curator - this is just the library
implementation.
Change-Id: I85506303538aacc137a28828ab39ccfd9ff72924
Reviewed-on: https://review.monogon.dev/c/monogon/+/541
Reviewed-by: Leopold Schabel <leo@nexantic.com>
diff --git a/metropolis/node/core/rpc/server_authentication.go b/metropolis/node/core/rpc/server_authentication.go
index 6ae2618..50cf36a 100644
--- a/metropolis/node/core/rpc/server_authentication.go
+++ b/metropolis/node/core/rpc/server_authentication.go
@@ -14,6 +14,7 @@
cpb "source.monogon.dev/metropolis/node/core/curator/proto/api"
"source.monogon.dev/metropolis/node/core/identity"
+ "source.monogon.dev/metropolis/pkg/logtree"
apb "source.monogon.dev/metropolis/proto/api"
)
@@ -41,34 +42,10 @@
getPeerInfoUnauthenticated(ctx context.Context) (*PeerInfo, error)
}
-// stream implements the gRPC StreamInterceptor interface for use with
-// grpc.NewServer, based on an authenticationStrategy.
-func streamInterceptor(a authenticationStrategy) grpc.StreamServerInterceptor {
- return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
- pi, err := check(ss.Context(), a, info.FullMethod)
- if err != nil {
- return err
- }
- return handler(srv, pi.serverStream(ss))
- }
-}
-
-// unaryInterceptor implements the gRPC UnaryInterceptor interface for use with
-// grpc.NewServer, based on an authenticationStrategy.
-func unaryInterceptor(a authenticationStrategy) grpc.UnaryServerInterceptor {
- return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
- pi, err := check(ctx, a, info.FullMethod)
- if err != nil {
- return nil, err
- }
- return handler(pi.apply(ctx), req)
- }
-}
-
-// check is called by the unary and server interceptors to perform
+// authenticationCheck is called by the unary and server interceptors to perform
// authentication and authorization checks for a given RPC, calling the
// serverInterceptors' authenticate function if needed.
-func check(ctx context.Context, a authenticationStrategy, methodName string) (*PeerInfo, error) {
+func authenticationCheck(ctx context.Context, a authenticationStrategy, methodName string) (*PeerInfo, error) {
mi, err := getMethodInfo(methodName)
if err != nil {
return nil, err
@@ -113,7 +90,7 @@
// metropolis.proto.ext.authorization options and authenticate/authorize
// incoming connections. It also runs the gRPC server with the correct TLS
// settings for authenticating itself to callers.
-func (l *ExternalServerSecurity) SetupExternalGRPC(impls ClusterExternalServices) *grpc.Server {
+func (l *ExternalServerSecurity) SetupExternalGRPC(logger logtree.LeveledLogger, impls ClusterExternalServices) *grpc.Server {
externalCreds := credentials.NewTLS(&tls.Config{
Certificates: []tls.Certificate{l.NodeCredentials.TLSCredentials()},
ClientAuth: tls.RequestClientCert,
@@ -121,8 +98,8 @@
s := grpc.NewServer(
grpc.Creds(externalCreds),
- grpc.UnaryInterceptor(unaryInterceptor(l)),
- grpc.StreamInterceptor(streamInterceptor(l)),
+ grpc.UnaryInterceptor(unaryInterceptor(logger, l)),
+ grpc.StreamInterceptor(streamInterceptor(logger, l)),
)
cpb.RegisterCuratorServer(s, impls)
apb.RegisterAAAServer(s, impls)
@@ -237,10 +214,10 @@
// SetupLocalGRPC returns a grpc.Server ready to listen on a local domain socket
// and serve the Curator service. All incoming RPCs will be authenticated as
// originating from the node for which LocalServerSecurity has been configured.
-func (l *LocalServerSecurity) SetupLocalGRPC(impls ClusterInternalServices) *grpc.Server {
+func (l *LocalServerSecurity) SetupLocalGRPC(logger logtree.LeveledLogger, impls ClusterInternalServices) *grpc.Server {
s := grpc.NewServer(
- grpc.UnaryInterceptor(unaryInterceptor(l)),
- grpc.StreamInterceptor(streamInterceptor(l)),
+ grpc.UnaryInterceptor(unaryInterceptor(logger, l)),
+ grpc.StreamInterceptor(streamInterceptor(logger, l)),
)
cpb.RegisterCuratorServer(s, impls)
return s