Make apiexportendpointslices consumer aware

Signed-off-by: Mangirdas Judeikis <[email protected]>

Makefile

@@ -336,6 +336,20 @@ test-e2e-sharded-minimal: build-all
 		$(SUITES_ARGS) \
 	$(if $(value WAIT),|| { echo "Terminated with $$?"; wait "$$PID"; },)
 
+# This is just easy target to run 2 shard test server locally until manually killed.
+# You can targer test to it by running:
+# go test ./test/e2e/apibinding/... --kcp-kubeconfig=$(pwd)/.kcp/admin.kubeconfig --shard-kubeconfigs=root=$(pwd)/.kcp-0/admin.kubeconfig -run=^TestAPIBindingEndpointSlicesSharded$
+test-run-sharded-server: WORK_DIR ?= .
+test-run-sharded-server: LOG_DIR ?= $(WORK_DIR)/.kcp
+test-run-sharded-server:
+	mkdir -p "$(LOG_DIR)" "$(WORK_DIR)/.kcp"
+	rm -f "$(WORK_DIR)/.kcp/ready-to-test"
+	UNSAFE_E2E_HACK_DISABLE_ETCD_FSYNC=true NO_GORUN=1 ./bin/sharded-test-server --quiet --v=2 --log-dir-path="$(LOG_DIR)" --work-dir-path="$(WORK_DIR)" --shard-run-virtual-workspaces=false --shard-feature-gates=$(TEST_FEATURE_GATES) $(TEST_SERVER_ARGS) --number-of-shards=2 2>&1 & PID=$$!; echo "PID $$PID" && \
+	trap 'kill -TERM $$PID' TERM INT EXIT && \
+	while [ ! -f "$(WORK_DIR)/.kcp/ready-to-test" ]; do sleep 1; done && \
+	echo 'Server started' && \
+	wait $$PID
+
 .PHONY: test
 ifdef USE_GOTESTSUM
 test: $(GOTESTSUM)

config/crds/apis.kcp.io_apiexportendpointslices.yaml

@@ -148,6 +148,15 @@ spec:
                   - url
                   type: object
                 type: array
+                x-kubernetes-list-map-keys:
+                - url
+                x-kubernetes-list-type: map
+              shardSelector:
+                description: |-
+                  shardSelector is the selector used to filter the shards. It is used to filter the shards
+                  when determining partition scope when deriving the endpoints. This is set by owning shard,
+                  and is used by follower shards to determine if its inscope or not.
+                type: string
             type: object
         type: object
     served: true

pkg/authorization/bootstrap/policy.go

@@ -24,6 +24,7 @@ import (
 	rbacrest "k8s.io/kubernetes/pkg/registry/rbac/rest"
 	"k8s.io/kubernetes/plugin/pkg/auth/authorizer/rbac/bootstrappolicy"
 
+	"github.com/kcp-dev/kcp/sdk/apis/apis"
 	"github.com/kcp-dev/kcp/sdk/apis/core"
 	"github.com/kcp-dev/kcp/sdk/apis/tenancy"
 )
@@ -101,6 +102,13 @@ func clusterRoles() []rbacv1.ClusterRole {
 				rbacv1helpers.NewRule("access").URLs("/").RuleOrDie(),
 			},
 		},
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: SystemExternalLogicalClusterAdmin},
+			Rules: []rbacv1.PolicyRule{
+				rbacv1helpers.NewRule("update", "patch", "get").Groups(apis.GroupName).Resources("apiexportendpointslices/status").RuleOrDie(),
+				rbacv1helpers.NewRule("get", "list", "watch").Groups(apis.GroupName).Resources("apiexportendpointslices").RuleOrDie(),
+			},
+		},
 	}
 }
 

pkg/indexers/apiexport.go

@@ -17,6 +17,8 @@ limitations under the License.
 package indexers
 
 import (
+	"fmt"
+
 	kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache"
 	"github.com/kcp-dev/logicalcluster/v3"
 
@@ -33,6 +35,8 @@ const (
 	// APIExportByClaimedIdentities is the indexer name for retrieving APIExports that have a permission claim for a
 	// particular identity hash.
 	APIExportByClaimedIdentities = "APIExportByClaimedIdentities"
+	// APIExportEndpointSliceByAPIExport is the indexer name for retrieving APIExportEndpointSlices by their APIExport's Reference Path and Name.
+	APIExportEndpointSliceByAPIExport = "APIExportEndpointSliceByAPIExport"
 )
 
 // IndexAPIExportByIdentity is an index function that indexes an APIExport by its identity hash.
@@ -72,3 +76,17 @@ func IndexAPIExportByClaimedIdentities(obj interface{}) ([]string, error) {
 	}
 	return sets.List[string](claimedIdentities), nil
 }
