server_interceptors.go

Documentation: github.com/grpc-ecosystem/go-grpc-middleware/logging/kit

     1  package kit
     2  
     3  import (
     4  	"path"
     5  	"time"
     6  
     7  	"context"
     8  
     9  	"github.com/go-kit/log"
    10  	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
    11  	"github.com/grpc-ecosystem/go-grpc-middleware/logging/kit/ctxkit"
    12  	"google.golang.org/grpc"
    13  	"google.golang.org/grpc/codes"
    14  )
    15  
    16  var (
    17  	// SystemField is used in every log statement made through grpc_zap. Can be overwritten before any initialization code.
    18  	SystemField = "grpc"
    19  	// ServerField is used in every server-side log statement made through grpc_zap.Can be overwritten before initialization.
    20  	ServerField = "server"
    21  )
    22  
    23  // UnaryServerInterceptor returns a new unary server interceptors that adds kit.Logger to the context.
    24  func UnaryServerInterceptor(logger log.Logger, opts ...Option) grpc.UnaryServerInterceptor {
    25  	o := evaluateServerOpt(opts)
    26  	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    27  		startTime := time.Now()
    28  		newCtx := injectLogger(ctx, logger, info.FullMethod, startTime, o.timestampFormat)
    29  
    30  		resp, err := handler(newCtx, req)
    31  		if !o.shouldLog(info.FullMethod, err) {
    32  			return resp, err
    33  		}
    34  
    35  		code := o.codeFunc(err)
    36  		logCall(newCtx, o, "finished unary call with code "+code.String(), code, startTime, err)
    37  
    38  		return resp, err
    39  	}
    40  }
    41  
    42  // StreamServerInterceptor returns a new stream server interceptors that adds kit.Logger to the context.
    43  func StreamServerInterceptor(logger log.Logger, opts ...Option) grpc.StreamServerInterceptor {
    44  	o := evaluateServerOpt(opts)
    45  	return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    46  		startTime := time.Now()
    47  		newCtx := injectLogger(stream.Context(), logger, info.FullMethod, startTime, o.timestampFormat)
    48  
    49  		wrapped := grpc_middleware.WrapServerStream(stream)
    50  		wrapped.WrappedContext = newCtx
    51  
    52  		err := handler(srv, wrapped)
    53  		if !o.shouldLog(info.FullMethod, err) {
    54  			return err
    55  		}
    56  
    57  		code := o.codeFunc(err)
    58  		logCall(newCtx, o, "finished streaming call with code "+code.String(), code, startTime, err)
    59  
    60  		return err
    61  	}
    62  }
    63  
    64  func injectLogger(ctx context.Context, logger log.Logger, fullMethodString string, start time.Time, timestampFormat string) context.Context {
    65  	f := ctxkit.TagsToFields(ctx)
    66  	f = append(f, "grpc.start_time", start.Format(timestampFormat))
    67  	if d, ok := ctx.Deadline(); ok {
    68  		f = append(f, "grpc.request.deadline", d.Format(timestampFormat))
    69  	}
    70  	f = append(f, serverCallFields(fullMethodString)...)
    71  	callLog := log.With(logger, f...)
    72  	return ctxkit.ToContext(ctx, callLog)
    73  }
    74  
    75  func serverCallFields(fullMethodString string) []interface{} {
    76  	service := path.Dir(fullMethodString)[1:]
    77  	method := path.Base(fullMethodString)
    78  	return []interface{}{
    79  		"system", SystemField,
    80  		"span.kind", ServerField,
    81  		"grpc.service", service,
    82  		"grpc.method", method,
    83  	}
    84  }
    85  
    86  func logCall(ctx context.Context, options *options, msg string, code codes.Code, startTime time.Time, err error) {
    87  	extractedLogger := ctxkit.Extract(ctx)
    88  	extractedLogger = options.levelFunc(code, extractedLogger)
    89  	args := []interface{}{"msg", msg, "error", err, "grpc.code", code.String()}
    90  	args = append(args, options.durationFunc(time.Since(startTime))...)
    91  	_ = extractedLogger.Log(args...)
    92  }
    93
View as plain text