clientv3: handle stale endpoints, clean up logging #8714
Conversation
clientv3/balancer.go
Outdated
| @@ -152,7 +152,9 @@ func (b *simpleBalancer) pinned() string { | |||
| return b.pinAddr | |||
| } | |||
|
|
|||
| func (b *simpleBalancer) endpointError(addr string, err error) { return } | |||
| func (b *simpleBalancer) endpointError(host string, err error) { | |||
| panic("'endpointError' not implemented") | |||
There was a problem hiding this comment.
why do we need to panic here? probably should keep it as a noop.
clientv3/health_balancer.go
Outdated
| // unhealthy tracks the last unhealthy time of endpoints. | ||
| unhealthy map[string]time.Time | ||
| // unhealthyHosts tracks unhealthy hosts. | ||
| unhealthyHosts map[string]time.Time |
There was a problem hiding this comment.
why do we rename this host? host does not include port. do we really track host here? for example, if we have two endpoints: localhost:1111, localhost:1112, we only track them as host (localhost)?
There was a problem hiding this comment.
Was trying to differentiate from up(addr grpc.Address) where addr is [scheme]://[host]:[port], where hb.host2ep contains [host]:[port]. addr is kind of confusing because we actually pass keys from host2ep. What do you think?
There was a problem hiding this comment.
addr == endpoint, which contains the fqdn (scheme://host:port). we need to make this accurate. if it is host, then use host. or use endpoint or addr.
There was a problem hiding this comment.
endpoint sounds better. Will change to endpoint.
There was a problem hiding this comment.
if you want to be accurate, then change everything to hostport if it only contains host:port.
clientv3/retry.go
Outdated
| return func(rpcCtx context.Context, f rpcFunc) error { | ||
| for { | ||
| pinned := c.balancer.pinned() | ||
| err := f(rpcCtx) | ||
| if err == nil { | ||
| return nil | ||
| } | ||
| if logger.V(4) { | ||
| logger.Infof("clientv3/auth-retry: error %v on pinned endpoint %s", err, pinned) |
clientv3/retry.go
Outdated
| // always stop retry on etcd errors other than invalid auth token | ||
| if rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken { | ||
| gterr := c.getToken(rpcCtx) | ||
| if gterr != nil { | ||
| if logger.V(4) { | ||
| logger.Infof("clientv3/auth-retry: error %q(%q) on pinned endpoint %q (returning)", err.Error(), gterr.Error(), pinned) |
There was a problem hiding this comment.
log here that we cannot retry due to error x
clientv3/retry.go
Outdated
| return err // return the original error for simplicity | ||
| } | ||
| if logger.V(4) { | ||
| logger.Infof("clientv3/auth-retry: error %q on pinned endpoint %q (retrying)", err, pinned) |
gyuho
force-pushed
the
aaa
branch
3 times, most recently
from
0976b3f to
832e4a0
Compare
|
@xiang90 All addressed. PTAL. Thanks. |
gyuho
force-pushed
the
aaa
branch
2 times, most recently
from
3ca37a5 to
fd7e638
Compare
clientv3/balancer.go
Outdated
| endpoints() []string | ||
| // pinned returns the current pinned endpoint. | ||
| pinned() string | ||
| // endpointError handles error from server-side. | ||
| endpointError(addr string, err error) | ||
| endpointError(hostPort string, err error) |
clientv3/health_balancer.go
Outdated
| if _, ok := hb.hostPort2ep[k]; !ok { | ||
| delete(hb.unhealthy, k) | ||
| if logger.V(4) { | ||
| logger.Infof("clientv3/health-balancer: removes stale endpoint %q from unhealthy", k) |
clientv3/health_balancer.go
Outdated
| } | ||
| } | ||
|
|
||
| func (hb *healthBalancer) mayPin(addr grpc.Address) bool { | ||
| hb.mu.RLock() | ||
| if _, ok := hb.hostPort2ep[addr.Addr]; !ok { // stale endpoint |
There was a problem hiding this comment.
add.Addr is in host:port format?
There was a problem hiding this comment.
stale hostport on comments
Signed-off-by: Gyu-Ho Lee <[email protected]>
clientv3/retry.go
Outdated
| } | ||
| // always stop retry on etcd errors other than invalid auth token | ||
| if rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken { | ||
| gterr := c.getToken(rpcCtx) | ||
| if gterr != nil { | ||
| if logger.V(4) { | ||
| logger.Infof("clientv3/auth-retry: cannot retry due to error %q(%q) on pinned endpoint %q (returning)", err.Error(), gterr.Error(), pinned) |
clientv3/balancer.go
Outdated
| @@ -40,12 +40,12 @@ type balancer interface { | |||
| grpc.Balancer | |||
| ConnectNotify() <-chan struct{} | |||
|
|
|||
| endpoint(host string) string | |||
| endpoint(hostPort string) string | |||
| endpoints() []string | |||
| // pinned returns the current pinned endpoint. | |||
| pinned() string | |||
| // endpointError handles error from server-side. | |||
|
lgtm |
Signed-off-by: Gyu-Ho Lee <[email protected]>
No description provided.