Repository: kubernetes-sigs/kubefed (go)

Issue:
repeated sync/status controller initialization
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!-->


**What happened**:
When one `federatedTypeConfig` is created with expectation of enabled sync and status controllers, I found sync/status controller will be initialized several times.
```
I0315 08:41:51.911705       1 controller.go:106] Starting sync controller for "FederatedService"
I0315 08:41:52.102628       1 controller.go:106] Starting sync controller for "FederatedService"
I0315 08:41:52.307827       1 controller.go:96] Starting status controller for "FederatedService"
I0315 08:41:52.331943       1 controller.go:106] Starting sync controller for "FederatedService"
I0315 08:41:52.420020       1 controller.go:96] Starting status controller for "FederatedService"
I0315 08:41:52.438640       1 controller.go:96] Starting status controller for "FederatedService"
```

**What you expected to happen**:
Each sync/status controller should be initialized only once.

**How to reproduce it (as minimally and precisely as possible)**:
First creating one `federatedTypeConfig` with enabled sync/status controllers.

<!-- DO NOT EDIT BELOW THIS LINE -->
/kind bug

Pull request description:
make objects in workqueue "comparable" to avoid multiple reconciliations on one object at same time
**What this PR does / why we need it**:
As we know, `workqueue` is using a map [dirty map[interface{}]struct{}](https://github.com/kubernetes/client-go/blob/f6ce18ae578c8cca64d14ab9687824d9e1305a67/util/workqueue/queue.go#L78) to deduplicate objects and prevent same object(with same namespace and name) to be reconciled by multiple workers at same time.
This means that the objects enqueued must be *comparable*.
But kubefed is just enqueuing objects of type `*DelayingDelivererItem` which will never be equal (pointer values with different addresses), [https://github.com/kubernetes-sigs/kubefed/blob/16b2f5cc4f05c9e9b696970295f353c56d33aa3a/pkg/controller/util/worker.go#L119-L121](https://github.com/kubernetes-sigs/kubefed/blob/16b2f5cc4f05c9e9b696970295f353c56d33aa3a/pkg/controller/util/worker.go#L119-L121).
So, even `federatedTypeConfig` controller has [a global locker](https://github.com/kubernetes-sigs/kubefed/blob/16b2f5cc4f05c9e9b696970295f353c56d33aa3a/pkg/controller/federatedtypeconfig/controller.go#L54-L55) for sync/status controllers, sync/status controllers will still be initialized several times.

Here is a timeline,
1. `federatedService` reconciliation starts in worker1
2. add finalizer to `federatedService` brings an update event
3. `federatedService` reconciliation starts in worker2 (workqueue regards this as a new object because they are not equal)
4. worker1 checks sync controller map(not found) and create one
5. worker2 checks sync controller map(not found) and create another
6. worker1 fills the stop channel to map
7. worker2 fills another stop channel to map(the first created sync controller will lose control)
....


**Which issue(s) this PR fixes**:
Fixes #1375 

**Special notes for your reviewer**:
1. "comparable" here means objects should be identical and equal with same namespace and name
2. precheck (type cast) object type before enqueuing and after getting
3. move `reconcile logic` into a new function to call `Done` in defer for more returns

Signed-off-by: Bruce Ma <brucema19901024@gmail.com>


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1009cde1..f7100397 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,29 +1,5 @@
 # Unreleased
 
-# v0.7.0
-- [#1380](https://github.com/kubernetes-sigs/kubefed/pull/1380) 
-  fix: infinite reconciliation loop.
-- [#1385](https://github.com/kubernetes-sigs/kubefed/pull/1385) 
-  fix: replica rebalance reconciliation.
-- [#1377](https://github.com/kubernetes-sigs/kubefed/pull/1377) 
-  feat: add a proxy url field to kubefed clusters.
-- [#1382](https://github.com/kubernetes-sigs/kubefed/pull/1382) 
-  fix: register workqueue metrics in controller-runtime firstly.
-- [#1371](https://github.com/kubernetes-sigs/kubefed/pull/1371) 
-  chore: enable workqueue metrics for all controllers.
-- [#1360](https://github.com/kubernetes-sigs/kubefed/pull/1360) 
-  chore: remove FederatedIngress feature.
-- [#1367](https://github.com/kubernetes-sigs/kubefed/pull/1367) 
-  fix: webhook command.
-- [#1361](https://github.com/kubernetes-sigs/kubefed/pull/1361) 
-  feature: kubefedcluster use cadata in kubeconfig file.
-- [#1355](https://github.com/kubernetes-sigs/kubefed/pull/1355) 
-  feature: support DeleteOptions when deleting resources in member clusters.
-- [#1357](https://github.com/kubernetes-sigs/kubefed/pull/1357) 
-  chore: Upgrade dependencies.
-- [#1351](https://github.com/kubernetes-sigs/kubefed/pull/1351) 
-  chore: remove feature CrossClusterServiceDiscovery.
-
 # v0.6.1
 - [#1346](https://github.com/kubernetes-sigs/kubefed/pull/1346)
   fix: upgrade path broken from older versions than v0.6.0.
diff --git a/charts/index.yaml b/charts/index.yaml
index 880ebd74..a44b495c 100644
--- a/charts/index.yaml
+++ b/charts/index.yaml
@@ -1,19 +1,6 @@
 apiVersion: v1
 entries:
   kubefed:
-  - apiVersion: v2
-    created: "2021-03-25T18:38:56.281653494+01:00"
-    dependencies:
-    - condition: controllermanager.enabled
-      name: controllermanager
-      repository: https://localhost/
-      version: 0.7.0
-    description: KubeFed helm chart
-    digest: 83c78f3d7f0609b258fb8b5a49958b4eccb6de59774e9a09ac914f18e51c9d26
-    name: kubefed
-    urls:
-    - https://github.com/kubernetes-sigs/kubefed/releases/download/v0.7.0/kubefed-0.7.0.tgz
-    version: 0.7.0
   - apiVersion: v2
     created: "2021-01-25T16:48:21.652921408+01:00"
     dependencies:
@@ -169,4 +156,4 @@ entries:
     urls:
     - https://github.com/kubernetes-sigs/kubefed/releases/download/v0.1.0-rc1/kubefed-0.1.0-rc1.tgz
     version: 0.1.0-rc1
-generated: "2021-03-25T18:38:56.280009551+01:00"
+generated: "2021-01-25T16:48:21.651203778+01:00"
diff --git a/charts/kubefed/charts/controllermanager/crds/crds.yaml b/charts/kubefed/charts/controllermanager/crds/crds.yaml
index 98858fdd..9c77a3ac 100644
--- a/charts/kubefed/charts/controllermanager/crds/crds.yaml
+++ b/charts/kubefed/charts/controllermanager/crds/crds.yaml
@@ -503,9 +503,6 @@ spec:
                 items:
                   type: string
                 type: array
-              proxyURL:
-                description: ProxyURL allows to set proxy URL for the cluster.
-                type: string
               secretRef:
                 description: Name of the secret containing the token required to access
                   the member cluster. The secret needs to exist in the same namespace
diff --git a/cmd/controller-manager/main.go b/cmd/controller-manager/main.go
index 0e3b4f16..2acb5934 100644
--- a/cmd/controller-manager/main.go
+++ b/cmd/controller-manager/main.go
@@ -20,8 +20,6 @@ import (
 	"fmt"
 	"os"
 
-	_ "sigs.k8s.io/controller-runtime/pkg/metrics" // for workqueue metrics registration
-
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	_ "k8s.io/client-go/plugin/pkg/client/auth" // Load all client auth plugins for GCP, Azure, Openstack, etc
 	"k8s.io/component-base/logs"
diff --git a/cmd/hyperfed/main.go b/cmd/hyperfed/main.go
index f22f4390..f34dad5c 100644
--- a/cmd/hyperfed/main.go
+++ b/cmd/hyperfed/main.go
@@ -31,8 +31,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	_ "sigs.k8s.io/controller-runtime/pkg/metrics" // for workqueue metrics registration
-
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	_ "k8s.io/client-go/plugin/pkg/client/auth" // Load all client auth plugins for GCP, Azure, Openstack, etc
 	utilflag "k8s.io/component-base/cli/flag"
diff --git a/go.mod b/go.mod
index 09964448..029155cb 100644
--- a/go.mod
+++ b/go.mod
@@ -17,7 +17,6 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.6.1
 	go.uber.org/zap v1.16.0 // indirect
-	golang.org/x/net v0.0.0-20201110031124-69a78807bb2b
 	k8s.io/api v0.20.2
 	k8s.io/apiextensions-apiserver v0.20.2
 	k8s.io/apimachinery v0.20.2
diff --git a/pkg/apis/core/v1beta1/kubefedcluster_types.go b/pkg/apis/core/v1beta1/kubefedcluster_types.go
index c3c59bc9..04da5748 100644
--- a/pkg/apis/core/v1beta1/kubefedcluster_types.go
+++ b/pkg/apis/core/v1beta1/kubefedcluster_types.go
@@ -51,10 +51,6 @@ type KubeFedClusterSpec struct {
 	// If * is specified, it is expected to be the only option in list.
 	// +optional
 	DisabledTLSValidations []TLSValidation `json:"disabledTLSValidations,omitempty"`
-
-	// ProxyURL allows to set proxy URL for the cluster.
-	// +optional
-	ProxyURL string `json:"proxyURL"`
 }
 
 // LocalSecretReference is a reference to a secret within the enclosing
diff --git a/pkg/apis/core/v1beta1/validation/validation.go b/pkg/apis/core/v1beta1/validation/validation.go
index 45ef4b19..5b58d126 100644
--- a/pkg/apis/core/v1beta1/validation/validation.go
+++ b/pkg/apis/core/v1beta1/validation/validation.go
@@ -18,7 +18,6 @@ package validation
 
 import (
 	"fmt"
-	"net/url"
 	"strconv"
 	"strings"
 	"time"
@@ -164,9 +163,6 @@ func validateKubeFedClusterSpec(spec *v1beta1.KubeFedClusterSpec, path *field.Pa
 	allErrs := validateAPIEndpoint(spec.APIEndpoint, path.Child("apiEndpoint"))
 	allErrs = append(allErrs, validateLocalSecretReference(&spec.SecretRef, path.Child("secretRef"))...)
 	allErrs = append(allErrs, validateDisabledTLSValidations(spec.DisabledTLSValidations, path.Child("disabledTLSValidations"))...)
-	if spec.ProxyURL != "" {
-		allErrs = append(allErrs, validateProxyURL(spec.ProxyURL, path.Child("proxyURL"))...)
-	}
 	return allErrs
 }
 
@@ -179,21 +175,6 @@ func validateKubeFedClusterStatus(status *v1beta1.KubeFedClusterStatus, path *fi
 	return allErrs
 }
 
-func validateProxyURL(proxyURL string, path *field.Path) field.ErrorList {
-	var allErrs field.ErrorList
-
-	u, err := url.Parse(proxyURL)
-	if err != nil {
-		allErrs = append(allErrs, field.Invalid(path, proxyURL, "error parsing the proxy URL"))
-	}
-	switch u.Scheme {
-	case "http", "https", "socks5":
-	default:
-		allErrs = append(allErrs, field.Invalid(path, proxyURL, "proxy URL scheme must be one of: [http, https, socks5]"))
-	}
-	return allErrs
-}
-
 func validateAPIEndpoint(endpoint string, path *field.Path) field.ErrorList {
 	if endpoint == "" {
 		return field.ErrorList{field.Required(path, "")}
diff --git a/pkg/controller/sync/controller.go b/pkg/controller/sync/controller.go
index 9169c86b..b2c8579f 100644
--- a/pkg/controller/sync/controller.go
+++ b/pkg/controller/sync/controller.go
@@ -387,7 +387,7 @@ func (s *KubeFedSyncController) syncToClusters(fedResource FederatedResource) ut
 	}
 
 	collectedStatus, collectedResourceStatus := dispatcher.CollectedStatus()
-	klog.V(4).Infof("Setting the federated status '%v' for %s %q", collectedResourceStatus, kind, key)
+	klog.V(4).Infof("Setting the federated status '%v'", collectedResourceStatus)
 	return s.setFederatedStatus(fedResource, status.AggregateSuccess, &collectedStatus, &collectedResourceStatus, enableRawResourceStatusCollection)
 }
 
diff --git a/pkg/controller/sync/status/status.go b/pkg/controller/sync/status/status.go
index 10916ee3..0614b9ad 100644
--- a/pkg/controller/sync/status/status.go
+++ b/pkg/controller/sync/status/status.go
@@ -125,25 +125,15 @@ type CollectedResourceStatus struct {
 // whether status should be written to the API.
 func SetFederatedStatus(fedObject *unstructured.Unstructured, reason AggregateReason, collectedStatus CollectedPropagationStatus, collectedResourceStatus CollectedResourceStatus, resourceStatusCollection bool) (bool, error) {
 	resource := &GenericFederatedResource{}
-
 	err := util.UnstructuredToInterface(fedObject, resource)
 	if err != nil {
 		return false, errors.Wrapf(err, "Failed to unmarshall to generic resource")
 	}
-
-	// we apply to collectedResourceStatus the same marshalling applied to GenericFederatedResource
-	// so the resources can be actually comparable later on
-	normalizedCollectedResourceStatus, err := normalizeStatus(collectedResourceStatus)
-	if err != nil {
-		return false, errors.Wrap(err, "Failed to normalize status")
-	}
-
 	if resource.Status == nil {
 		resource.Status = &GenericFederatedStatus{}
 	}
 
-	changed := resource.Status.update(fedObject.GetGeneration(), reason, collectedStatus, *normalizedCollectedResourceStatus, resourceStatusCollection)
-
+	changed := resource.Status.update(fedObject.GetGeneration(), reason, collectedStatus, collectedResourceStatus, resourceStatusCollection)
 	if !changed {
 		return false, nil
 	}
@@ -187,7 +177,7 @@ func (s *GenericFederatedStatus) update(generation int64, reason AggregateReason
 		}
 	}
 
-	clustersChanged := s.setClusters(collectedStatus.StatusMap, collectedResourceStatus.StatusMap, resourceStatusCollection)
+	clustersChanged := s.setClusters(collectedStatus.StatusMap, collectedResourceStatus.StatusMap)
 
 	// Indicate that changes were propagated if either status.clusters
 	// was changed or if existing resources were updated (which could
@@ -206,8 +196,8 @@ func (s *GenericFederatedStatus) update(generation int64, reason AggregateReason
 // setClusters sets the status.clusters slice from propagation and resource status
 // maps. Returns a boolean indication of whether the status.clusters was
 // modified.
-func (s *GenericFederatedStatus) setClusters(statusMap PropagationStatusMap, resourceStatusMap map[string]interface{}, resourceStatusCollection bool) bool {
-	if !s.clustersDiffer(statusMap, resourceStatusMap, resourceStatusCollection) {
+func (s *GenericFederatedStatus) setClusters(statusMap PropagationStatusMap, resourceStatusMap map[string]interface{}) bool {
+	if !s.clustersDiffer(statusMap, resourceStatusMap) {
 		return false
 	}
 	s.Clusters = []GenericClusterStatus{}
@@ -224,9 +214,9 @@ func (s *GenericFederatedStatus) setClusters(statusMap PropagationStatusMap, res
 
 // clustersDiffer checks whether `status.clusters` differs from the
 // given status map.
-func (s *GenericFederatedStatus) clustersDiffer(statusMap PropagationStatusMap, resourceStatusMap map[string]interface{}, resourceStatusCollection bool) bool {
-	if len(s.Clusters) != len(statusMap) || resourceStatusCollection && len(s.Clusters) != len(resourceStatusMap) {
-		klog.V(4).Infof("Clusters differs from the size: clusters = %v, statusMap = %v, resourceStatusMap = %v", s.Clusters, statusMap, resourceStatusMap)
+func (s *GenericFederatedStatus) clustersDiffer(statusMap PropagationStatusMap, resourceStatusMap map[string]interface{}) bool {
+	if len(s.Clusters) != len(statusMap) || len(s.Clusters) != len(resourceStatusMap) {
+		klog.V(4).Info("Clusters differs from the size")
 		return true
 	}
 	for _, status := range s.Clusters {
@@ -288,28 +278,3 @@ func (s *GenericFederatedStatus) setPropagationCondition(reason AggregateReason,
 
 	return updateRequired
 }
-
-func normalizeStatus(collectedResourceStatus CollectedResourceStatus) (*CollectedResourceStatus, error) {
-	if len(collectedResourceStatus.StatusMap) == 0 {
-		return &collectedResourceStatus, nil
-	}
-	cleanedStatus := CollectedResourceStatus{
-		StatusMap:        map[string]interface{}{},
-		ResourcesUpdated: collectedResourceStatus.ResourcesUpdated,
-	}
-
-	for key, value := range collectedResourceStatus.StatusMap {
-		content, err := json.Marshal(value)
-		if err != nil {
-			return nil, errors.Wrapf(err, "Failed to marshall collected resource status for cluster %s", key)
-		}
-		var status interface{}
-		err = json.Unmarshal(content, &status)
-		if err != nil {
-			return nil, errors.Wrapf(err, "Failed to unmarshall collected resource status as interface for cluster %s", key)
-		}
-		cleanedStatus.StatusMap[key] = status
-	}
-
-	return &cleanedStatus, nil
-}
diff --git a/pkg/controller/util/cluster_util.go b/pkg/controller/util/cluster_util.go
index 124ae87e..1ca1fef9 100644
--- a/pkg/controller/util/cluster_util.go
+++ b/pkg/controller/util/cluster_util.go
@@ -84,14 +84,6 @@ func BuildClusterConfig(fedCluster *fedv1b1.KubeFedCluster, client generic.Clien
 	clusterConfig.QPS = KubeAPIQPS
 	clusterConfig.Burst = KubeAPIBurst
 
-	if fedCluster.Spec.ProxyURL != "" {
-		proxyURL, err := url.Parse(fedCluster.Spec.ProxyURL)
-		if err != nil {
-			return nil, errors.Errorf("Failed to parse provided proxy URL %s: %v", fedCluster.Spec.ProxyURL, err)
-		}
-		clusterConfig.Proxy = http.ProxyURL(proxyURL)
-	}
-
 	if len(fedCluster.Spec.DisabledTLSValidations) != 0 {
 		klog.V(1).Infof("Cluster %s will use a custom transport for TLS certificate validation", fedCluster.Name)
 		if err = CustomizeTLSTransport(fedCluster, clusterConfig); err != nil {
diff --git a/pkg/controller/util/podanalyzer/pod_helper.go b/pkg/controller/util/podanalyzer/pod_helper.go
index 4a6fcc64..bc2b13b0 100644
--- a/pkg/controller/util/podanalyzer/pod_helper.go
+++ b/pkg/controller/util/podanalyzer/pod_helper.go
@@ -20,8 +20,6 @@ import (
 	"time"
 
 	api_v1 "k8s.io/api/core/v1"
-
-	ctlutil "sigs.k8s.io/kubefed/pkg/controller/util"
 )
 
 type PodAnalysisResult struct {
@@ -43,9 +41,8 @@ const (
 // AnalyzePods calculates how many pods from the list are in one of
 // the meaningful (from the replica set perspective) states. This function is
 // a temporary workaround against the current lack of ownerRef in pods.
-func AnalyzePods(podList *api_v1.PodList, currentTime time.Time) (PodAnalysisResult, ctlutil.ReconciliationStatus) {
+func AnalyzePods(podList *api_v1.PodList, currentTime time.Time) PodAnalysisResult {
 	result := PodAnalysisResult{}
-	unschedulableRightNow := 0
 	for _, pod := range podList.Items {
 		result.Total++
 		for _, condition := range pod.Status.Conditions {
@@ -55,23 +52,11 @@ func AnalyzePods(podList *api_v1.PodList, currentTime time.Time) (PodAnalysisRes
 				}
 			} else if condition.Type == api_v1.PodScheduled &&
 				condition.Status == api_v1.ConditionFalse &&
-				condition.Reason == api_v1.PodReasonUnschedulable {
-				unschedulableRightNow++
-				if condition.LastTransitionTime.Add(UnschedulableThreshold).Before(currentTime) {
-					result.Unschedulable++
-				}
+				condition.Reason == api_v1.PodReasonUnschedulable &&
+				condition.LastTransitionTime.Add(UnschedulableThreshold).Before(currentTime) {
+				result.Unschedulable++
 			}
 		}
 	}
-	if unschedulableRightNow != result.Unschedulable {
-		// We get the reconcile event almost immediately after  the status of a
-		// pod changes, however we will not consider the unschedulable pods as
-		// unschedulable immediately (until 60 secs), because we don't  want to
-		// change state frequently (it can lead to continuously moving replicas
-		// around). We need to reconcile again after a timeout. We use the return
-		// status to indicate retry for reconcile.
-		return result, ctlutil.StatusNeedsRecheck
-	}
-
-	return result, ctlutil.StatusAllOK
+	return result
 }
diff --git a/pkg/controller/util/worker.go b/pkg/controller/util/worker.go
index 17326e76..d38ddc92 100644
--- a/pkg/controller/util/worker.go
+++ b/pkg/controller/util/worker.go
@@ -117,7 +117,10 @@ func (w *asyncWorker) EnqueueWithDelay(qualifiedName QualifiedName, delay time.D
 func (w *asyncWorker) Run(stopChan <-chan struct{}) {
 	StartBackoffGC(w.backoff, stopChan)
 	w.deliverer.StartWithHandler(func(item *DelayingDelivererItem) {
-		w.queue.Add(item)
+		qualifiedName, ok := item.Value.(*QualifiedName)
+		if ok {
+			w.queue.Add(*qualifiedName)
+		}
 	})
 	go wait.Until(w.worker, w.timing.Interval, stopChan)
 
@@ -148,26 +151,32 @@ func (w *asyncWorker) deliver(qualifiedName QualifiedName, delay time.Duration,
 }
 
 func (w *asyncWorker) worker() {
-	for {
-		obj, quit := w.queue.Get()
-		if quit {
-			return
-		}
+	for w.reconcileOnce() {
+	}
+}
 
-		item := obj.(*DelayingDelivererItem)
-		qualifiedName := item.Value.(*QualifiedName)
-		status := w.reconcile(*qualifiedName)
-		w.queue.Done(item)
-
-		switch status {
-		case StatusAllOK:
-			break
-		case StatusError:
-			w.EnqueueForError(*qualifiedName)
-		case StatusNeedsRecheck:
-			w.EnqueueForRetry(*qualifiedName)
-		case StatusNotSynced:
-			w.EnqueueForClusterSync(*qualifiedName)
-		}
+func (w *asyncWorker) reconcileOnce() bool {
+	obj, quit := w.queue.Get()
+	if quit {
+		return false
+	}
+	defer w.queue.Done(obj)
+
+	qualifiedName, ok := obj.(QualifiedName)
+	if !ok {
+		return true
+	}
+
+	status := w.reconcile(qualifiedName)
+	switch status {
+	case StatusAllOK:
+		break
+	case StatusError:
+		w.EnqueueForError(qualifiedName)
+	case StatusNeedsRecheck:
+		w.EnqueueForRetry(qualifiedName)
+	case StatusNotSynced:
+		w.EnqueueForClusterSync(qualifiedName)
 	}
+	return true
 }
diff --git a/pkg/kubefedctl/join.go b/pkg/kubefedctl/join.go
index 9eedb976..85e49053 100644
--- a/pkg/kubefedctl/join.go
+++ b/pkg/kubefedctl/join.go
@@ -276,20 +276,8 @@ func joinClusterForNamespace(hostConfig, clusterConfig *rest.Config, kubefedName
 		caBundle = clusterConfig.CAData
 	}
 
-	var proxyURL string
-	if clusterConfig.Proxy != nil {
-		url, err := clusterConfig.Proxy(nil)
-		if err != nil {
-			klog.V(2).Infof("Error getting proxy URL for host %s: %w", clusterConfig.Host, err)
-			return nil, errors.Errorf("failed to create proxy URL request for kubefed cluster: %v", err)
-		}
-		if url != nil {
-			proxyURL = url.String()
-		}
-	}
-
 	kubefedCluster, err := createKubeFedCluster(client, joiningClusterName, clusterConfig.Host,
-		secret.Name, kubefedNamespace, caBundle, disabledTLSValidations, proxyURL, dryRun, errorOnExisting)
+		secret.Name, kubefedNamespace, caBundle, disabledTLSValidations, dryRun, errorOnExisting)
 	if err != nil {
 		klog.V(2).Infof("Failed to create federated cluster resource: %v", err)
 		return nil, err
@@ -331,7 +319,7 @@ func performPreflightChecks(clusterClientset kubeclient.Interface, name, hostClu
 // the cluster and secret.
 func createKubeFedCluster(client genericclient.Client, joiningClusterName, apiEndpoint,
 	secretName, kubefedNamespace string, caBundle []byte, disabledTLSValidations []fedv1b1.TLSValidation,
-	proxyURL string, dryRun, errorOnExisting bool) (*fedv1b1.KubeFedCluster, error) {
+	dryRun, errorOnExisting bool) (*fedv1b1.KubeFedCluster, error) {
 	fedCluster := &fedv1b1.KubeFedCluster{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: kubefedNamespace,
@@ -344,7 +332,6 @@ func createKubeFedCluster(client genericclient.Client, joiningClusterName, apiEn
 				Name: secretName,
 			},
 			DisabledTLSValidations: disabledTLSValidations,
-			ProxyURL:               proxyURL,
 		},
 	}
 
diff --git a/pkg/schedulingtypes/replicascheduler.go b/pkg/schedulingtypes/replicascheduler.go
index 1a5da96f..c30af5b4 100644
--- a/pkg/schedulingtypes/replicascheduler.go
+++ b/pkg/schedulingtypes/replicascheduler.go
@@ -190,7 +190,7 @@ func (s *ReplicaScheduler) Reconcile(obj pkgruntime.Object, qualifiedName ctluti
 	}
 
 	key := qualifiedName.String()
-	result, status, err := s.GetSchedulingResult(rsp, qualifiedName, clusterNames)
+	result, err := s.GetSchedulingResult(rsp, qualifiedName, clusterNames)
 	if err != nil {
 		runtime.HandleError(errors.Wrapf(err, "Failed to compute the schedule information while reconciling RSP named %q", key))
 		return ctlutil.StatusError
@@ -202,7 +202,7 @@ func (s *ReplicaScheduler) Reconcile(obj pkgruntime.Object, qualifiedName ctluti
 		return ctlutil.StatusError
 	}
 
-	return status
+	return ctlutil.StatusAllOK
 }
 
 // The list of clusters could come from any target informer
@@ -219,8 +219,7 @@ func (s *ReplicaScheduler) clusterNames() ([]string, error) {
 	return clusterNames, nil
 }
 
-func (s *ReplicaScheduler) GetSchedulingResult(rsp *fedschedulingv1a1.ReplicaSchedulingPreference,
-	qualifiedName ctlutil.QualifiedName, clusterNames []string) (map[string]int64, ctlutil.ReconciliationStatus, error) {
+func (s *ReplicaScheduler) GetSchedulingResult(rsp *fedschedulingv1a1.ReplicaSchedulingPreference, qualifiedName ctlutil.QualifiedName, clusterNames []string) (map[string]int64, error) {
 	key := qualifiedName.String()
 
 	objectGetter := func(clusterName, key string) (interface{}, bool, error) {
@@ -251,9 +250,9 @@ func (s *ReplicaScheduler) GetSchedulingResult(rsp *fedschedulingv1a1.ReplicaSch
 		return podList, nil
 	}
 
-	currentReplicasPerCluster, estimatedCapacity, status, err := clustersReplicaState(clusterNames, key, objectGetter, podsGetter)
+	currentReplicasPerCluster, estimatedCapacity, err := clustersReplicaState(clusterNames, key, objectGetter, podsGetter)
 	if err != nil {
-		return nil, status, err
+		return nil, err
 	}
 
 	// TODO: Move this to API defaulting logic
@@ -264,11 +263,7 @@ func (s *ReplicaScheduler) GetSchedulingResult(rsp *fedschedulingv1a1.ReplicaSch
 	}
 
 	plnr := planner.NewPlanner(rsp)
-	scheduleResult, err := schedule(plnr, key, clusterNames, currentReplicasPerCluster, estimatedCapacity)
-	if err != nil {
-		return nil, status, err
-	}
-	return scheduleResult, status, err
+	return schedule(plnr, key, clusterNames, currentReplicasPerCluster, estimatedCapacity)
 }
 
 func schedule(planner *planner.Planner, key string, clusterNames []string, currentReplicasPerCluster map[string]int64, estimatedCapacity map[string]int64) (map[string]int64, error) {
@@ -316,16 +311,14 @@ func clustersReplicaState(
 	clusterNames []string,
 	key string,
 	objectGetter func(clusterName string, key string) (interface{}, bool, error),
-	podsGetter func(clusterName string, obj *unstructured.Unstructured) (*corev1.PodList, error)) (
-	currentReplicasPerCluster map[string]int64, estimatedCapacity map[string]int64,
-	status ctlutil.ReconciliationStatus, err error) {
+	podsGetter func(clusterName string, obj *unstructured.Unstructured) (*corev1.PodList, error)) (currentReplicasPerCluster map[string]int64, estimatedCapacity map[string]int64, err error) {
 	currentReplicasPerCluster = make(map[string]int64)
 	estimatedCapacity = make(map[string]int64)
 
 	for _, clusterName := range clusterNames {
 		obj, exists, err := objectGetter(clusterName, key)
 		if err != nil {
-			return nil, nil, status, err
+			return nil, nil, err
 		}
 		if !exists {
 			continue
@@ -334,14 +327,14 @@ func clustersReplicaState(
 		unstructuredObj := obj.(*unstructured.Unstructured)
 		replicas, ok, err := unstructured.NestedInt64(unstructuredObj.Object, "spec", "replicas")
 		if err != nil {
-			return nil, nil, status, errors.Wrap(err, "Error retrieving 'replicas' field")
+			return nil, nil, errors.Wrap(err, "Error retrieving 'replicas' field")
 		}
 		if !ok {
 			replicas = int64(0)
 		}
-		readyReplicas, ok, err := unstructured.NestedInt64(unstructuredObj.Object, "status", "readyReplicas")
+		readyReplicas, ok, err := unstructured.NestedInt64(unstructuredObj.Object, "status", "readyreplicas")
 		if err != nil {
-			return nil, nil, status, errors.Wrap(err, "Error retrieving 'readyreplicas' field")
+			return nil, nil, errors.Wrap(err, "Error retrieving 'readyreplicas' field")
 		}
 		if !ok {
 			readyReplicas = int64(0)
@@ -353,17 +346,16 @@ func clustersReplicaState(
 			currentReplicasPerCluster[clusterName] = int64(0)
 			podList, err := podsGetter(clusterName, unstructuredObj)
 			if err != nil {
-				return nil, nil, status, err
+				return nil, nil, err
 			}
 
-			podResult := podanalyzer.PodAnalysisResult{}
-			podResult, status = podanalyzer.AnalyzePods(podList, time.Now())
-			currentReplicasPerCluster[clusterName] = int64(podResult.RunningAndReady) // include pending as well?
-			unschedulable := int64(podResult.Unschedulable)
+			podStatus := podanalyzer.AnalyzePods(podList, time.Now())
+			currentReplicasPerCluster[clusterName] = int64(podStatus.RunningAndReady) // include pending as well?
+			unschedulable := int64(podStatus.Unschedulable)
 			if unschedulable > 0 {
 				estimatedCapacity[clusterName] = replicas - unschedulable
 			}
 		}
 	}
-	return currentReplicasPerCluster, estimatedCapacity, status, nil
+	return currentReplicasPerCluster, estimatedCapacity, nil
 }


Test patch:
diff --git a/pkg/apis/core/v1beta1/kubefedcluster_types_test.go b/pkg/apis/core/v1beta1/kubefedcluster_types_test.go
deleted file mode 100644
index e76df092..00000000
--- a/pkg/apis/core/v1beta1/kubefedcluster_types_test.go
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
-Copyright 2021 The Kubernetes 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 v1beta1
-
-import (
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"golang.org/x/net/context"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-)
-
-var _ = Describe("KubefedCluster", func() {
-	var (
-		key              types.NamespacedName
-		created, fetched *KubeFedCluster
-	)
-
-	Context("Create API", func() {
-		It("should create a kubefed cluster setting a proxy url", func() {
-
-			created = &KubeFedCluster{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "foo",
-					Namespace: metav1.NamespaceDefault,
-				},
-				Spec: KubeFedClusterSpec{
-					APIEndpoint: "https://my.example.com:80/path/to/endpoint",
-					ProxyURL:    "socks5://example.com",
-					SecretRef: LocalSecretReference{
-						Name: "foo",
-					},
-				},
-			}
-
-			key, _ = client.ObjectKeyFromObject(created)
-
-			By("creating an API obj")
-			Expect(k8sClient.Create(context.TODO(), created)).To(Succeed())
-
-			fetched = &KubeFedCluster{}
-			Expect(k8sClient.Get(context.TODO(), key, fetched)).To(Succeed())
-			Expect(fetched).To(Equal(created))
-
-			By("deleting the created object")
-			Expect(k8sClient.Delete(context.TODO(), created)).To(Succeed())
-			Expect(k8sClient.Get(context.TODO(), key, created)).ToNot(Succeed())
-		})
-
-		It("should create a kubefed cluster without a proxy url and an empty secret", func() {
-
-			created = &KubeFedCluster{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "foo",
-					Namespace: metav1.NamespaceDefault,
-				},
-				Spec: KubeFedClusterSpec{
-					APIEndpoint: "https://my.example.com:80/path/to/endpoint",
-					SecretRef: LocalSecretReference{
-						Name: "",
-					},
-				},
-			}
-
-			key, _ = client.ObjectKeyFromObject(created)
-
-			By("creating an API obj")
-			Expect(k8sClient.Create(context.TODO(), created)).To(Succeed())
-
-			fetched = &KubeFedCluster{}
-			Expect(k8sClient.Get(context.TODO(), key, fetched)).To(Succeed())
-			Expect(fetched).To(Equal(created))
-
-			By("deleting the created object")
-			Expect(k8sClient.Delete(context.TODO(), created)).To(Succeed())
-			Expect(k8sClient.Get(context.TODO(), key, created)).ToNot(Succeed())
-		})
-
-	})
-
-})
diff --git a/pkg/apis/core/v1beta1/suite_test.go b/pkg/apis/core/v1beta1/suite_test.go
deleted file mode 100644
index 020ea4b4..00000000
--- a/pkg/apis/core/v1beta1/suite_test.go
+++ /dev/null
@@ -1,65 +0,0 @@
-package v1beta1
-
-import (
-	"path/filepath"
-	"testing"
-
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-
-	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/rest"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/envtest"
-	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
-	logf "sigs.k8s.io/controller-runtime/pkg/log"
-	"sigs.k8s.io/controller-runtime/pkg/log/zap"
-)
-
-var cfg *rest.Config
-var k8sClient client.Client
-var testEnv *envtest.Environment
-
-func TestAPIs(t *testing.T) {
-	RegisterFailHandler(Fail)
-
-	RunSpecsWithDefaultAndCustomReporters(t,
-		"v1beta1 Suite",
-		[]Reporter{printer.NewlineReporter{}})
-}
-
-var _ = BeforeSuite(func(done Done) {
-	logf.SetLogger(zap.New(func(o *zap.Options) {
-		o.Development = true
-		o.DestWritter = GinkgoWriter
-	}))
-
-	By("bootstrapping test environment")
-	testEnv = &envtest.Environment{
-		CRDInstallOptions: envtest.CRDInstallOptions{
-			ErrorIfPathMissing: true,
-			Paths: []string{
-				filepath.Join("..", "..", "..", "..", "charts", "kubefed", "charts", "controllermanager", "crds"),
-			},
-		},
-	}
-
-	err := SchemeBuilder.AddToScheme(scheme.Scheme)
-	Expect(err).NotTo(HaveOccurred())
-
-	cfg, err = testEnv.Start()
-	Expect(err).ToNot(HaveOccurred())
-	Expect(cfg).ToNot(BeNil())
-
-	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
-	Expect(err).ToNot(HaveOccurred())
-	Expect(k8sClient).ToNot(BeNil())
-
-	close(done)
-}, 60)
-
-var _ = AfterSuite(func() {
-	By("tearing down the test environment")
-	err := testEnv.Stop()
-	Expect(err).ToNot(HaveOccurred())
-})
diff --git a/pkg/apis/core/v1beta1/validation/validation_test.go b/pkg/apis/core/v1beta1/validation/validation_test.go
index a6e6a350..3ddf72de 100644
--- a/pkg/apis/core/v1beta1/validation/validation_test.go
+++ b/pkg/apis/core/v1beta1/validation/validation_test.go
@@ -480,56 +480,6 @@ func TestValidateAPIEndpoint(t *testing.T) {
 	}
 }
 
-func TestProxyURL(t *testing.T) {
-	tests := []struct {
-		proxyURL       string
-		expectedErrMsg string
-	}{
-		{
-			proxyURL: "socks5://example.com",
-		},
-		{
-			proxyURL: "https://example.com",
-		},
-		{
-			proxyURL: "http://example.com",
-		},
-		{
-			proxyURL:       "socks6://example.com",
-			expectedErrMsg: "proxyURL: Invalid value: \"socks6://example.com\": proxy URL scheme must be one of: [http, https, socks5]",
-		},
-		{
-			proxyURL:       "example.com",
-			expectedErrMsg: "proxyURL: Invalid value: \"example.com\": proxy URL scheme must be one of: [http, https, socks5]",
-		},
-		{
-			proxyURL:       "chewbacca@example.com",
-			expectedErrMsg: "proxyURL: Invalid value: \"chewbacca@example.com\": proxy URL scheme must be one of: [http, https, socks5]",
-		},
-	}
-
-	for _, test := range tests {
-		errs := validateProxyURL(test.proxyURL, field.NewPath("proxyURL"))
-		if len(errs) == 0 && test.expectedErrMsg == "" {
-			continue
-		}
-		if len(errs) == 0 && test.expectedErrMsg != "" {
-			t.Errorf("[%s] expected failure", test.expectedErrMsg)
-		} else {
-			matchedErr := false
-			for _, err := range errs {
-				if strings.Contains(err.Error(), test.expectedErrMsg) {
-					matchedErr = true
-					break
-				}
-			}
-			if !matchedErr {
-				t.Errorf("unexpected error: %v, expected: %q", errs, test.expectedErrMsg)
-			}
-		}
-	}
-}
-
 func TestValidateLocalSecretReference(t *testing.T) {
 	testCases := []struct {
 		secretName     string
diff --git a/pkg/controller/sync/status/status_test.go b/pkg/controller/sync/status/status_test.go
index 92b43a88..7100a403 100644
--- a/pkg/controller/sync/status/status_test.go
+++ b/pkg/controller/sync/status/status_test.go
@@ -17,7 +17,6 @@ limitations under the License.
 package status
 
 import (
-	"reflect"
 	"testing"
 
 	apiv1 "k8s.io/api/core/v1"
@@ -25,170 +24,52 @@ import (
 
 func TestGenericPropagationStatusUpdateChanged(t *testing.T) {
 	testCases := map[string]struct {
-		generation               int64
-		reason                   AggregateReason
-		statusMap                PropagationStatusMap
-		resourceStatusMap        map[string]interface{}
-		remoteStatus             interface{}
-		resourcesUpdated         bool
-		expectedChanged          bool
-		resourceStatusCollection bool
+		generation        int64
+		reason            AggregateReason
+		statusMap         PropagationStatusMap
+		resourceStatusMap map[string]interface{}
+		resourcesUpdated  bool
+		expectedChanged   bool
 	}{
-		"Cluster not propagated indicates changed with status collected enabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterNotReady,
-			},
-			resourceStatusMap: map[string]interface{}{
-				"cluster1": map[string]interface{}{},
-			},
-			reason:                   AggregateSuccess,
-			resourcesUpdated:         false,
-			resourceStatusCollection: true,
-			expectedChanged:          true,
-		},
-		"Cluster not propagated indicates changed with status collected disabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterNotReady,
-			},
-			resourceStatusMap: map[string]interface{}{
-				"cluster1": map[string]interface{}{},
-			},
-			reason:                   AggregateSuccess,
-			resourcesUpdated:         false,
-			resourceStatusCollection: false,
-			expectedChanged:          true,
-		},
-		"Cluster status not retrieved indicates changed with status collected enabled": {
-			statusMap:                PropagationStatusMap{},
-			reason:                   AggregateSuccess,
-			resourcesUpdated:         false,
-			resourceStatusCollection: true,
-			expectedChanged:          true,
-		},
-		"Cluster status not retrieved indicates changed with status collected disabled": {
-			statusMap:                PropagationStatusMap{},
-			reason:                   AggregateSuccess,
-			resourcesUpdated:         false,
-			resourceStatusCollection: false,
-			expectedChanged:          true,
-		},
-		"No collected remote status indicates changed with status collected enabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterPropagationOK,
-			},
-			reason:                   AggregateSuccess,
-			resourcesUpdated:         false,
-			resourceStatusCollection: true,
-			expectedChanged:          true,
-		},
-		"No collected remote status indicates unchanged with status collected disabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterPropagationOK,
-			},
-			reason:                   AggregateSuccess,
-			resourcesUpdated:         false,
-			resourceStatusCollection: false,
-			expectedChanged:          false,
-		},
-		"No change in clusters indicates unchanged with status collected enabled": {
+		"No change in clusters indicates unchanged": {
 			statusMap: PropagationStatusMap{
 				"cluster1": ClusterPropagationOK,
 			},
 			resourceStatusMap: map[string]interface{}{
-				"cluster1": map[string]interface{}{},
+				"ready": false,
+				"stage": "absent",
 			},
-			resourcesUpdated:         false,
-			resourceStatusCollection: true,
-			expectedChanged:          true,
+			resourcesUpdated: false,
+			expectedChanged:  true,
 		},
-		"No change in clusters indicates unchanged with status collected disabled": {
+		"No change in clusters with update indicates changed": {
 			statusMap: PropagationStatusMap{
 				"cluster1": ClusterPropagationOK,
 			},
 			resourceStatusMap: map[string]interface{}{
-				"cluster1": map[string]interface{}{},
+				"ready": false,
+				"stage": "absent",
 			},
-			resourcesUpdated:         false,
-			resourceStatusCollection: false,
-			expectedChanged:          true,
+			resourcesUpdated: true,
+			expectedChanged:  true,
 		},
-		"No change in clusters with update indicates changed with status collected enabled": {
+		"Change in clusters indicates changed": {
 			statusMap: PropagationStatusMap{
 				"cluster1": ClusterPropagationOK,
 			},
 			resourceStatusMap: map[string]interface{}{
-				"cluster1": map[string]interface{}{},
+				"ready": true,
+				"stage": "deployed",
 			},
-			resourcesUpdated:         true,
-			resourceStatusCollection: true,
-			expectedChanged:          true,
+			expectedChanged: true,
 		},
-		"No change in clusters with update indicates changed with status collected disabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterPropagationOK,
-			},
-			resourceStatusMap: map[string]interface{}{
-				"cluster1": map[string]interface{}{},
-			},
-			resourcesUpdated:         true,
-			resourceStatusCollection: false,
-			expectedChanged:          true,
+		"Transition indicates changed": {
+			reason:          NamespaceNotFederated,
+			expectedChanged: true,
 		},
-		"No change in remote status indicates unchanged with status collected enabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterPropagationOK,
-			},
-			resourceStatusMap: map[string]interface{}{
-				"cluster1": map[string]interface{}{
-					"status": map[string]interface{}{
-						"status": "remoteStatus",
-					},
-				},
-			},
-			remoteStatus: map[string]interface{}{
-				"status": map[string]interface{}{
-					"status": "remoteStatus",
-				},
-			},
-			resourcesUpdated:         false,
-			resourceStatusCollection: true,
-			expectedChanged:          false,
-		},
-		"Change in clusters indicates changed with status collected enabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterPropagationOK,
-				"cluster2": ClusterPropagationOK,
-			},
-			resourceStatusCollection: true,
-			expectedChanged:          true,
-		},
-		"Change in clusters indicates changed with status collected disabled": {
-			statusMap: PropagationStatusMap{
-				"cluster1": ClusterPropagationOK,
-				"cluster2": ClusterPropagationOK,
-			},
-			resourceStatusCollection: false,
-			expectedChanged:          true,
-		},
-		"Transition indicates changed with remote status collection enabled": {
-			reason:                   NamespaceNotFederated,
-			resourceStatusCollection: true,
-			expectedChanged:          true,
-		},
-		"Transition indicates changed with remote status collection disabled": {
-			reason:                   NamespaceNotFederated,
-			resourceStatusCollection: false,
-			expectedChanged:          true,
-		},
-		"Changed generation indicates changed with remote status collection enabled": {
-			generation:               1,
-			resourceStatusCollection: true,
-			expectedChanged:          true,
-		},
-		"Changed generation indicates changed with remote status collection disabled": {
-			generation:               1,
-			resourceStatusCollection: false,
-			expectedChanged:          true,
+		"Changed generation indicates changed": {
+			generation:      1,
+			expectedChanged: true,
 		},
 	}
 	for testName, tc := range testCases {
@@ -196,8 +77,7 @@ func TestGenericPropagationStatusUpdateChanged(t *testing.T) {
 			fedStatus := &GenericFederatedStatus{
 				Clusters: []GenericClusterStatus{
 					{
-						Name:         "cluster1",
-						RemoteStatus: tc.remoteStatus,
+						Name: "cluster1",
 					},
 				},
 				Conditions: []*GenericCondition{
@@ -215,68 +95,10 @@ func TestGenericPropagationStatusUpdateChanged(t *testing.T) {
 				StatusMap:        tc.resourceStatusMap,
 				ResourcesUpdated: tc.resourcesUpdated,
 			}
-			changed := fedStatus.update(tc.generation, tc.reason, collectedStatus, collectedResourceStatus, tc.resourceStatusCollection)
+			changed := fedStatus.update(tc.generation, tc.reason, collectedStatus, collectedResourceStatus, true)
 			if tc.expectedChanged != changed {
 				t.Fatalf("Expected changed to be %v, got %v", tc.expectedChanged, changed)
 			}
 		})
 	}
 }
-
-func TestNormalizeStatus(t *testing.T) {
-	testCases := []struct {
-		name           string
-		input          CollectedResourceStatus
-		expectedResult *CollectedResourceStatus
-		expectedError  error
-	}{
-		{
-			name:           "CollectedResourceStatus is not modified if StatusMap is nil",
-			input:          CollectedResourceStatus{},
-			expectedResult: &CollectedResourceStatus{},
-		},
-		{
-			name: "CollectedResourceStatus is not modified if StatusMap is empty",
-			input: CollectedResourceStatus{
-				StatusMap: map[string]interface{}{},
-			},
-			expectedResult: &CollectedResourceStatus{
-				StatusMap: map[string]interface{}{},
-			},
-		},
-		{
-			name: "CollectedResourceStatus StatusMap is correctly normalized and numbers are converted to float64",
-			input: CollectedResourceStatus{
-				StatusMap: map[string]interface{}{
-					"number": 1,
-					"string": "value",
-					"complexobj": map[string]int{
-						"one": 1,
-					},
-				},
-			},
-			expectedResult: &CollectedResourceStatus{
-				StatusMap: map[string]interface{}{
-					"number": float64(1),
-					"string": "value",
-					"complexobj": map[string]interface{}{
-						"one": float64(1),
-					},
-				},
-			},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			actual, err := normalizeStatus(tc.input)
-			if err != tc.expectedError {
-				t.Fatalf("Expected error to be %v, got %v", tc.expectedError, err)
-			}
-
-			if !reflect.DeepEqual(tc.expectedResult, actual) {
-				t.Fatalf("Expected result to be %#v, got %#v", tc.expectedResult, actual)
-			}
-		})
-	}
-}
diff --git a/pkg/controller/util/podanalyzer/pod_helper_test.go b/pkg/controller/util/podanalyzer/pod_helper_test.go
index c206ee7e..5f2f175f 100644
--- a/pkg/controller/util/podanalyzer/pod_helper_test.go
+++ b/pkg/controller/util/podanalyzer/pod_helper_test.go
@@ -24,8 +24,6 @@ import (
 
 	api_v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	ctlutil "sigs.k8s.io/kubefed/pkg/controller/util"
 )
 
 func TestAnalyze(t *testing.T) {
@@ -40,7 +38,7 @@ func TestAnalyze(t *testing.T) {
 				},
 			},
 		})
-	podUnschedulableTimeElapsed := newPod("pU",
+	podUnschedulable := newPod("pU",
 		api_v1.PodStatus{
 			Phase: api_v1.PodPending,
 			Conditions: []api_v1.PodCondition{
@@ -52,40 +50,25 @@ func TestAnalyze(t *testing.T) {
 				},
 			},
 		})
-	podUnschedulableRightNow := newPod("pU",
-		api_v1.PodStatus{
-			Phase: api_v1.PodPending,
-			Conditions: []api_v1.PodCondition{
-				{
-					Type:               api_v1.PodScheduled,
-					Status:             api_v1.ConditionFalse,
-					Reason:             api_v1.PodReasonUnschedulable,
-					LastTransitionTime: metav1.Time{Time: now},
-				},
-			},
-		})
 	podOther := newPod("pO",
 		api_v1.PodStatus{
 			Phase:      api_v1.PodPending,
 			Conditions: []api_v1.PodCondition{},
 		})
 
-	result, status := AnalyzePods(&api_v1.PodList{Items: []api_v1.Pod{*podRunning, *podRunning,
-		*podRunning, *podUnschedulableTimeElapsed, *podUnschedulableTimeElapsed, *podUnschedulableRightNow}}, now)
+	result := AnalyzePods(&api_v1.PodList{Items: []api_v1.Pod{*podRunning, *podRunning, *podRunning, *podUnschedulable, *podUnschedulable}}, now)
 	assert.Equal(t, PodAnalysisResult{
-		Total:           6,
+		Total:           5,
 		RunningAndReady: 3,
 		Unschedulable:   2,
 	}, result)
-	assert.Equal(t, status, ctlutil.StatusNeedsRecheck)
 
-	result, status = AnalyzePods(&api_v1.PodList{Items: []api_v1.Pod{*podOther}}, now)
+	result = AnalyzePods(&api_v1.PodList{Items: []api_v1.Pod{*podOther}}, now)
 	assert.Equal(t, PodAnalysisResult{
 		Total:           1,
 		RunningAndReady: 0,
 		Unschedulable:   0,
 	}, result)
-	assert.Equal(t, status, ctlutil.StatusAllOK)
 }
 
 func newPod(name string, status api_v1.PodStatus) *api_v1.Pod {
diff --git a/pkg/controller/util/worker_test.go b/pkg/controller/util/worker_test.go
new file mode 100644
index 00000000..1d2c1138
--- /dev/null
+++ b/pkg/controller/util/worker_test.go
@@ -0,0 +1,88 @@
+/*
+Copyright 2021 The Kubernetes 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 util
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestDeduplicate(t *testing.T) {
+	// single event for reconciliation
+	singleEvent := QualifiedName{
+		Namespace: "ns",
+		Name:      "name",
+	}
+
+	// counters for reconciliation and enqueue
+	var reconcileCount, enqueueCount int32 = 0, 0
+	addReconcileCount := func() {
+		atomic.AddInt32(&reconcileCount, 1)
+	}
+	addEnqueueCount := func() {
+		atomic.AddInt32(&enqueueCount, 1)
+	}
+
+	var worker ReconcileWorker
+
+	// an utility function to enqueue single event for specified times
+	duplicateEnqueue := func(times int) {
+		for i := 0; i < times; i++ {
+			worker.Enqueue(singleEvent)
+			addEnqueueCount()
+		}
+	}
+
+	// new worker
+	doOnce := sync.Once{}
+	worker = NewReconcileWorker("test deduplicate",
+		func(qualifiedName QualifiedName) ReconciliationStatus {
+			addReconcileCount()
+			// enqueue same events for 5 times during the reconciliation itself
+			doOnce.Do(func() {
+				duplicateEnqueue(5)
+			})
+			return StatusAllOK
+		},
+		WorkerTiming{},
+	)
+
+	// run worker
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	worker.Run(ctx.Done())
+
+	// enqueue single event for 10 times
+	duplicateEnqueue(10)
+
+	// wait for all events enqueued and reconciled
+	// FIXME: there is not a good way to detect whether all enqueued events have been reconciled, so we just wait for a while
+	time.Sleep(1 * time.Second)
+
+	// check counters
+	if enqueueCount != 15 {
+		t.Errorf("expected enqueue count 15 but got %d", enqueueCount)
+	}
+	if reconcileCount != 2 {
+		t.Errorf("expected reconcile count 2 but got %d", reconcileCount)
+	}
+
+	t.Logf("the enqueued (before or during reconciliation) 15 same events have been squashed to 2")
+}
diff --git a/pkg/kubefedctl/join_test.go b/pkg/kubefedctl/join_test.go
deleted file mode 100644
index 145fbc1e..00000000
--- a/pkg/kubefedctl/join_test.go
+++ /dev/null
@@ -1,145 +0,0 @@
-/*
-Copyright 2018 The Kubernetes 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 kubefedctl
-
-import (
-	"net/http"
-	"net/http/httptest"
-	"net/http/httputil"
-	"net/url"
-
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-
-	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/rest"
-	utiltesting "k8s.io/client-go/util/testing"
-
-	"sigs.k8s.io/kubefed/pkg/kubefedctl/util"
-
-	"sigs.k8s.io/kubefed/pkg/apis/core/v1beta1"
-)
-
-var (
-	joiningClusterName = "joiningcluster"
-	hostClusterName    = "hostnamecluster"
-)
-
-var _ = Describe("Kubefedctl", func() {
-
-	Context("Join Cluster", func() {
-		It("should create a kubefed cluster successfully with a proxyURL", func() {
-			testServer, _, _ := testServerEnv(200)
-			defer testServer.Close()
-
-			clusterConfig := &rest.Config{
-				Host: testServer.URL,
-				ContentConfig: rest.ContentConfig{
-					GroupVersion:         &v1.SchemeGroupVersion,
-					NegotiatedSerializer: scheme.Codecs.WithoutConversion(),
-				},
-			}
-
-			expectedKubefedCluster := &v1beta1.KubeFedCluster{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      joiningClusterName,
-					Namespace: metav1.NamespaceDefault,
-				},
-				Spec: v1beta1.KubeFedClusterSpec{
-					APIEndpoint: testServer.URL,
-					SecretRef: v1beta1.LocalSecretReference{
-						Name: "secret",
-					},
-				},
-			}
-
-			kubefedCluster, err := joinClusterForNamespace(
-				cfg,
-				clusterConfig,
-				metav1.NamespaceDefault,
-				metav1.NamespaceDefault,
-				hostClusterName,
-				joiningClusterName,
-				"secret",
-				v1.ClusterScoped, true, false)
-
-			Expect(err).NotTo(HaveOccurred())
-			Expect(kubefedCluster).To(Equal(expectedKubefedCluster))
-
-			By("Using a joining cluster behind a proxy-url")
-			testProxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-				to, err := url.Parse(req.RequestURI)
-				Expect(err).To(BeNil())
-				httputil.NewSingleHostReverseProxy(to).ServeHTTP(w, req)
-			}))
-			defer testProxyServer.Close()
-
-			u, err := url.Parse(testProxyServer.URL)
-			if err != nil {
-				Expect(err).To(BeNil())
-			}
-
-			// Set a fakeProxyServer as URL
-			clusterConfig.Proxy = http.ProxyURL(u)
-
-			expectedKubefedClusterWithProxyURL := &v1beta1.KubeFedCluster{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      joiningClusterName,
-					Namespace: metav1.NamespaceDefault,
-				},
-				Spec: v1beta1.KubeFedClusterSpec{
-					APIEndpoint: testServer.URL,
-					ProxyURL:    u.String(),
-				},
-			}
-
-			kubefedClusterWithProxyURL, err := joinClusterForNamespace(
-				cfg,
-				clusterConfig,
-				metav1.NamespaceDefault,
-				metav1.NamespaceDefault,
-				hostClusterName,
-				joiningClusterName,
-				"",
-				v1.ClusterScoped, true, false)
-
-			Expect(err).NotTo(HaveOccurred())
-			Expect(kubefedClusterWithProxyURL).To(Equal(expectedKubefedClusterWithProxyURL))
-
-		})
-	})
-})
-
-func testServerEnv(statusCode int) (*httptest.Server, *utiltesting.FakeHandler, *corev1.ServiceAccount) {
-	status := &corev1.ServiceAccount{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      util.ClusterServiceAccountName(joiningClusterName, hostClusterName),
-			Namespace: metav1.NamespaceDefault,
-		},
-	}
-	expectedBody, _ := runtime.Encode(scheme.Codecs.LegacyCodec(corev1.SchemeGroupVersion), status)
-	fakeHandler := utiltesting.FakeHandler{
-		StatusCode:   statusCode,
-		ResponseBody: string(expectedBody),
-	}
-	testServer := httptest.NewServer(&fakeHandler)
-	return testServer, &fakeHandler, status
-}
diff --git a/pkg/kubefedctl/suite_test.go b/pkg/kubefedctl/suite_test.go
deleted file mode 100644
index 4238069b..00000000
--- a/pkg/kubefedctl/suite_test.go
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
-Copyright 2018 The Kubernetes 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 kubefedctl
-
-import (
-	"path/filepath"
-	"testing"
-
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/rest"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/envtest"
-	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
-	logf "sigs.k8s.io/controller-runtime/pkg/log"
-	"sigs.k8s.io/controller-runtime/pkg/log/zap"
-
-	"sigs.k8s.io/kubefed/pkg/apis/core/v1beta1"
-)
-
-var (
-	cfg          *rest.Config
-	k8sClient    client.Client
-	clientScheme *runtime.Scheme
-	testEnv      *envtest.Environment
-)
-
-func TestKubefedCTL(t *testing.T) {
-	RegisterFailHandler(Fail)
-	RunSpecsWithDefaultAndCustomReporters(t, "Kubefedctl Suite", []Reporter{printer.NewlineReporter{}})
-}
-
-var _ = BeforeSuite(func(done Done) {
-	logf.SetLogger(zap.New(func(o *zap.Options) {
-		o.Development = true
-		o.DestWritter = GinkgoWriter
-	}))
-
-	testEnv = &envtest.Environment{
-		CRDInstallOptions: envtest.CRDInstallOptions{
-			ErrorIfPathMissing: true,
-			Paths: []string{
-				filepath.Join("..", "..", "charts", "kubefed", "charts", "controllermanager", "crds"),
-			},
-		},
-	}
-
-	clientScheme = runtime.NewScheme()
-
-	Expect(v1beta1.AddToScheme(clientScheme)).To(Succeed())
-	Expect(corev1.AddToScheme(clientScheme)).To(Succeed())
-
-	var err error
-	cfg, err = testEnv.Start()
-	Expect(err).ToNot(HaveOccurred())
-	Expect(cfg).ToNot(BeNil())
-
-	k8sClient, err = client.New(cfg, client.Options{Scheme: clientScheme})
-	Expect(err).ToNot(HaveOccurred())
-	Expect(k8sClient).ToNot(BeNil())
-
-	close(done)
-}, 60)
-
-var _ = AfterSuite(func() {
-	By("tearing down the test environment")
-	err := testEnv.Stop()
-	Expect(err).ToNot(HaveOccurred())
-})