*: limit request size for v3

Documentation/rfc/v3api.md

@@ -34,6 +34,15 @@ To prove out the design of the v3 API the team has also built [a number of examp
     - easy for people to write simple etcd application
 
 
+## Notes
+
+### Request Size Limitation
+
+The max request size is around 1MB. Since etcd replicates requests in a streaming fashion, a very large
+request might block other requests for a long time. The use case for etcd is to store small configuration
+values, so we prevent user from submitting large requests. This also applies to Txn requests. We might loosen
+the size in the future a little bit or make it configurable.
+
 ## Protobuf Defined API
 
 [api protobuf](../../etcdserver/etcdserverpb/rpc.proto)

etcdserver/api/v3rpc/error.go

@@ -34,4 +34,6 @@ var (
 	ErrPeerURLExist   = grpc.Errorf(codes.FailedPrecondition, "etcdserver: Peer URLs already exists")
 	ErrMemberBadURLs  = grpc.Errorf(codes.InvalidArgument, "etcdserver: given member URLs are invalid")
 	ErrMemberNotFound = grpc.Errorf(codes.NotFound, "etcdserver: member not found")
+
+	ErrRequestTooLarge = grpc.Errorf(codes.InvalidArgument, "etcdserver: request is too large")
 )

etcdserver/api/v3rpc/key.go

@@ -293,6 +293,8 @@ func togRPCError(err error) error {
 	case lease.ErrLeaseNotFound:
 		return ErrLeaseNotFound
 	// TODO: handle error from raft and timeout
+	case etcdserver.ErrRequestTooLarge:
+		return ErrRequestTooLarge
 	default:
 		return grpc.Errorf(codes.Internal, err.Error())
 	}

etcdserver/errors.go

@@ -34,6 +34,7 @@ var (
 	ErrTimeoutDueToConnectionLost = errors.New("etcdserver: request timed out, possibly due to connection lost")
 	ErrNotEnoughStartedMembers    = errors.New("etcdserver: re-configuration failed due to not enough started members")
 	ErrNoLeader                   = errors.New("etcdserver: no leader")
+	ErrRequestTooLarge            = errors.New("etcdserver: request is too large")
 )
 
 func isKeyNotFound(err error) bool {

etcdserver/v3demo_server.go

@@ -29,6 +29,14 @@ import (
 	"github.com/coreos/etcd/storage/storagepb"
 )
 
+const (
+	// the max request size that raft accepts.
+	// TODO: make this a flag? But we probably do not want to
+	// accept large request which might block raft stream. User
+	// specify a large value might end up with shooting in the foot.
+	maxRequestBytes = 1.5 * 1024 * 1024
+)
+
 type RaftKV interface {
 	Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error)
 	Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error)
@@ -165,6 +173,11 @@ func (s *EtcdServer) processInternalRaftRequest(ctx context.Context, r pb.Intern
 	if err != nil {
 		return nil, err
 	}
+
+	if len(data) > maxRequestBytes {
+		return nil, ErrRequestTooLarge
+	}
+
 	ch := s.w.Register(r.ID)
 
 	s.r.Propose(ctx, data)

integration/v3_grpc_test.go

@@ -413,6 +413,24 @@ func TestV3TxnInvaildRange(t *testing.T) {
 	}
 }
 
+func TestV3TooLargeRequest(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	clus := NewClusterV3(t, &ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+
+	kvc := clus.RandClient().KV
+
+	// 2MB request value
+	largeV := make([]byte, 2*1024*1024)
+	preq := &pb.PutRequest{Key: []byte("foo"), Value: largeV}
+
+	_, err := kvc.Put(context.Background(), preq)
+	if err != v3rpc.ErrRequestTooLarge {
+		t.Errorf("err = %v, want %v", err, v3rpc.ErrRequestTooLarge)
+	}
+}
+
 // TestV3Hash tests hash.
 func TestV3Hash(t *testing.T) {
 	defer testutil.AfterTest(t)