Repository: clusternet/clusternet
Pull request #396 (2022-06-28 06:37:36)

aggregate common labels from nodes
Signed-off-by: lmxia <xialingming@gmail.com>

<!--  Thanks for sending a pull request!

-->

#### What type of PR is this?
Add feature about aggregate nodes common labels to mcls.

#### What this PR does / why we need it:
Managedcluster's labels precisely represent nodes in child cluster.
#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #

#### Special notes for your reviewer:
1. Aggreate labels start with ``node.clusternet.io/`` 
2. Keep managedclusters existed labels.
3. Updated 3 minites periodly default.

codecov-commenter: # [Codecov](https://codecov.io/gh/clusternet/clusternet/pull/396?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) Report
> Merging [#396](https://codecov.io/gh/clusternet/clusternet/pull/396?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (07a47f6) into [main](https://codecov.io/gh/clusternet/clusternet/commit/2eda0b49ab38d94650745f97284c654ccb72e820?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (2eda0b4) will **increase** coverage by `0.16%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 07a47f6 differs from pull request most recent head 42d22b4. Consider uploading reports for the commit 42d22b4 to get more accurate results

```diff
@@            Coverage Diff             @@
##             main     #396      +/-   ##
==========================================
+ Coverage   11.58%   11.74%   +0.16%     
==========================================
  Files          60       60              
  Lines        6334     6383      +49     
==========================================
+ Hits          734      750      +16     
- Misses       5550     5579      +29     
- Partials       50       54       +4     
```


| [Impacted Files](https://codecov.io/gh/clusternet/clusternet/pull/396?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) | Coverage Δ | |
|---|---|---|
| [.../clusternet/clusternet/pkg/agent/status\_manager.go](https://codecov.io/gh/clusternet/clusternet/pull/396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9hZ2VudC9zdGF0dXNfbWFuYWdlci5nbw==) | `0.00% <0.00%> (ø)` | |
| [...clusters/clusterstatus/clusterstatus\_controller.go](https://codecov.io/gh/clusternet/clusternet/pull/396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9jb250cm9sbGVycy9jbHVzdGVycy9jbHVzdGVyc3RhdHVzL2NsdXN0ZXJzdGF0dXNfY29udHJvbGxlci5nbw==) | `5.70% <0.00%> (+5.70%)` | :arrow_up: |
| [...ernet/clusternet/pkg/hub/apiserver/shadow/utils.go](https://codecov.io/gh/clusternet/clusternet/pull/396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9odWIvYXBpc2VydmVyL3NoYWRvdy91dGlscy5nbw==) | `100.00% <0.00%> (+10.71%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/clusternet/clusternet/pull/396?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/clusternet/clusternet/pull/396?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Last update [2eda0b4...42d22b4](https://codecov.io/gh/clusternet/clusternet/pull/396?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet).


Patch:
diff --git a/go.mod b/go.mod
index 0ed43ff..22ad883 100644
--- a/go.mod
+++ b/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.3.0
 	github.com/spf13/pflag v1.0.5
+	github.com/stretchr/testify v1.7.0
 	helm.sh/helm/v3 v3.8.0
 	k8s.io/api v0.23.1
 	k8s.io/apiextensions-apiserver v0.23.1
@@ -125,7 +126,6 @@ require (
 	github.com/russross/blackfriday v1.5.2 // indirect
 	github.com/shopspring/decimal v1.2.0 // indirect
 	github.com/spf13/cast v1.4.1 // indirect
-	github.com/stretchr/testify v1.7.0 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
 	github.com/xeipuuv/gojsonschema v1.2.0 // indirect
diff --git a/pkg/agent/status_manager.go b/pkg/agent/status_manager.go
index 3d1127c..410b959 100644
--- a/pkg/agent/status_manager.go
+++ b/pkg/agent/status_manager.go
@@ -18,13 +18,17 @@ package agent
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"os"
+	"reflect"
+	"strings"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
@@ -131,6 +135,15 @@ func (mgr *Manager) updateClusterStatus(ctx context.Context, namespace, clusterI
 			return false, nil
 		}
 
+		newLabels := mgr.getManagedClusterNewLabels()
+		if !reflect.DeepEqual(mgr.managedCluster.GetLabels(), newLabels) {
+			mcls, lastError = patchManagedClusterTwoWayMergeLabels(client, mgr.managedCluster, newLabels)
+			if lastError == nil {
+				mgr.managedCluster = mcls
+			} else {
+				return false, nil
+			}
+		}
 		mgr.managedCluster.Status = *status
 		mcls, lastError = client.ClustersV1beta1().ManagedClusters(namespace).UpdateStatus(ctx, mgr.managedCluster, metav1.UpdateOptions{})
 		if lastError == nil {
@@ -149,3 +162,38 @@ func (mgr *Manager) updateClusterStatus(ctx context.Context, namespace, clusterI
 		klog.WarningDepth(2, "failed to update status of ManagedCluster: %v", lastError)
 	}
 }
+
+func (mgr *Manager) getManagedClusterNewLabels() map[string]string {
+	originLabels := mgr.managedCluster.GetLabels()
+	modifiedLabels := map[string]string{}
+	for k, v := range originLabels {
+		if strings.HasPrefix(k, known.NodeLabelsKeyPrefix) {
+			continue
+		}
+		modifiedLabels[k] = v
+	}
+	return labels.Merge(modifiedLabels, mgr.clusterStatusController.GetManagedClusterLabels())
+}
+
+func patchManagedClusterTwoWayMergeLabels(client clusternetclientset.Interface, mcls *clusterapi.ManagedCluster,
+	newLabels map[string]string) (result *clusterapi.ManagedCluster, err error) {
+	actualCopy := mcls.DeepCopy()
+	actualCopy.Labels = newLabels
+
+	oldData, err := json.Marshal(mcls)
+	if err != nil {
+		return nil, err
+	}
+
+	newData, err := json.Marshal(actualCopy)
+	if err != nil {
+		return nil, err
+	}
+
+	patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, clusterapi.ManagedCluster{})
+	if err != nil {
+		return nil, err
+	}
+
+	return client.ClustersV1beta1().ManagedClusters(mcls.Namespace).Patch(context.TODO(), mcls.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
+}
diff --git a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
index 9f7ef35..d41166b 100644
--- a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
+++ b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
@@ -19,6 +19,7 @@ package clusterstatus
 import (
 	"context"
 	"net/http"
+	"strings"
 	"sync"
 
 	corev1 "k8s.io/api/core/v1"
@@ -185,6 +186,15 @@ func (c *Controller) GetClusterStatus() *clusterapi.ManagedClusterStatus {
 	return c.clusterStatus.DeepCopy()
 }
 
+func (c *Controller) GetManagedClusterLabels() labels.Set {
+	nodes, err := c.nodeLister.List(labels.Everything())
+	if err != nil {
+		klog.Warningf("failed to list nodes: %v", err)
+		return nil
+	}
+	return getCommonNodeLabels(nodes)
+}
+
 func (c *Controller) getKubernetesVersion(_ context.Context) (*version.Info, error) {
 	return c.kubeClient.Discovery().ServerVersion()
 }
@@ -332,3 +342,30 @@ func getNodeCondition(status *corev1.NodeStatus, conditionType corev1.NodeCondit
 	}
 	return -1, nil
 }
+
+// getCommonNodeLabels return the common labels from nodes meta.label
+func getCommonNodeLabels(nodes []*corev1.Node) map[string]string {
+	if len(nodes) == 0 {
+		return nil
+	}
+	if len(nodes) == 1 {
+		return nodes[0].Labels
+	}
+	initLabels := nodes[0].Labels
+	for _, node := range nodes[1:] {
+		currentLabels := map[string]string{}
+		for k, v := range node.Labels {
+			c, ok := initLabels[k]
+			if ok && c == v {
+				if strings.HasPrefix(k, known.NodeLabelsKeyPrefix) {
+					currentLabels[k] = v
+				}
+			}
+		}
+		if len(currentLabels) == 0 {
+			return nil
+		}
+		initLabels = currentLabels
+	}
+	return initLabels
+}
diff --git a/pkg/known/labels.go b/pkg/known/labels.go
index 55afb70..0a676f8 100644
--- a/pkg/known/labels.go
+++ b/pkg/known/labels.go
@@ -18,6 +18,7 @@ package known
 
 // label key
 const (
+	NodeLabelsKeyPrefix       = "node.clusternet.io/"
 	ClusterRegisteredByLabel  = "clusters.clusternet.io/registered-by"
 	ClusterIDLabel            = "clusters.clusternet.io/cluster-id"
 	ClusterNameLabel          = "clusters.clusternet.io/cluster-name"
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index b2d1adb..752ee9e 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -240,7 +240,10 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 
 	sub, err := sched.subsLister.Subscriptions(ns).Get(name)
 	if err != nil {
-		utilruntime.HandleError(err)
+		if !errors.IsNotFound(err) {
+			utilruntime.HandleError(err)
+			sched.SchedulingQueue.AddRateLimited(key)
+		}
 		return
 	}
 	klog.V(3).InfoS("Attempting to schedule subscription", "subscription", klog.KObj(sub))
@@ -251,6 +254,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 		if err != nil {
 			if !errors.IsNotFound(err) {
 				utilruntime.HandleError(err)
+				sched.SchedulingQueue.AddRateLimited(key)
 			}
 			return
 		}
@@ -276,7 +280,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 
 	scheduleResult, err := sched.scheduleAlgorithm.Schedule(schedulingCycleCtx, sched.framework, state, sub, finv)
 	if err != nil {
-		sched.recordSchedulingFailure(sub, err, ReasonUnschedulable)
+		sched.handleSchedulingFailure(sub, err, ReasonUnschedulable)
 		if !strings.Contains(err.Error(), "clusters are available") {
 			return
 		}
@@ -289,7 +293,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 		metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 		// trigger un-reserve to clean up state associated with the reserved subscription
 		sched.framework.RunReservePluginsUnreserve(schedulingCycleCtx, state, sub, targetClusters)
-		sched.recordSchedulingFailure(sub, sts.AsError(), SchedulerError)
+		sched.handleSchedulingFailure(sub, sts.AsError(), SchedulerError)
 		return
 	}
 
@@ -306,7 +310,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 		}
 		// One of the plugins returned status different from success or wait.
 		sched.framework.RunReservePluginsUnreserve(schedulingCycleCtx, state, sub, targetClusters)
-		sched.recordSchedulingFailure(sub, runPermitStatus.AsError(), reason)
+		sched.handleSchedulingFailure(sub, runPermitStatus.AsError(), reason)
 		return
 	}
 
@@ -329,7 +333,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 			}
 			// trigger un-reserve plugins to clean up state associated with the reserved subscription
 			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, state, sub, targetClusters)
-			sched.recordSchedulingFailure(sub, waitOnPermitStatus.AsError(), reason)
+			sched.handleSchedulingFailure(sub, waitOnPermitStatus.AsError(), reason)
 			return
 		}
 
@@ -339,7 +343,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 			metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 			// trigger un-reserve plugins to clean up state associated with the reserved subscription
 			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, state, sub, targetClusters)
-			sched.recordSchedulingFailure(sub, preBindStatus.AsError(), SchedulerError)
+			sched.handleSchedulingFailure(sub, preBindStatus.AsError(), SchedulerError)
 			return
 		}
 