+
+// IndexAPIExportEndpointSliceByAPIExportFunc indexes the APIExportEndpointSlice by their APIExport's Reference Path and Name.
+func IndexAPIExportEndpointSliceByAPIExport(obj interface{}) ([]string, error) {
+	apiExportEndpointSlice, ok := obj.(*apisv1alpha1.APIExportEndpointSlice)
+	if !ok {
+		return []string{}, fmt.Errorf("obj %T is not an APIExportEndpointSlice", obj)
+	}
+
+	path := logicalcluster.NewPath(apiExportEndpointSlice.Spec.APIExport.Path)
+	if path.Empty() {
+		path = logicalcluster.From(apiExportEndpointSlice).Path()
+	}
+	return []string{path.Join(apiExportEndpointSlice.Spec.APIExport.Name).String()}, nil
+}

pkg/reconciler/apis/apiexportendpointslice/apiexportendpointslice_controller.go

@@ -45,7 +45,7 @@ import (
 	topologyv1alpha1 "github.com/kcp-dev/kcp/sdk/apis/topology/v1alpha1"
 	kcpclientset "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/cluster"
 	apisv1alpha1client "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/typed/apis/v1alpha1"
-	apisinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/apis/v1alpha1"
+	apisv1alpha1informers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/apis/v1alpha1"
 	corev1alpha1informers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/core/v1alpha1"
 	topologyinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/topology/v1alpha1"
 )
