Merge pull request #4698 from xiang90/fix_w

*: fix watch full key range

clientv3/op.go

@@ -28,6 +28,10 @@ const (
 	tDeleteRange
 )
 
+var (
+	noPrefixEnd = []byte{0}
+)
+
 // Op represents an Operation that kv can execute.
 type Op struct {
 	t   opType
@@ -175,8 +179,7 @@ func getPrefix(key []byte) []byte {
 	}
 	// next prefix does not exist (e.g., 0xffff);
 	// default to WithFromKey policy
-	end = []byte{0}
-	return end
+	return noPrefixEnd
 }
 
 // WithPrefix enables 'Get', 'Delete', or 'Watch' requests to operate

etcdserver/api/v3rpc/watch.go

@@ -111,6 +111,10 @@ func (sws *serverWatchStream) recvLoop() error {
 			}
 
 			creq := uv.CreateRequest
+			if len(creq.Key) == 0 {
+				// \x00 is the smallest key
+				creq.Key = []byte{0}
+			}
 			if len(creq.RangeEnd) == 1 && creq.RangeEnd[0] == 0 {
 				// support  >= key queries
 				creq.RangeEnd = []byte{}

integration/v3_watch_test.go

@@ -98,6 +98,27 @@ func TestV3WatchFromCurrentRevision(t *testing.T) {
 
 			[]*pb.WatchResponse{},
 		},
+		// watch full range, matching
+		{
+			[]string{"fooLong"},
+			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
+				CreateRequest: &pb.WatchCreateRequest{
+					Key:      []byte(""),
+					RangeEnd: []byte("\x00")}}},
+
+			[]*pb.WatchResponse{
+				{
+					Header:  &pb.ResponseHeader{Revision: 2},
+					Created: false,
+					Events: []*storagepb.Event{
+						{
+							Type: storagepb.PUT,
+							Kv:   &storagepb.KeyValue{Key: []byte("fooLong"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1},
+						},
+					},
+				},
+			},
+		},
 		// multiple puts, one watcher with matching key
 		{
 			[]string{"foo", "foo", "foo"},