Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

storage: start initial revision at 1

Browse Source 
When the start revision was 0, there was no way to safely watch
starting from the first store revision.
This commit is contained in:
Anthony Romano 2016-01-14 15:50:42 -08:00
parent 02ab7be106
commit ecba9b61cf
3 changed files with 80 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

147
storage/kv_test.go
View File

@ -83,16 +83,9 @@ func testKVRange(t *testing.T, f rangeFunc) {
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
s.Put([]byte("foo"), []byte("bar"), 1) kvs := put3TestKVs(s)
s.Put([]byte("foo1"), []byte("bar1"), 2)
s.Put([]byte("foo2"), []byte("bar2"), 3)
kvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1, Lease: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1, Lease: 2},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1, Lease: 3},
}
wrev := int64(3) wrev := int64(4)
tests := []struct { tests := []struct {
key, end []byte key, end []byte
wkvs []storagepb.KeyValue wkvs []storagepb.KeyValue
@ -151,25 +144,18 @@ func testKVRangeRev(t *testing.T, f rangeFunc) {
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
s.Put([]byte("foo"), []byte("bar"), 1) kvs := put3TestKVs(s)
s.Put([]byte("foo1"), []byte("bar1"), 2)
s.Put([]byte("foo2"), []byte("bar2"), 3)
kvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1, Lease: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1, Lease: 2},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1, Lease: 3},
}
tests := []struct { tests := []struct {
rev int64 rev int64
wrev int64 wrev int64
wkvs []storagepb.KeyValue wkvs []storagepb.KeyValue
}{ }{
{-1, 3, kvs}, {-1, 4, kvs},
{0, 3, kvs}, {0, 4, kvs},
{1, 1, kvs[:1]}, {2, 2, kvs[:1]},
{2, 2, kvs[:2]}, {3, 3, kvs[:2]},
{3, 3, kvs}, {4, 4, kvs},
} }
for i, tt := range tests { for i, tt := range tests {
@ -194,10 +180,8 @@ func testKVRangeBadRev(t *testing.T, f rangeFunc) {
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
s.Put([]byte("foo"), []byte("bar"), lease.NoLease) put3TestKVs(s)
s.Put([]byte("foo1"), []byte("bar1"), lease.NoLease) if err := s.Compact(4); err != nil {
s.Put([]byte("foo2"), []byte("bar2"), lease.NoLease)
if err := s.Compact(3); err != nil {
t.Fatalf("compact error (%v)", err) t.Fatalf("compact error (%v)", err)
} }
@ -206,9 +190,9 @@ func testKVRangeBadRev(t *testing.T, f rangeFunc) {
werr error werr error
}{ }{
{-1, ErrCompacted}, {-1, ErrCompacted},
{1, ErrCompacted},
{2, ErrCompacted}, {2, ErrCompacted},
{3, ErrCompacted}, {5, ErrFutureRev},
{4, ErrFutureRev},
{100, ErrFutureRev}, {100, ErrFutureRev},
} }
for i, tt := range tests { for i, tt := range tests {
@ -227,16 +211,9 @@ func testKVRangeLimit(t *testing.T, f rangeFunc) {
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
s.Put([]byte("foo"), []byte("bar"), 1) kvs := put3TestKVs(s)
s.Put([]byte("foo1"), []byte("bar1"), 2)
s.Put([]byte("foo2"), []byte("bar2"), 3)
kvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1, Lease: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1, Lease: 2},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1, Lease: 3},
}
wrev := int64(3) wrev := int64(4)
tests := []struct { tests := []struct {
limit int64 limit int64
wkvs []storagepb.KeyValue wkvs []storagepb.KeyValue
@ -276,8 +253,8 @@ func testKVPutMultipleTimes(t *testing.T, f putFunc) {
base := int64(i + 1) base := int64(i + 1)
rev := f(s, []byte("foo"), []byte("bar"), lease.LeaseID(base)) rev := f(s, []byte("foo"), []byte("bar"), lease.LeaseID(base))
if rev != base { if rev != base+1 {
t.Errorf("#%d: rev = %d, want %d", i, rev, base) t.Errorf("#%d: rev = %d, want %d", i, rev, base+1)
} }
kvs, _, err := s.Range([]byte("foo"), nil, 0, 0) kvs, _, err := s.Range([]byte("foo"), nil, 0, 0)
@ -285,7 +262,7 @@ func testKVPutMultipleTimes(t *testing.T, f putFunc) {
t.Fatal(err) t.Fatal(err)
} }
wkvs := []storagepb.KeyValue{ wkvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: base, Version: base, Lease: base}, {Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: base + 1, Version: base, Lease: base},
} }
if !reflect.DeepEqual(kvs, wkvs) { if !reflect.DeepEqual(kvs, wkvs) {
t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs) t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs)
@ -305,27 +282,27 @@ func testKVDeleteRange(t *testing.T, f deleteRangeFunc) {
}{ }{
{ {
[]byte("foo"), nil, []byte("foo"), nil,
4, 1, 5, 1,
}, },
{ {
[]byte("foo"), []byte("foo1"), []byte("foo"), []byte("foo1"),
4, 1, 5, 1,
}, },
{ {
[]byte("foo"), []byte("foo2"), []byte("foo"), []byte("foo2"),
4, 2, 5, 2,
}, },
{ {
[]byte("foo"), []byte("foo3"), []byte("foo"), []byte("foo3"),
4, 3, 5, 3,
}, },
{ {
[]byte("foo3"), []byte("foo8"), []byte("foo3"), []byte("foo8"),
3, 0, 4, 0,
}, },
{ {
[]byte("foo3"), nil, []byte("foo3"), nil,
3, 0, 4, 0,
}, },
} }
@ -357,14 +334,14 @@ func testKVDeleteMultipleTimes(t *testing.T, f deleteRangeFunc) {
s.Put([]byte("foo"), []byte("bar"), lease.NoLease) s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
n, rev := f(s, []byte("foo"), nil) n, rev := f(s, []byte("foo"), nil)
if n != 1 || rev != 2 { if n != 1 || rev != 3 {
t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 2) t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 3)
} }
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
n, rev := f(s, []byte("foo"), nil) n, rev := f(s, []byte("foo"), nil)
if n != 0 || rev != 2 { if n != 0 || rev != 3 {
t.Fatalf("#%d: n = %d, rev = %d, want (%d, %d)", i, n, rev, 0, 2) t.Fatalf("#%d: n = %d, rev = %d, want (%d, %d)", i, n, rev, 0, 3)
} }
} }
} }
@ -376,7 +353,7 @@ func TestKVOperationInSequence(t *testing.T) {
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
base := int64(i * 2) base := int64(i*2 + 1)
// put foo // put foo
rev := s.Put([]byte("foo"), []byte("bar"), lease.NoLease) rev := s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
@ -417,7 +394,7 @@ func TestKVOperationInSequence(t *testing.T) {
} }
} }
func TestKVTxnBlockNonTnxOperations(t *testing.T) { func TestKVTxnBlockNonTxnOperations(t *testing.T) {
b, tmpPath := backend.NewDefaultTmpBackend() b, tmpPath := backend.NewDefaultTmpBackend()
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
@ -488,14 +465,14 @@ func TestKVTxnWrongID(t *testing.T) {
} }
// test that txn range, put, delete on single key in sequence repeatedly works correctly. // test that txn range, put, delete on single key in sequence repeatedly works correctly.
func TestKVTnxOperationInSequence(t *testing.T) { func TestKVTxnOperationInSequence(t *testing.T) {
b, tmpPath := backend.NewDefaultTmpBackend() b, tmpPath := backend.NewDefaultTmpBackend()
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
id := s.TxnBegin() id := s.TxnBegin()
base := int64(i) base := int64(i + 1)
// put foo // put foo
rev, err := s.TxnPut(id, []byte("foo"), []byte("bar"), lease.NoLease) rev, err := s.TxnPut(id, []byte("foo"), []byte("bar"), lease.NoLease)
@ -560,26 +537,26 @@ func TestKVCompactReserveLastValue(t *testing.T) {
// wanted kvs right after the compacted rev // wanted kvs right after the compacted rev
wkvs []storagepb.KeyValue wkvs []storagepb.KeyValue
}{ }{
{
0,
[]storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar0"), CreateRevision: 1, ModRevision: 1, Version: 1, Lease: 1},
},
},
{ {
1, 1,
[]storagepb.KeyValue{ []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar1"), CreateRevision: 1, ModRevision: 2, Version: 2, Lease: 2}, {Key: []byte("foo"), Value: []byte("bar0"), CreateRevision: 2, ModRevision: 2, Version: 1, Lease: 1},
}, },
}, },
{ {
2, 2,
nil, []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 3, Version: 2, Lease: 2},
},
}, },
{ {
3, 3,
nil,
},
{
4,
[]storagepb.KeyValue{ []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar2"), CreateRevision: 4, ModRevision: 4, Version: 1, Lease: 3}, {Key: []byte("foo"), Value: []byte("bar2"), CreateRevision: 5, ModRevision: 5, Version: 1, Lease: 3},
}, },
}, },
} }
@ -615,8 +592,8 @@ func TestKVCompactBad(t *testing.T) {
{0, nil}, {0, nil},
{1, nil}, {1, nil},
{1, ErrCompacted}, {1, ErrCompacted},
{3, nil}, {4, nil},
{4, ErrFutureRev}, {5, ErrFutureRev},
{100, ErrFutureRev}, {100, ErrFutureRev},
} }
for i, tt := range tests { for i, tt := range tests {
@ -701,14 +678,7 @@ func TestKVSnapshot(t *testing.T) {
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
s.Put([]byte("foo"), []byte("bar"), 1) wkvs := put3TestKVs(s)
s.Put([]byte("foo1"), []byte("bar1"), 2)
s.Put([]byte("foo2"), []byte("bar2"), 3)
wkvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1, Lease: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1, Lease: 2},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1, Lease: 3},
}
newPath := "new_test" newPath := "new_test"
f, err := os.Create(newPath) f, err := os.Create(newPath)
@ -734,8 +704,8 @@ func TestKVSnapshot(t *testing.T) {
if !reflect.DeepEqual(kvs, wkvs) { if !reflect.DeepEqual(kvs, wkvs) {
t.Errorf("kvs = %+v, want %+v", kvs, wkvs) t.Errorf("kvs = %+v, want %+v", kvs, wkvs)
} }
if rev != 3 { if rev != 4 {
t.Errorf("rev = %d, want %d", rev, 3) t.Errorf("rev = %d, want %d", rev, 4)
} }
} }
@ -757,8 +727,8 @@ func TestWatchableKVWatch(t *testing.T) {
Kv: &storagepb.KeyValue{ Kv: &storagepb.KeyValue{
Key: []byte("foo"), Key: []byte("foo"),
Value: []byte("bar"), Value: []byte("bar"),
CreateRevision: 1, CreateRevision: 2,
ModRevision: 1, ModRevision: 2,
Version: 1, Version: 1,
Lease: 1, Lease: 1,
}, },
@ -783,8 +753,8 @@ func TestWatchableKVWatch(t *testing.T) {
Kv: &storagepb.KeyValue{ Kv: &storagepb.KeyValue{
Key: []byte("foo1"), Key: []byte("foo1"),
Value: []byte("bar1"), Value: []byte("bar1"),
CreateRevision: 2, CreateRevision: 3,
ModRevision: 2, ModRevision: 3,
Version: 1, Version: 1,
Lease: 2, Lease: 2,
}, },
@ -810,8 +780,8 @@ func TestWatchableKVWatch(t *testing.T) {
Kv: &storagepb.KeyValue{ Kv: &storagepb.KeyValue{
Key: []byte("foo1"), Key: []byte("foo1"),
Value: []byte("bar1"), Value: []byte("bar1"),
CreateRevision: 2, CreateRevision: 3,
ModRevision: 2, ModRevision: 3,
Version: 1, Version: 1,
Lease: 2, Lease: 2,
}, },
@ -835,8 +805,8 @@ func TestWatchableKVWatch(t *testing.T) {
Kv: &storagepb.KeyValue{ Kv: &storagepb.KeyValue{
Key: []byte("foo1"), Key: []byte("foo1"),
Value: []byte("bar11"), Value: []byte("bar11"),
CreateRevision: 2, CreateRevision: 3,
ModRevision: 3, ModRevision: 4,
Version: 2, Version: 2,
Lease: 3, Lease: 3,
}, },
@ -858,3 +828,14 @@ func cleanup(s KV, b backend.Backend, path string) {
b.Close() b.Close()
os.Remove(path) os.Remove(path)
} }
func put3TestKVs(s KV) []storagepb.KeyValue {
s.Put([]byte("foo"), []byte("bar"), 1)
s.Put([]byte("foo1"), []byte("bar1"), 2)
s.Put([]byte("foo2"), []byte("bar2"), 3)
return []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1, Lease: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 3, ModRevision: 3, Version: 1, Lease: 2},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 4, ModRevision: 4, Version: 1, Lease: 3},
}
}

