Skip to content

Commit

Permalink
refresh rangePermCache when snapshot is applied
Browse files Browse the repository at this point in the history
Signed-off-by: Oleg Guba <oleg@dropbox.com>
  • Loading branch information
veshij committed Oct 26, 2022
1 parent 548bacb commit 0438192
Show file tree
Hide file tree
Showing 3 changed files with 133 additions and 0 deletions.
2 changes: 2 additions & 0 deletions server/auth/range_perm_cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,8 @@ func (as *authStore) refreshRangePermCache(tx AuthReadTx) {
as.rangePermCacheMu.Lock()
defer as.rangePermCacheMu.Unlock()

as.lg.Debug("Refreshing rangePermCache")

as.rangePermCache = make(map[string]*unifiedRangePermissions)

users := tx.UnsafeGetAllUsers()
Expand Down
1 change: 1 addition & 0 deletions server/auth/store.go
Original file line number Diff line number Diff line change
Expand Up @@ -386,6 +386,7 @@ func (as *authStore) Recover(be AuthBackend) {

enabled := tx.UnsafeReadAuthEnabled()
as.setRevision(tx.UnsafeReadAuthRevision())
as.refreshRangePermCache(tx)

tx.Unlock()

Expand Down
130 changes: 130 additions & 0 deletions tests/e2e/ctl_v3_auth_cluster_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
package e2e

import (
"context"
"fmt"
clientv3 "go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/tests/v3/framework/config"
"go.etcd.io/etcd/tests/v3/framework/e2e"
"sync"
"testing"
)

func TestAuthCluster(t *testing.T) {
e2e.BeforeTest(t)
cfg := &e2e.EtcdProcessClusterConfig{
ClusterSize: 1,
InitialToken: "new",
SnapshotCount: 2,
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

epc, err := e2e.NewEtcdProcessCluster(ctx, t, cfg)
if err != nil {
t.Fatalf("could not start etcd process cluster (%v)", err)
}
defer func() {
if err := epc.Close(); err != nil {
t.Fatalf("could not close test cluster (%v)", err)
}
}()

epcClient := epc.Client()
createUsers(ctx, t, epcClient)

if _, err := epcClient.AuthEnable(ctx); err != nil {
t.Fatalf("could not enable Auth: (%v)", err)
}

numKeys := 10
testUserClientOpts := e2e.WithAuth("test", "testPassword")
rootUserClientOpts := e2e.WithAuth("root", "rootPassword")
writeKeys(ctx, epc.Client(testUserClientOpts), numKeys, "test")

if err := epc.StartNewProc(ctx, t, rootUserClientOpts); err != nil {
t.Fatalf("could not start second etcd process (%v)", err)
}

writeKeys(ctx, epc.Client(testUserClientOpts), numKeys, "test_two_nodes")

hashKvs, err := epc.Client(rootUserClientOpts).HashKV(ctx, int64(numKeys*2))
if err != nil {
t.Fatalf("could not HashKV (%v)", err)
}

revisionSet := false
var revision int64
var hash uint32
for _, v := range hashKvs {
if !revisionSet {
revision = v.Header.GetRevision()
hash = v.Hash
revisionSet = true
continue
}
if revision != v.Header.GetRevision() || hash != v.Hash {
t.Errorf("Inconsistent revisions found:")
for _, v := range hashKvs {
t.Errorf("%+v", v)
}
t.Fail()
}
}
}

func createUsers(ctx context.Context, t *testing.T, client *e2e.EtcdctlV3) {
if _, err := client.UserAdd(ctx, "root", "rootPassword", config.UserAddOptions{}); err != nil {
t.Fatalf("could not add root user (%v)", err)
}
if _, err := client.UserGrantRole(ctx, "root", "root"); err != nil {
t.Fatalf("could not grant root role to root user (%v)", err)
}

if _, err := client.RoleAdd(ctx, "test"); err != nil {
t.Fatalf("could not create 'test' role (%v)", err)
}
if _, err := client.RoleGrantPermission(ctx, "test", "/test/", "/test0", clientv3.PermissionType(clientv3.PermReadWrite)); err != nil {
t.Fatalf("could not RoleGrantPermission (%v)", err)
}
if _, err := client.UserAdd(ctx, "test", "testPassword", config.UserAddOptions{}); err != nil {
t.Fatalf("could not add user test (%v)", err)
}
if _, err := client.UserGrantRole(ctx, "test", "test"); err != nil {
t.Fatalf("could not grant test role user (%v)", err)
}
}

func writeKeys(ctx context.Context, client *e2e.EtcdctlV3, numKeys int, value string) {
writeThreads := 10

type kv struct {
key string
value string
}
q := make(chan kv)

wg := sync.WaitGroup{}
for i := 0; i <= writeThreads; i++ {
wg.Add(1)
go func() {
for {
item, ok := <-q
if !ok {
break
}
if err := client.Put(ctx, item.key, item.value, config.PutOptions{}); err != nil {
panic(err)
}
}
wg.Done()
}()
}

for i := 0; i <= numKeys; i++ {
q <- kv{key: fmt.Sprintf("/test/%d", i), value: value}
}
close(q)

wg.Wait()
}

0 comments on commit 0438192

Please sign in to comment.