Properly enforce RangeEnd + Limit restriction

Required for compatibility with Kubernetes 1.24, which when searching
within a namespace for a resource using FieldSelector on metatada.name,
sends a Get request for the fully qualified resource name, but with a
non-zero limit.

Signed-off-by: Brad Davidson <[email protected]>

Author: brandond

pkg/drivers/jetstream/jetstream.go

@@ -280,7 +280,7 @@ func (j *JetStream) isKeyExpired(_ context.Context, createTime time.Time, value
 }
 
 // Get returns the associated server.KeyValue
-func (j *JetStream) Get(ctx context.Context, key string, revision int64) (revRet int64, kvRet *server.KeyValue, errRet error) {
+func (j *JetStream) Get(ctx context.Context, key, rangeEnd string, limit, revision int64) (revRet int64, kvRet *server.KeyValue, errRet error) {
 	//logrus.Tracef("GET %s, rev=%d", key, revision)
 	start := time.Now()
 	defer func() {
@@ -674,7 +674,7 @@ func (j *JetStream) listAfter(ctx context.Context, prefix string, revision int64
 				KV:     kv.KV,
 				PrevKV: &server.KeyValue{},
 			}
-			if _, prevKV, err := j.Get(ctx, kv.KV.Key, kv.PrevRevision); err == nil && prevKV != nil {
+			if _, prevKV, err := j.Get(ctx, kv.KV.Key, "", 1, kv.PrevRevision); err == nil && prevKV != nil {
 				event.PrevKV = prevKV
 			}
 

pkg/logstructured/logstructured.go

@@ -44,21 +44,21 @@ func (l *LogStructured) Start(ctx context.Context) error {
 	return nil
 }
 
-func (l *LogStructured) Get(ctx context.Context, key string, revision int64) (revRet int64, kvRet *server.KeyValue, errRet error) {
+func (l *LogStructured) Get(ctx context.Context, key, rangeEnd string, limit, revision int64) (revRet int64, kvRet *server.KeyValue, errRet error) {
 	defer func() {
 		l.adjustRevision(ctx, &revRet)
 		logrus.Tracef("GET %s, rev=%d => rev=%d, kv=%v, err=%v", key, revision, revRet, kvRet != nil, errRet)
 	}()
 
-	rev, event, err := l.get(ctx, key, revision, false)
+	rev, event, err := l.get(ctx, key, rangeEnd, limit, revision, false)
 	if event == nil {
 		return rev, nil, err
 	}
 	return rev, event.KV, err
 }
 
-func (l *LogStructured) get(ctx context.Context, key string, revision int64, includeDeletes bool) (int64, *server.Event, error) {
-	rev, events, err := l.log.List(ctx, key, "", 1, revision, includeDeletes)
+func (l *LogStructured) get(ctx context.Context, key, rangeEnd string, limit, revision int64, includeDeletes bool) (int64, *server.Event, error) {
+	rev, events, err := l.log.List(ctx, key, rangeEnd, limit, revision, includeDeletes)
 	if err == server.ErrCompacted {
 		// ignore compacted when getting by revision
 		err = nil
@@ -91,7 +91,7 @@ func (l *LogStructured) Create(ctx context.Context, key string, value []byte, le
 		logrus.Tracef("CREATE %s, size=%d, lease=%d => rev=%d, err=%v", key, len(value), lease, revRet, errRet)
 	}()
 
-	rev, prevEvent, err := l.get(ctx, key, 0, true)
+	rev, prevEvent, err := l.get(ctx, key, "", 1, 0, true)
 	if err != nil {
 		return 0, err
 	}
@@ -123,7 +123,7 @@ func (l *LogStructured) Delete(ctx context.Context, key string, revision int64)
 		logrus.Tracef("DELETE %s, rev=%d => rev=%d, kv=%v, deleted=%v, err=%v", key, revision, revRet, kvRet != nil, deletedRet, errRet)
 	}()
 
-	rev, event, err := l.get(ctx, key, 0, true)
+	rev, event, err := l.get(ctx, key, "", 1, 0, true)
 	if err != nil {
 		return 0, nil, false, err
 	}
@@ -150,7 +150,7 @@ func (l *LogStructured) Delete(ctx context.Context, key string, revision int64)
 	if err != nil {
 		// If error on Append we assume it's a UNIQUE constraint error, so we fetch the latest (if we can)
 		// and return that the delete failed
-		latestRev, latestEvent, latestErr := l.get(ctx, key, 0, true)
+		latestRev, latestEvent, latestErr := l.get(ctx, key, "", 1, 0, true)
 		if latestErr != nil || latestEvent == nil {
 			return rev, event.KV, false, nil
 		}
@@ -220,7 +220,7 @@ func (l *LogStructured) Update(ctx context.Context, key string, value []byte, re
 		logrus.Tracef("UPDATE %s, value=%d, rev=%d, lease=%v => rev=%d, kvrev=%d, updated=%v, err=%v", key, len(value), revision, lease, revRet, kvRev, updateRet, errRet)
 	}()
 
-	rev, event, err := l.get(ctx, key, 0, false)
+	rev, event, err := l.get(ctx, key, "", 1, 0, false)
 	if err != nil {
 		return 0, nil, false, err
 	}
@@ -245,7 +245,7 @@ func (l *LogStructured) Update(ctx context.Context, key string, value []byte, re
 
 	rev, err = l.log.Append(ctx, updateEvent)
 	if err != nil {
-		rev, event, err := l.get(ctx, key, 0, false)
+		rev, event, err := l.get(ctx, key, "", 1, 0, false)
 		if event == nil {
 			return rev, nil, false, err
 		}

pkg/server/get.go

@@ -8,11 +8,11 @@ import (
 )
 
 func (l *LimitedServer) get(ctx context.Context, r *etcdserverpb.RangeRequest) (*RangeResponse, error) {
-	if r.Limit != 0 {
+	if r.Limit != 0 && len(r.RangeEnd) != 0 {
 		return nil, fmt.Errorf("invalid combination of rangeEnd and limit, limit should be 0 got %d", r.Limit)
 	}
 
-	rev, kv, err := l.backend.Get(ctx, string(r.Key), r.Revision)
+	rev, kv, err := l.backend.Get(ctx, string(r.Key), string(r.RangeEnd), r.Limit, r.Revision)
 	if err != nil {
 		return nil, err
 	}

pkg/server/types.go

@@ -14,7 +14,7 @@ var (
 
 type Backend interface {
 	Start(ctx context.Context) error
-	Get(ctx context.Context, key string, revision int64) (int64, *KeyValue, error)
+	Get(ctx context.Context, key, rangeEnd string, limit, revision int64) (int64, *KeyValue, error)
 	Create(ctx context.Context, key string, value []byte, lease int64) (int64, error)
 	Delete(ctx context.Context, key string, revision int64) (int64, *KeyValue, bool, error)
 	List(ctx context.Context, prefix, startKey string, limit, revision int64) (int64, []*KeyValue, error)