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

Merge pull request #3434 from xiang90/index_revision

Browse Source 
*: v3api index->revision
This commit is contained in:
Xiang Li 2015-09-04 10:48:59 -07:00
commit 778f8d8fea
10 changed files with 183 additions and 183 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
Documentation/rfc/v3api.md
View File

@ -42,7 +42,7 @@ Put( PutRequest { key = foo, value = bar } )
PutResponse {
cluster_id = 0x1000,
member_id = 0x1,
index = 1,
revision = 1,
raft_term = 0x1,
}
```
@ -54,14 +54,14 @@ Get ( RangeRequest { key = foo } )
RangeResponse {
cluster_id = 0x1000,
member_id = 0x1,
index = 1,
revision = 1,
raft_term = 0x1,
kvs = {
{
key = foo,
value = bar,
create_index = 1,
mod_index = 1,
create_revision = 1,
mod_revision = 1,
version = 1;
},
},
@ -75,22 +75,22 @@ Range ( RangeRequest { key = foo, end_key = foo80, limit = 30 } )
RangeResponse {
cluster_id = 0x1000,
member_id = 0x1,
index = 100,
revision = 100,
raft_term = 0x1,
kvs = {
{
key = foo0,
value = bar0,
create_index = 1,
mod_index = 1,
create_revision = 1,
mod_revision = 1,
version = 1;
},
...,
{
key = foo30,
value = bar30,
create_index = 30,
mod_index = 30,
create_revision = 30,
mod_revision = 30,
version = 1;
},
},
@ -100,10 +100,10 @@ RangeResponse {
#### Finish a txn (assume we have foo0=bar0, foo1=bar1)
```
Txn(TxnRequest {
// mod_index of foo0 is equal to 1, mod_index of foo1 is greater than 1
// mod_revision of foo0 is equal to 1, mod_revision of foo1 is greater than 1
compare = {
{compareType = equal, key = foo0, mod_index = 1},
{compareType = greater, key = foo1, mod_index = 1}}
{compareType = equal, key = foo0, mod_revision = 1},
{compareType = greater, key = foo1, mod_revision = 1}}
},
// if the comparison succeeds, put foo2 = bar2
success = {PutRequest { key = foo2, value = success }},
@ -114,7 +114,7 @@ Txn(TxnRequest {
TxnResponse {
cluster_id = 0x1000,
member_id = 0x1,
index = 3,
revision = 3,
raft_term = 0x1,
succeeded = true,
responses = {
@ -122,7 +122,7 @@ TxnResponse {
{
cluster_id = 0x1000,
member_id = 0x1,
index = 3,
revision = 3,
raft_term = 0x1,
}
}
@ -135,8 +135,8 @@ TxnResponse {
Watch( WatchRequest{
key = foo,
end_key = fop, // prefix foo
start_index = 20,
end_index = 10000,
start_revision = 20,
end_revision = 10000,
// server decided notification frequency
progress_notification = true,
}
@ -147,14 +147,14 @@ Watch( WatchRequest{
WatchResponse {
cluster_id = 0x1000,
member_id = 0x1,
index = 3,
revision = 3,
raft_term = 0x1,
event_type = put,
kv = {
key = foo0,
value = bar0,
create_index = 1,
mod_index = 1,
create_revision = 1,
mod_revision = 1,
version = 1;
},
}
@ -164,7 +164,7 @@ WatchResponse {
WatchResponse {
cluster_id = 0x1000,
member_id = 0x1,
index = 2000,
revision = 2000,
raft_term = 0x1,
// nil event as notification
}
@ -175,14 +175,14 @@ WatchResponse {
WatchResponse {
cluster_id = 0x1000,
member_id = 0x1,
index = 3000,
revision = 3000,
raft_term = 0x1,
event_type = put,
kv = {
key = foo0,
value = bar3000,
create_index = 1,
mod_index = 3000,
create_revision = 1,
mod_revision = 3000,
version = 2;
},
}

42
Documentation/rfc/v3api.proto
View File

@ -6,18 +6,18 @@ service etcd {
rpc Range(RangeRequest) returns (RangeResponse) {}
// Put puts the given key into the store.
// A put request increases the index of the store,
// A put request increases the revision of the store,
// and generates one event in the event history.
rpc Put(PutRequest) returns (PutResponse) {}
// Delete deletes the given range from the store.
// A delete request increase the index of the store,
// A delete request increase the revision of the store,
// and generates one event in the event history.
rpc DeleteRange(DeleteRangeRequest) returns (DeleteRangeResponse) {}
// Txn processes all the requests in one transaction.
// A txn request increases the index of the store,
// and generates events with the same index in the event history.
// A txn request increases the revision of the store,
// and generates events with the same revision in the event history.
rpc Txn(TxnRequest) returns (TxnResponse) {}
// Watch watches the events happening or happened in etcd. Both input and output
@ -55,8 +55,8 @@ message ResponseHeader {
string error = 1;
uint64 cluster_id = 2;
uint64 member_id = 3;
// index of the store when the request was applied.
int64 index = 4;
// revision of the store when the request was applied.
int64 revision = 4;
// term of raft when the request was applied.
uint64 raft_term = 5;
}
@ -137,10 +137,10 @@ message Compare {
oneof target_union {
// version of the given key
int64 version = 4;
// create index of the given key
int64 create_index = 5;
// last modified index of the given key
int64 mod_index = 6;
// create revision of the given key
int64 create_revision = 5;
// last modified revision of the given key
int64 mod_revision = 6;
// value of the given key
bytes value = 7;
}
@ -180,9 +180,9 @@ message TxnResponse {
message KeyValue {
bytes key = 1;
int64 create_index = 2;
// mod_index is the last modified index of the key.
int64 mod_index = 3;
int64 create_revision = 2;
// mod_revision is the last modified revision of the key.
int64 mod_revision = 3;
// version is the version of the key. A deletion resets
// the version to zero and any modification of the key
// increases its version.
@ -195,10 +195,10 @@ message WatchRangeRequest {
bytes key = 1;
// if the range_end is given, it gets the keys in range [key, range_end).
bytes range_end = 2;
// start_index is an optional index (including) to watch from. No start_index is "now".
int64 start_index = 3;
// end_index is an optional index (excluding) to end watch. No end_index is "forever".
int64 end_index = 4;
// start_revision is an optional revision (including) to watch from. No start_revision is "now".
int64 start_revision = 3;
// end_revision is an optional revision (excluding) to end watch. No end_revision is "forever".
int64 end_revision = 4;
bool progress_notification = 5;
}
@ -220,12 +220,12 @@ message Event {
KeyValue kv = 2;
}
// Compaction compacts the kv store upto the given index (including).
// Compaction compacts the kv store upto the given revision (including).
// It removes the old versions of a key. It keeps the newest version of
// the key even if its latest modification index is smaller than the given
// index.
// the key even if its latest modification revision is smaller than the given
// revision.
message CompactionRequest {
int64 index = 1;
int64 revision = 1;
}
message CompactionResponse {

92
etcdserver/etcdserverpb/rpc.pb.go
View File

@ -74,8 +74,8 @@ type ResponseHeader struct {
Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
ClusterId uint64 `protobuf:"varint,2,opt,name=cluster_id,proto3" json:"cluster_id,omitempty"`
MemberId uint64 `protobuf:"varint,3,opt,name=member_id,proto3" json:"member_id,omitempty"`
// index of the store when the request was applied.
Index int64 `protobuf:"varint,4,opt,name=index,proto3" json:"index,omitempty"`
// revision of the store when the request was applied.
Revision int64 `protobuf:"varint,4,opt,name=revision,proto3" json:"revision,omitempty"`
// term of raft when the request was applied.
RaftTerm uint64 `protobuf:"varint,5,opt,name=raft_term,proto3" json:"raft_term,omitempty"`
}
@ -246,10 +246,10 @@ type Compare struct {
Key []byte `protobuf:"bytes,3,opt,name=key,proto3" json:"key,omitempty"`
// version of the given key
Version int64 `protobuf:"varint,4,opt,name=version,proto3" json:"version,omitempty"`
// create index of the given key
CreateIndex int64 `protobuf:"varint,5,opt,name=create_index,proto3" json:"create_index,omitempty"`
// last modified index of the given key
ModIndex int64 `protobuf:"varint,6,opt,name=mod_index,proto3" json:"mod_index,omitempty"`
// create revision of the given key
CreateRevision int64 `protobuf:"varint,5,opt,name=create_revision,proto3" json:"create_revision,omitempty"`
// last modified revision of the given key
ModRevision int64 `protobuf:"varint,6,opt,name=mod_revision,proto3" json:"mod_revision,omitempty"`
// value of the given key
Value []byte `protobuf:"bytes,7,opt,name=value,proto3" json:"value,omitempty"`
}
@ -328,12 +328,12 @@ func (m *TxnResponse) GetResponses() []*ResponseUnion {
return nil
}
// Compaction compacts the kv store upto the given index (including).
// Compaction compacts the kv store upto the given revision (including).
// It removes the old versions of a key. It keeps the newest version of
// the key even if its latest modification index is smaller than the given
// index.
// the key even if its latest modification revision is smaller than the given
// revision.
type CompactionRequest struct {
Index int64 `protobuf:"varint,1,opt,name=index,proto3" json:"index,omitempty"`
Revision int64 `protobuf:"varint,1,opt,name=revision,proto3" json:"revision,omitempty"`
}
func (m *CompactionRequest) Reset() { *m = CompactionRequest{} }
@ -370,16 +370,16 @@ type EtcdClient interface {
// Range gets the keys in the range from the store.
Range(ctx context.Context, in *RangeRequest, opts ...grpc.CallOption) (*RangeResponse, error)
// Put puts the given key into the store.
// A put request increases the index of the store,
// A put request increases the revision of the store,
// and generates one event in the event history.
Put(ctx context.Context, in *PutRequest, opts ...grpc.CallOption) (*PutResponse, error)
// Delete deletes the given range from the store.
// A delete request increase the index of the store,
// A delete request increase the revision of the store,
// and generates one event in the event history.
DeleteRange(ctx context.Context, in *DeleteRangeRequest, opts ...grpc.CallOption) (*DeleteRangeResponse, error)
// Txn processes all the requests in one transaction.
// A txn request increases the index of the store,
// and generates events with the same index in the event history.
// A txn request increases the revision of the store,
// and generates events with the same revision in the event history.
Txn(ctx context.Context, in *TxnRequest, opts ...grpc.CallOption) (*TxnResponse, error)
// Compact compacts the event history in etcd. User should compact the
// event history periodically, or it will grow infinitely.
@ -445,16 +445,16 @@ type EtcdServer interface {
// Range gets the keys in the range from the store.
Range(context.Context, *RangeRequest) (*RangeResponse, error)
// Put puts the given key into the store.
// A put request increases the index of the store,
// A put request increases the revision of the store,
// and generates one event in the event history.
Put(context.Context, *PutRequest) (*PutResponse, error)
// Delete deletes the given range from the store.
// A delete request increase the index of the store,
// A delete request increase the revision of the store,
// and generates one event in the event history.
DeleteRange(context.Context, *DeleteRangeRequest) (*DeleteRangeResponse, error)
// Txn processes all the requests in one transaction.
// A txn request increases the index of the store,
// and generates events with the same index in the event history.
// A txn request increases the revision of the store,
// and generates events with the same revision in the event history.
Txn(context.Context, *TxnRequest) (*TxnResponse, error)
// Compact compacts the event history in etcd. User should compact the
// event history periodically, or it will grow infinitely.
@ -584,10 +584,10 @@ func (m *ResponseHeader) MarshalTo(data []byte) (int, error) {
i++
i = encodeVarintRpc(data, i, uint64(m.MemberId))
}
if m.Index != 0 {
if m.Revision != 0 {
data[i] = 0x20
i++
i = encodeVarintRpc(data, i, uint64(m.Index))
i = encodeVarintRpc(data, i, uint64(m.Revision))
}
if m.RaftTerm != 0 {
data[i] = 0x28
@ -949,15 +949,15 @@ func (m *Compare) MarshalTo(data []byte) (int, error) {
i++
i = encodeVarintRpc(data, i, uint64(m.Version))
}
if m.CreateIndex != 0 {
if m.CreateRevision != 0 {
data[i] = 0x28
i++
i = encodeVarintRpc(data, i, uint64(m.CreateIndex))
i = encodeVarintRpc(data, i, uint64(m.CreateRevision))
}
if m.ModIndex != 0 {
if m.ModRevision != 0 {
data[i] = 0x30
i++
i = encodeVarintRpc(data, i, uint64(m.ModIndex))
i = encodeVarintRpc(data, i, uint64(m.ModRevision))
}
if m.Value != nil {
if len(m.Value) > 0 {
@ -1089,10 +1089,10 @@ func (m *CompactionRequest) MarshalTo(data []byte) (int, error) {
_ = i
var l int
_ = l
if m.Index != 0 {
if m.Revision != 0 {
data[i] = 0x8
i++
i = encodeVarintRpc(data, i, uint64(m.Index))
i = encodeVarintRpc(data, i, uint64(m.Revision))
}
return i, nil
}
@ -1165,8 +1165,8 @@ func (m *ResponseHeader) Size() (n int) {
if m.MemberId != 0 {
n += 1 + sovRpc(uint64(m.MemberId))
}
if m.Index != 0 {
n += 1 + sovRpc(uint64(m.Index))
if m.Revision != 0 {
n += 1 + sovRpc(uint64(m.Revision))
}
if m.RaftTerm != 0 {
n += 1 + sovRpc(uint64(m.RaftTerm))
@ -1327,11 +1327,11 @@ func (m *Compare) Size() (n int) {
if m.Version != 0 {
n += 1 + sovRpc(uint64(m.Version))
}
if m.CreateIndex != 0 {
n += 1 + sovRpc(uint64(m.CreateIndex))
if m.CreateRevision != 0 {
n += 1 + sovRpc(uint64(m.CreateRevision))
}
if m.ModIndex != 0 {
n += 1 + sovRpc(uint64(m.ModIndex))
if m.ModRevision != 0 {
n += 1 + sovRpc(uint64(m.ModRevision))
}
if m.Value != nil {
l = len(m.Value)
@ -1388,8 +1388,8 @@ func (m *TxnResponse) Size() (n int) {
func (m *CompactionRequest) Size() (n int) {
var l int
_ = l
if m.Index != 0 {
n += 1 + sovRpc(uint64(m.Index))
if m.Revision != 0 {
n += 1 + sovRpc(uint64(m.Revision))
}
return n
}
@ -1496,16 +1496,16 @@ func (m *ResponseHeader) Unmarshal(data []byte) error {
}
case 4:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Index", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field Revision", wireType)
}
m.Index = 0
m.Revision = 0
for shift := uint(0); ; shift += 7 {
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := data[iNdEx]
iNdEx++
m.Index |= (int64(b) & 0x7F) << shift
m.Revision |= (int64(b) & 0x7F) << shift
if b < 0x80 {
break
}
@ -2503,32 +2503,32 @@ func (m *Compare) Unmarshal(data []byte) error {
}
case 5:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field CreateIndex", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field CreateRevision", wireType)
}
m.CreateIndex = 0
m.CreateRevision = 0
for shift := uint(0); ; shift += 7 {
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := data[iNdEx]
iNdEx++
m.CreateIndex |= (int64(b) & 0x7F) << shift
m.CreateRevision |= (int64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
case 6:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field ModIndex", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field ModRevision", wireType)
}
m.ModIndex = 0
m.ModRevision = 0
for shift := uint(0); ; shift += 7 {
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := data[iNdEx]
iNdEx++
m.ModIndex |= (int64(b) & 0x7F) << shift
m.ModRevision |= (int64(b) & 0x7F) << shift
if b < 0x80 {
break
}
@ -2854,16 +2854,16 @@ func (m *CompactionRequest) Unmarshal(data []byte) error {
switch fieldNum {
case 1:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Index", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field Revision", wireType)
}
m.Index = 0
m.Revision = 0
for shift := uint(0); ; shift += 7 {
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := data[iNdEx]
iNdEx++
m.Index |= (int64(b) & 0x7F) << shift
m.Revision |= (int64(b) & 0x7F) << shift
if b < 0x80 {
break
}

28
etcdserver/etcdserverpb/rpc.proto
View File

@ -13,18 +13,18 @@ service etcd {
rpc Range(RangeRequest) returns (RangeResponse) {}
// Put puts the given key into the store.
// A put request increases the index of the store,
// A put request increases the revision of the store,
// and generates one event in the event history.
rpc Put(PutRequest) returns (PutResponse) {}
// Delete deletes the given range from the store.
// A delete request increase the index of the store,
// A delete request increase the revision of the store,
// and generates one event in the event history.
rpc DeleteRange(DeleteRangeRequest) returns (DeleteRangeResponse) {}
// Txn processes all the requests in one transaction.
// A txn request increases the index of the store,
// and generates events with the same index in the event history.
// A txn request increases the revision of the store,
// and generates events with the same revision in the event history.
rpc Txn(TxnRequest) returns (TxnResponse) {}
// Compact compacts the event history in etcd. User should compact the
@ -37,8 +37,8 @@ message ResponseHeader {
string error = 1;
uint64 cluster_id = 2;
uint64 member_id = 3;
// index of the store when the request was applied.
int64 index = 4;
// revision of the store when the request was applied.
int64 revision = 4;
// term of raft when the request was applied.
uint64 raft_term = 5;
}
@ -119,10 +119,10 @@ message Compare {
oneof target_union {
// version of the given key
int64 version = 4;
// create index of the given key
int64 create_index = 5;
// last modified index of the given key
int64 mod_index = 6;
// create revision of the given key
int64 create_revision = 5;
// last modified revision of the given key
int64 mod_revision = 6;
// value of the given key
bytes value = 7;
}
@ -160,12 +160,12 @@ message TxnResponse {
repeated ResponseUnion responses = 3;
}
// Compaction compacts the kv store upto the given index (including).
// Compaction compacts the kv store upto the given revision (including).
// It removes the old versions of a key. It keeps the newest version of
// the key even if its latest modification index is smaller than the given
// index.
// the key even if its latest modification revision is smaller than the given
// revision.
message CompactionRequest {
int64 index = 1;
int64 revision = 1;
}
message CompactionResponse {

18
etcdserver/v3demo_server.go
View File

@ -36,7 +36,7 @@ func (s *EtcdServer) V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) pro
case r.DeleteRange != nil:
return doDeleteRange(s.kv, r.DeleteRange)
case r.Txn != nil:
var index int64
var revision int64
rt := r.Txn
ok := true
@ -46,7 +46,7 @@ func (s *EtcdServer) V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) pro
ok = false
break
}
index = rev
revision = rev
kv := kvs[0]
// -1 is less, 0 is equal, 1 is greater
@ -55,9 +55,9 @@ func (s *EtcdServer) V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) pro
case pb.Compare_VALUE:
result = bytes.Compare(kv.Value, c.Value)
case pb.Compare_CREATE:
result = compareInt64(kv.CreateIndex, c.CreateIndex)
result = compareInt64(kv.CreateRevision, c.CreateRevision)
case pb.Compare_MOD:
result = compareInt64(kv.ModIndex, c.ModIndex)
result = compareInt64(kv.ModRevision, c.ModRevision)
case pb.Compare_VERSION:
result = compareInt64(kv.Version, c.Version)
}
@ -93,12 +93,12 @@ func (s *EtcdServer) V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) pro
resps[i] = doUnion(s.kv, reqs[i])
}
if len(resps) != 0 {
index += 1
revision += 1
}
txnResp := &pb.TxnResponse{}
txnResp.Header = &pb.ResponseHeader{}
txnResp.Header.Index = index
txnResp.Header.Revision = revision
txnResp.Responses = resps
txnResp.Succeeded = ok
return txnResp
@ -122,7 +122,7 @@ func doPut(kv dstorage.KV, p *pb.PutRequest) *pb.PutResponse {
resp := &pb.PutResponse{}
resp.Header = &pb.ResponseHeader{}
rev := kv.Put(p.Key, p.Value)
resp.Header.Index = rev
resp.Header.Revision = rev
return resp
}
@ -134,7 +134,7 @@ func doRange(kv dstorage.KV, r *pb.RangeRequest) *pb.RangeResponse {
panic("not handled error")
}
resp.Header.Index = rev
resp.Header.Revision = rev
for i := range kvs {
resp.Kvs = append(resp.Kvs, &kvs[i])
}
@ -145,7 +145,7 @@ func doDeleteRange(kv dstorage.KV, dr *pb.DeleteRangeRequest) *pb.DeleteRangeRes
resp := &pb.DeleteRangeResponse{}
resp.Header = &pb.ResponseHeader{}
_, rev := kv.DeleteRange(dr.Key, dr.RangeEnd)
resp.Header.Index = rev
resp.Header.Revision = rev
return resp
}

36
storage/kv_test.go
View File

@ -82,9 +82,9 @@ func testKVRange(t *testing.T, f rangeFunc) {
s.Put([]byte("foo1"), []byte("bar1"))
s.Put([]byte("foo2"), []byte("bar2"))
kvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
}
wrev := int64(3)
@ -149,9 +149,9 @@ func testKVRangeRev(t *testing.T, f rangeFunc) {
s.Put([]byte("foo1"), []byte("bar1"))
s.Put([]byte("foo2"), []byte("bar2"))
kvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
}
tests := []struct {
@ -223,9 +223,9 @@ func testKVRangeLimit(t *testing.T, f rangeFunc) {
s.Put([]byte("foo1"), []byte("bar1"))
s.Put([]byte("foo2"), []byte("bar2"))
kvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
}
wrev := int64(3)
@ -276,7 +276,7 @@ func testKVPutMultipleTimes(t *testing.T, f putFunc) {
t.Fatal(err)
}
wkvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: base, Version: base},
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: base, Version: base},
}
if !reflect.DeepEqual(kvs, wkvs) {
t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs)
@ -377,7 +377,7 @@ func TestKVOperationInSequence(t *testing.T) {
t.Fatal(err)
}
wkvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: base + 1, ModIndex: base + 1, Version: 1},
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: base + 1, ModRevision: base + 1, Version: 1},
}
if !reflect.DeepEqual(kvs, wkvs) {
t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs)
@ -494,7 +494,7 @@ func TestKVTnxOperationInSequence(t *testing.T) {
t.Fatal(err)
}
wkvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: base + 1, ModIndex: base + 1, Version: 1},
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: base + 1, ModRevision: base + 1, Version: 1},
}
if !reflect.DeepEqual(kvs, wkvs) {
t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs)
@ -545,13 +545,13 @@ func TestKVCompactReserveLastValue(t *testing.T) {
{
0,
[]storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar0"), CreateIndex: 1, ModIndex: 1, Version: 1},
{Key: []byte("foo"), Value: []byte("bar0"), CreateRevision: 1, ModRevision: 1, Version: 1},
},
},
{
1,
[]storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar1"), CreateIndex: 1, ModIndex: 2, Version: 2},
{Key: []byte("foo"), Value: []byte("bar1"), CreateRevision: 1, ModRevision: 2, Version: 2},
},
},
{
@ -561,7 +561,7 @@ func TestKVCompactReserveLastValue(t *testing.T) {
{
3,
[]storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar2"), CreateIndex: 4, ModIndex: 4, Version: 1},
{Key: []byte("foo"), Value: []byte("bar2"), CreateRevision: 4, ModRevision: 4, Version: 1},
},
},
}
@ -661,9 +661,9 @@ func TestKVSnapshot(t *testing.T) {
s.Put([]byte("foo1"), []byte("bar1"))
s.Put([]byte("foo2"), []byte("bar2"))
wkvs := []storagepb.KeyValue{
{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
}
f, err := os.Create("new_test")

12
storage/kvstore.go
View File

@ -238,7 +238,7 @@ func (s *store) Restore() error {
// restore index
switch e.Type {
case storagepb.PUT:
s.kvindex.Restore(e.Kv.Key, revision{e.Kv.CreateIndex, 0}, rev, e.Kv.Version)
s.kvindex.Restore(e.Kv.Key, revision{e.Kv.CreateRevision, 0}, rev, e.Kv.Version)
case storagepb.DELETE:
s.kvindex.Tombstone(e.Kv.Key, rev)
default:
@ -344,11 +344,11 @@ func (s *store) put(key, value []byte) {
event := storagepb.Event{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{
Key: key,
Value: value,
CreateIndex: c,
ModIndex: rev,
Version: ver,
Key: key,
Value: value,
CreateRevision: c,
ModRevision: rev,
Version: ver,
},
}

50
storage/kvstore_test.go
View File

@ -32,11 +32,11 @@ func TestStorePut(t *testing.T) {
storagepb.Event{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{
Key: []byte("foo"),
Value: []byte("bar"),
CreateIndex: 2,
ModIndex: 2,
Version: 1,
Key: []byte("foo"),
Value: []byte("bar"),
CreateRevision: 2,
ModRevision: 2,
Version: 1,
},
},
revision{2, 0},
@ -48,11 +48,11 @@ func TestStorePut(t *testing.T) {
storagepb.Event{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{
Key: []byte("foo"),
Value: []byte("bar"),
CreateIndex: 2,
ModIndex: 2,
Version: 2,
Key: []byte("foo"),
Value: []byte("bar"),
CreateRevision: 2,
ModRevision: 2,
Version: 2,
},
},
revision{2, 1},
@ -64,11 +64,11 @@ func TestStorePut(t *testing.T) {
storagepb.Event{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{
Key: []byte("foo"),
Value: []byte("bar"),
CreateIndex: 2,
ModIndex: 3,
Version: 3,
Key: []byte("foo"),
Value: []byte("bar"),
CreateRevision: 2,
ModRevision: 3,
Version: 3,
},
},
revision{3, 0},
@ -108,11 +108,11 @@ func TestStoreRange(t *testing.T) {
ev := storagepb.Event{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{
Key: []byte("foo"),
Value: []byte("bar"),
CreateIndex: 1,
ModIndex: 2,
Version: 1,
Key: []byte("foo"),
Value: []byte("bar"),
CreateRevision: 1,
ModRevision: 2,
Version: 1,
},
}
evb, err := ev.Marshal()
@ -269,11 +269,11 @@ func TestStoreRestore(t *testing.T) {
putev := storagepb.Event{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{
Key: []byte("foo"),
Value: []byte("bar"),
CreateIndex: 3,
ModIndex: 3,
Version: 1,
Key: []byte("foo"),
Value: []byte("bar"),
CreateRevision: 3,
ModRevision: 3,
Version: 1,
},
}
putevb, err := putev.Marshal()

36
storage/storagepb/kv.pb.go
View File

@ -48,10 +48,10 @@ func (x Event_EventType) String() string {
}
type KeyValue struct {
Key []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
CreateIndex int64 `protobuf:"varint,2,opt,name=create_index,proto3" json:"create_index,omitempty"`
// mod_index is the last modified index of the key.
ModIndex int64 `protobuf:"varint,3,opt,name=mod_index,proto3" json:"mod_index,omitempty"`
Key []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
CreateRevision int64 `protobuf:"varint,2,opt,name=create_revision,proto3" json:"create_revision,omitempty"`
// mod_revision is the last modified revision of the key.
ModRevision int64 `protobuf:"varint,3,opt,name=mod_revision,proto3" json:"mod_revision,omitempty"`
// version is the version of the key. A deletion resets
// the version to zero and any modification of the key
// increases its version.
@ -101,15 +101,15 @@ func (m *KeyValue) MarshalTo(data []byte) (int, error) {
i += copy(data[i:], m.Key)
}
}
if m.CreateIndex != 0 {
if m.CreateRevision != 0 {
data[i] = 0x10
i++
i = encodeVarintKv(data, i, uint64(m.CreateIndex))
i = encodeVarintKv(data, i, uint64(m.CreateRevision))
}
if m.ModIndex != 0 {
if m.ModRevision != 0 {
data[i] = 0x18
i++
i = encodeVarintKv(data, i, uint64(m.ModIndex))
i = encodeVarintKv(data, i, uint64(m.ModRevision))
}
if m.Version != 0 {
data[i] = 0x20
@ -196,11 +196,11 @@ func (m *KeyValue) Size() (n int) {
n += 1 + l + sovKv(uint64(l))
}
}
if m.CreateIndex != 0 {
n += 1 + sovKv(uint64(m.CreateIndex))
if m.CreateRevision != 0 {
n += 1 + sovKv(uint64(m.CreateRevision))
}
if m.ModIndex != 0 {
n += 1 + sovKv(uint64(m.ModIndex))
if m.ModRevision != 0 {
n += 1 + sovKv(uint64(m.ModRevision))
}
if m.Version != 0 {
n += 1 + sovKv(uint64(m.Version))
@ -286,32 +286,32 @@ func (m *KeyValue) Unmarshal(data []byte) error {
iNdEx = postIndex
case 2:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field CreateIndex", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field CreateRevision", wireType)
}
m.CreateIndex = 0
m.CreateRevision = 0
for shift := uint(0); ; shift += 7 {
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := data[iNdEx]
iNdEx++
m.CreateIndex |= (int64(b) & 0x7F) << shift
m.CreateRevision |= (int64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field ModIndex", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field ModRevision", wireType)
}
m.ModIndex = 0
m.ModRevision = 0
for shift := uint(0); ; shift += 7 {
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := data[iNdEx]
iNdEx++
m.ModIndex |= (int64(b) & 0x7F) << shift
m.ModRevision |= (int64(b) & 0x7F) << shift
if b < 0x80 {
break
}

6
storage/storagepb/kv.proto
View File

@ -11,9 +11,9 @@ option (gogoproto.goproto_enum_prefix_all) = false;
message KeyValue {
bytes key = 1;
int64 create_index = 2;
// mod_index is the last modified index of the key.
int64 mod_index = 3;
int64 create_revision = 2;
// mod_revision is the last modified revision of the key.
int64 mod_revision = 3;
// version is the version of the key. A deletion resets
// the version to zero and any modification of the key
// increases its version.