Merge pull request #955 from ncdc/e2e-pcluster-kubeconfig

Towards more stable syncer

Author: sttts

pkg/reconciler/workload/namespace/namespace_controller.go

@@ -32,7 +32,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
 	coreinformers "k8s.io/client-go/informers/core/v1"
@@ -392,11 +391,7 @@ func (c *Controller) processCluster(ctx context.Context, key string) error {
 		// controller. Rescheduling will always happen eventually due
 		// to namespace informer resync.
 
-		clusterName, err := request.ClusterNameFrom(ctx)
-		if err != nil {
-			klog.Errorf("Unable to enqueue namespaces related to WorkloadCluster %s: %v", key, err)
-			return nil
-		}
+		clusterName, _ := clusters.SplitClusterAwareKey(key)
 
 		return c.enqueueNamespaces(clusterName, labels.Everything())
 	} else if err != nil {

pkg/reconciler/workload/namespace/namespace_reconcile.go

@@ -127,8 +127,13 @@ func (c *Controller) reconcileResource(ctx context.Context, lclusterName logical
 	}
 
 	old, new := lbls[ClusterLabel], ns.Labels[ClusterLabel]
+
+	klog.Infof("ANDY comparing %s %s|%s/%s clusterlabel %q to ns clusterlabel %q",
+		gvr.String(), lclusterName, unstr.GetNamespace(), unstr.GetName(), old, new)
+
 	if old == new {
 		// Already assigned to the right cluster.
+		klog.Infof("ANDY no-op %s %s|%s/%s", gvr.String(), lclusterName, unstr.GetNamespace(), unstr.GetName())
 		return nil
 	}
 
@@ -167,7 +172,7 @@ func (c *Controller) reconcileGVR(ctx context.Context, gvr schema.GroupVersionRe
 // will succeed without error if a cluster is assigned or if there are no viable clusters
 // to assign to. The condition of not being scheduled to a cluster will be reflected in
 // the namespace's status rather than by returning an error.
-func (c *Controller) ensureScheduled(ctx context.Context, ns *corev1.Namespace) error {
+func (c *Controller) ensureScheduled(ctx context.Context, ns *corev1.Namespace) (*corev1.Namespace, error) {
 	oldPClusterName := ns.Labels[ClusterLabel]
 
 	scheduler := namespaceScheduler{
@@ -176,47 +181,46 @@ func (c *Controller) ensureScheduled(ctx context.Context, ns *corev1.Namespace)
 	}
 	newPClusterName, err := scheduler.AssignCluster(ns)
 	if err != nil {
-		return err
+		return ns, err
 	}
 
 	if oldPClusterName == newPClusterName {
-		return nil
+		return ns, nil
 	}
 
 	klog.Infof("Patching to update cluster assignment for namespace %s|%s: %s -> %s",
 		ns.ClusterName, ns.Name, oldPClusterName, newPClusterName)
 	patchType, patchBytes := clusterLabelPatchBytes(newPClusterName)
 	patchedNamespace, err := c.kubeClient.Cluster(logicalcluster.From(ns)).CoreV1().Namespaces().
 		Patch(ctx, ns.Name, patchType, patchBytes, metav1.PatchOptions{})
-	if err == nil {
-		// Update the label to enable the caller to detect a scheduling change.
-		ns.Labels[ClusterLabel] = patchedNamespace.Labels[ClusterLabel]
+	if err != nil {
+		return ns, err
 	}
 
-	return err
+	return patchedNamespace, nil
 }
 
 // ensureScheduledStatus ensures the status of the given namespace reflects the
 // namespace's scheduled state.
-func (c *Controller) ensureScheduledStatus(ctx context.Context, ns *corev1.Namespace) error {
+func (c *Controller) ensureScheduledStatus(ctx context.Context, ns *corev1.Namespace) (*corev1.Namespace, error) {
 	updatedNs := setScheduledCondition(ns)
 
 	if equality.Semantic.DeepEqual(ns.Status, updatedNs.Status) {
-		return nil
+		return ns, nil
 	}
 
 	patchBytes, err := statusPatchBytes(ns, updatedNs)
 	if err != nil {
-		return err
+		return ns, err
 	}
 
-	_, err = c.kubeClient.Cluster(logicalcluster.From(ns)).CoreV1().Namespaces().
+	patchedNamespace, err := c.kubeClient.Cluster(logicalcluster.From(ns)).CoreV1().Namespaces().
 		Patch(ctx, ns.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{}, "status")
 	if err != nil {
-		return fmt.Errorf("failed to patch status on namespace %s|%s: %w", ns.ClusterName, ns.Name, err)
+		return ns, fmt.Errorf("failed to patch status on namespace %s|%s: %w", ns.ClusterName, ns.Name, err)
 	}
 
-	return nil
+	return patchedNamespace, nil
 }
 
 // reconcileNamespace is responsible for assigning a namespace to a cluster, if
@@ -240,11 +244,13 @@ func (c *Controller) reconcileNamespace(ctx context.Context, lclusterName logica
 		ns.Labels = map[string]string{}
 	}
 
-	if err := c.ensureScheduled(ctx, ns); err != nil {
+	ns, err = c.ensureScheduled(ctx, ns)
+	if err != nil {
 		return err
 	}
 
-	if err := c.ensureScheduledStatus(ctx, ns); err != nil {
+	ns, err = c.ensureScheduledStatus(ctx, ns)
+	if err != nil {
 		return err
 	}
 
@@ -259,6 +265,7 @@ func (c *Controller) reconcileNamespace(ctx context.Context, lclusterName logica
 func (c *Controller) enqueueResourcesForNamespace(ns *corev1.Namespace) error {
 	lastScheduling, previouslyEnqueued := c.namepaceContentsEnqueuedFor(ns)
 	if previouslyEnqueued && lastScheduling == ns.Labels[ClusterLabel] {
+		klog.Infof("ANDY enqueueResourcesForNamespace short-circuiting %s|%s", logicalcluster.From(ns), ns.Name)
 		return nil
 	}
 

pkg/syncer/statussyncer.go

@@ -91,10 +91,10 @@ func (c *Controller) updateStatusInUpstream(ctx context.Context, gvr schema.Grou
 
 	upstreamObj.SetResourceVersion(existing.GetResourceVersion())
 	if _, err := c.toClient.Resource(gvr).Namespace(upstreamNamespace).UpdateStatus(ctx, upstreamObj, metav1.UpdateOptions{}); err != nil {
-		klog.Errorf("Failed updating status of resource %s|%s/%s from pcluster namespace %s: %v", c.upstreamClusterName, upstreamNamespace, upstreamObj.GetName(), downstreamObj.GetNamespace(), err)
+		klog.Errorf("Failed updating status of resource %q %s|%s/%s from pcluster namespace %s: %v", gvr.String(), c.upstreamClusterName, upstreamNamespace, upstreamObj.GetName(), downstreamObj.GetNamespace(), err)
 		return err
 	}
-	klog.Infof("Updated status of resource %s|%s/%s from pcluster namespace %s", c.upstreamClusterName, upstreamNamespace, upstreamObj.GetName(), downstreamObj.GetNamespace())
+	klog.Infof("Updated status of resource %q %s|%s/%s from pcluster namespace %s", gvr.String(), c.upstreamClusterName, upstreamNamespace, upstreamObj.GetName(), downstreamObj.GetNamespace())
 
 	return nil
 }

test/e2e/framework/fixture.go

@@ -32,18 +32,23 @@ import (
 	"github.com/kcp-dev/apimachinery/pkg/logicalcluster"
 	"github.com/stretchr/testify/require"
 
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	kubernetesclientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/klog/v2"
 
+	apiresourcev1alpha1 "github.com/kcp-dev/kcp/pkg/apis/apiresource/v1alpha1"
 	tenancyv1alpha1 "github.com/kcp-dev/kcp/pkg/apis/tenancy/v1alpha1"
 	kcpclient "github.com/kcp-dev/kcp/pkg/client/clientset/versioned"
 	kcpclientset "github.com/kcp-dev/kcp/pkg/client/clientset/versioned"
@@ -183,6 +188,10 @@ func LogToConsoleEnvSet() bool {
 	return inProcess
 }
 
+func preserveTestResources() bool {
+	return os.Getenv("PRESERVE") != ""
+}
+
 func NewOrganizationFixture(t *testing.T, server RunningServer) (orgClusterName logicalcluster.LogicalCluster) {
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	t.Cleanup(cancelFunc)
@@ -202,6 +211,10 @@ func NewOrganizationFixture(t *testing.T, server RunningServer) (orgClusterName
 	require.NoError(t, err, "failed to create organization workspace")
 
 	t.Cleanup(func() {
+		if preserveTestResources() {
+			return
+		}
+
 		ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))
 		defer cancelFn()
 
@@ -257,6 +270,10 @@ func NewWorkspaceWithWorkloads(t *testing.T, server RunningServer, orgClusterNam
 	require.NoError(t, err, "failed to create workspace")
 
 	t.Cleanup(func() {
+		if preserveTestResources() {
+			return
+		}
+
 		ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))
 		defer cancelFn()
 
@@ -385,35 +402,86 @@ func (sf SyncerFixture) Start(t *testing.T) *StartedSyncerFixture {
 	downstreamKubeClient, err := kubernetesclientset.NewForConfig(downstreamConfig)
 	require.NoError(t, err)
 
-	if useDeployedSyncer {
-		// Ensure cleanup of pcluster resources
+	artifactDir, err := CreateTempDirForTest(t, "artifacts")
+	if err != nil {
+		t.Errorf("failed to create temp dir for syncer artifacts: %v", err)
+	}
+
+	// collect both in deployed and in-process mode
+	t.Cleanup(func() {
+		ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))
+		defer cancelFn()
+
+		t.Logf("Collecting imported resource info")
+		upstreamCfg := sf.UpstreamServer.DefaultConfig(t)
+
+		gather := func(client dynamic.Interface, gvr schema.GroupVersionResource) {
+			resourceClient := client.Resource(gvr)
+
+			t.Logf("gathering %q", gvr)
+			list, err := resourceClient.List(ctx, metav1.ListOptions{})
+			if err != nil {
+				// Don't fail the test
+				t.Logf("Error gathering %s: %v", gvr, err)
+				return
+			}
 
+			t.Logf("got %d items", len(list.Items))
+
+			for i := range list.Items {
+				item := list.Items[i]
+				sf.UpstreamServer.Artifact(t, func() (runtime.Object, error) {
+					return &item, nil
+				})
+			}
+		}
+
+		upstreamDynamic, err := dynamic.NewClusterForConfig(upstreamCfg)
+		require.NoError(t, err, "error creating upstream dynamic client")
+
+		downstreamDynamic, err := dynamic.NewForConfig(downstreamConfig)
+		require.NoError(t, err, "error creating downstream dynamic client")
+
+		gather(upstreamDynamic.Cluster(sf.WorkspaceClusterName), apiresourcev1alpha1.SchemeGroupVersion.WithResource("apiresourceimports"))
+		gather(upstreamDynamic.Cluster(sf.WorkspaceClusterName), apiresourcev1alpha1.SchemeGroupVersion.WithResource("negotiatedapiresources"))
+		gather(upstreamDynamic.Cluster(sf.WorkspaceClusterName), corev1.SchemeGroupVersion.WithResource("namespaces"))
+		gather(downstreamDynamic, corev1.SchemeGroupVersion.WithResource("namespaces"))
+		gather(upstreamDynamic.Cluster(sf.WorkspaceClusterName), appsv1.SchemeGroupVersion.WithResource("deployments"))
+		gather(downstreamDynamic, appsv1.SchemeGroupVersion.WithResource("deployments"))
+	})
+
+	if useDeployedSyncer {
 		syncerID := syncerConfig.ID()
 		t.Cleanup(func() {
-			if useDeployedSyncer {
-				t.Logf("Collecting syncer %s logs", syncerID)
-
-				ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))
-				defer cancelFn()
+			ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))
+			defer cancelFn()
 
+			// collect syncer logs
+			t.Logf("Collecting syncer pod logs")
+			func() {
+				t.Logf("Listing downstream pods in namespace %s", syncerID)
 				pods, err := downstreamKubeClient.CoreV1().Pods(syncerID).List(ctx, metav1.ListOptions{})
 				if err != nil {
-					t.Errorf("failed to list pods in %s: %v", syncerID, err)
-				}
-				artifactDir, err := CreateTempDirForTest(t, "artifacts")
-				if err != nil {
-					t.Errorf("failed to create temp dir for syncer artifacts: %v", err)
+					t.Logf("failed to list pods in %s: %v", syncerID, err)
+					return
 				}
+
 				for _, pod := range pods.Items {
 					t.Logf("Collecting downstream logs for pod %s/%s", syncerID, pod.Name)
 					logs := Kubectl(t, downstreamKubeconfigPath, "-n", syncerID, "logs", pod.Name)
+
 					artifactPath := filepath.Join(artifactDir, fmt.Sprintf("syncer-%s-%s.log", syncerID, pod.Name))
+
 					err = ioutil.WriteFile(artifactPath, logs, 0644)
 					if err != nil {
 						t.Logf("failed to write logs for pod %s in %s to %s: %v", pod.Name, syncerID, artifactPath, err)
 						continue // not fatal
 					}
 				}
+			}()
+
+			if preserveTestResources() {
+				return
 			}
 
 			t.Logf("Deleting syncer resources for logical cluster %q, workload cluster %q", syncerConfig.KCPClusterName, syncerConfig.WorkloadClusterName)
@@ -455,10 +523,8 @@ func (sf SyncerFixture) Start(t *testing.T) *StartedSyncerFixture {
 				}
 			}
 		})
-
 	} else {
 		// Start an in-process syncer
-
 		err := syncer.StartSyncer(ctx, syncerConfig, 2, 5*time.Second)
 		require.NoError(t, err, "syncer failed to start")
 	}

test/e2e/framework/util.go

@@ -40,10 +40,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer/json"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/client-go/discovery"
-	cacheddiscovery "k8s.io/client-go/discovery/cached"
 	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/restmapper"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 
 	apiresourcev1alpha1 "github.com/kcp-dev/kcp/pkg/apis/apiresource/v1alpha1"
@@ -197,16 +194,14 @@ func artifact(t *testing.T, server RunningServer, producer func() (runtime.Objec
 		gvk := gvks[0]
 		data.GetObjectKind().SetGroupVersionKind(gvk)
 
-		cfg := server.DefaultConfig(t) // TODO(sttts): this doesn't make sense: discovery from a random workspace
-		discoveryClient, err := discovery.NewDiscoveryClientForConfig(cfg)
-		require.NoError(t, err, "could not get discovery client for server")
+		group := gvk.Group
+		if group == "" {
+			group = "core"
+		}
 
-		scopedDiscoveryClient := discoveryClient.WithCluster(logicalcluster.From(accessor))
-		mapper := restmapper.NewDeferredDiscoveryRESTMapper(cacheddiscovery.NewMemCacheClient(scopedDiscoveryClient))
-		mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
-		require.NoError(t, err, "could not get REST mapping for artifact's GVK")
+		gvkForFilename := fmt.Sprintf("%s_%s", group, gvk.Kind)
 
-		file := path.Join(dir, fmt.Sprintf("%s_%s.yaml", mapping.Resource.GroupResource().String(), accessor.GetName()))
+		file := path.Join(dir, fmt.Sprintf("%s-%s.yaml", gvkForFilename, accessor.GetName()))
 		t.Logf("saving artifact to %s", file)
 
 		serializer := json.NewSerializerWithOptions(json.DefaultMetaFactory, scheme.Scheme, scheme.Scheme, json.SerializerOptions{Yaml: true})