6
storage/kvstore.go
View File

@ -75,7 +75,7 @@ func NewStore(b backend.Backend, le lease.Lessor) *store {
le: le, le: le,
currentRev: revision{}, currentRev: revision{main: 1},
compactMainRev: -1, compactMainRev: -1,
stopc: make(chan struct{}), stopc: make(chan struct{}),
} }
@ -256,7 +256,7 @@ func (s *store) Restore(b backend.Backend) error {
s.b = b s.b = b
s.kvindex = newTreeIndex() s.kvindex = newTreeIndex()
s.currentRev = revision{} s.currentRev = revision{main: 1}
s.compactMainRev = -1 s.compactMainRev = -1
s.tx = b.BatchTx() s.tx = b.BatchTx()
s.txnID = -1 s.txnID = -1
@ -267,7 +267,7 @@ func (s *store) Restore(b backend.Backend) error {
func (s *store) restore() error { func (s *store) restore() error {
min, max := newRevBytes(), newRevBytes() min, max := newRevBytes(), newRevBytes()
revToBytes(revision{}, min) revToBytes(revision{main: 1}, min)
revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max) revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max)
// restore index // restore index

26
storage/kvstore_test.go
View File

@ -34,7 +34,7 @@ func TestStoreRev(t *testing.T) {
s := NewStore(b, &lease.FakeLessor{}) s := NewStore(b, &lease.FakeLessor{})
defer os.Remove(tmpPath) defer os.Remove(tmpPath)
for i := 0; i < 3; i++ { for i := 1; i <= 3; i++ {
s.Put([]byte("foo"), []byte("bar"), lease.NoLease) s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
if r := s.Rev(); r != int64(i+1) { if r := s.Rev(); r != int64(i+1) {
t.Errorf("#%d: rev = %d, want %d", i, r, i+1) t.Errorf("#%d: rev = %d, want %d", i, r, i+1)
@ -312,15 +312,15 @@ func TestStoreRestore(t *testing.T) {
putkv := storagepb.KeyValue{ putkv := storagepb.KeyValue{
Key: []byte("foo"), Key: []byte("foo"),
Value: []byte("bar"), Value: []byte("bar"),
CreateRevision: 3, CreateRevision: 4,
ModRevision: 3, ModRevision: 4,
Version: 1, Version: 1,
} }
putkvb, err := putkv.Marshal() putkvb, err := putkv.Marshal()
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
delkey := newTestKeyBytes(revision{4, 0}, true) delkey := newTestKeyBytes(revision{5, 0}, true)
delkv := storagepb.KeyValue{ delkv := storagepb.KeyValue{
Key: []byte("foo"), Key: []byte("foo"),
} }
@ -328,30 +328,30 @@ func TestStoreRestore(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{2, 0})}} b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}} b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}}
b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{2, 0})}} b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
s.restore() s.restore()
if s.compactMainRev != 2 { if s.compactMainRev != 3 {
t.Errorf("compact rev = %d, want 4", s.compactMainRev) t.Errorf("compact rev = %d, want 5", s.compactMainRev)
} }
wrev := revision{4, 0} wrev := revision{5, 0}
if !reflect.DeepEqual(s.currentRev, wrev) { if !reflect.DeepEqual(s.currentRev, wrev) {
t.Errorf("current rev = %v, want %v", s.currentRev, wrev) t.Errorf("current rev = %v, want %v", s.currentRev, wrev)
} }
wact := []testutil.Action{ wact := []testutil.Action{
{"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}}, {"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
{"range", []interface{}{keyBucketName, newTestRevBytes(revision{}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(0)}}, {"range", []interface{}{keyBucketName, newTestRevBytes(revision{1, 0}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(0)}},
{"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}}, {"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
} }
if g := b.tx.Action(); !reflect.DeepEqual(g, wact) { if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
t.Errorf("tx actions = %+v, want %+v", g, wact) t.Errorf("tx actions = %+v, want %+v", g, wact)
} }
wact = []testutil.Action{ wact = []testutil.Action{
{"restore", []interface{}{[]byte("foo"), revision{3, 0}, revision{3, 0}, int64(1)}}, {"restore", []interface{}{[]byte("foo"), revision{4, 0}, revision{3, 0}, int64(1)}},
{"tombstone", []interface{}{[]byte("foo"), revision{4, 0}}}, {"tombstone", []interface{}{[]byte("foo"), revision{5, 0}}},
} }
if g := fi.Action(); !reflect.DeepEqual(g, wact) { if g := fi.Action(); !reflect.DeepEqual(g, wact) {
t.Errorf("index action = %+v, want %+v", g, wact) t.Errorf("index action = %+v, want %+v", g, wact)
@ -420,7 +420,7 @@ func TestTxnPut(t *testing.T) {
for i := 0; i < sliceN; i++ { for i := 0; i < sliceN; i++ {
id := s.TxnBegin() id := s.TxnBegin()
base := int64(i + 1) base := int64(i + 2)
rev, err := s.TxnPut(id, keys[i], vals[i], lease.NoLease) rev, err := s.TxnPut(id, keys[i], vals[i], lease.NoLease)
if err != nil { if err != nil {