diff --git a/auth/store.go b/auth/store.go index e1ba64a96..5c6d37dbe 100644 --- a/auth/store.go +++ b/auth/store.go @@ -91,6 +91,9 @@ type AuthenticateParamIndex struct{} // AuthenticateParamSimpleTokenPrefix is used for a key of context in the parameters of Authenticate() type AuthenticateParamSimpleTokenPrefix struct{} +// saveConsistentIndexFunc is used to sync consistentIndex to backend, now reusing store.saveIndex +type saveConsistentIndexFunc func(tx backend.BatchTx) + // AuthStore defines auth storage interface. type AuthStore interface { // AuthEnable turns on the authentication feature @@ -183,6 +186,9 @@ type AuthStore interface { // HasRole checks that user has role HasRole(user, role string) bool + + // SetConsistentIndexSyncer sets consistentIndex syncer + SetConsistentIndexSyncer(syncer saveConsistentIndexFunc) } type TokenProvider interface { @@ -206,10 +212,14 @@ type authStore struct { rangePermCache map[string]*unifiedRangePermissions // username -> unifiedRangePermissions - tokenProvider TokenProvider - bcryptCost int // the algorithm cost / strength for hashing auth passwords + tokenProvider TokenProvider + syncConsistentIndex saveConsistentIndexFunc + bcryptCost int // the algorithm cost / strength for hashing auth passwords } +func (as *authStore) SetConsistentIndexSyncer(syncer saveConsistentIndexFunc) { + as.syncConsistentIndex = syncer +} func (as *authStore) AuthEnable() error { as.enabledMu.Lock() defer as.enabledMu.Unlock() @@ -258,6 +268,7 @@ func (as *authStore) AuthDisable() { tx.Lock() tx.UnsafePut(authBucketName, enableFlagKey, authDisabled) as.commitRevision(tx) + as.saveConsistentIndex(tx) tx.Unlock() b.ForceCommit() @@ -403,6 +414,7 @@ func (as *authStore) UserAdd(r *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, putUser(as.lg, tx, newUser) as.commitRevision(tx) + as.saveConsistentIndex(tx) as.lg.Info("added a user", zap.String("user-name", r.Name)) return &pb.AuthUserAddResponse{}, nil @@ -426,6 +438,7 @@ func (as *authStore) UserDelete(r *pb.AuthUserDeleteRequest) (*pb.AuthUserDelete delUser(tx, r.Name) as.commitRevision(tx) + as.saveConsistentIndex(tx) as.invalidateCachedPerm(r.Name) as.tokenProvider.invalidateUser(r.Name) @@ -470,6 +483,7 @@ func (as *authStore) UserChangePassword(r *pb.AuthUserChangePasswordRequest) (*p putUser(as.lg, tx, updatedUser) as.commitRevision(tx) + as.saveConsistentIndex(tx) as.invalidateCachedPerm(r.Name) as.tokenProvider.invalidateUser(r.Name) @@ -518,6 +532,7 @@ func (as *authStore) UserGrantRole(r *pb.AuthUserGrantRoleRequest) (*pb.AuthUser as.invalidateCachedPerm(r.User) as.commitRevision(tx) + as.saveConsistentIndex(tx) as.lg.Info( "granted a role to a user", @@ -596,6 +611,7 @@ func (as *authStore) UserRevokeRole(r *pb.AuthUserRevokeRoleRequest) (*pb.AuthUs as.invalidateCachedPerm(r.Name) as.commitRevision(tx) + as.saveConsistentIndex(tx) as.lg.Info( "revoked a role from a user", @@ -666,6 +682,7 @@ func (as *authStore) RoleRevokePermission(r *pb.AuthRoleRevokePermissionRequest) as.clearCachedPerm() as.commitRevision(tx) + as.saveConsistentIndex(tx) as.lg.Info( "revoked a permission on range", @@ -717,6 +734,7 @@ func (as *authStore) RoleDelete(r *pb.AuthRoleDeleteRequest) (*pb.AuthRoleDelete } as.commitRevision(tx) + as.saveConsistentIndex(tx) as.lg.Info("deleted a role", zap.String("role-name", r.Role)) return &pb.AuthRoleDeleteResponse{}, nil @@ -743,6 +761,7 @@ func (as *authStore) RoleAdd(r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, putRole(as.lg, tx, newRole) as.commitRevision(tx) + as.saveConsistentIndex(tx) as.lg.Info("created a role", zap.String("role-name", r.Name)) return &pb.AuthRoleAddResponse{}, nil @@ -781,6 +800,16 @@ func (as *authStore) RoleGrantPermission(r *pb.AuthRoleGrantPermissionRequest) ( }) if idx < len(role.KeyPermission) && bytes.Equal(role.KeyPermission[idx].Key, r.Perm.Key) && bytes.Equal(role.KeyPermission[idx].RangeEnd, r.Perm.RangeEnd) { + if role.KeyPermission[idx].PermType == r.Perm.PermType { + as.lg.Warn( + "ignored grant permission request to a role, existing permission", + zap.String("role-name", r.Name), + zap.ByteString("key", r.Perm.Key), + zap.ByteString("range-end", r.Perm.RangeEnd), + zap.String("permission-type", authpb.Permission_Type_name[int32(r.Perm.PermType)]), + ) + return &pb.AuthRoleGrantPermissionResponse{}, nil + } // update existing permission role.KeyPermission[idx].PermType = r.Perm.PermType } else { @@ -802,6 +831,7 @@ func (as *authStore) RoleGrantPermission(r *pb.AuthRoleGrantPermissionRequest) ( as.clearCachedPerm() as.commitRevision(tx) + as.saveConsistentIndex(tx) as.lg.Info( "granted/updated a permission to a user", @@ -1035,6 +1065,7 @@ func NewAuthStore(lg *zap.Logger, be backend.Backend, tp TokenProvider, bcryptCo if as.Revision() == 0 { as.commitRevision(tx) + as.saveConsistentIndex(tx) } tx.Unlock() @@ -1279,3 +1310,11 @@ func (as *authStore) HasRole(user, role string) bool { func (as *authStore) BcryptCost() int { return as.bcryptCost } + +func (as *authStore) saveConsistentIndex(tx backend.BatchTx) { + if as.syncConsistentIndex != nil { + as.syncConsistentIndex(tx) + } else { + as.lg.Error("failed to save consistentIndex,syncConsistentIndex is nil") + } +} diff --git a/auth/store_test.go b/auth/store_test.go index 86128173f..d523b1905 100644 --- a/auth/store_test.go +++ b/auth/store_test.go @@ -414,6 +414,55 @@ func TestListUsers(t *testing.T) { } } +func TestRoleGrantPermissionRevision(t *testing.T) { + as, tearDown := setupAuthStore(t) + defer tearDown(t) + + _, err := as.RoleAdd(&pb.AuthRoleAddRequest{Name: "role-test-1"}) + if err != nil { + t.Fatal(err) + } + + perm := &authpb.Permission{ + PermType: authpb.WRITE, + Key: []byte("Keys"), + RangeEnd: []byte("RangeEnd"), + } + _, err = as.RoleGrantPermission(&pb.AuthRoleGrantPermissionRequest{ + Name: "role-test-1", + Perm: perm, + }) + + if err != nil { + t.Fatal(err) + } + + r, err := as.RoleGet(&pb.AuthRoleGetRequest{Role: "role-test-1"}) + if err != nil { + t.Fatal(err) + } + + if !reflect.DeepEqual(perm, r.Perm[0]) { + t.Errorf("expected %v, got %v", perm, r.Perm[0]) + } + + oldRevision := as.Revision() + + _, err = as.RoleGrantPermission(&pb.AuthRoleGrantPermissionRequest{ + Name: "role-test-1", + Perm: perm, + }) + + if err != nil { + t.Error(err) + } + newRevision := as.Revision() + + if oldRevision != newRevision { + t.Errorf("expected revision diff is 0, got %d", newRevision-oldRevision) + } +} + func TestRoleGrantPermission(t *testing.T) { as, tearDown := setupAuthStore(t) defer tearDown(t) diff --git a/etcdserver/backend.go b/etcdserver/backend.go index 67fd21063..99b06a825 100644 --- a/etcdserver/backend.go +++ b/etcdserver/backend.go @@ -95,7 +95,7 @@ func openBackend(cfg ServerConfig) backend.Backend { // case, replace the db with the snapshot db sent by the leader. func recoverSnapshotBackend(cfg ServerConfig, oldbe backend.Backend, snapshot raftpb.Snapshot) (backend.Backend, error) { var cIndex consistentIndex - kv := mvcc.New(cfg.Logger, oldbe, &lease.FakeLessor{}, &cIndex, mvcc.StoreConfig{CompactionBatchLimit: cfg.CompactionBatchLimit}) + kv := mvcc.New(cfg.Logger, oldbe, &lease.FakeLessor{}, nil, &cIndex, mvcc.StoreConfig{CompactionBatchLimit: cfg.CompactionBatchLimit}) defer kv.Close() if snapshot.Metadata.Index <= kv.ConsistentIndex() { return oldbe, nil diff --git a/etcdserver/server.go b/etcdserver/server.go index 234ddde23..a4fe18c14 100644 --- a/etcdserver/server.go +++ b/etcdserver/server.go @@ -514,7 +514,19 @@ func NewServer(cfg ServerConfig) (srv *EtcdServer, err error) { CheckpointInterval: cfg.LeaseCheckpointInterval, ExpiredLeasesRetryInterval: srv.Cfg.ReqTimeout(), }) - srv.kv = mvcc.New(srv.getLogger(), srv.be, srv.lessor, &srv.consistIndex, mvcc.StoreConfig{CompactionBatchLimit: cfg.CompactionBatchLimit}) + + tp, err := auth.NewTokenProvider(cfg.Logger, cfg.AuthToken, + func(index uint64) <-chan struct{} { + return srv.applyWait.Wait(index) + }, + ) + if err != nil { + cfg.Logger.Warn("failed to create token provider", zap.Error(err)) + return nil, err + } + srv.authStore = auth.NewAuthStore(srv.getLogger(), srv.be, tp, int(cfg.BcryptCost)) + + srv.kv = mvcc.New(srv.getLogger(), srv.be, srv.lessor, srv.authStore, &srv.consistIndex, mvcc.StoreConfig{CompactionBatchLimit: cfg.CompactionBatchLimit}) if beExist { kvindex := srv.kv.ConsistentIndex() // TODO: remove kvindex != 0 checking when we do not expect users to upgrade @@ -539,16 +551,7 @@ func NewServer(cfg ServerConfig) (srv *EtcdServer, err error) { }() srv.consistIndex.setConsistentIndex(srv.kv.ConsistentIndex()) - tp, err := auth.NewTokenProvider(cfg.Logger, cfg.AuthToken, - func(index uint64) <-chan struct{} { - return srv.applyWait.Wait(index) - }, - ) - if err != nil { - cfg.Logger.Warn("failed to create token provider", zap.Error(err)) - return nil, err - } - srv.authStore = auth.NewAuthStore(srv.getLogger(), srv.be, tp, int(cfg.BcryptCost)) + if num := cfg.AutoCompactionRetention; num != 0 { srv.compactor, err = v3compactor.New(cfg.Logger, cfg.AutoCompactionMode, num, srv.kv, srv) if err != nil { diff --git a/etcdserver/server_test.go b/etcdserver/server_test.go index f54cfc148..3a5f4fdd8 100644 --- a/etcdserver/server_test.go +++ b/etcdserver/server_test.go @@ -983,7 +983,7 @@ func TestSnapshot(t *testing.T) { r: *r, v2store: st, } - srv.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, &srv.consistIndex, mvcc.StoreConfig{}) + srv.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, nil, &srv.consistIndex, mvcc.StoreConfig{}) srv.be = be ch := make(chan struct{}, 2) @@ -1064,7 +1064,7 @@ func TestSnapshotOrdering(t *testing.T) { be, tmpPath := backend.NewDefaultTmpBackend() defer os.RemoveAll(tmpPath) - s.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, &s.consistIndex, mvcc.StoreConfig{}) + s.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, nil, &s.consistIndex, mvcc.StoreConfig{}) s.be = be s.start() @@ -1125,7 +1125,7 @@ func TestTriggerSnap(t *testing.T) { } srv.applyV2 = &applierV2store{store: srv.v2store, cluster: srv.cluster} - srv.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, &srv.consistIndex, mvcc.StoreConfig{}) + srv.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, nil, &srv.consistIndex, mvcc.StoreConfig{}) srv.be = be srv.start() @@ -1197,7 +1197,7 @@ func TestConcurrentApplyAndSnapshotV3(t *testing.T) { defer func() { os.RemoveAll(tmpPath) }() - s.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, &s.consistIndex, mvcc.StoreConfig{}) + s.kv = mvcc.New(zap.NewExample(), be, &lease.FakeLessor{}, nil, &s.consistIndex, mvcc.StoreConfig{}) s.be = be s.start() diff --git a/mvcc/kv_test.go b/mvcc/kv_test.go index 466040790..06f82636b 100644 --- a/mvcc/kv_test.go +++ b/mvcc/kv_test.go @@ -712,7 +712,7 @@ func TestKVSnapshot(t *testing.T) { func TestWatchableKVWatch(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})) + s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{})) defer cleanup(s, b, tmpPath) w := s.NewWatchStream() diff --git a/mvcc/watchable_store.go b/mvcc/watchable_store.go index 6cedd5b18..0529993c6 100644 --- a/mvcc/watchable_store.go +++ b/mvcc/watchable_store.go @@ -15,6 +15,7 @@ package mvcc import ( + "go.etcd.io/etcd/auth" "sync" "time" @@ -69,11 +70,11 @@ type watchableStore struct { // cancel operations. type cancelFunc func() -func New(lg *zap.Logger, b backend.Backend, le lease.Lessor, ig ConsistentIndexGetter, cfg StoreConfig) ConsistentWatchableKV { - return newWatchableStore(lg, b, le, ig, cfg) +func New(lg *zap.Logger, b backend.Backend, le lease.Lessor, as auth.AuthStore, ig ConsistentIndexGetter, cfg StoreConfig) ConsistentWatchableKV { + return newWatchableStore(lg, b, le, as, ig, cfg) } -func newWatchableStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, ig ConsistentIndexGetter, cfg StoreConfig) *watchableStore { +func newWatchableStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, as auth.AuthStore, ig ConsistentIndexGetter, cfg StoreConfig) *watchableStore { if lg == nil { lg = zap.NewNop() } @@ -90,6 +91,10 @@ func newWatchableStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, ig Co // use this store as the deleter so revokes trigger watch events s.le.SetRangeDeleter(func() lease.TxnDelete { return s.Write(traceutil.TODO()) }) } + if as != nil { + // TODO: encapsulating consistentindex into a separate package + as.SetConsistentIndexSyncer(s.store.saveIndex) + } s.wg.Add(2) go s.syncWatchersLoop() go s.syncVictimsLoop() diff --git a/mvcc/watchable_store_bench_test.go b/mvcc/watchable_store_bench_test.go index 0f553493f..1b0581b98 100644 --- a/mvcc/watchable_store_bench_test.go +++ b/mvcc/watchable_store_bench_test.go @@ -28,7 +28,7 @@ import ( func BenchmarkWatchableStorePut(b *testing.B) { be, tmpPath := backend.NewDefaultTmpBackend() - s := New(zap.NewExample(), be, &lease.FakeLessor{}, nil, StoreConfig{}) + s := New(zap.NewExample(), be, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer cleanup(s, be, tmpPath) // arbitrary number of bytes @@ -49,7 +49,7 @@ func BenchmarkWatchableStorePut(b *testing.B) { func BenchmarkWatchableStoreTxnPut(b *testing.B) { var i fakeConsistentIndex be, tmpPath := backend.NewDefaultTmpBackend() - s := New(zap.NewExample(), be, &lease.FakeLessor{}, &i, StoreConfig{}) + s := New(zap.NewExample(), be, &lease.FakeLessor{}, nil, &i, StoreConfig{}) defer cleanup(s, be, tmpPath) // arbitrary number of bytes @@ -80,7 +80,7 @@ func BenchmarkWatchableStoreWatchPutUnsync(b *testing.B) { func benchmarkWatchableStoreWatchPut(b *testing.B, synced bool) { be, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), be, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), be, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer cleanup(s, be, tmpPath) k := []byte("testkey") @@ -180,7 +180,7 @@ func BenchmarkWatchableStoreUnsyncedCancel(b *testing.B) { func BenchmarkWatchableStoreSyncedCancel(b *testing.B) { be, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), be, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), be, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() diff --git a/mvcc/watchable_store_test.go b/mvcc/watchable_store_test.go index e4d0cd62e..67d0611f2 100644 --- a/mvcc/watchable_store_test.go +++ b/mvcc/watchable_store_test.go @@ -32,7 +32,7 @@ import ( func TestWatch(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() @@ -54,7 +54,7 @@ func TestWatch(t *testing.T) { func TestNewWatcherCancel(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() @@ -224,7 +224,7 @@ func TestSyncWatchers(t *testing.T) { // TestWatchCompacted tests a watcher that watches on a compacted revision. func TestWatchCompacted(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() @@ -261,7 +261,7 @@ func TestWatchCompacted(t *testing.T) { func TestWatchFutureRev(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() @@ -302,7 +302,7 @@ func TestWatchRestore(t *testing.T) { test := func(delay time.Duration) func(t *testing.T) { return func(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer cleanup(s, b, tmpPath) testKey := []byte("foo") @@ -310,7 +310,7 @@ func TestWatchRestore(t *testing.T) { rev := s.Put(testKey, testValue, lease.NoLease) newBackend, newPath := backend.NewDefaultTmpBackend() - newStore := newWatchableStore(zap.NewExample(), newBackend, &lease.FakeLessor{}, nil, StoreConfig{}) + newStore := newWatchableStore(zap.NewExample(), newBackend, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer cleanup(newStore, newBackend, newPath) w := newStore.NewWatchStream() @@ -348,11 +348,11 @@ func TestWatchRestore(t *testing.T) { // 5. choose the watcher from step 1, without panic func TestWatchRestoreSyncedWatcher(t *testing.T) { b1, b1Path := backend.NewDefaultTmpBackend() - s1 := newWatchableStore(zap.NewExample(), b1, &lease.FakeLessor{}, nil, StoreConfig{}) + s1 := newWatchableStore(zap.NewExample(), b1, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer cleanup(s1, b1, b1Path) b2, b2Path := backend.NewDefaultTmpBackend() - s2 := newWatchableStore(zap.NewExample(), b2, &lease.FakeLessor{}, nil, StoreConfig{}) + s2 := newWatchableStore(zap.NewExample(), b2, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer cleanup(s2, b2, b2Path) testKey, testValue := []byte("foo"), []byte("bar") @@ -399,7 +399,7 @@ func TestWatchRestoreSyncedWatcher(t *testing.T) { // TestWatchBatchUnsynced tests batching on unsynced watchers func TestWatchBatchUnsynced(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) oldMaxRevs := watchBatchMaxRevs defer func() { @@ -533,7 +533,7 @@ func TestWatchVictims(t *testing.T) { oldChanBufLen, oldMaxWatchersPerSync := chanBufLen, maxWatchersPerSync b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() @@ -611,7 +611,7 @@ func TestWatchVictims(t *testing.T) { // canceling its watches. func TestStressWatchCancelClose(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() diff --git a/mvcc/watcher_bench_test.go b/mvcc/watcher_bench_test.go index 901a1ec0d..f2e6c6f6e 100644 --- a/mvcc/watcher_bench_test.go +++ b/mvcc/watcher_bench_test.go @@ -26,7 +26,7 @@ import ( func BenchmarkKVWatcherMemoryUsage(b *testing.B) { be, tmpPath := backend.NewDefaultTmpBackend() - watchable := newWatchableStore(zap.NewExample(), be, &lease.FakeLessor{}, nil, StoreConfig{}) + watchable := newWatchableStore(zap.NewExample(), be, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer cleanup(watchable, be, tmpPath) diff --git a/mvcc/watcher_test.go b/mvcc/watcher_test.go index 679ab441d..b350af99b 100644 --- a/mvcc/watcher_test.go +++ b/mvcc/watcher_test.go @@ -32,7 +32,7 @@ import ( // and the watched event attaches the correct watchID. func TestWatcherWatchID(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})) + s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{})) defer cleanup(s, b, tmpPath) w := s.NewWatchStream() @@ -82,7 +82,7 @@ func TestWatcherWatchID(t *testing.T) { func TestWatcherRequestsCustomID(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})) + s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{})) defer cleanup(s, b, tmpPath) w := s.NewWatchStream() @@ -119,7 +119,7 @@ func TestWatcherRequestsCustomID(t *testing.T) { // and returns events with matching prefixes. func TestWatcherWatchPrefix(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})) + s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{})) defer cleanup(s, b, tmpPath) w := s.NewWatchStream() @@ -193,7 +193,7 @@ func TestWatcherWatchPrefix(t *testing.T) { // does not create watcher, which panics when canceling in range tree. func TestWatcherWatchWrongRange(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})) + s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{})) defer cleanup(s, b, tmpPath) w := s.NewWatchStream() @@ -213,7 +213,7 @@ func TestWatcherWatchWrongRange(t *testing.T) { func TestWatchDeleteRange(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{}) + s := newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{}) defer func() { s.store.Close() @@ -252,7 +252,7 @@ func TestWatchDeleteRange(t *testing.T) { // with given id inside watchStream. func TestWatchStreamCancelWatcherByID(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})) + s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{})) defer cleanup(s, b, tmpPath) w := s.NewWatchStream() @@ -344,7 +344,7 @@ func TestWatcherRequestProgress(t *testing.T) { func TestWatcherWatchWithFilter(t *testing.T) { b, tmpPath := backend.NewDefaultTmpBackend() - s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})) + s := WatchableKV(newWatchableStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, nil, StoreConfig{})) defer cleanup(s, b, tmpPath) w := s.NewWatchStream()