@@ -57,9 +57,9 @@ const (
 // NewController returns a new controller for APIExportEndpointSlices.
 // Shards and APIExports are read from the cache server.
 func NewController(
-	apiExportEndpointSliceClusterInformer apisinformers.APIExportEndpointSliceClusterInformer,
+	apiExportEndpointSliceClusterInformer apisv1alpha1informers.APIExportEndpointSliceClusterInformer,
 	globalShardClusterInformer corev1alpha1informers.ShardClusterInformer,
-	globalAPIExportClusterInformer apisinformers.APIExportClusterInformer,
+	globalAPIExportClusterInformer apisv1alpha1informers.APIExportClusterInformer,
 	partitionClusterInformer topologyinformers.PartitionClusterInformer,
 	kcpClusterClient kcpclientset.ClusterInterface,
 ) (*controller, error) {
@@ -76,8 +76,8 @@ func NewController(
 		listShards: func(selector labels.Selector) ([]*corev1alpha1.Shard, error) {
 			return globalShardClusterInformer.Lister().List(selector)
 		},
-		getAPIExportEndpointSlice: func(clusterName logicalcluster.Name, name string) (*apisv1alpha1.APIExportEndpointSlice, error) {
-			return apiExportEndpointSliceClusterInformer.Lister().Cluster(clusterName).Get(name)
+		getAPIExportEndpointSlice: func(ctx context.Context, path logicalcluster.Path, name string) (*apisv1alpha1.APIExportEndpointSlice, error) {
+			return indexers.ByPathAndName[*apisv1alpha1.APIExportEndpointSlice](apisv1alpha1.Resource("apiexportendpointslices"), apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), path, name)
 		},
 		getAPIExport: func(path logicalcluster.Path, name string) (*apisv1alpha1.APIExport, error) {
 			return indexers.ByPathAndName[*apisv1alpha1.APIExport](apisv1alpha1.Resource("apiexports"), globalAPIExportClusterInformer.Informer().GetIndexer(), path, name)
@@ -168,12 +168,12 @@ type controller struct {
 
 	listShards                            func(selector labels.Selector) ([]*corev1alpha1.Shard, error)
 	listAPIExportEndpointSlices           func() ([]*apisv1alpha1.APIExportEndpointSlice, error)
-	getAPIExportEndpointSlice             func(clusterName logicalcluster.Name, name string) (*apisv1alpha1.APIExportEndpointSlice, error)
+	getAPIExportEndpointSlice             func(ctx context.Context, path logicalcluster.Path, name string) (*apisv1alpha1.APIExportEndpointSlice, error)
 	getAPIExport                          func(path logicalcluster.Path, name string) (*apisv1alpha1.APIExport, error)
 	getPartition                          func(clusterName logicalcluster.Name, name string) (*topologyv1alpha1.Partition, error)
 	getAPIExportEndpointSlicesByPartition func(key string) ([]*apisv1alpha1.APIExportEndpointSlice, error)
 
-	apiExportEndpointSliceClusterInformer apisinformers.APIExportEndpointSliceClusterInformer
+	apiExportEndpointSliceClusterInformer apisv1alpha1informers.APIExportEndpointSliceClusterInformer
 	commit                                CommitFunc
 }
 
@@ -204,15 +204,15 @@ func (c *controller) enqueueAPIExportEndpointSlicesForAPIExport(obj interface{})
 	// binding keys by full path
 	keys := sets.New[string]()
 	if path := logicalcluster.NewPath(export.Annotations[core.LogicalClusterPathAnnotationKey]); !path.Empty() {
-		pathKeys, err := c.apiExportEndpointSliceClusterInformer.Informer().GetIndexer().IndexKeys(indexAPIExportEndpointSliceByAPIExport, path.Join(export.Name).String())
+		pathKeys, err := c.apiExportEndpointSliceClusterInformer.Informer().GetIndexer().IndexKeys(indexers.APIExportEndpointSliceByAPIExport, path.Join(export.Name).String())
 		if err != nil {
 			utilruntime.HandleError(err)
 			return
 		}
 		keys.Insert(pathKeys...)
 	}
 
-	clusterKeys, err := c.apiExportEndpointSliceClusterInformer.Informer().GetIndexer().IndexKeys(indexAPIExportEndpointSliceByAPIExport, logicalcluster.From(export).Path().Join(export.Name).String())
+	clusterKeys, err := c.apiExportEndpointSliceClusterInformer.Informer().GetIndexer().IndexKeys(indexers.APIExportEndpointSliceByAPIExport, logicalcluster.From(export).Path().Join(export.Name).String())
 	if err != nil {
 		utilruntime.HandleError(err)
 		return
@@ -328,7 +328,7 @@ func (c *controller) process(ctx context.Context, key string) error {
 		utilruntime.HandleError(err)
 		return nil
 	}
-	obj, err := c.getAPIExportEndpointSlice(clusterName, name)
+	obj, err := c.getAPIExportEndpointSlice(ctx, clusterName.Path(), name)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			return nil // object deleted before we handled it
@@ -353,6 +353,7 @@ func (c *controller) process(ctx context.Context, key string) error {
 	// If the object being reconciled changed as a result, update it.
 	oldResource := &Resource{ObjectMeta: old.ObjectMeta, Spec: &old.Spec, Status: &old.Status}
 	newResource := &Resource{ObjectMeta: obj.ObjectMeta, Spec: &obj.Spec, Status: &obj.Status}
+
 	if err := c.commit(ctx, oldResource, newResource); err != nil {
 		errs = append(errs, err)
 	}
@@ -380,15 +381,19 @@ func filterShardEvent(oldObj, newObj interface{}) bool {
 }
 
 // InstallIndexers adds the additional indexers that this controller requires to the informers.
-func InstallIndexers(globalAPIExportClusterInformer apisinformers.APIExportClusterInformer, apiExportEndpointSliceClusterInformer apisinformers.APIExportEndpointSliceClusterInformer) {
+func InstallIndexers(
+	globalAPIExportClusterInformer apisv1alpha1informers.APIExportClusterInformer,
+	apiExportEndpointSliceClusterInformer apisv1alpha1informers.APIExportEndpointSliceClusterInformer,
+) {
 	indexers.AddIfNotPresentOrDie(globalAPIExportClusterInformer.Informer().GetIndexer(), cache.Indexers{
 		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
 	})
-
 	indexers.AddIfNotPresentOrDie(apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), cache.Indexers{
-		indexAPIExportEndpointSliceByAPIExport: indexAPIExportEndpointSliceByAPIExportFunc,
+		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
+	})
+	indexers.AddIfNotPresentOrDie(apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.APIExportEndpointSliceByAPIExport: indexers.IndexAPIExportEndpointSliceByAPIExport,
 	})
-
 	indexers.AddIfNotPresentOrDie(apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), cache.Indexers{
 		indexAPIExportEndpointSlicesByPartition: indexAPIExportEndpointSlicesByPartitionFunc,
 	})

pkg/reconciler/apis/apiexportendpointslice/apiexportendpointslice_controller_test.go

@@ -79,7 +79,7 @@ func TestReconcile(t *testing.T) {
 			wantPartitionNotValid:               true,
 			wantAPIExportEndpointSliceURLsError: true,
 		},
-		"APIExportEndpointSliceURLs set when no issue": {
+		"APIExportEndpointSliceReadyForURLs set when no issue": {
 			wantAPIExportEndpointSliceURLsReady: true,
 			wantAPIExportValid:                  true,
 			wantPartitionValid:                  true,
@@ -189,7 +189,7 @@ func TestReconcile(t *testing.T) {
 			if tc.wantAPIExportEndpointSliceURLsError {
 				requireConditionMatches(t, apiExportEndpointSlice,
 					conditions.FalseCondition(
-						apisv1alpha1.APIExportEndpointSliceURLsReady,
+						apisv1alpha1.APIExportEndpointSliceReadyForURLs,
 						apisv1alpha1.ErrorGeneratingURLsReason,
 						conditionsv1alpha1.ConditionSeverityError,
 						"",
@@ -198,17 +198,13 @@ func TestReconcile(t *testing.T) {
 			}
 
 			if tc.wantAPIExportEndpointSliceURLsReady {
-				requireConditionMatches(t, apiExportEndpointSlice, conditions.TrueCondition(apisv1alpha1.APIExportEndpointSliceURLsReady))
-				require.Equal(t, []apisv1alpha1.APIExportEndpoint{
-					{URL: "https://server-1.kcp.dev/services/apiexport/root:org:ws/my-export"},
-					{URL: "https://server-2.kcp.dev/services/apiexport/root:org:ws/my-export"},
-				}, apiExportEndpointSlice.Status.APIExportEndpoints)
+				requireConditionMatches(t, apiExportEndpointSlice, conditions.TrueCondition(apisv1alpha1.APIExportEndpointSliceReadyForURLs))
 			}
 
 			if tc.wantAPIExportEndpointSliceURLsUnknown {
 				requireConditionMatches(t, apiExportEndpointSlice,
 					conditions.UnknownCondition(
-						apisv1alpha1.APIExportEndpointSliceURLsReady,
+						apisv1alpha1.APIExportEndpointSliceReadyForURLs,
 						apisv1alpha1.ErrorGeneratingURLsReason,
 						"",
 					),

pkg/reconciler/apis/apiexportendpointslice/apiexportendpointslice_indexes.go

@@ -26,24 +26,9 @@ import (
 )
 
 const (
-	indexAPIExportEndpointSliceByAPIExport  = "indexAPIExportEndpointSliceByAPIExport"
 	indexAPIExportEndpointSlicesByPartition = "indexAPIExportEndpointSlicesByPartition"
 )
 
-// indexAPIExportEndpointSliceByAPIExportFunc indexes the APIExportEndpointSlice by their APIExport's Reference Path and Name.
-func indexAPIExportEndpointSliceByAPIExportFunc(obj interface{}) ([]string, error) {
-	apiExportEndpointSlice, ok := obj.(*apisv1alpha1.APIExportEndpointSlice)
-	if !ok {
-		return []string{}, fmt.Errorf("obj %T is not an APIExportEndpointSlice", obj)
-	}
-
-	path := logicalcluster.NewPath(apiExportEndpointSlice.Spec.APIExport.Path)
-	if path.Empty() {
-		path = logicalcluster.From(apiExportEndpointSlice).Path()
-	}
-	return []string{path.Join(apiExportEndpointSlice.Spec.APIExport.Name).String()}, nil
-}
-
 // indexAPIExportEndpointSlicesByPartitionFunc is an index function that maps a Partition to the key for its
 // spec.partition.
 func indexAPIExportEndpointSlicesByPartitionFunc(obj interface{}) ([]string, error) {