Skip to content

Commit

Permalink
Issue argoproj#340 - add gRPC payload logging interceptor
Browse files Browse the repository at this point in the history
  • Loading branch information
Alexander Matyushentsev authored and Alexander Matyushentsev committed Jul 19, 2018
1 parent 82fda1c commit fbf7891
Show file tree
Hide file tree
Showing 3 changed files with 101 additions and 4 deletions.
7 changes: 7 additions & 0 deletions reposerver/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"github.com/grpc-ecosystem/go-grpc-middleware"
"github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus"
log "github.com/sirupsen/logrus"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
)
Expand All @@ -34,10 +35,16 @@ func (a *ArgoCDRepoServer) CreateGRPC() *grpc.Server {
server := grpc.NewServer(
grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
grpc_logrus.StreamServerInterceptor(a.log),
grpc_util.PayloadStreamServerInterceptor(a.log, false, func(ctx context.Context, fullMethodName string, servingObject interface{}) bool {
return true
}),
grpc_util.PanicLoggerStreamServerInterceptor(a.log),
)),
grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
grpc_logrus.UnaryServerInterceptor(a.log),
grpc_util.PayloadUnaryServerInterceptor(a.log, false, func(ctx context.Context, fullMethodName string, servingObject interface{}) bool {
return true
}),
grpc_util.PanicLoggerUnaryServerInterceptor(a.log),
)),
)
Expand Down
16 changes: 12 additions & 4 deletions server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,9 @@ import (

"github.com/gobuffalo/packr"
golang_proto "github.com/golang/protobuf/proto"
grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
grpc_auth "github.com/grpc-ecosystem/go-grpc-middleware/auth"
grpc_logrus "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus"
"github.com/grpc-ecosystem/go-grpc-middleware"
"github.com/grpc-ecosystem/go-grpc-middleware/auth"
"github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus"
"github.com/grpc-ecosystem/grpc-gateway/runtime"
log "github.com/sirupsen/logrus"
"github.com/soheilhy/cmux"
Expand All @@ -28,7 +28,7 @@ import (
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/kubernetes"

argocd "github.com/argoproj/argo-cd"
"github.com/argoproj/argo-cd"
"github.com/argoproj/argo-cd/common"
"github.com/argoproj/argo-cd/errors"

Expand All @@ -55,6 +55,7 @@ import (
"github.com/argoproj/argo-cd/util/swagger"
tlsutil "github.com/argoproj/argo-cd/util/tls"
"github.com/argoproj/argo-cd/util/webhook"
netCtx "golang.org/x/net/context"
)

var (
Expand Down Expand Up @@ -299,18 +300,25 @@ func (a *ArgoCDServer) useTLS() bool {

func (a *ArgoCDServer) newGRPCServer() *grpc.Server {
var sOpts []grpc.ServerOption
loginMethodName := "/session.SessionService/Create"
// NOTE: notice we do not configure the gRPC server here with TLS (e.g. grpc.Creds(creds))
// This is because TLS handshaking occurs in cmux handling
sOpts = append(sOpts, grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
grpc_logrus.StreamServerInterceptor(a.log),
grpc_auth.StreamServerInterceptor(a.authenticate),
grpc_util.PayloadStreamServerInterceptor(a.log, true, func(ctx netCtx.Context, fullMethodName string, servingObject interface{}) bool {
return fullMethodName != loginMethodName
}),
grpc_util.ErrorCodeStreamServerInterceptor(),
grpc_util.PanicLoggerStreamServerInterceptor(a.log),
)))
sOpts = append(sOpts, grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
bug21955WorkaroundInterceptor,
grpc_logrus.UnaryServerInterceptor(a.log),
grpc_auth.UnaryServerInterceptor(a.authenticate),
grpc_util.PayloadUnaryServerInterceptor(a.log, true, func(ctx netCtx.Context, fullMethodName string, servingObject interface{}) bool {
return fullMethodName != loginMethodName
}),
grpc_util.ErrorCodeUnaryServerInterceptor(),
grpc_util.PanicLoggerUnaryServerInterceptor(a.log),
)))
Expand Down
82 changes: 82 additions & 0 deletions util/grpc/logging.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
package grpc

import (
"bytes"
"encoding/json"
"fmt"

"golang.org/x/net/context"
"google.golang.org/grpc"

"github.com/gogo/protobuf/proto"
"github.com/grpc-ecosystem/go-grpc-middleware/logging"
"github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus"
"github.com/grpc-ecosystem/go-grpc-middleware/tags/logrus"
"github.com/sirupsen/logrus"
)

func logRequest(entry *logrus.Entry, info string, pbMsg interface{}, ctx context.Context, logClaims bool) {
if logClaims {
if data, err := json.Marshal(ctx.Value("claims")); err == nil {
entry = entry.WithField("grpc.request.claims", string(data))
}
}
if p, ok := pbMsg.(proto.Message); ok {
entry = entry.WithField("grpc.request.content", &jsonpbMarshalleble{p})
}
entry.Info(info)
}

type jsonpbMarshalleble struct {
proto.Message
}

func (j *jsonpbMarshalleble) MarshalJSON() ([]byte, error) {
b := &bytes.Buffer{}
if err := grpc_logrus.JsonPbMarshaller.Marshal(b, j.Message); err != nil {
return nil, fmt.Errorf("jsonpb serializer failed: %v", err)
}
return b.Bytes(), nil
}

type loggingServerStream struct {
grpc.ServerStream
entry *logrus.Entry
logClaims bool
info string
}

func (l *loggingServerStream) SendMsg(m interface{}) error {
return l.ServerStream.SendMsg(m)
}

func (l *loggingServerStream) RecvMsg(m interface{}) error {
err := l.ServerStream.RecvMsg(m)
if err == nil {
logRequest(l.entry, l.info, m, l.ServerStream.Context(), l.logClaims)
}
return err
}

func PayloadStreamServerInterceptor(entry *logrus.Entry, logClaims bool, decider grpc_logging.ServerPayloadLoggingDecider) grpc.StreamServerInterceptor {
return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
if !decider(stream.Context(), info.FullMethod, srv) {
return handler(srv, stream)
}
logEntry := entry.WithFields(ctx_logrus.Extract(stream.Context()).Data)
newStream := &loggingServerStream{ServerStream: stream, entry: logEntry, logClaims: logClaims, info: fmt.Sprintf("received streaming call %s", info.FullMethod)}
return handler(srv, newStream)
}
}

func PayloadUnaryServerInterceptor(entry *logrus.Entry, logClaims bool, decider grpc_logging.ServerPayloadLoggingDecider) grpc.UnaryServerInterceptor {
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
if !decider(ctx, info.FullMethod, info.Server) {
return handler(ctx, req)
}
logEntry := entry.WithFields(ctx_logrus.Extract(ctx).Data)
logRequest(logEntry, fmt.Sprintf("received unary call %s", info.FullMethod), req, ctx, logClaims)
resp, err := handler(ctx, req)
return resp, err
}
}

0 comments on commit fbf7891

Please sign in to comment.