|
|
|
@ -15,7 +15,7 @@ import (
|
|
|
|
|
"github.com/zeromicro/go-zero/core/logx"
|
|
|
|
|
"github.com/zeromicro/go-zero/core/syncx"
|
|
|
|
|
"github.com/zeromicro/go-zero/core/threading"
|
|
|
|
|
v3rpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
|
|
|
|
|
"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
|
|
|
|
|
clientv3 "go.etcd.io/etcd/client/v3"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
@ -24,6 +24,7 @@ var (
|
|
|
|
|
clusters: make(map[string]*cluster),
|
|
|
|
|
}
|
|
|
|
|
connManager = syncx.NewResourceManager()
|
|
|
|
|
errClosed = errors.New("etcd monitor chan has been closed")
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// A Registry is a registry that manages the etcd client connections.
|
|
|
|
@ -294,36 +295,38 @@ func (c *cluster) watch(cli EtcdClient, key string, rev int64) {
|
|
|
|
|
if err == nil {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
if rev != 0 && errors.Is(err, v3rpc.ErrCompacted) {
|
|
|
|
|
logx.Errorf("etcd watch stream has been compacted, try to reload, rev %v", rev)
|
|
|
|
|
|
|
|
|
|
if rev != 0 && errors.Is(err, rpctypes.ErrCompacted) {
|
|
|
|
|
logx.Errorf("etcd watch stream has been compacted, try to reload, rev %d", rev)
|
|
|
|
|
rev = c.load(cli, key)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// log the error and retry
|
|
|
|
|
logx.Error(err)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (c *cluster) watchStream(cli EtcdClient, key string, rev int64) error {
|
|
|
|
|
var rch clientv3.WatchChan
|
|
|
|
|
if rev != 0 {
|
|
|
|
|
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key), clientv3.WithPrefix(),
|
|
|
|
|
clientv3.WithRev(rev+1))
|
|
|
|
|
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
|
|
|
|
|
clientv3.WithPrefix(), clientv3.WithRev(rev+1))
|
|
|
|
|
} else {
|
|
|
|
|
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key), clientv3.WithPrefix())
|
|
|
|
|
rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
|
|
|
|
|
clientv3.WithPrefix())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
for {
|
|
|
|
|
select {
|
|
|
|
|
case wresp, ok := <-rch:
|
|
|
|
|
if !ok {
|
|
|
|
|
logx.Error("etcd monitor chan has been closed")
|
|
|
|
|
return errors.New("etcd monitor chan has been closed")
|
|
|
|
|
return errClosed
|
|
|
|
|
}
|
|
|
|
|
if wresp.Canceled {
|
|
|
|
|
logx.Errorf("etcd monitor chan has been canceled, error: %v", wresp.Err())
|
|
|
|
|
return wresp.Err()
|
|
|
|
|
return fmt.Errorf("etcd monitor chan has been canceled, error: %w", wresp.Err())
|
|
|
|
|
}
|
|
|
|
|
if wresp.Err() != nil {
|
|
|
|
|
logx.Error(fmt.Sprintf("etcd monitor chan error: %v", wresp.Err()))
|
|
|
|
|
return wresp.Err()
|
|
|
|
|
return fmt.Errorf("etcd monitor chan error: %w", wresp.Err())
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
c.handleWatchEvents(key, wresp.Events)
|
|
|
|
|