Repository: clusternet/clusternet
Pull request #679 (2023-05-08 10:44:22)

user-defined prefixes for label aggregation
#### What type of PR is this?
kind/feature

#### What this PR does / why we need it:
In old version , we hard code label prefix for aggregate.
We need aggregate label by custom in more cases.

#### Which issue(s) this PR fixes:
If there are multiple key-value pairs with same key whose occurrence ratio is greater than the threshold, we need to aggregate the key-value pairs that appear more times, not the last one.

#### Special notes for your reviewer:


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

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)

```diff
@@            Coverage Diff             @@
##             main     #679      +/-   ##
==========================================
+ Coverage   13.60%   14.10%   +0.50%     
==========================================
  Files          66       72       +6     
  Lines        7492     7921     +429     
==========================================
+ Hits         1019     1117      +98     
- Misses       6391     6720     +329     
- Partials       82       84       +2     
```


[see 42 files with indirect coverage changes](https://app.codecov.io/gh/clusternet/clusternet/pull/679/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)


dixudx: cc @lmxia for a review
yinsenyan: @dixudx How to re-run make test ? 

Patch:
diff --git a/pkg/agent/options/cluster_reg_options.go b/pkg/agent/options/cluster_reg_options.go
index d465e81..b5da3b0 100644
--- a/pkg/agent/options/cluster_reg_options.go
+++ b/pkg/agent/options/cluster_reg_options.go
@@ -31,6 +31,7 @@ import (
 
 	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
 	"github.com/clusternet/clusternet/pkg/features"
+	"github.com/clusternet/clusternet/pkg/known"
 )
 
 var validateClusterNameRegex = regexp.MustCompile(nameFmt)
@@ -67,6 +68,9 @@ type ClusterRegistrationOptions struct {
 	// e.g LabelAggregateThreshold 0.8  means 80% of work nodes in child clusters labels in common will be aggregated to parent.
 	LabelAggregateThreshold float32
 
+	// LabelAggregatePrefix  specifies the prefix of node label which can be aggregate
+	LabelAggregatePrefix []string
+
 	// TODO: check ca hash
 }
 
@@ -79,6 +83,7 @@ func NewClusterRegistrationOptions() *ClusterRegistrationOptions {
 		ClusterStatusReportFrequency:  metav1.Duration{Duration: DefaultClusterStatusReportFrequency},
 		ClusterStatusCollectFrequency: metav1.Duration{Duration: DefaultClusterStatusCollectFrequency},
 		LabelAggregateThreshold:       0.8,
+		LabelAggregatePrefix:          []string{known.NodeLabelsKeyPrefix},
 	}
 }
 
@@ -109,6 +114,8 @@ func (opts *ClusterRegistrationOptions) AddFlagSets(fss *cliflag.NamedFlagSets)
 		"Specify the labels for the child cluster, split by `,`")
 	mgmtfs.Float32Var(&opts.LabelAggregateThreshold, LabelAggregateThreshold, opts.LabelAggregateThreshold,
 		"Specifies the threshold of common labels in child cluster nodes should be aggregated to parent")
