syncing resources from the same apiexport works

On-behalf-of: @SAP [email protected]

Author: xrstf

internal/controller/apiexport/controller.go

@@ -130,15 +130,22 @@ func (r *Reconciler) reconcile(ctx context.Context, apiExport *kcpdevv1alpha1.AP
 		return fmt.Errorf("failed to list PublishedResources: %w", err)
 	}
 
-	// filter out those PRs that have not yet been processed into an ARS or that are invalid
+	// filter out those PRs that are invalid; we keep those that are not yet converted into ARS,
+	// just to reduce the amount of re-reconciles when the agent processes a number of PRs in a row
+	// and would constantly update the APIExport; instead we rely on kcp to handle the eventual
+	// consistency.
+	// This allows us to already determine all resources we "own", which helps us in
+	// bilding the proper permission claims if one of the related resources is using a resource
+	// type managed via PublishedResource. Otherwise the controller might not see the PR for the
+	// related resource and temporarily incorrectly assume it needs to add a permission claim.
 	filteredPubResources := slices.DeleteFunc(pubResources.Items, func(pr syncagentv1alpha1.PublishedResource) bool {
 		// TODO: Turn this into a webhook or CEL expressions.
-		if err := validation.ValidatePublishedResource(&pr); err != nil {
+		err := validation.ValidatePublishedResource(&pr)
+		if err != nil {
 			r.log.With("pr", pr.Name, "error", err).Warn("Ignoring invalid PublishedResource.")
-			return true
 		}
 
-		return pr.Status.ResourceSchemaName == ""
+		return err != nil
 	})
 
 	// for each PR, we note down the created ARS and also the GVKs of related resources

internal/sync/syncer_related.go

@@ -81,6 +81,8 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 		eventObjSide = syncSideSource
 	}
 
