-
Notifications
You must be signed in to change notification settings - Fork 394
/
interceptors.go
117 lines (106 loc) · 3.98 KB
/
interceptors.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package server
import (
"context"
"encoding/json"
"fmt"
"io"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"storj.io/common/identity"
"storj.io/common/rpc/rpcpeer"
"storj.io/storj/storage"
)
func (server *Server) logOnErrorStreamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) (err error) {
err = handler(srv, ss)
if err != nil {
// no zap errors for canceled or wrong file downloads
if storage.ErrKeyNotFound.Has(err) ||
status.Code(err) == codes.Canceled ||
status.Code(err) == codes.Unavailable ||
err == io.EOF {
return err
}
server.log.Error("gRPC stream error response", zap.Error(err))
}
return err
}
func (server *Server) logOnErrorUnaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
resp, err = handler(ctx, req)
if err != nil {
// no zap errors for wrong file downloads
if status.Code(err) == codes.NotFound {
return resp, err
}
server.log.Error("gRPC unary error response", zap.Error(err))
}
return resp, err
}
// CombineInterceptors combines two UnaryServerInterceptors so they act as one
// (because grpc only allows you to pass one in).
func CombineInterceptors(a, b grpc.UnaryServerInterceptor) grpc.UnaryServerInterceptor {
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
return a(ctx, req, info, func(actx context.Context, areq interface{}) (interface{}, error) {
return b(actx, areq, info, func(bctx context.Context, breq interface{}) (interface{}, error) {
return handler(bctx, breq)
})
})
}
}
type nodeRequestLog struct {
GRPCService string `json:"grpc_service"`
GRPCMethod string `json:"grpc_method"`
PeerAddress string `json:"peer_address"`
PeerNodeID string `json:"peer_node_id"`
Msg interface{} `json:"msg"`
}
func prepareRequestLog(ctx context.Context, req, server interface{}, methodName string) ([]byte, error) {
reqLog := nodeRequestLog{
GRPCService: fmt.Sprintf("%T", server),
GRPCMethod: methodName,
PeerAddress: "<no peer???>",
Msg: req,
}
if peer, err := rpcpeer.FromContext(ctx); err == nil {
reqLog.PeerAddress = peer.Addr.String()
if peerIdentity, err := identity.PeerIdentityFromPeer(peer); err == nil {
reqLog.PeerNodeID = peerIdentity.ID.String()
} else {
reqLog.PeerNodeID = fmt.Sprintf("<no peer id: %v>", err)
}
}
return json.Marshal(reqLog)
}
// UnaryMessageLoggingInterceptor creates a UnaryServerInterceptor which
// logs the full contents of incoming unary requests.
func UnaryMessageLoggingInterceptor(log *zap.Logger) grpc.UnaryServerInterceptor {
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
if jsonReq, err := prepareRequestLog(ctx, req, info.Server, info.FullMethod); err == nil {
log.Info(string(jsonReq))
} else {
log.Error("Failed to marshal request to JSON.",
zap.String("method", info.FullMethod), zap.Error(err),
)
}
return handler(ctx, req)
}
}
// StreamMessageLoggingInterceptor creates a StreamServerInterceptor which
// logs the full contents of incoming streaming requests.
func StreamMessageLoggingInterceptor(log *zap.Logger) grpc.StreamServerInterceptor {
return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
// Are we even using any of these yet? I'm only guessing at how best to pass in things
// so that they make sense.
if jsonReq, err := prepareRequestLog(ss.Context(), srv, nil, info.FullMethod); err == nil {
log.Info(string(jsonReq))
} else {
log.Error("Failed to marshal request to JSON.",
zap.String("method", info.FullMethod), zap.Error(err),
)
}
return handler(srv, ss)
}
}