Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

clientv3: log warning in case of error sending request

Browse Source
This commit is contained in:
Ted Yu 2019-12-20 15:08:33 -08:00
parent 322c38e169
commit cb8bf089f8
1 changed files with 18 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
clientv3/watch.go
View File

@ -25,6 +25,7 @@ import (
pb "go.etcd.io/etcd/etcdserver/etcdserverpb" pb "go.etcd.io/etcd/etcdserver/etcdserverpb"
mvccpb "go.etcd.io/etcd/mvcc/mvccpb" mvccpb "go.etcd.io/etcd/mvcc/mvccpb"
"go.uber.org/zap"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
@ -140,6 +141,7 @@ type watcher struct {
// streams holds all the active grpc streams keyed by ctx value. // streams holds all the active grpc streams keyed by ctx value.
streams map[string]*watchGrpcStream streams map[string]*watchGrpcStream
lg *zap.Logger
} }
// watchGrpcStream tracks all watch resources attached to a single grpc stream. // watchGrpcStream tracks all watch resources attached to a single grpc stream.
@ -242,6 +244,7 @@ func NewWatchFromWatchClient(wc pb.WatchClient, c *Client) Watcher {
} }
if c != nil { if c != nil {
w.callOpts = c.callOpts w.callOpts = c.callOpts
w.lg = c.lg
} }
return w return w
} }
@ -544,10 +547,14 @@ func (w *watchGrpcStream) run() {
w.resuming = append(w.resuming, ws) w.resuming = append(w.resuming, ws)
if len(w.resuming) == 1 { if len(w.resuming) == 1 {
// head of resume queue, can register a new watcher // head of resume queue, can register a new watcher
wc.Send(ws.initReq.toPB()) if err := wc.Send(ws.initReq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
case *progressRequest: case *progressRequest:
wc.Send(wreq.toPB()) if err := wc.Send(wreq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
// new events from the watch client // new events from the watch client
@ -571,7 +578,9 @@ func (w *watchGrpcStream) run() {
} }
if ws := w.nextResume(); ws != nil { if ws := w.nextResume(); ws != nil {
wc.Send(ws.initReq.toPB()) if err := wc.Send(ws.initReq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
// reset for next iteration // reset for next iteration
@ -616,7 +625,9 @@ func (w *watchGrpcStream) run() {
}, },
} }
req := &pb.WatchRequest{RequestUnion: cr} req := &pb.WatchRequest{RequestUnion: cr}
wc.Send(req) if err := wc.Send(req); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
// watch client failed on Recv; spawn another if possible // watch client failed on Recv; spawn another if possible
@ -629,7 +640,9 @@ func (w *watchGrpcStream) run() {
return return
} }
if ws := w.nextResume(); ws != nil { if ws := w.nextResume(); ws != nil {
wc.Send(ws.initReq.toPB()) if err := wc.Send(ws.initReq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
cancelSet = make(map[int64]struct{}) cancelSet = make(map[int64]struct{})