@@ -348,7 +352,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 			metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 			// trigger un-reserve plugins to clean up state associated with the reserved subscription
 			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, state, sub, targetClusters)
-			sched.recordSchedulingFailure(sub, fmt.Errorf("binding rejected: %w", err), SchedulerError)
+			sched.handleSchedulingFailure(sub, fmt.Errorf("binding rejected: %w", err), SchedulerError)
 		} else {
 			metrics.SubscriptionScheduled(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 
@@ -386,9 +390,9 @@ func (sched *Scheduler) bind(ctx context.Context, state *framework.CycleState, s
 	return fmt.Errorf("bind status: %s, %v", bindStatus.Code().String(), bindStatus.Message())
 }
 
-// recordSchedulingFailure records an event for the subscription that indicates the
+// handleSchedulingFailure records an event for the subscription that indicates the
 // subscription has failed to schedule. Also, update the subscription condition.
-func (sched *Scheduler) recordSchedulingFailure(sub *appsapi.Subscription, err error, _ string) {
+func (sched *Scheduler) handleSchedulingFailure(sub *appsapi.Subscription, err error, _ string) {
 	klog.V(2).InfoS("Unable to schedule subscription; waiting", "subscription", klog.KObj(sub), "err", err)
 
 	msg := truncateMessage(err.Error())
@@ -434,7 +438,7 @@ func (sched *Scheduler) addAllEventHandlers() {
 				sched.lock.Lock()
 				defer sched.lock.Unlock()
 				sched.subscribersMap[klog.KObj(sub).String()] = sub.Spec.Subscribers
-				sched.SchedulingQueue.AddRateLimited(klog.KObj(sub).String())
+				sched.SchedulingQueue.Add(klog.KObj(sub).String())
 			},
 			UpdateFunc: func(oldObj, newObj interface{}) {
 				oldSub := oldObj.(*appsapi.Subscription)
@@ -449,7 +453,7 @@ func (sched *Scheduler) addAllEventHandlers() {
 				sched.lock.Lock()
 				defer sched.lock.Unlock()
 				sched.subscribersMap[klog.KObj(newSub).String()] = newSub.Spec.Subscribers
-				sched.SchedulingQueue.AddRateLimited(klog.KObj(newSub).String())
+				sched.SchedulingQueue.Add(klog.KObj(newSub).String())
 			},
 		},
 	})
@@ -457,7 +461,7 @@ func (sched *Scheduler) addAllEventHandlers() {
 	sched.ClusternetInformerFactory.Apps().V1alpha1().FeedInventories().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			finv := obj.(*appsapi.FeedInventory)
-			sched.SchedulingQueue.AddRateLimited(klog.KObj(finv).String())
+			sched.SchedulingQueue.Add(klog.KObj(finv).String())
 		},
 		UpdateFunc: func(oldObj, newObj interface{}) {
 			oldInventory := oldObj.(*appsapi.FeedInventory)
@@ -466,7 +470,7 @@ func (sched *Scheduler) addAllEventHandlers() {
 				// Periodic resync will send update events for all known Inventory.
 				return
 			}
-			sched.SchedulingQueue.AddRateLimited(klog.KObj(newInventory).String())
+			sched.SchedulingQueue.Add(klog.KObj(newInventory).String())
 		},
 	})
 
