Merge pull request #3514 from xiang90/v3_raft

support clustered v3 api

etcdserver/api/v3rpc/key.go

@@ -29,23 +29,23 @@ func New(s etcdserver.V3DemoServer) pb.EtcdServer {
 }
 
 func (h *handler) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
-	resp := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{Range: r})
-	return resp.(*pb.RangeResponse), nil
+	resp, err := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{Range: r})
+	return resp.(*pb.RangeResponse), err
 }
 
 func (h *handler) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {
-	resp := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{Put: r})
-	return resp.(*pb.PutResponse), nil
+	resp, err := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{Put: r})
+	return resp.(*pb.PutResponse), err
 }
 
 func (h *handler) DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) {
-	resp := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{DeleteRange: r})
-	return resp.(*pb.DeleteRangeResponse), nil
+	resp, err := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{DeleteRange: r})
+	return resp.(*pb.DeleteRangeResponse), err
 }
 
 func (h *handler) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) {
-	resp := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{Txn: r})
-	return resp.(*pb.TxnResponse), nil
+	resp, err := h.server.V3DemoDo(ctx, pb.InternalRaftRequest{Txn: r})
+	return resp.(*pb.TxnResponse), err
 }
 
 func (h *handler) Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error) {

etcdserver/etcdserverpb/raft_internal.pb.go

@@ -17,17 +17,25 @@ var _ = proto.Marshal
 // An InternalRaftRequest is the union of all requests which can be
 // sent via raft.
 type InternalRaftRequest struct {
-	V2          *Request            `protobuf:"bytes,1,opt,name=v2" json:"v2,omitempty"`
-	Range       *RangeRequest       `protobuf:"bytes,2,opt,name=range" json:"range,omitempty"`
-	Put         *PutRequest         `protobuf:"bytes,3,opt,name=put" json:"put,omitempty"`
-	DeleteRange *DeleteRangeRequest `protobuf:"bytes,4,opt,name=delete_range" json:"delete_range,omitempty"`
-	Txn         *TxnRequest         `protobuf:"bytes,5,opt,name=txn" json:"txn,omitempty"`
+	ID          uint64              `protobuf:"varint,1,opt,proto3" json:"ID,omitempty"`
+	V2          *Request            `protobuf:"bytes,2,opt,name=v2" json:"v2,omitempty"`
+	Range       *RangeRequest       `protobuf:"bytes,3,opt,name=range" json:"range,omitempty"`
+	Put         *PutRequest         `protobuf:"bytes,4,opt,name=put" json:"put,omitempty"`
+	DeleteRange *DeleteRangeRequest `protobuf:"bytes,5,opt,name=delete_range" json:"delete_range,omitempty"`
+	Txn         *TxnRequest         `protobuf:"bytes,6,opt,name=txn" json:"txn,omitempty"`
 }
 
 func (m *InternalRaftRequest) Reset()         { *m = InternalRaftRequest{} }
 func (m *InternalRaftRequest) String() string { return proto.CompactTextString(m) }
 func (*InternalRaftRequest) ProtoMessage()    {}
 
+type EmptyResponse struct {
+}
+
+func (m *EmptyResponse) Reset()         { *m = EmptyResponse{} }
+func (m *EmptyResponse) String() string { return proto.CompactTextString(m) }
+func (*EmptyResponse) ProtoMessage()    {}
+
 func (m *InternalRaftRequest) Marshal() (data []byte, err error) {
 	size := m.Size()
 	data = make([]byte, size)
@@ -43,8 +51,13 @@ func (m *InternalRaftRequest) MarshalTo(data []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.ID != 0 {
+		data[i] = 0x8
+		i++
+		i = encodeVarintRaftInternal(data, i, uint64(m.ID))
+	}
 	if m.V2 != nil {
-		data[i] = 0xa
+		data[i] = 0x12
 		i++
 		i = encodeVarintRaftInternal(data, i, uint64(m.V2.Size()))
 		n1, err := m.V2.MarshalTo(data[i:])
@@ -54,7 +67,7 @@ func (m *InternalRaftRequest) MarshalTo(data []byte) (int, error) {
 		i += n1
 	}
 	if m.Range != nil {
-		data[i] = 0x12
+		data[i] = 0x1a
 		i++
 		i = encodeVarintRaftInternal(data, i, uint64(m.Range.Size()))
 		n2, err := m.Range.MarshalTo(data[i:])
@@ -64,7 +77,7 @@ func (m *InternalRaftRequest) MarshalTo(data []byte) (int, error) {
 		i += n2
 	}
 	if m.Put != nil {
-		data[i] = 0x1a
+		data[i] = 0x22
 		i++
 		i = encodeVarintRaftInternal(data, i, uint64(m.Put.Size()))
 		n3, err := m.Put.MarshalTo(data[i:])
@@ -74,7 +87,7 @@ func (m *InternalRaftRequest) MarshalTo(data []byte) (int, error) {
 		i += n3
 	}
 	if m.DeleteRange != nil {
-		data[i] = 0x22
+		data[i] = 0x2a
 		i++
 		i = encodeVarintRaftInternal(data, i, uint64(m.DeleteRange.Size()))
 		n4, err := m.DeleteRange.MarshalTo(data[i:])
@@ -84,7 +97,7 @@ func (m *InternalRaftRequest) MarshalTo(data []byte) (int, error) {
 		i += n4
 	}
 	if m.Txn != nil {
-		data[i] = 0x2a
+		data[i] = 0x32
 		i++
 		i = encodeVarintRaftInternal(data, i, uint64(m.Txn.Size()))
 		n5, err := m.Txn.MarshalTo(data[i:])
@@ -96,6 +109,24 @@ func (m *InternalRaftRequest) MarshalTo(data []byte) (int, error) {
 	return i, nil
 }
 
+func (m *EmptyResponse) Marshal() (data []byte, err error) {
+	size := m.Size()
+	data = make([]byte, size)
+	n, err := m.MarshalTo(data)
+	if err != nil {
+		return nil, err
+	}
+	return data[:n], nil
+}
+
+func (m *EmptyResponse) MarshalTo(data []byte) (int, error) {
+	var i int
+	_ = i
+	var l int
+	_ = l
+	return i, nil
+}
+
 func encodeFixed64RaftInternal(data []byte, offset int, v uint64) int {
 	data[offset] = uint8(v)
 	data[offset+1] = uint8(v >> 8)
@@ -126,6 +157,9 @@ func encodeVarintRaftInternal(data []byte, offset int, v uint64) int {
 func (m *InternalRaftRequest) Size() (n int) {
 	var l int
 	_ = l
+	if m.ID != 0 {
+		n += 1 + sovRaftInternal(uint64(m.ID))
+	}
 	if m.V2 != nil {
 		l = m.V2.Size()
 		n += 1 + l + sovRaftInternal(uint64(l))
@@ -149,6 +183,12 @@ func (m *InternalRaftRequest) Size() (n int) {
 	return n
 }
 
+func (m *EmptyResponse) Size() (n int) {
+	var l int
+	_ = l
+	return n
+}
+
 func sovRaftInternal(x uint64) (n int) {
 	for {
 		n++
@@ -162,42 +202,6 @@ func sovRaftInternal(x uint64) (n int) {
 func sozRaftInternal(x uint64) (n int) {
 	return sovRaftInternal(uint64((x << 1) ^ uint64((int64(x) >> 63))))
 }
-func (this *InternalRaftRequest) GetValue() interface{} {
-	if this.V2 != nil {
-		return this.V2
-	}
-	if this.Range != nil {
-		return this.Range
-	}
-	if this.Put != nil {
-		return this.Put
-	}
-	if this.DeleteRange != nil {
-		return this.DeleteRange
-	}
-	if this.Txn != nil {
-		return this.Txn
-	}
-	return nil
-}
-
-func (this *InternalRaftRequest) SetValue(value interface{}) bool {
-	switch vt := value.(type) {
-	case *Request:
-		this.V2 = vt
-	case *RangeRequest:
-		this.Range = vt
-	case *PutRequest:
-		this.Put = vt
-	case *DeleteRangeRequest:
-		this.DeleteRange = vt
-	case *TxnRequest:
-		this.Txn = vt
-	default:
-		return false
-	}
-	return true
-}
 func (m *InternalRaftRequest) Unmarshal(data []byte) error {
 	l := len(data)
 	iNdEx := 0
@@ -218,6 +222,22 @@ func (m *InternalRaftRequest) Unmarshal(data []byte) error {
 		wireType := int(wire & 0x7)
 		switch fieldNum {
 		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ID", wireType)
+			}
+			m.ID = 0
+			for shift := uint(0); ; shift += 7 {
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := data[iNdEx]
+				iNdEx++
+				m.ID |= (uint64(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field V2", wireType)
 			}
@@ -247,7 +267,7 @@ func (m *InternalRaftRequest) Unmarshal(data []byte) error {
 				return err
 			}
 			iNdEx = postIndex
-		case 2:
+		case 3:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field Range", wireType)
 			}
@@ -277,7 +297,7 @@ func (m *InternalRaftRequest) Unmarshal(data []byte) error {
 				return err
 			}
 			iNdEx = postIndex
-		case 3:
+		case 4:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field Put", wireType)
 			}
@@ -307,7 +327,7 @@ func (m *InternalRaftRequest) Unmarshal(data []byte) error {
 				return err
 			}
 			iNdEx = postIndex
-		case 4:
+		case 5:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field DeleteRange", wireType)
 			}
@@ -337,7 +357,7 @@ func (m *InternalRaftRequest) Unmarshal(data []byte) error {
 				return err
 			}
 			iNdEx = postIndex
-		case 5:
+		case 6:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field Txn", wireType)
 			}
@@ -393,6 +413,50 @@ func (m *InternalRaftRequest) Unmarshal(data []byte) error {
 
 	return nil
 }
+func (m *EmptyResponse) Unmarshal(data []byte) error {
+	l := len(data)
+	iNdEx := 0
+	for iNdEx < l {
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := data[iNdEx]
+			iNdEx++
+			wire |= (uint64(b) & 0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		switch fieldNum {
+		default:
+			var sizeOfWire int
+			for {
+				sizeOfWire++
+				wire >>= 7
+				if wire == 0 {
+					break
+				}
+			}
+			iNdEx -= sizeOfWire
+			skippy, err := skipRaftInternal(data[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthRaftInternal
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	return nil
+}
 func skipRaftInternal(data []byte) (n int, err error) {
 	l := len(data)
 	iNdEx := 0

etcdserver/etcdserverpb/raft_internal.proto

@@ -13,12 +13,13 @@ option (gogoproto.goproto_getters_all) = false;
 // An InternalRaftRequest is the union of all requests which can be
 // sent via raft.
 message InternalRaftRequest {
-  option (gogoproto.onlyone) = true;
-  oneof value {
-    Request v2 = 1;
-    RangeRequest range = 2;
-    PutRequest put = 3;
-    DeleteRangeRequest delete_range = 4;
-    TxnRequest txn = 5;
-  }
+  uint64 ID = 1;
+  Request v2 = 2;
+  RangeRequest range = 3;
+  PutRequest put = 4;
+  DeleteRangeRequest delete_range = 5;
+  TxnRequest txn = 6;
+}
+
+message EmptyResponse {
 }

etcdserver/server.go

@@ -776,6 +776,8 @@ func (s *EtcdServer) apply(es []raftpb.Entry, confState *raftpb.ConfState) (uint
 				case raftReq.V2 != nil:
 					req := raftReq.V2
 					s.w.Trigger(req.ID, s.applyRequest(*req))
+				default:
+					s.w.Trigger(raftReq.ID, s.applyV3Request(&raftReq))
 				}
 			}
 		case raftpb.EntryConfChange:

etcdserver/v3demo_server.go

@@ -24,25 +24,48 @@ import (
 )
 
 type V3DemoServer interface {
-	V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) proto.Message
+	V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) (proto.Message, error)
 }
 
-func (s *EtcdServer) V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) proto.Message {
+func (s *EtcdServer) V3DemoDo(ctx context.Context, r pb.InternalRaftRequest) (proto.Message, error) {
+	r.ID = s.reqIDGen.Next()
+
+	data, err := r.Marshal()
+	if err != nil {
+		return &pb.EmptyResponse{}, err
+	}
+	ch := s.w.Register(r.ID)
+
+	s.r.Propose(ctx, data)
+
+	select {
+	case x := <-ch:
+		resp := x.(proto.Message)
+		return resp, nil
+	case <-ctx.Done():
+		s.w.Trigger(r.ID, nil) // GC wait
+		return &pb.EmptyResponse{}, ctx.Err()
+	case <-s.done:
+		return &pb.EmptyResponse{}, ErrStopped
+	}
+}
+
+func (s *EtcdServer) applyV3Request(r *pb.InternalRaftRequest) interface{} {
 	switch {
 	case r.Range != nil:
-		return doRange(s.kv, r.Range)
+		return applyRange(s.kv, r.Range)
 	case r.Put != nil:
-		return doPut(s.kv, r.Put)
+		return applyPut(s.kv, r.Put)
 	case r.DeleteRange != nil:
-		return doDeleteRange(s.kv, r.DeleteRange)
+		return applyDeleteRange(s.kv, r.DeleteRange)
 	case r.Txn != nil:
-		return doTxn(s.kv, r.Txn)
+		return applyTxn(s.kv, r.Txn)
 	default:
 		panic("not implemented")
 	}
 }
 
-func doPut(kv dstorage.KV, p *pb.PutRequest) *pb.PutResponse {
+func applyPut(kv dstorage.KV, p *pb.PutRequest) *pb.PutResponse {
 	resp := &pb.PutResponse{}
 	resp.Header = &pb.ResponseHeader{}
 	rev := kv.Put(p.Key, p.Value)
@@ -50,7 +73,7 @@ func doPut(kv dstorage.KV, p *pb.PutRequest) *pb.PutResponse {
 	return resp
 }
 
-func doRange(kv dstorage.KV, r *pb.RangeRequest) *pb.RangeResponse {
+func applyRange(kv dstorage.KV, r *pb.RangeRequest) *pb.RangeResponse {
 	resp := &pb.RangeResponse{}
 	resp.Header = &pb.ResponseHeader{}
 	kvs, rev, err := kv.Range(r.Key, r.RangeEnd, r.Limit, 0)
@@ -65,7 +88,7 @@ func doRange(kv dstorage.KV, r *pb.RangeRequest) *pb.RangeResponse {
 	return resp
 }
 
-func doDeleteRange(kv dstorage.KV, dr *pb.DeleteRangeRequest) *pb.DeleteRangeResponse {
+func applyDeleteRange(kv dstorage.KV, dr *pb.DeleteRangeRequest) *pb.DeleteRangeResponse {
 	resp := &pb.DeleteRangeResponse{}
 	resp.Header = &pb.ResponseHeader{}
 	_, rev := kv.DeleteRange(dr.Key, dr.RangeEnd)
@@ -73,12 +96,12 @@ func doDeleteRange(kv dstorage.KV, dr *pb.DeleteRangeRequest) *pb.DeleteRangeRes
 	return resp
 }
 
-func doTxn(kv dstorage.KV, rt *pb.TxnRequest) *pb.TxnResponse {
+func applyTxn(kv dstorage.KV, rt *pb.TxnRequest) *pb.TxnResponse {
 	var revision int64
 
 	ok := true
 	for _, c := range rt.Compare {
-		if revision, ok = doCompare(kv, c); !ok {
+		if revision, ok = applyCompare(kv, c); !ok {
 			break
 		}
 	}
@@ -91,7 +114,7 @@ func doTxn(kv dstorage.KV, rt *pb.TxnRequest) *pb.TxnResponse {
 	}
 	resps := make([]*pb.ResponseUnion, len(reqs))
 	for i := range reqs {
-		resps[i] = doUnion(kv, reqs[i])
+		resps[i] = applyUnion(kv, reqs[i])
 	}
 	if len(resps) != 0 {
 		revision += 1
@@ -105,21 +128,21 @@ func doTxn(kv dstorage.KV, rt *pb.TxnRequest) *pb.TxnResponse {
 	return txnResp
 }
 
-func doUnion(kv dstorage.KV, union *pb.RequestUnion) *pb.ResponseUnion {
+func applyUnion(kv dstorage.KV, union *pb.RequestUnion) *pb.ResponseUnion {
 	switch {
 	case union.RequestRange != nil:
-		return &pb.ResponseUnion{ResponseRange: doRange(kv, union.RequestRange)}
+		return &pb.ResponseUnion{ResponseRange: applyRange(kv, union.RequestRange)}
 	case union.RequestPut != nil:
-		return &pb.ResponseUnion{ResponsePut: doPut(kv, union.RequestPut)}
+		return &pb.ResponseUnion{ResponsePut: applyPut(kv, union.RequestPut)}
 	case union.RequestDeleteRange != nil:
-		return &pb.ResponseUnion{ResponseDeleteRange: doDeleteRange(kv, union.RequestDeleteRange)}
+		return &pb.ResponseUnion{ResponseDeleteRange: applyDeleteRange(kv, union.RequestDeleteRange)}
 	default:
 		// empty union
 		return nil
 	}
 }
 
-func doCompare(kv dstorage.KV, c *pb.Compare) (int64, bool) {
+func applyCompare(kv dstorage.KV, c *pb.Compare) (int64, bool) {
 	ckvs, rev, err := kv.Range(c.Key, nil, 1, 0)
 	if err != nil {
 		return rev, false