consul/agent/consul/state/kvs_test.go

1458 lines
38 KiB
Go

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")
}
}()
}