@@ -484,7 +488,7 @@ func (sched *Scheduler) addAllEventHandlers() {
 				if !selector.Matches(labels.Set(newMcls.Labels)) && oldMcls != nil && !selector.Matches(labels.Set(oldMcls.Labels)) {
 					continue
 				}
-				sched.SchedulingQueue.AddRateLimited(key)
+				sched.SchedulingQueue.Add(key)
 				break
 			}
 		}


Test patch:
diff --git a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller_test.go b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller_test.go
new file mode 100644
index 0000000..7dbe3cd
--- /dev/null
+++ b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller_test.go
@@ -0,0 +1,61 @@
+package clusterstatus
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func BuildNode(name string, labels map[string]string) *corev1.Node {
+	node := &corev1.Node{
+		TypeMeta: metav1.TypeMeta{Kind: "Node"},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: labels,
+		},
+	}
+	return node
+}
+
+type StatusSuite struct {
+	nodes []*corev1.Node
+	suite.Suite
+}
+
+func TestDeployment(t *testing.T) {
+	suite.Run(t, new(StatusSuite))
+}
+
+func (suite *StatusSuite) SetupTest() {
+	suite.nodes = []*corev1.Node{
+		0: BuildNode("node0", map[string]string{
+			"node.clusternet.io/k1": "v1",
+			"node.clusternet.io/k2": "v2",
+		}),
+		1: BuildNode("node1", map[string]string{
+			"node.clusternet.io/k1": "v1",
+			"node.clusternet.io/k2": "v2",
+			"node.clusternet.io/k3": "v3",
+		}),
+		2: BuildNode("node2", map[string]string{
+			"node.clusternet.io/k1": "v1",
+			"node.clusternet.io/k2": "v2",
+			"node.clusternet.io/k4": "v4",
+		}),
+		3: BuildNode("node3", map[string]string{
+			"node.clusternet.io/k1": "v1",
+			"node.clusternet.io/k2": "v2",
+			"node.clusternet.io/k3": "v3",
+			"node.clusternet.io/k4": "v4",
+		}),
+	}
+}
+
+func (suite *StatusSuite) TestGetCommonNodeLabels() {
+	commonLabels := getCommonNodeLabels(suite.nodes)
+	suite.Equal(
+		map[string]string{"node.clusternet.io/k1": "v1", "node.clusternet.io/k2": "v2"},
+		commonLabels, "failed get common labels from nodes")
+}