WIP: Rebase 1.32 Pre.1 Diff

staging/src/k8s.io/apiextensions-apiserver/pkg/registry/customresource/strategy.go

@@ -156,7 +156,7 @@ func (a customResourceStrategy) PrepareForCreate(ctx context.Context, obj runtim
 	}
 
 	accessor, _ := meta.Accessor(obj)
-	if _, found := accessor.GetAnnotations()[genericapirequest.AnnotationKey]; found {
+	if _, found := accessor.GetAnnotations()[genericapirequest.ShardAnnotationKey]; found {
 		// to avoid an additional UPDATE request (mismatch on the generation field) replicated objects have the generation field already set
 		return
 	}
@@ -190,7 +190,7 @@ func (a customResourceStrategy) PrepareForUpdate(ctx context.Context, obj, old r
 	if !apiequality.Semantic.DeepEqual(newCopyContent, oldCopyContent) {
 		oldAccessor, _ := meta.Accessor(oldCustomResourceObject)
 		newAccessor, _ := meta.Accessor(newCustomResourceObject)
-		if _, found := oldAccessor.GetAnnotations()[genericapirequest.AnnotationKey]; found {
+		if _, found := oldAccessor.GetAnnotations()[genericapirequest.ShardAnnotationKey]; found {
 			// the presence of the annotation indicates the object is from the cache server.
 			// since the objects from the cache should not be modified in any way, just return early.
 			return

staging/src/k8s.io/apiserver/pkg/endpoints/request/context_shard.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2022 The KCP Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package request
+
+import (
+	"context"
+)
+
+type shardKey int
+
+const (
+	// shardKey is the context key for the request.
+	shardContextKey shardKey = iota
+
+	// ShardAnnotationKey is the name of the annotation key used to denote an object's shard name.
+	ShardAnnotationKey = "kcp.io/shard"
+)
+
+// Shard describes a shard
+type Shard string
+
+// Empty returns true if the name of the shard is empty.
+func (s Shard) Empty() bool {
+	return s == ""
+}
+
+// Wildcard checks if the given shard name matches wildcard.
+// If true the query applies to all shards.
+func (s Shard) Wildcard() bool {
+	return s == "*"
+}
+
+// String casts Shard to string type
+func (s Shard) String() string {
+	return string(s)
+}
+
+// WithShard returns a context that holds the given shard.
+func WithShard(parent context.Context, shard Shard) context.Context {
+	return context.WithValue(parent, shardContextKey, shard)
+}
+
+// ShardFrom returns the value of the shard key in the context, or an empty value if there is no shard key.
+func ShardFrom(ctx context.Context) Shard {
+	shard, ok := ctx.Value(shardContextKey).(Shard)
+	if !ok {
+		return ""
+	}
+	return shard
+}

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -267,11 +267,18 @@ const (
 // to resource directories enforcing namespace rules.
 func NoNamespaceKeyRootFunc(ctx context.Context, prefix string) string {
 	key := prefix
+	shard := genericapirequest.ShardFrom(ctx)
+	if shard.Wildcard() {
+		return key
+	}
 	cluster, err := genericapirequest.ValidClusterFrom(ctx)
 	if err != nil {
 		klog.Errorf("invalid context cluster value: %v", err)
 		return key
 	}
+	if !shard.Empty() {
+		key += "/" + shard.String()
+	}
 	if !cluster.Wildcard {
 		key += "/" + cluster.Name.String()
 	}
@@ -514,6 +521,12 @@ func (e *Store) create(ctx context.Context, obj runtime.Object, createValidation
 	if err := rest.BeforeCreate(e.CreateStrategy, ctx, obj); err != nil {
 		return nil, err
 	}
+
+	if _, found := objectMeta.GetAnnotations()[genericapirequest.ShardAnnotationKey]; found {
+		// Remove the shard annotation so it is not persisted
+		delete(objectMeta.GetAnnotations(), genericapirequest.ShardAnnotationKey)
+	}
+
 	// at this point we have a fully formed object.  It is time to call the validators that the apiserver
 	// handling chain wants to enforce.
 	if createValidation != nil {
@@ -1081,6 +1094,13 @@ func (e *Store) updateForGracefulDeletionAndFinalizers(ctx context.Context, name
 			if err != nil {
 				return nil, err
 			}
+
+			// the following annotation key indicates that the request is from the cache server
+			// in that case we've decided not to require finalization, the object will be deleted immediately
+			if _, hasShardAnnotation := existingAccessor.GetAnnotations()[genericapirequest.ShardAnnotationKey]; hasShardAnnotation {
+				return existing, nil
+			}
+
 			needsUpdate, newFinalizers := deletionFinalizersForGarbageCollection(ctx, e, existingAccessor, options)
 			if needsUpdate {
 				existingAccessor.SetFinalizers(newFinalizers)

staging/src/k8s.io/apiserver/pkg/registry/rest/meta.go

@@ -21,10 +21,15 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 )
 
 // WipeObjectMetaSystemFields erases fields that are managed by the system on ObjectMeta.
 func WipeObjectMetaSystemFields(meta metav1.Object) {
+	if _, found := meta.GetAnnotations()[genericapirequest.ShardAnnotationKey]; found {
+		// Do not wipe system fields if we are storing a cached object
+		return
+	}
 	meta.SetCreationTimestamp(metav1.Time{})
 	meta.SetUID("")
 	meta.SetDeletionTimestamp(nil)
@@ -34,6 +39,12 @@ func WipeObjectMetaSystemFields(meta metav1.Object) {
 
 // FillObjectMetaSystemFields populates fields that are managed by the system on ObjectMeta.
 func FillObjectMetaSystemFields(meta metav1.Object) {
+	if _, found := meta.GetAnnotations()[genericapirequest.ShardAnnotationKey]; found {
+		// In general the shard annotation is not attached to objects. Instead, it is assigned by the storage layer on the fly.
+		// To avoid an additional UPDATE request (mismatch on the creationTime and UID fields) replicated objects have those fields already set.
+		// Thus all we have to do is to simply return early.
+		return
+	}
 	meta.SetCreationTimestamp(metav1.Now())
 	meta.SetUID(uuid.NewUUID())
 }

staging/src/k8s.io/apiserver/pkg/registry/rest/update.go

@@ -124,7 +124,12 @@ func BeforeUpdate(strategy RESTUpdateStrategy, ctx context.Context, obj, old run
 	if err != nil {
 		return err
 	}
-	objectMeta.SetGeneration(oldMeta.GetGeneration())
+	if len(oldMeta.GetAnnotations()[genericapirequest.ShardAnnotationKey]) == 0 || objectMeta.GetGeneration() == 0 {
+		// the absence of the annotation indicates the object is NOT from the cache server,
+		// if the new object doesn't have its generation set, just rewrite it from the old object
+		// otherwise  we are dealing with an object from the cache server that wants its generation to be updated
+		objectMeta.SetGeneration(oldMeta.GetGeneration())
+	}
 
 	strategy.PrepareForUpdate(ctx, obj, old)
 

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go

@@ -237,7 +237,8 @@ func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, ou
 		return storage.NewInternalError(err)
 	}
 
-	err = decode(s.codec, s.versioner, data, out, getResp.KV.ModRevision, clusterName)
+	shardName := endpointsrequest.ShardFrom(ctx)
+	err = decode(s.codec, s.versioner, data, out, getResp.KV.ModRevision, clusterName, shardName)
 	if err != nil {
 		recordDecodeError(s.groupResourceString, preparedKey)
 		return err
@@ -306,7 +307,8 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
 	}
 
 	if out != nil {
-		err = decode(s.codec, s.versioner, data, out, txnResp.Revision, clusterName)
+		shardName := endpointsrequest.ShardFrom(ctx)
+		err = decode(s.codec, s.versioner, data, out, txnResp.Revision, clusterName, shardName)
 		if err != nil {
 			span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error()))
 			recordDecodeError(s.groupResourceString, preparedKey)
@@ -344,7 +346,9 @@ func (s *store) conditionalDelete(
 	if err != nil {
 		klog.Errorf("No cluster defined in conditionalDelete action for key %s : %s", key, err.Error())
 	}
-	getCurrentState := s.getCurrentState(ctx, key, v, false, skipTransformDecode, clusterName)
+	shardName := endpointsrequest.ShardFrom(ctx)
+
+	getCurrentState := s.getCurrentState(ctx, key, v, false, skipTransformDecode, clusterName, shardName)
 
 	var origState *objState
 	var origStateIsCurrent bool
@@ -422,7 +426,7 @@ func (s *store) conditionalDelete(
 		}
 		if !txnResp.Succeeded {
 			klog.V(4).Infof("deletion of %s failed because of a conflict, going to retry", key)
-			origState, err = s.getState(ctx, txnResp.KV, key, v, false, skipTransformDecode, clusterName)
+			origState, err = s.getState(ctx, txnResp.KV, key, v, false, skipTransformDecode, clusterName, shardName)
 			if err != nil {
 				return err
 			}
@@ -431,14 +435,7 @@ func (s *store) conditionalDelete(
 		}
 
 		if !skipTransformDecode {
-			if len(txnResp.Responses) == 0 || txnResp.Responses[0].GetResponseDeleteRange() == nil {
-				return errors.New(fmt.Sprintf("invalid DeleteRange response: %v", txnResp.Responses))
-			}
-			deleteResp := txnResp.Responses[0].GetResponseDeleteRange()
-			if deleteResp.Header == nil {
-				return errors.New("invalid DeleteRange response - nil header")
-			}
-			err = decode(s.codec, s.versioner, origState.data, out, deleteResp.Header.Revision, clusterName)
+			err = decode(s.codec, s.versioner, origState.data, out, txnResp.Revision, clusterName, shardName)
 			if err != nil {
 				recordDecodeError(s.groupResourceString, key)
 				return err
@@ -456,6 +453,8 @@ func (s *store) GuaranteedUpdate(
 	if err != nil {
 		klog.Errorf("No cluster defined in GuaranteedUpdate action for key %s : %s", key, err.Error())
 	}
+	shardName := endpointsrequest.ShardFrom(ctx)
+
 	preparedKey, err := s.prepareKey(key)
 	if err != nil {
 		return err
@@ -473,7 +472,7 @@ func (s *store) GuaranteedUpdate(
 	}
 
 	skipTransformDecode := false
-	getCurrentState := s.getCurrentState(ctx, preparedKey, v, ignoreNotFound, skipTransformDecode, clusterName)
+	getCurrentState := s.getCurrentState(ctx, preparedKey, v, ignoreNotFound, skipTransformDecode, clusterName, shardName)
 
 	var origState *objState
 	var origStateIsCurrent bool
@@ -559,7 +558,7 @@ func (s *store) GuaranteedUpdate(
 			}
 			// recheck that the data from etcd is not stale before short-circuiting a write
 			if !origState.stale {
-				err = decode(s.codec, s.versioner, origState.data, destination, origState.rev, clusterName)
+				err = decode(s.codec, s.versioner, origState.data, destination, origState.rev, clusterName, shardName)
 				if err != nil {
 					recordDecodeError(s.groupResourceString, preparedKey)
 					return err
@@ -599,7 +598,7 @@ func (s *store) GuaranteedUpdate(
 		span.AddEvent("Transaction committed")
 		if !txnResp.Succeeded {
 			klog.V(4).Infof("GuaranteedUpdate of %s failed because of a conflict, going to retry", preparedKey)
-			origState, err = s.getState(ctx, txnResp.KV, preparedKey, v, ignoreNotFound, skipTransformDecode, clusterName)
+			origState, err = s.getState(ctx, txnResp.KV, preparedKey, v, ignoreNotFound, skipTransformDecode, clusterName, shardName)
 			if err != nil {
 				return err
 			}
@@ -608,7 +607,7 @@ func (s *store) GuaranteedUpdate(
 			continue
 		}
 
-		err = decode(s.codec, s.versioner, data, destination, txnResp.Revision, clusterName)
+		err = decode(s.codec, s.versioner, data, destination, txnResp.Revision, clusterName, shardName)
 		if err != nil {
 			span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error()))
 			recordDecodeError(s.groupResourceString, preparedKey)
@@ -757,7 +756,9 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 	if err != nil {
 		return storage.NewInternalError(fmt.Errorf("unable to get cluster for list key %q: %v", keyPrefix, err))
 	}
+	shard := endpointsrequest.ShardFrom(ctx)
 	crdIndicator := kcp.CustomResourceIndicatorFrom(ctx)
+	// end kcp
 
 	// loop until we have filled the requested limit from etcd or there are no more results
 	var lastKey []byte
@@ -835,8 +836,10 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
 			default:
 			}
 
-			clusterName := adjustClusterNameIfWildcard(cluster, keyPrefix, string(kv.Key))
-			obj, err := decodeListItem(ctx, data, uint64(kv.ModRevision), s.codec, s.versioner, newItemFunc, clusterName)
+			// kcp
+			clusterName := adjustClusterNameIfWildcard(shard, cluster, crdIndicator, keyPrefix, string(kv.Key))
+			shardName := adjustShardNameIfWildcard(shard, keyPrefix, string(kv.Key))
+			obj, err := decodeListItem(ctx, data, uint64(kv.ModRevision), s.codec, s.versioner, newItemFunc, clusterName, shardName)
 			if err != nil {
 				recordDecodeError(s.groupResourceString, string(kv.Key))
 				if done := aggregator.Aggregate(string(kv.Key), err); done {
@@ -967,15 +970,15 @@ func (s *store) watchContext(ctx context.Context) context.Context {
 	return clientv3.WithRequireLeader(ctx)
 }
 
-func (s *store) getCurrentState(ctx context.Context, key string, v reflect.Value, ignoreNotFound bool, skipTransformDecode bool, clusterName logicalcluster.Name) func() (*objState, error) {
+func (s *store) getCurrentState(ctx context.Context, key string, v reflect.Value, ignoreNotFound bool, skipTransformDecode bool, clusterName logicalcluster.Name, shardName endpointsrequest.Shard) func() (*objState, error) {
 	return func() (*objState, error) {
 		startTime := time.Now()
 		getResp, err := s.client.Kubernetes.Get(ctx, key, kubernetes.GetOptions{})
 		metrics.RecordEtcdRequest("get", s.groupResourceString, err, startTime)
 		if err != nil {
 			return nil, err
 		}
-		return s.getState(ctx, getResp.KV, key, v, ignoreNotFound, skipTransformDecode, clusterName)
+		return s.getState(ctx, getResp.KV, key, v, ignoreNotFound, skipTransformDecode, clusterName, shardName)
 	}
 }
 
@@ -985,7 +988,7 @@ func (s *store) getCurrentState(ctx context.Context, key string, v reflect.Value
 // storage will be transformed and decoded.
 // NOTE: when skipTransformDecode is true, the 'data', and the 'obj' fields
 // of the objState will be nil, and 'stale' will be set to true.
-func (s *store) getState(ctx context.Context, kv *mvccpb.KeyValue, key string, v reflect.Value, ignoreNotFound bool, skipTransformDecode bool, clusterName logicalcluster.Name) (*objState, error) {
+func (s *store) getState(ctx context.Context, kv *mvccpb.KeyValue, key string, v reflect.Value, ignoreNotFound bool, skipTransformDecode bool, clusterName logicalcluster.Name, shardName endpointsrequest.Shard) (*objState, error) {
 	state := &objState{
 		meta: &storage.ResponseMeta{},
 	}
@@ -1021,7 +1024,7 @@ func (s *store) getState(ctx context.Context, kv *mvccpb.KeyValue, key string, v
 
 		state.data = data
 		state.stale = stale
-		if err := decode(s.codec, s.versioner, state.data, state.obj, state.rev, clusterName); err != nil {
+		if err := decode(s.codec, s.versioner, state.data, state.obj, state.rev, clusterName, shardName); err != nil {
 			recordDecodeError(s.groupResourceString, key)
 			return nil, err
 		}
@@ -1117,7 +1120,7 @@ func (s *store) prepareKey(key string) (string, error) {
 
 // decode decodes value of bytes into object. It will also set the object resource version to rev.
 // On success, objPtr would be set to the object.
-func decode(codec runtime.Codec, versioner storage.Versioner, value []byte, objPtr runtime.Object, rev int64, clusterName logicalcluster.Name) error {
+func decode(codec runtime.Codec, versioner storage.Versioner, value []byte, objPtr runtime.Object, rev int64, clusterName logicalcluster.Name, shardName endpointsrequest.Shard) error {
 	if _, err := conversion.EnforcePtr(objPtr); err != nil {
 		return fmt.Errorf("unable to convert output object to pointer: %v", err)
 	}
@@ -1131,13 +1134,13 @@ func decode(codec runtime.Codec, versioner storage.Versioner, value []byte, objP
 	}
 
 	// kcp: apply clusterName to the decoded object, as the name is not persisted in storage.
-	annotateDecodedObjectWith(objPtr, clusterName)
+	annotateDecodedObjectWith(objPtr, clusterName, shardName)
 
 	return nil
 }
 
 // decodeListItem decodes bytes value in array into object.
-func decodeListItem(ctx context.Context, data []byte, rev uint64, codec runtime.Codec, versioner storage.Versioner, newItemFunc func() runtime.Object) (runtime.Object, error) {
+func decodeListItem(ctx context.Context, data []byte, rev uint64, codec runtime.Codec, versioner storage.Versioner, newItemFunc func() runtime.Object, clusterName logicalcluster.Name, shardName endpointsrequest.Shard) (runtime.Object, error) {
 	startedAt := time.Now()
 	defer func() {
 		endpointsrequest.TrackDecodeLatency(ctx, time.Since(startedAt))
@@ -1152,7 +1155,8 @@ func decodeListItem(ctx context.Context, data []byte, rev uint64, codec runtime.
 		klog.Errorf("failed to update object version: %v", err)
 	}
 
-	annotateDecodedObjectWith(obj, clusterName)
+	// kcp: apply clusterName and shardName to the decoded object, as the name is not persisted in storage.
+	annotateDecodedObjectWith(obj, clusterName, shardName)
 
 	return obj, nil
 }

staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_patch.go

@@ -28,9 +28,10 @@ import (
 
 // adjustClusterNameIfWildcard determines the logical cluster name. If this is not a cluster-wildcard list/watch request,
 // the cluster name is returned unmodified. Otherwise, the cluster name is extracted from the key based on whether it is
+// - a shard-wildcard request: <prefix>/shardName/clusterName/<remainder>
 // - CR partial metadata request: <prefix>/identity/clusterName/<remainder>
 // - any other request: <prefix>/clusterName/<remainder>.
-func adjustClusterNameIfWildcard(cluster *genericapirequest.Cluster, crdRequest bool, keyPrefix, key string) logicalcluster.Name {
+func adjustClusterNameIfWildcard(shard genericapirequest.Shard, cluster *genericapirequest.Cluster, crdRequest bool, keyPrefix, key string) logicalcluster.Name {
 	if !cluster.Wildcard {
 		return cluster.Name
 	}
@@ -51,6 +52,9 @@ func adjustClusterNameIfWildcard(cluster *genericapirequest.Cluster, crdRequest
 		// expecting 2699f4d273d342adccdc8a32663408226ecf66de7d191113ed3d4dc9bccec2f2/root:org:ws/<remainder>
 		// OR customresources/root:org:ws/<remainder>
 		return extract(3, 1)
+	case shard.Wildcard():
+		// expecting shardName/clusterName/<remainder>
+		return extract(3, 1)
 	default:
 		// expecting root:org:ws/<remainder>
 		return extract(2, 0)
@@ -101,7 +105,7 @@ func annotateDecodedObjectWith(obj interface{}, clusterName logicalcluster.Name,
 	}
 	annotations[logicalcluster.AnnotationKey] = clusterName.String()
 	if !shardName.Empty() {
-		annotations[genericapirequest.AnnotationKey] = shardName.String()
+		annotations[genericapirequest.ShardAnnotationKey] = shardName.String()
 	}
 	s.SetAnnotations(annotations)
 }