+	fmt.Println("processRelatedResource()")
+
 	// find the all objects on the origin side that match the given criteria
 	resolvedObjects, err := resolveRelatedResourceObjects(ctx, origin, dest, relRes)
 	if err != nil {
@@ -89,6 +91,7 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 
 	// no objects were found yet, that's okay
 	if len(resolvedObjects) == 0 {
+		fmt.Println("stop: no objects found")
 		return false, nil
 	}
 
@@ -101,12 +104,18 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 
 	// Synchronize related objects the same way the parent object was synchronized.
 	projectedGVR := projection.RelatedResourceProjectedGVR(&relRes)
-	// projectedAPIVersion, projectedKind := projection.RelatedResourceProjectedGVR(&relRes).ToAPIVersionAndKind()
+
+	projectedGVK, err := dest.client.RESTMapper().KindFor(projectedGVR)
+	if err != nil {
+		return false, fmt.Errorf("failed to lookup %v: %w", projectedGVR, err)
+	}
+
+	fmt.Printf("projectedGVK: %+v\n", projectedGVK)
 
 	for idx, resolved := range resolvedObjects {
 		destObject := &unstructured.Unstructured{}
-		destObject.SetAPIVersion(projectedGVR.GroupVersion().String())
-		destObject.SetKind(projectedGVR.Resource)
+		destObject.SetAPIVersion(projectedGVK.GroupVersion().String())
+		destObject.SetKind(projectedGVK.Kind)
 
 		if err = dest.client.Get(ctx, resolved.destination, destObject); err != nil {
 			destObject = nil
@@ -136,8 +145,8 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 			destCreator: func(source *unstructured.Unstructured) (*unstructured.Unstructured, error) {
 				fmt.Printf("source: %+v\n", *source)
 				dest := source.DeepCopy()
-				dest.SetAPIVersion(projectedGVR.GroupVersion().String())
-				dest.SetKind(projectedGVR.Resource)
+				dest.SetAPIVersion(projectedGVK.GroupVersion().String())
+				dest.SetKind(projectedGVK.Kind)
 				dest.SetName(resolved.destination.Name)
 				dest.SetNamespace(resolved.destination.Namespace)
 
@@ -358,6 +367,8 @@ func mapSlices(a, b []string) map[string]string {
 func resolveRelatedResourceObjectsInNamespaces(ctx context.Context, relatedOrigin, relatedDest syncSide, relRes syncagentv1alpha1.RelatedResourceSpec, spec syncagentv1alpha1.RelatedResourceObjectSpec, namespaceMap map[string]string) ([]resolvedObject, error) {
 	result := []resolvedObject{}
 
+	fmt.Printf("resolveRelatedResourceObjectsInNamespaces(%+v)\n", namespaceMap)
+
 	for originNamespace, destNamespace := range namespaceMap {
 		nameMap, err := resolveRelatedResourceObjectsInNamespace(ctx, relatedOrigin, relatedDest, relRes, spec, originNamespace)
 		if err != nil {
@@ -366,14 +377,19 @@ func resolveRelatedResourceObjectsInNamespaces(ctx context.Context, relatedOrigi
 
 		for originName, destName := range nameMap {
 			originGVR := projection.RelatedResourceGVR(&relRes)
-			// apiVersion := schema.GroupVersion{Group: relRes.Group, Version: relRes.Version}.String()
+
+			originGVK, err := relatedOrigin.client.RESTMapper().KindFor(originGVR)
+			if err != nil {
+				return nil, fmt.Errorf("failed to lookup %v: %w", originGVR, err)
+			}
 
 			originObj := &unstructured.Unstructured{}
-			originObj.SetAPIVersion(originGVR.GroupVersion().String())
-			originObj.SetKind(originGVR.Resource)
+			originObj.SetAPIVersion(originGVK.GroupVersion().String())
+			originObj.SetKind(originGVK.Kind)
 
 			err = relatedOrigin.client.Get(ctx, types.NamespacedName{Name: originName, Namespace: originNamespace}, originObj)
 			if err != nil {
+				fmt.Printf("err: %v\n", err)
 				// this should rarely happen, only if an object was deleted in between the .List() call
 				// above and the .Get() call here.
 				if apierrors.IsNotFound(err) {

test/e2e/sync/related_test.go

@@ -1048,6 +1048,10 @@ func TestSyncNonStandardRelatedResources(t *testing.T) {
 			name:      "turn a related ConfigMap into a Secret",
 			workspace: "sync-related-configmap-to-secret",
 			mainResource: crds.Crontab{
+				TypeMeta: metav1.TypeMeta{
+					APIVersion: "kcp.example.com/v1",
+					Kind:       "CronTab",
+				},
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "my-crontab",
 					Namespace: "default",
@@ -1107,6 +1111,10 @@ func TestSyncNonStandardRelatedResources(t *testing.T) {
 			name:      "use a resource from the same APIExport as a related resource",
 			workspace: "sync-related-same-apiexport",
 			mainResource: crds.Crontab{
+				TypeMeta: metav1.TypeMeta{
+					APIVersion: "kcp.example.com/v1",
+					Kind:       "CronTab",
+				},
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "my-crontab",
 					Namespace: "default",
@@ -1117,7 +1125,7 @@ func TestSyncNonStandardRelatedResources(t *testing.T) {
 				},
 			},
 			relatedConfig: syncagentv1alpha1.RelatedResourceSpec{
-				Identifier: "credentials",
+				Identifier: "the-backup",
 				Origin:     syncagentv1alpha1.RelatedResourceOriginService,
 				Resource:   "backups",
 				Group:      "eksempel.no",
@@ -1135,13 +1143,21 @@ func TestSyncNonStandardRelatedResources(t *testing.T) {
 				},
 			},
 			sourceRelatedObject: &crds.Backup{
+				TypeMeta: metav1.TypeMeta{
+					APIVersion: "eksempel.no/v1",
+					Kind:       "Backup",
+				},
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "my-backup",
-					Namespace: "default",
+					Namespace: "synced-default",
 				},
 				Spec: crds.BackupSpec{},
 			},
 			expectedSyncedRelatedObject: &crds.Backup{
+				TypeMeta: metav1.TypeMeta{
+					APIVersion: "kcp.example.com/v1",
+					Kind:       "Backup",
+				},
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "my-backup",
 					Namespace: "default",
@@ -1669,14 +1685,11 @@ func TestSyncNonStandardRelatedResources(t *testing.T) {
 			t.Log("Creating CronTab in kcp…")
 
 			crontab := utils.ToUnstructured(t, &testcase.mainResource)
-			crontab.SetAPIVersion("kcp.example.com/v1")
-			crontab.SetKind("CronTab")
-
 			if err := teamClient.Create(ctx, crontab); err != nil {
 				t.Fatalf("Failed to create CronTab in kcp: %v", err)
 			}
 
-			// fake operator: create a credential Secret
+			// fake operator: create a related object as a result of processing the main object
 			t.Logf("Creating original related object on the %s side…", testcase.relatedConfig.Origin)
 
 			originClient := envtestClient
@@ -1688,20 +1701,28 @@ func TestSyncNonStandardRelatedResources(t *testing.T) {
 
 			ensureNamespace(t, ctx, originClient, testcase.sourceRelatedObject.GetNamespace())
 
-			if err := originClient.Create(ctx, testcase.sourceRelatedObject); err != nil {
+			relatedObject := utils.ToUnstructured(t, &testcase.sourceRelatedObject)
+			if err := originClient.Create(ctx, relatedObject); err != nil {
 				t.Fatalf("Failed to create related object: %v", err)
 			}
 
 			// wait for the agent to do its magic
 			t.Log("Wait for related object to be synced…")
 			projectedGVR := projection.RelatedResourceProjectedGVR(&testcase.relatedConfig)
+			projectedGVK, err := destClient.RESTMapper().KindFor(projectedGVR)
+			if err != nil {
+				t.Fatalf("Failed to resolve projected GVR %v: %v", projectedGVR, err)
+			}
+
 			copiedRelatedObject := &unstructured.Unstructured{}
-			copiedRelatedObject.SetAPIVersion(projectedGVR.GroupVersion().String())
-			copiedRelatedObject.SetKind(projectedGVR.Resource)
+			copiedRelatedObject.SetAPIVersion(projectedGVK.GroupVersion().String())
+			copiedRelatedObject.SetKind(projectedGVK.Kind)
 
-			err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (done bool, err error) {
+			err = wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (done bool, err error) {
 				copyKey := ctrlruntimeclient.ObjectKeyFromObject(testcase.expectedSyncedRelatedObject)
-				return destClient.Get(ctx, copyKey, copiedRelatedObject) == nil, nil
+				err = destClient.Get(ctx, copyKey, copiedRelatedObject)
+				fmt.Printf("poll err: %v\n", err)
+				return err == nil, nil
 			})
 			if err != nil {
 				t.Fatalf("Failed to wait for related object to be synced: %v", err)