// Copyright (c) HashiCorp, Inc. // SPDX-License-Identifier: MPL-2.0 package state import ( "reflect" "strings" "testing" "time" "github.com/stretchr/testify/require" "github.com/hashicorp/go-memdb" "github.com/hashicorp/consul/agent/structs" ) func TestStateStore_ReapTombstones(t *testing.T) { s := testStateStore(t) // Create some KV pairs. testSetKey(t, s, 1, "foo", "foo", nil) testSetKey(t, s, 2, "foo/bar", "bar", nil) testSetKey(t, s, 3, "foo/baz", "bar", nil) testSetKey(t, s, 4, "foo/moo", "bar", nil) testSetKey(t, s, 5, "foo/zoo", "bar", nil) // Call a delete on some specific keys. if err := s.KVSDelete(6, "foo/baz", nil); err != nil { t.Fatalf("err: %s", err) } if err := s.KVSDelete(7, "foo/moo", nil); err != nil { t.Fatalf("err: %s", err) } // Pull out the list and check the index, which should come from the // tombstones. idx, _, err := s.KVSList(nil, "foo/", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 7 { t.Fatalf("bad index: %d", idx) } // Reap the tombstones <= 6. if err := s.ReapTombstones(8, 6); err != nil { t.Fatalf("err: %s", err) } // Should still be good because 7 is in there. idx, _, err = s.KVSList(nil, "foo/", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 7 { t.Fatalf("bad index: %d", idx) } // Now reap them all. if err := s.ReapTombstones(9, 7); err != nil { t.Fatalf("err: %s", err) } // At this point the sub index will slide backwards. idx, _, err = s.KVSList(nil, "foo/", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 5 { t.Fatalf("bad index: %d", idx) } // Make sure the tombstones are actually gone. snap := s.Snapshot() defer snap.Close() stones, err := snap.Tombstones() if err != nil { t.Fatalf("err: %s", err) } if stones.Next() != nil { t.Fatalf("unexpected extra tombstones") } } func TestStateStore_GC(t *testing.T) { // Build up a fast GC. ttl := 10 * time.Millisecond gran := 5 * time.Millisecond gc, err := NewTombstoneGC(ttl, gran) if err != nil { t.Fatalf("err: %s", err) } // Enable it and attach it to the state store. gc.SetEnabled(true) s := NewStateStore(gc) // Create some KV pairs. testSetKey(t, s, 1, "foo", "foo", nil) testSetKey(t, s, 2, "foo/bar", "bar", nil) testSetKey(t, s, 3, "foo/baz", "bar", nil) testSetKey(t, s, 4, "foo/moo", "bar", nil) testSetKey(t, s, 5, "foo/zoo", "bar", nil) // Delete a key and make sure the GC sees it. if err := s.KVSDelete(6, "foo/zoo", nil); err != nil { t.Fatalf("err: %s", err) } select { case idx := <-gc.ExpireCh(): if idx != 6 { t.Fatalf("bad index: %d", idx) } case <-time.After(2 * ttl): t.Fatalf("GC never fired") } // Check for the same behavior with a tree delete. if err := s.KVSDeleteTree(7, "foo/moo", nil); err != nil { t.Fatalf("err: %s", err) } select { case idx := <-gc.ExpireCh(): if idx != 7 { t.Fatalf("bad index: %d", idx) } case <-time.After(2 * ttl): t.Fatalf("GC never fired") } // Check for the same behavior with a CAS delete. if ok, err := s.KVSDeleteCAS(8, 3, "foo/baz", nil); !ok || err != nil { t.Fatalf("err: %s", err) } select { case idx := <-gc.ExpireCh(): if idx != 8 { t.Fatalf("bad index: %d", idx) } case <-time.After(2 * ttl): t.Fatalf("GC never fired") } // Finally, try it with an expiring session. testRegisterNode(t, s, 9, "node1") session := &structs.Session{ ID: testUUID(), Node: "node1", Behavior: structs.SessionKeysDelete, } if err := s.SessionCreate(10, session); err != nil { t.Fatalf("err: %s", err) } d := &structs.DirEntry{ Key: "lock", Session: session.ID, } if ok, err := s.KVSLock(11, d); !ok || err != nil { t.Fatalf("err: %v", err) } if err := s.SessionDestroy(12, session.ID, nil); err != nil { t.Fatalf("err: %s", err) } select { case idx := <-gc.ExpireCh(): if idx != 12 { t.Fatalf("bad index: %d", idx) } case <-time.After(2 * ttl): t.Fatalf("GC never fired") } } func TestStateStore_KVSSet_KVSGet(t *testing.T) { s := testStateStore(t) // Get on an nonexistent key returns nil. ws := memdb.NewWatchSet() idx, result, err := s.KVSGet(ws, "foo", nil) if result != nil || err != nil || idx != 0 { t.Fatalf("expected (0, nil, nil), got : (%#v, %#v, %#v)", idx, result, err) } // Write a new K/V entry to the store. entry := &structs.DirEntry{ Key: "foo", Value: []byte("bar"), } if err := s.KVSSet(1, entry); err != nil { t.Fatalf("err: %s", err) } if !watchFired(ws) { t.Fatalf("bad") } // Retrieve the K/V entry again. ws = memdb.NewWatchSet() idx, result, err = s.KVSGet(ws, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result == nil { t.Fatalf("expected k/v pair, got nothing") } if idx != 1 { t.Fatalf("bad index: %d", idx) } // Check that the index was injected into the result. if result.CreateIndex != 1 || result.ModifyIndex != 1 { t.Fatalf("bad index: %d, %d", result.CreateIndex, result.ModifyIndex) } // Check that the value matches. if v := string(result.Value); v != "bar" { t.Fatalf("expected 'bar', got: '%s'", v) } // Updating the entry works and changes the index. update := &structs.DirEntry{ Key: "foo", Value: []byte("baz"), } if err := s.KVSSet(2, update); err != nil { t.Fatalf("err: %s", err) } if !watchFired(ws) { t.Fatalf("bad") } // Fetch the kv pair and check. ws = memdb.NewWatchSet() idx, result, err = s.KVSGet(ws, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.CreateIndex != 1 || result.ModifyIndex != 2 { t.Fatalf("bad index: %d, %d", result.CreateIndex, result.ModifyIndex) } if v := string(result.Value); v != "baz" { t.Fatalf("expected 'baz', got '%s'", v) } if idx != 2 { t.Fatalf("bad index: %d", idx) } // Attempt to set the session during an update. update = &structs.DirEntry{ Key: "foo", Value: []byte("zoo"), Session: "nope", } if err := s.KVSSet(3, update); err != nil { t.Fatalf("err: %s", err) } if !watchFired(ws) { t.Fatalf("bad") } // Fetch the kv pair and check. ws = memdb.NewWatchSet() idx, result, err = s.KVSGet(ws, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.CreateIndex != 1 || result.ModifyIndex != 3 { t.Fatalf("bad index: %d, %d", result.CreateIndex, result.ModifyIndex) } if v := string(result.Value); v != "zoo" { t.Fatalf("expected 'zoo', got '%s'", v) } if result.Session != "" { t.Fatalf("expected empty session, got '%s", result.Session) } if idx != 3 { t.Fatalf("bad index: %d", idx) } // Make a real session and then lock the key to set the session. testRegisterNode(t, s, 4, "node1") session := testUUID() if err := s.SessionCreate(5, &structs.Session{ID: session, Node: "node1"}); err != nil { t.Fatalf("err: %s", err) } update = &structs.DirEntry{ Key: "foo", Value: []byte("locked"), Session: session, } ok, err := s.KVSLock(6, update) if !ok || err != nil { t.Fatalf("didn't get the lock: %v %s", ok, err) } if !watchFired(ws) { t.Fatalf("bad") } // Fetch the kv pair and check. ws = memdb.NewWatchSet() idx, result, err = s.KVSGet(ws, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.CreateIndex != 1 || result.ModifyIndex != 6 { t.Fatalf("bad index: %d, %d", result.CreateIndex, result.ModifyIndex) } if v := string(result.Value); v != "locked" { t.Fatalf("expected 'zoo', got '%s'", v) } if result.Session != session { t.Fatalf("expected session, got '%s", result.Session) } if idx != 6 { t.Fatalf("bad index: %d", idx) } // Now make an update without the session and make sure it gets applied // and doesn't take away the session (it is allowed to change the value). update = &structs.DirEntry{ Key: "foo", Value: []byte("stoleit"), } if err := s.KVSSet(7, update); err != nil { t.Fatalf("err: %s", err) } if !watchFired(ws) { t.Fatalf("bad") } // Fetch the kv pair and check. ws = memdb.NewWatchSet() idx, result, err = s.KVSGet(ws, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.CreateIndex != 1 || result.ModifyIndex != 7 { t.Fatalf("bad index: %d, %d", result.CreateIndex, result.ModifyIndex) } if v := string(result.Value); v != "stoleit" { t.Fatalf("expected 'zoo', got '%s'", v) } if result.Session != session { t.Fatalf("expected session, got '%s", result.Session) } if idx != 7 { t.Fatalf("bad index: %d", idx) } // Setting some unrelated key should not fire the watch. testSetKey(t, s, 8, "other", "yup", nil) if watchFired(ws) { t.Fatalf("bad") } // Fetch a key that doesn't exist and make sure we get the right // response. idx, result, err = s.KVSGet(nil, "nope", nil) if result != nil || err != nil || idx != 8 { t.Fatalf("expected (8, nil, nil), got : (%#v, %#v, %#v)", idx, result, err) } // setting the same value again does not update the index ws = memdb.NewWatchSet() // Write a new K/V entry to the store. entry = &structs.DirEntry{ Key: "foo", Value: []byte("bar"), } require.Nil(t, s.KVSSet(1, entry)) require.Nil(t, s.KVSSet(2, entry)) idx, _, err = s.KVSGet(ws, entry.Key, nil) require.Nil(t, err) require.Equal(t, uint64(1), idx) } func TestStateStore_KVSList(t *testing.T) { s := testStateStore(t) // Listing an empty KVS returns nothing ws := memdb.NewWatchSet() idx, entries, err := s.KVSList(ws, "", nil) if idx != 0 || entries != nil || err != nil { t.Fatalf("expected (0, nil, nil), got: (%d, %#v, %#v)", idx, entries, err) } // Create some KVS entries testSetKey(t, s, 1, "foo", "foo", nil) testSetKey(t, s, 2, "foo/bar", "bar", nil) testSetKey(t, s, 3, "foo/bar/zip", "zip", nil) testSetKey(t, s, 4, "foo/bar/zip/zorp", "zorp", nil) testSetKey(t, s, 5, "foo/bar/baz", "baz", nil) if !watchFired(ws) { t.Fatalf("bad") } // List out all of the keys idx, entries, err = s.KVSList(nil, "", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 5 { t.Fatalf("bad index: %d", idx) } // Check that all of the keys were returned if n := len(entries); n != 5 { t.Fatalf("expected 5 kvs entries, got: %d", n) } // Try listing with a provided prefix idx, entries, err = s.KVSList(nil, "foo/bar/zip", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 4 { t.Fatalf("bad index: %d", idx) } // Check that only the keys in the prefix were returned if n := len(entries); n != 2 { t.Fatalf("expected 2 kvs entries, got: %d", n) } if entries[0].Key != "foo/bar/zip" || entries[1].Key != "foo/bar/zip/zorp" { t.Fatalf("bad: %#v", entries) } // Delete a key and make sure the index comes from the tombstone. ws = memdb.NewWatchSet() _, _, err = s.KVSList(ws, "foo/bar/baz", nil) if err != nil { t.Fatalf("err: %s", err) } if err := s.KVSDelete(6, "foo/bar/baz", nil); err != nil { t.Fatalf("err: %s", err) } if !watchFired(ws) { t.Fatalf("bad") } ws = memdb.NewWatchSet() idx, _, err = s.KVSList(ws, "foo/bar/baz", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 6 { t.Fatalf("bad index: %d", idx) } // Set a different key to bump the index. This shouldn't fire the // watch since there's a different prefix. testSetKey(t, s, 7, "some/other/key", "", nil) if watchFired(ws) { t.Fatalf("bad") } // Make sure we get the right index from the tombstone. idx, _, err = s.KVSList(nil, "foo/bar/baz", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 6 { t.Fatalf("bad index: %d", idx) } // Now reap the tombstones and make sure we get the latest index // since there are no matching keys. if err := s.ReapTombstones(8, 6); err != nil { t.Fatalf("err: %s", err) } idx, _, err = s.KVSList(nil, "foo/bar/baz", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 7 { t.Fatalf("bad index: %d", idx) } // List all the keys to make sure the index is also correct. idx, _, err = s.KVSList(nil, "", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 7 { t.Fatalf("bad index: %d", idx) } } func TestStateStore_KVSDelete(t *testing.T) { s := testStateStore(t) // Create some KV pairs testSetKey(t, s, 1, "foo", "foo", nil) testSetKey(t, s, 2, "foo/bar", "bar", nil) // Call a delete on a specific key if err := s.KVSDelete(3, "foo", nil); err != nil { t.Fatalf("err: %s", err) } // The entry was removed from the state store tx := s.db.Txn(false) defer tx.Abort() e, err := tx.First(tableKVs, indexID, Query{Value: "foo"}) if err != nil { t.Fatalf("err: %s", err) } if e != nil { t.Fatalf("expected kvs entry to be deleted, got: %#v", e) } // Try fetching the other keys to ensure they still exist e, err = tx.First(tableKVs, indexID, Query{Value: "foo/bar"}) if err != nil { t.Fatalf("err: %s", err) } if e == nil || string(e.(*structs.DirEntry).Value) != "bar" { t.Fatalf("bad kvs entry: %#v", e) } // Check that the index table was updated if idx := s.maxIndex(partitionedIndexEntryName(tableKVs, "default")); idx != 3 { t.Fatalf("bad index: %d", idx) } // Check that the tombstone was created and that prevents the index // from sliding backwards. idx, _, err := s.KVSList(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 3 { t.Fatalf("bad index: %d", idx) } // Now reap the tombstone and watch the index revert to the remaining // foo/bar key's index. if err := s.ReapTombstones(4, 3); err != nil { t.Fatalf("err: %s", err) } idx, _, err = s.KVSList(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 2 { t.Fatalf("bad index: %d", idx) } // Deleting a nonexistent key should be idempotent and not return an // error if err := s.KVSDelete(5, "foo", nil); err != nil { t.Fatalf("err: %s", err) } if idx := s.maxIndex(partitionedIndexEntryName(tableKVs, "default")); idx != 3 { t.Fatalf("bad index: %d", idx) } } func TestStateStore_KVSDeleteCAS(t *testing.T) { s := testStateStore(t) // Create some KV entries testSetKey(t, s, 1, "foo", "foo", nil) testSetKey(t, s, 2, "bar", "bar", nil) testSetKey(t, s, 3, "baz", "baz", nil) // Do a CAS delete with an index lower than the entry ok, err := s.KVSDeleteCAS(4, 1, "bar", nil) if ok || err != nil { t.Fatalf("expected (false, nil), got: (%v, %#v)", ok, err) } // Check that the index is untouched and the entry // has not been deleted. idx, e, err := s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if e == nil { t.Fatalf("expected a kvs entry, got nil") } if idx != 3 { t.Fatalf("bad index: %d", idx) } // Do another CAS delete, this time with the correct index // which should cause the delete to take place. ok, err = s.KVSDeleteCAS(4, 2, "bar", nil) if !ok || err != nil { t.Fatalf("expected (true, nil), got: (%v, %#v)", ok, err) } // Entry was deleted and index was updated idx, e, err = s.KVSGet(nil, "bar", nil) if err != nil { t.Fatalf("err: %s", err) } if e != nil { t.Fatalf("entry should be deleted") } if idx != 4 { t.Fatalf("bad index: %d", idx) } // Add another key to bump the index. testSetKey(t, s, 5, "some/other/key", "baz", nil) // Check that the tombstone was created and that prevents the index // from sliding backwards. idx, _, err = s.KVSList(nil, "bar", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 4 { t.Fatalf("bad index: %d", idx) } // Now reap the tombstone and watch the index move up to the table // index since there are no matching keys. if err := s.ReapTombstones(6, 4); err != nil { t.Fatalf("err: %s", err) } idx, _, err = s.KVSList(nil, "bar", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 5 { t.Fatalf("bad index: %d", idx) } // A delete on a nonexistent key should be idempotent and not return an // error ok, err = s.KVSDeleteCAS(7, 2, "bar", nil) if !ok || err != nil { t.Fatalf("expected (true, nil), got: (%v, %#v)", ok, err) } if idx := s.maxIndex(partitionedIndexEntryName(tableKVs, "default")); idx != 5 { t.Fatalf("bad index: %d", idx) } } func TestStateStore_KVSSetCAS(t *testing.T) { s := testStateStore(t) // Doing a CAS with ModifyIndex != 0 and no existing entry // is a no-op. entry := &structs.DirEntry{ Key: "foo", Value: []byte("foo"), RaftIndex: structs.RaftIndex{ CreateIndex: 1, ModifyIndex: 1, }, } ok, err := s.KVSSetCAS(2, entry) if ok || err != nil { t.Fatalf("expected (false, nil), got: (%#v, %#v)", ok, err) } // Check that nothing was actually stored tx := s.db.Txn(false) if e, err := tx.First(tableKVs, indexID, Query{Value: "foo"}); e != nil || err != nil { t.Fatalf("expected (nil, nil), got: (%#v, %#v)", e, err) } tx.Abort() // Index was not updated if idx := s.maxIndex("kvs"); idx != 0 { t.Fatalf("bad index: %d", idx) } // Doing a CAS with a ModifyIndex of zero when no entry exists // performs the set and saves into the state store. entry = &structs.DirEntry{ Key: "foo", Value: []byte("foo"), RaftIndex: structs.RaftIndex{ CreateIndex: 0, ModifyIndex: 0, }, } ok, err = s.KVSSetCAS(2, entry) if !ok || err != nil { t.Fatalf("expected (true, nil), got: (%#v, %#v)", ok, err) } // Entry was inserted idx, entry, err := s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if string(entry.Value) != "foo" || entry.CreateIndex != 2 || entry.ModifyIndex != 2 { t.Fatalf("bad entry: %#v", entry) } if idx != 2 { t.Fatalf("bad index: %d", idx) } // Doing a CAS with a ModifyIndex of zero when an entry exists does // not do anything. entry = &structs.DirEntry{ Key: "foo", Value: []byte("foo"), RaftIndex: structs.RaftIndex{ CreateIndex: 0, ModifyIndex: 0, }, } ok, err = s.KVSSetCAS(3, entry) if ok || err != nil { t.Fatalf("expected (false, nil), got: (%#v, %#v)", ok, err) } // Doing a CAS with a ModifyIndex which does not match the current // index does not do anything. entry = &structs.DirEntry{ Key: "foo", Value: []byte("bar"), RaftIndex: structs.RaftIndex{ CreateIndex: 3, ModifyIndex: 3, }, } ok, err = s.KVSSetCAS(3, entry) if ok || err != nil { t.Fatalf("expected (false, nil), got: (%#v, %#v)", ok, err) } // Entry was not updated in the store idx, entry, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if string(entry.Value) != "foo" || entry.CreateIndex != 2 || entry.ModifyIndex != 2 { t.Fatalf("bad entry: %#v", entry) } if idx != 2 { t.Fatalf("bad index: %d", idx) } // Doing a CAS with the proper current index should make the // modification. entry = &structs.DirEntry{ Key: "foo", Value: []byte("bar"), RaftIndex: structs.RaftIndex{ CreateIndex: 2, ModifyIndex: 2, }, } ok, err = s.KVSSetCAS(3, entry) if !ok || err != nil { t.Fatalf("expected (true, nil), got: (%#v, %#v)", ok, err) } // Entry was updated idx, entry, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if string(entry.Value) != "bar" || entry.CreateIndex != 2 || entry.ModifyIndex != 3 { t.Fatalf("bad entry: %#v", entry) } if idx != 3 { t.Fatalf("bad index: %d", idx) } // Attempt to update the session during the CAS. entry = &structs.DirEntry{ Key: "foo", Value: []byte("zoo"), Session: "nope", RaftIndex: structs.RaftIndex{ CreateIndex: 2, ModifyIndex: 3, }, } ok, err = s.KVSSetCAS(4, entry) if !ok || err != nil { t.Fatalf("expected (true, nil), got: (%#v, %#v)", ok, err) } // Entry was updated, but the session should have been ignored. idx, entry, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if string(entry.Value) != "zoo" || entry.CreateIndex != 2 || entry.ModifyIndex != 4 || entry.Session != "" { t.Fatalf("bad entry: %#v", entry) } if idx != 4 { t.Fatalf("bad index: %d", idx) } // Now lock it and try the update, which should keep the session. testRegisterNode(t, s, 5, "node1") session := testUUID() if err := s.SessionCreate(6, &structs.Session{ID: session, Node: "node1"}); err != nil { t.Fatalf("err: %s", err) } entry = &structs.DirEntry{ Key: "foo", Value: []byte("locked"), Session: session, RaftIndex: structs.RaftIndex{ CreateIndex: 2, ModifyIndex: 4, }, } ok, err = s.KVSLock(6, entry) if !ok || err != nil { t.Fatalf("didn't get the lock: %v %s", ok, err) } entry = &structs.DirEntry{ Key: "foo", Value: []byte("locked"), RaftIndex: structs.RaftIndex{ CreateIndex: 2, ModifyIndex: 6, }, } ok, err = s.KVSSetCAS(7, entry) if !ok || err != nil { t.Fatalf("expected (true, nil), got: (%#v, %#v)", ok, err) } // Entry was updated, and the lock status should have stayed the same. idx, entry, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if string(entry.Value) != "locked" || entry.CreateIndex != 2 || entry.ModifyIndex != 7 || entry.Session != session { t.Fatalf("bad entry: %#v", entry) } if idx != 7 { t.Fatalf("bad index: %d", idx) } } func TestStateStore_KVSDeleteTree(t *testing.T) { s := testStateStore(t) // Create kvs entries in the state store. testSetKey(t, s, 1, "foo/bar", "bar", nil) testSetKey(t, s, 2, "foo/bar/baz", "baz", nil) testSetKey(t, s, 3, "foo/bar/zip", "zip", nil) testSetKey(t, s, 4, "foo/zorp", "zorp", nil) // Calling tree deletion which affects nothing does not // modify the table index. if err := s.KVSDeleteTree(9, "bar", nil); err != nil { t.Fatalf("err: %s", err) } if idx := s.maxIndex(partitionedIndexEntryName(tableKVs, "default")); idx != 4 { t.Fatalf("bad index: %d", idx) } // Call tree deletion with a nested prefix. if err := s.KVSDeleteTree(5, "foo/bar", nil); err != nil { t.Fatalf("err: %s", err) } // Check that all the matching keys were deleted tx := s.db.Txn(false) defer tx.Abort() entries, err := tx.Get(tableKVs, indexID) if err != nil { t.Fatalf("err: %s", err) } num := 0 for entry := entries.Next(); entry != nil; entry = entries.Next() { if entry.(*structs.DirEntry).Key != "foo/zorp" { t.Fatalf("unexpected kvs entry: %#v", entry) } num++ } if num != 1 { t.Fatalf("expected 1 key, got: %d", num) } // Index should be updated if modifications are made if idx := s.maxIndex(partitionedIndexEntryName(tableKVs, "default")); idx != 5 { t.Fatalf("bad index: %d", idx) } // Check that the tombstones ware created and that prevents the index // from sliding backwards. idx, _, err := s.KVSList(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 5 { t.Fatalf("bad index: %d", idx) } // Now reap the tombstones and watch the index revert to the remaining // foo/zorp key's index. if err := s.ReapTombstones(6, 5); err != nil { t.Fatalf("err: %s", err) } idx, _, err = s.KVSList(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 4 { t.Fatalf("bad index: %d", idx) } } func TestStateStore_Watches_PrefixDelete(t *testing.T) { s := testStateStore(t) // Create some KVS entries testSetKey(t, s, 1, "foo", "foo", nil) testSetKey(t, s, 2, "foo/bar", "bar", nil) testSetKey(t, s, 3, "foo/bar/zip", "zip", nil) testSetKey(t, s, 4, "foo/bar/zip/zorp", "zorp", nil) testSetKey(t, s, 5, "foo/bar/zip/zap", "zap", nil) testSetKey(t, s, 6, "foo/nope", "nope", nil) ws := memdb.NewWatchSet() got, _, err := s.KVSList(ws, "foo/bar", nil) if err != nil { t.Fatalf("unexpected err: %s", err) } var wantIndex uint64 = 5 if got != wantIndex { t.Fatalf("bad index: %d, expected %d", wantIndex, got) } // Delete a key and make sure the index comes from the tombstone. if err := s.KVSDeleteTree(7, "foo/bar/zip", nil); err != nil { t.Fatalf("unexpected err: %s", err) } // Make sure watch fires if !watchFired(ws) { t.Fatalf("expected watch to fire but it did not") } //Verify index matches tombstone got, _, err = s.KVSList(ws, "foo/bar", nil) if err != nil { t.Fatalf("unexpected err: %s", err) } wantIndex = 7 if got != wantIndex { t.Fatalf("bad index: %d, expected %d", got, wantIndex) } // Make sure watch fires if !watchFired(ws) { t.Fatalf("expected watch to fire but it did not") } // Reap tombstone and verify list on the same key reverts its index value if err := s.ReapTombstones(8, wantIndex); err != nil { t.Fatalf("err: %s", err) } got, _, err = s.KVSList(nil, "foo/bar", nil) wantIndex = 2 if err != nil { t.Fatalf("err: %s", err) } if got != wantIndex { t.Fatalf("bad index: %d, expected %d", got, wantIndex) } // Set a different key to bump the index. This shouldn't fire the // watch since there's a different prefix. testSetKey(t, s, 9, "some/other/key", "", nil) // Now ask for the index for a node within the prefix that was deleted // We expect to get the max index in the tree wantIndex = 9 ws = memdb.NewWatchSet() got, _, err = s.KVSList(ws, "foo/bar/baz", nil) if err != nil { t.Fatalf("err: %s", err) } if watchFired(ws) { t.Fatalf("Watch should not have fired") } if got != wantIndex { t.Fatalf("bad index: %d, expected %d", got, wantIndex) } // List all the keys to make sure the index returned is the max index got, _, err = s.KVSList(nil, "", nil) if err != nil { t.Fatalf("err: %s", err) } if got != wantIndex { t.Fatalf("bad index: %d, expected %d", got, wantIndex) } // Delete all the keys, special case where tombstones are not inserted if err := s.KVSDeleteTree(10, "", nil); err != nil { t.Fatalf("unexpected err: %s", err) } wantIndex = 10 got, _, err = s.KVSList(nil, "/foo/bar", nil) if err != nil { t.Fatalf("err: %s", err) } if got != wantIndex { t.Fatalf("bad index: %d, expected %d", got, wantIndex) } } func TestStateStore_KVSLockDelay(t *testing.T) { s := testStateStore(t) // KVSLockDelay is exercised in the lock/unlock and session invalidation // cases below, so we just do a basic check on a nonexistent key here. expires := s.KVSLockDelay("/not/there", nil) if expires.After(time.Now()) { t.Fatalf("bad: %v", expires) } } func TestStateStore_KVSLock(t *testing.T) { s := testStateStore(t) // Lock with no session should fail. ok, err := s.KVSLock(0, &structs.DirEntry{Key: "foo", Value: []byte("foo")}) if ok || err == nil || !strings.Contains(err.Error(), "missing session") { t.Fatalf("didn't detect missing session: %v %s", ok, err) } // Now try with a bogus session. ok, err = s.KVSLock(1, &structs.DirEntry{Key: "foo", Value: []byte("foo"), Session: testUUID()}) if ok || err == nil || !strings.Contains(err.Error(), "invalid session") { t.Fatalf("didn't detect invalid session: %v %s", ok, err) } // Make a real session. testRegisterNode(t, s, 2, "node1") session1 := testUUID() if err := s.SessionCreate(3, &structs.Session{ID: session1, Node: "node1"}); err != nil { t.Fatalf("err: %s", err) } // Lock and make the key at the same time. ok, err = s.KVSLock(4, &structs.DirEntry{Key: "foo", Value: []byte("foo"), Session: session1}) if !ok || err != nil { t.Fatalf("didn't get the lock: %v %s", ok, err) } // Make sure the indexes got set properly. idx, result, err := s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 1 || result.CreateIndex != 4 || result.ModifyIndex != 4 || string(result.Value) != "foo" { t.Fatalf("bad entry: %#v", result) } if idx != 4 { t.Fatalf("bad index: %d", idx) } // Re-locking with the same session should update the value and report // success. ok, err = s.KVSLock(5, &structs.DirEntry{Key: "foo", Value: []byte("bar"), Session: session1}) if !ok || err != nil { t.Fatalf("didn't handle locking an already-locked key: %v %s", ok, err) } // Make sure the indexes got set properly, note that the lock index // won't go up since we didn't lock it again. idx, result, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 1 || result.CreateIndex != 4 || result.ModifyIndex != 5 || string(result.Value) != "bar" { t.Fatalf("bad entry: %#v", result) } if idx != 5 { t.Fatalf("bad index: %d", idx) } // Unlock and the re-lock. ok, err = s.KVSUnlock(6, &structs.DirEntry{Key: "foo", Value: []byte("baz"), Session: session1}) if !ok || err != nil { t.Fatalf("didn't handle unlocking a locked key: %v %s", ok, err) } ok, err = s.KVSLock(7, &structs.DirEntry{Key: "foo", Value: []byte("zoo"), Session: session1}) if !ok || err != nil { t.Fatalf("didn't get the lock: %v %s", ok, err) } // Make sure the indexes got set properly. idx, result, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 2 || result.CreateIndex != 4 || result.ModifyIndex != 7 || string(result.Value) != "zoo" { t.Fatalf("bad entry: %#v", result) } if idx != 7 { t.Fatalf("bad index: %d", idx) } // Lock an existing key. testSetKey(t, s, 8, "bar", "bar", nil) ok, err = s.KVSLock(9, &structs.DirEntry{Key: "bar", Value: []byte("xxx"), Session: session1}) if !ok || err != nil { t.Fatalf("didn't get the lock: %v %s", ok, err) } // Make sure the indexes got set properly. idx, result, err = s.KVSGet(nil, "bar", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 1 || result.CreateIndex != 8 || result.ModifyIndex != 9 || string(result.Value) != "xxx" { t.Fatalf("bad entry: %#v", result) } if idx != 9 { t.Fatalf("bad index: %d", idx) } // Attempting a re-lock with a different session should also fail. session2 := testUUID() if err := s.SessionCreate(10, &structs.Session{ID: session2, Node: "node1"}); err != nil { t.Fatalf("err: %s", err) } // Re-locking should not return an error, but will report that it didn't // get the lock. ok, err = s.KVSLock(11, &structs.DirEntry{Key: "bar", Value: []byte("nope"), Session: session2}) if ok || err != nil { t.Fatalf("didn't handle locking an already-locked key: %v %s", ok, err) } // Make sure the indexes didn't update. idx, result, err = s.KVSGet(nil, "bar", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 1 || result.CreateIndex != 8 || result.ModifyIndex != 9 || string(result.Value) != "xxx" { t.Fatalf("bad entry: %#v", result) } if idx != 9 { t.Fatalf("bad index: %d", idx) } } func TestStateStore_KVSUnlock(t *testing.T) { s := testStateStore(t) // Unlock with no session should fail. ok, err := s.KVSUnlock(0, &structs.DirEntry{Key: "foo", Value: []byte("bar")}) if ok || err == nil || !strings.Contains(err.Error(), "missing session") { t.Fatalf("didn't detect missing session: %v %s", ok, err) } // Make a real session. testRegisterNode(t, s, 1, "node1") session1 := testUUID() if err := s.SessionCreate(2, &structs.Session{ID: session1, Node: "node1"}); err != nil { t.Fatalf("err: %s", err) } // Unlock with a real session but no key should not return an error, but // will report it didn't unlock anything. ok, err = s.KVSUnlock(3, &structs.DirEntry{Key: "foo", Value: []byte("bar"), Session: session1}) if ok || err != nil { t.Fatalf("didn't handle unlocking a missing key: %v %s", ok, err) } // Make a key and unlock it, without it being locked. testSetKey(t, s, 4, "foo", "bar", nil) ok, err = s.KVSUnlock(5, &structs.DirEntry{Key: "foo", Value: []byte("baz"), Session: session1}) if ok || err != nil { t.Fatalf("didn't handle unlocking a non-locked key: %v %s", ok, err) } // Make sure the indexes didn't update. idx, result, err := s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 0 || result.CreateIndex != 4 || result.ModifyIndex != 4 || string(result.Value) != "bar" { t.Fatalf("bad entry: %#v", result) } if idx != 4 { t.Fatalf("bad index: %d", idx) } // Lock it with the first session. ok, err = s.KVSLock(6, &structs.DirEntry{Key: "foo", Value: []byte("bar"), Session: session1}) if !ok || err != nil { t.Fatalf("didn't get the lock: %v %s", ok, err) } // Attempt an unlock with another session. session2 := testUUID() if err := s.SessionCreate(7, &structs.Session{ID: session2, Node: "node1"}); err != nil { t.Fatalf("err: %s", err) } ok, err = s.KVSUnlock(8, &structs.DirEntry{Key: "foo", Value: []byte("zoo"), Session: session2}) if ok || err != nil { t.Fatalf("didn't handle unlocking with the wrong session: %v %s", ok, err) } // Make sure the indexes didn't update. idx, result, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 1 || result.CreateIndex != 4 || result.ModifyIndex != 6 || string(result.Value) != "bar" { t.Fatalf("bad entry: %#v", result) } if idx != 6 { t.Fatalf("bad index: %d", idx) } // Now do the unlock with the correct session. ok, err = s.KVSUnlock(9, &structs.DirEntry{Key: "foo", Value: []byte("zoo"), Session: session1}) if !ok || err != nil { t.Fatalf("didn't handle unlocking with the correct session: %v %s", ok, err) } // Make sure the indexes got set properly. idx, result, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 1 || result.CreateIndex != 4 || result.ModifyIndex != 9 || string(result.Value) != "zoo" { t.Fatalf("bad entry: %#v", result) } if idx != 9 { t.Fatalf("bad index: %d", idx) } // Unlocking again should fail and not change anything. ok, err = s.KVSUnlock(10, &structs.DirEntry{Key: "foo", Value: []byte("nope"), Session: session1}) if ok || err != nil { t.Fatalf("didn't handle unlocking with the previous session: %v %s", ok, err) } // Make sure the indexes didn't update. idx, result, err = s.KVSGet(nil, "foo", nil) if err != nil { t.Fatalf("err: %s", err) } if result.LockIndex != 1 || result.CreateIndex != 4 || result.ModifyIndex != 9 || string(result.Value) != "zoo" { t.Fatalf("bad entry: %#v", result) } if idx != 9 { t.Fatalf("bad index: %d", idx) } } func TestStateStore_KVS_Snapshot_Restore(t *testing.T) { s := testStateStore(t) // Build up some entries to seed. entries := structs.DirEntries{ &structs.DirEntry{ Key: "aaa", Flags: 23, Value: []byte("hello"), }, &structs.DirEntry{ Key: "bar/a", Value: []byte("one"), }, &structs.DirEntry{ Key: "bar/b", Value: []byte("two"), }, &structs.DirEntry{ Key: "bar/c", Value: []byte("three"), }, } for i, entry := range entries { if err := s.KVSSet(uint64(i+1), entry); err != nil { t.Fatalf("err: %s", err) } } // Make a node and session so we can test a locked key. testRegisterNode(t, s, 5, "node1") session := testUUID() if err := s.SessionCreate(6, &structs.Session{ID: session, Node: "node1"}); err != nil { t.Fatalf("err: %s", err) } entries[3].Session = session if ok, err := s.KVSLock(7, entries[3]); !ok || err != nil { t.Fatalf("didn't get the lock: %v %s", ok, err) } // This is required for the compare later. entries[3].LockIndex = 1 // Snapshot the keys. snap := s.Snapshot() defer snap.Close() // Alter the real state store. if err := s.KVSSet(8, &structs.DirEntry{Key: "aaa", Value: []byte("nope")}); err != nil { t.Fatalf("err: %s", err) } // Verify the snapshot. if idx := snap.LastIndex(); idx != 6 { t.Fatalf("bad index: %d", idx) } iter, err := snap.KVs() if err != nil { t.Fatalf("err: %s", err) } var dump structs.DirEntries for entry := iter.Next(); entry != nil; entry = iter.Next() { dump = append(dump, entry.(*structs.DirEntry)) } if !reflect.DeepEqual(dump, entries) { t.Fatalf("bad: %#v", dump) } // Restore the values into a new state store. func() { s := testStateStore(t) restore := s.Restore() for _, entry := range dump { if err := restore.KVS(entry); err != nil { t.Fatalf("err: %s", err) } } restore.Commit() // Read the restored keys back out and verify they match. idx, res, err := s.KVSList(nil, "", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 7 { t.Fatalf("bad index: %d", idx) } if !reflect.DeepEqual(res, entries) { t.Fatalf("bad: %#v", res) } // Check that the index was updated. if idx := s.maxIndex(partitionedIndexEntryName(tableKVs, "default")); idx != 7 { t.Fatalf("bad index: %d", idx) } }() } func TestStateStore_Tombstone_Snapshot_Restore(t *testing.T) { s := testStateStore(t) // Insert a key and then delete it to create a tombstone. testSetKey(t, s, 1, "foo/bar", "bar", nil) testSetKey(t, s, 2, "foo/bar/baz", "bar", nil) testSetKey(t, s, 3, "foo/bar/zoo", "bar", nil) if err := s.KVSDelete(4, "foo/bar", nil); err != nil { t.Fatalf("err: %s", err) } // Snapshot the Tombstones. snap := s.Snapshot() defer snap.Close() // Alter the real state store. if err := s.ReapTombstones(5, 4); err != nil { t.Fatalf("err: %s", err) } idx, _, err := s.KVSList(nil, "foo/bar", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 3 { t.Fatalf("bad index: %d", idx) } // Verify the snapshot. stones, err := snap.Tombstones() if err != nil { t.Fatalf("err: %s", err) } var dump []*Tombstone for stone := stones.Next(); stone != nil; stone = stones.Next() { dump = append(dump, stone.(*Tombstone)) } if len(dump) != 1 { t.Fatalf("bad %#v", dump) } stone := dump[0] if stone.Key != "foo/bar" || stone.Index != 4 { t.Fatalf("bad: %#v", stone) } // Restore the values into a new state store. func() { s := testStateStore(t) restore := s.Restore() for _, stone := range dump { if err := restore.Tombstone(stone); err != nil { t.Fatalf("err: %s", err) } } restore.Commit() // See if the stone works properly in a list query. idx, _, err := s.KVSList(nil, "foo/bar", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 4 { t.Fatalf("bad index: %d", idx) } // Make sure it reaps correctly. We should still get a 4 for // the index here because it will be using the last index from // the tombstone table. if err := s.ReapTombstones(6, 4); err != nil { t.Fatalf("err: %s", err) } idx, _, err = s.KVSList(nil, "foo/bar", nil) if err != nil { t.Fatalf("err: %s", err) } if idx != 4 { t.Fatalf("bad index: %d", idx) } // But make sure the tombstone is actually gone. snap := s.Snapshot() defer snap.Close() stones, err := snap.Tombstones() if err != nil { t.Fatalf("err: %s", err) } if stones.Next() != nil { t.Fatalf("unexpected extra tombstones") } }() }