+	mgmtfs.StringArrayVar(&opts.LabelAggregatePrefix, LabelAggregatePrefix, opts.LabelAggregatePrefix,
+		"Specifies the desired prefixes of node labels to be aggregated")
 
 	mfs := fss.FlagSet("cluster health status")
 	mfs.DurationVar(&opts.ClusterStatusReportFrequency.Duration, ClusterStatusReportFrequency, opts.ClusterStatusReportFrequency.Duration,
diff --git a/pkg/agent/options/constant.go b/pkg/agent/options/constant.go
index cb49ffa..9e7d192 100644
--- a/pkg/agent/options/constant.go
+++ b/pkg/agent/options/constant.go
@@ -67,6 +67,9 @@ const (
 
 	// LabelAggregateThreshold 0.8 means 80% of work nodes in child clusters labels in common will be aggregated to parent.
 	LabelAggregateThreshold = "labels-aggregate-threshold"
+
+	// LabelAggregatePrefix means prefix can be aggregate
+	LabelAggregatePrefix = "labels-aggregate-prefix"
 )
 
 // default values
diff --git a/pkg/agent/status_manager.go b/pkg/agent/status_manager.go
index cda4e1e..d1d8ad2 100644
--- a/pkg/agent/status_manager.go
+++ b/pkg/agent/status_manager.go
@@ -22,7 +22,6 @@ import (
 	"errors"
 	"os"
 	"reflect"
-	"strings"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -42,6 +41,7 @@ import (
 	"github.com/clusternet/clusternet/pkg/controllers/clusters/clusterstatus"
 	clusternetclientset "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
 	"github.com/clusternet/clusternet/pkg/known"
+	"github.com/clusternet/clusternet/pkg/utils"
 )
 
 type Manager struct {
@@ -51,6 +51,8 @@ type Manager struct {
 	clusterStatusController *clusterstatus.Controller
 
 	managedCluster *clusterapi.ManagedCluster
+
+	labelAggregatePrefix []string
 }
 
 func NewStatusManager(
@@ -74,6 +76,7 @@ func NewStatusManager(
 			opts.ClusterRegistrationOptions.ClusterStatusReportFrequency,
 			opts.ClusterRegistrationOptions.LabelAggregateThreshold,
 		),
+		labelAggregatePrefix: opts.ClusterRegistrationOptions.LabelAggregatePrefix,
 	}
 }
 
@@ -167,14 +170,14 @@ func (mgr *Manager) updateClusterStatus(ctx context.Context, namespace, clusterI
 
 func (mgr *Manager) getManagedClusterNewLabels() map[string]string {
 	originLabels := mgr.managedCluster.GetLabels()
-	modifiedLabels := map[string]string{}
+	modifiedLabels := make(map[string]string)
 	for k, v := range originLabels {
-		if strings.HasPrefix(k, known.NodeLabelsKeyPrefix) {
+		if utils.ContainsPrefix(mgr.labelAggregatePrefix, k) {
 			continue
 		}
 		modifiedLabels[k] = v
 	}
-	return labels.Merge(modifiedLabels, mgr.clusterStatusController.GetManagedClusterLabels())
+	return labels.Merge(modifiedLabels, mgr.clusterStatusController.GetManagedClusterLabels(mgr.labelAggregatePrefix))
 }
 
 func patchManagedClusterTwoWayMergeLabels(client clusternetclientset.Interface, mcls *clusterapi.ManagedCluster,
diff --git a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
index 318d124..0b4acd9 100644
--- a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
+++ b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
@@ -40,6 +40,7 @@ import (
 	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
 	"github.com/clusternet/clusternet/pkg/features"
 	"github.com/clusternet/clusternet/pkg/known"
+	"github.com/clusternet/clusternet/pkg/utils"
 )
 
 // Controller is a controller that collects cluster status
@@ -188,13 +189,13 @@ func (c *Controller) GetClusterStatus() *clusterapi.ManagedClusterStatus {
 	return c.clusterStatus.DeepCopy()
 }
 
-func (c *Controller) GetManagedClusterLabels() labels.Set {
+func (c *Controller) GetManagedClusterLabels(labelPrefixes []string) labels.Set {
 	nodes, err := c.nodeLister.List(labels.Everything())
 	if err != nil {
 		klog.Warningf("failed to list nodes: %v", err)
 		return nil
 	}
-	return aggregateLimitedLabels(nodes, c.labelAggregateThreshold)
+	return aggregateLimitedLabels(nodes, c.labelAggregateThreshold, labelPrefixes)
 }
 
 func (c *Controller) getKubernetesVersion(_ context.Context) (*version.Info, error) {
@@ -388,37 +389,57 @@ func isMasterNode(labels map[string]string) bool {
 	return false
 }
 
-func statisticNodesLabels(nodes []*corev1.Node) (map[string]int, int) {
+func statisticNodesLabels(nodes []*corev1.Node, labelPrefixes []string) (map[string]int, int) {
 	// statistic map.
-	countMap := make(map[string]int, 0)
+	countMap := make(map[string]map[string]int, 0)
 	masterNodeNum := 0
 	for _, node := range nodes {
 		if isMasterNode(node.Labels) {
 			masterNodeNum += 1
 			continue
 		}
-		for k, v := range node.Labels {
-			if strings.HasPrefix(k, known.NodeLabelsKeyPrefix) {
-				countKey := strings.Join([]string{k, v}, "/")
-				if total, exist := countMap[countKey]; exist {
-					countMap[countKey] = total + 1
+		for labelKey, labelValue := range node.Labels {
+			if utils.ContainsPrefix(labelPrefixes, labelKey) {
+				label := strings.Join([]string{labelKey, labelValue}, "=")
+				if labelMap, isok := countMap[labelKey]; isok {
+					if count, isok2 := labelMap[label]; isok2 {
+						countMap[labelKey][label] = count + 1
+					} else {
+						countMap[labelKey][label] = 1
+					}
 				} else {
-					countMap[countKey] = 1
+					countMap[labelKey] = map[string]int{label: 1}
 				}
 			}
 		}
 	}
-	return countMap, masterNodeNum
+	return filterMaxPair(countMap), masterNodeNum
+}
+
+func filterMaxPair(m map[string]map[string]int) map[string]int {
+	result := make(map[string]int)
+	for _, labelMap := range m {
+		var label string
+		var count int
+		for k, v := range labelMap {
+			if count < v {
+				label = k
+				count = v
+			}
+		}
+		result[label] = count
+	}
+	return result
 }
 
-func aggregateLimitedLabels(nodes []*corev1.Node, threshold float32) map[string]string {
+func aggregateLimitedLabels(nodes []*corev1.Node, threshold float32, labelPrefixes []string) map[string]string {
 	newMap := make(map[string]string, 0)
-	countMap, masterNum := statisticNodesLabels(nodes)
+	countMap, masterNum := statisticNodesLabels(nodes, labelPrefixes)
 	workNodeNum := len(nodes) - masterNum
 	for k, v := range countMap {
 		// the key is higher than threshold
 		if float32(v)/float32(workNodeNum) >= threshold {
-			lastInd := strings.LastIndex(k, "/")
+			lastInd := strings.LastIndex(k, "=")
 			newMap[k[:lastInd]] = k[lastInd+1:]
 		}
 	}


Test patch:
diff --git a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller_test.go b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller_test.go
index 29218c1..4e26243 100644
--- a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller_test.go
+++ b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller_test.go
@@ -1,11 +1,14 @@
 package clusterstatus
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/suite"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/clusternet/clusternet/pkg/known"
 )
 
 func BuildNode(name string, labels map[string]string) *corev1.Node {
@@ -85,7 +88,7 @@ func (suite *StatusSuite) TestGetSingleNodeCommonNodeLabels() {
 }
 
 func (suite *StatusSuite) TestAggregateLimitedLabels1() {
-	commonLabels := aggregateLimitedLabels(suite.nodes, 0.8)
+	commonLabels := aggregateLimitedLabels(suite.nodes, 0.8, []string{known.NodeLabelsKeyPrefix})
 	suite.Equal(
 		map[string]string{
 			"node.clusternet.io/k1": "v1",
@@ -95,7 +98,7 @@ func (suite *StatusSuite) TestAggregateLimitedLabels1() {
 }
 
 func (suite *StatusSuite) TestAggregateLimitedLabels2() {
-	commonLabels := aggregateLimitedLabels(suite.nodes, 0.7)
+	commonLabels := aggregateLimitedLabels(suite.nodes, 0.7, []string{known.NodeLabelsKeyPrefix})
 	suite.Equal(
 		map[string]string{
 			"node.clusternet.io/k1": "v1",
@@ -106,7 +109,7 @@ func (suite *StatusSuite) TestAggregateLimitedLabels2() {
 }
 
 func (suite *StatusSuite) TestAggregateLimitedLabels3() {
-	commonLabels := aggregateLimitedLabels(suite.nodes, 0.5)
+	commonLabels := aggregateLimitedLabels(suite.nodes, 0.5, []string{known.NodeLabelsKeyPrefix})
 	suite.Equal(
 		map[string]string{
 			"node.clusternet.io/k1": "v1",
@@ -118,7 +121,7 @@ func (suite *StatusSuite) TestAggregateLimitedLabels3() {
 }
 
 func (suite *StatusSuite) TestAggregateLimitedSingleLabels() {
-	commonLabels := aggregateLimitedLabels(suite.nodes[0:1], 1)
+	commonLabels := aggregateLimitedLabels(suite.nodes[0:1], 1, []string{known.NodeLabelsKeyPrefix})
 	suite.Equal(
 		map[string]string{
 			"node.clusternet.io/k1": "v1",
@@ -127,3 +130,97 @@ func (suite *StatusSuite) TestAggregateLimitedSingleLabels() {
 		},
 		commonLabels, "failed to aggregate single node")
 }
+
+func TestAggregateLimitedLabels(t *testing.T) {
+	type arg struct {
+		nodes         []*corev1.Node
+		threshold     float32
+		labelPrefixes []string
+	}
+	var tests = []struct {
+		name string
+		args arg
+		want map[string]string
+	}{
+		{
+			name: "aggregate one",
+			args: arg{
+				nodes:         []*corev1.Node{BuildNode("n0", map[string]string{"foo": "bar", "name": "n0"})},
+				threshold:     0,
+				labelPrefixes: []string{"foo"},
+			},
+			want: map[string]string{"foo": "bar"},
+		},
+		{
+			name: "aggretate with 100% threshold case 1",
+			args: arg{
+				nodes: []*corev1.Node{
+					BuildNode("n0", map[string]string{"foo": "bar", "name": "n0"}),
+					BuildNode("n1", map[string]string{"foo": "bar", "name": "n1"}),
+					BuildNode("n3", map[string]string{"name": "n2"}),
+				},
+				threshold:     1,
+				labelPrefixes: []string{"foo"},
+			},
+			want: map[string]string{},
+		},
+		{
+			name: "aggretate with 100% threshold case 2",
+			args: arg{
+				nodes: []*corev1.Node{
+					BuildNode("n0", map[string]string{"foo": "bar", "name": "n0"}),
+					BuildNode("n1", map[string]string{"foo": "bar", "name": "n1"}),
+					BuildNode("n3", map[string]string{"foo": "bar", "name": "n2"}),
+				},
+				threshold:     1,
+				labelPrefixes: []string{"foo"},
+			},
+			want: map[string]string{"foo": "bar"},
+		},
+		{
+			name: "aggretate with mutil prefix case 1",
+			args: arg{
+				nodes: []*corev1.Node{
+					BuildNode("n0", map[string]string{"foo": "bar", "zone": "zone0"}),
+					BuildNode("n1", map[string]string{"foo": "bar", "zone": "zone0"}),
+					BuildNode("n3", map[string]string{"foo": "bar", "zone": "zone1"}),
+				},
+				threshold:     0.5,
+				labelPrefixes: []string{"foo", "zone"},
+			},
+			want: map[string]string{"foo": "bar", "zone": "zone0"},
+		},
+		{
+			name: "aggretate with mutil prefix case 2",
+			args: arg{
+				nodes: []*corev1.Node{
+					BuildNode("n0", map[string]string{"foo": "bar", "zone": "zone0"}),
+					BuildNode("n1", map[string]string{"foo": "bar", "zone": "zone0"}),
+					BuildNode("n3", map[string]string{"foo": "bar", "zone": "zone1"}),
+				},
+				threshold:     0.1,
+				labelPrefixes: []string{"foo", "zone"},
+			},
+			want: map[string]string{"foo": "bar", "zone": "zone0"},
+		},
+		{
+			name: "aggretate with mutil prefix case 3",
+			args: arg{
+				nodes: []*corev1.Node{
+					BuildNode("n0", map[string]string{"foo/abc": "bar", "zone": "zone0"}),
+					BuildNode("n1", map[string]string{"foo/def": "bar", "zone": "zone1"}),
+					BuildNode("n3", map[string]string{"zone": "zone1"}),
+				},
+				threshold:     0.1,
+				labelPrefixes: []string{"foo", "zone"},
+			},
+			want: map[string]string{"foo/abc": "bar", "foo/def": "bar", "zone": "zone1"},
+		},
+	}
+	for _, tt := range tests {
+		newMap := aggregateLimitedLabels(tt.args.nodes, tt.args.threshold, tt.args.labelPrefixes)
+		if !reflect.DeepEqual(newMap, tt.want) {
+			t.Errorf("error of aggregateLimitedLabels %s,\n want %v\n  got %v", tt.name, tt.want, newMap)
+		}
+	}
+}