Feat: poll multi-cluster metrics and export to prometheus (#3429)

* Feat: poll multi-cluster metrics and export to prometheus Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * pass context to polling loop Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * move metrics definition to montitor/metrics/multicluster.go Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * remove pod usage metric and make reviewable Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * revert the change of GetClusterMetricsFromMetricsAPI Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * revert the change of GetClusterMetricsFromMetricsAPI Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * Separate the polling logic into a function Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * add start menber function Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * make refreshPeriod a menber var Signed-off-by: zhukunshuai <jookunshuai@gmail.com> * fix typo Signed-off-by: zhukunshuai <jookunshuai@gmail.com>
2026-08-18 20:17:04 +00:00 · 2022-03-15 13:03:10 +08:00
parent 1e0f329304
commit 38665e319d
5 changed files with 187 additions and 12 deletions
@@ -88,6 +88,8 @@ func main() {
 	var renewDeadline time.Duration
 	var retryPeriod time.Duration
 	var enableClusterGateway bool
+	var enableClusterMetrics bool
+	var clusterMetricsInterval time.Duration

 	flag.BoolVar(&useWebhook, "use-webhook", false, "Enable Admission Webhook")
 	flag.StringVar(&certDir, "webhook-cert-dir", "/k8s-webhook-server/serving-certs", "Admission webhook cert/key dir.")
@@ -135,6 +137,8 @@ func main() {
 	flag.DurationVar(&retryPeriod, "leader-election-retry-period", 2*time.Second,
 		"The duration the LeaderElector clients should wait between tries of actions")
 	flag.BoolVar(&enableClusterGateway, "enable-cluster-gateway", false, "Enable cluster-gateway to use multicluster, disabled by default.")
+	flag.BoolVar(&enableClusterMetrics, "enable-cluster-metrics", false, "Enable cluster-metrics-management to collect metrics from clusters with cluster-gateway, disabled by default. When this param is enabled, enable-cluster-gateway should be enabled")
+	flag.DurationVar(&clusterMetricsInterval, "cluster-metrics-interval", 15*time.Second, "The interval that ClusterMetricsMgr will collect metrics from clusters, default value is 15 seconds.")
 	flag.BoolVar(&controllerArgs.EnableCompatibility, "enable-asi-compatibility", false, "enable compatibility for asi")
 	flag.BoolVar(&controllerArgs.IgnoreAppWithoutControllerRequirement, "ignore-app-without-controller-version", false, "If true, application controller will not process the app without 'app.oam.dev/controller-version-require' annotation")
 	standardcontroller.AddOptimizeFlags()
@@ -203,10 +207,19 @@ func main() {

 	// wrapper the round tripper by multi cluster rewriter
 	if enableClusterGateway {
-		if _, err := multicluster.Initialize(restConfig, true); err != nil {
+		client, err := multicluster.Initialize(restConfig, true)
+		if err != nil {
 			klog.ErrorS(err, "failed to enable multi-cluster capability")
 			os.Exit(1)
 		}
+
+		if enableClusterMetrics {
+			_, err := multicluster.NewClusterMetricsMgr(context.Background(), client, clusterMetricsInterval)
+			if err != nil {
+				klog.ErrorS(err, "failed to enable multi-cluster-metrics capability")
+				os.Exit(1)
+			}
+		}
 	}
 	ctrl.SetLogger(klogr.New())

@@ -0,0 +1,97 @@
+/*
+Copyright 2022 The KubeVela 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 metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var (
+	// ClusterIsConnectedGauge report if the cluster is connected
+	ClusterIsConnectedGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "cluster_isconnected",
+		Help: "if cluster is connected.",
+	}, []string{"cluster"})
+
+	// ClusterWorkerNumberGauge report the number of WorkerNumber in cluster
+	ClusterWorkerNumberGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_worker_node_number",
+		Help:        "cluster worker node number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterMasterNumberGauge report the number of MasterNumber in cluster
+	ClusterMasterNumberGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_master_node_number",
+		Help:        "cluster master node number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterMemoryCapacityGauge report the number of MemoryCapacity in cluster
+	ClusterMemoryCapacityGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_memory_capacity",
+		Help:        "cluster memory capacity number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterCPUCapacityGauge report the number of CPUCapacity in cluster
+	ClusterCPUCapacityGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_cpu_capacity",
+		Help:        "cluster cpu capacity number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterPodCapacityGauge report the number of PodCapacity in cluster
+	ClusterPodCapacityGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_pod_capacity",
+		Help:        "cluster pod capacity number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterMemoryAllocatableGauge report the number of MemoryAllocatable in cluster
+	ClusterMemoryAllocatableGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_memory_allocatable",
+		Help:        "cluster memory allocatable number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterCPUAllocatableGauge report the number of CPUAllocatable in cluster
+	ClusterCPUAllocatableGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_cpu_allocatable",
+		Help:        "cluster cpu allocatable number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterPodAllocatableGauge report the number of PodAllocatable in cluster
+	ClusterPodAllocatableGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_pod_allocatable",
+		Help:        "cluster pod allocatable number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterMemoryUsageGauge report the number of MemoryUsage in cluster
+	ClusterMemoryUsageGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_memory_usage",
+		Help:        "cluster memory usage number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+
+	// ClusterCPUUsageGauge report the number of CPUUsage in cluster
+	ClusterCPUUsageGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        "cluster_cpu_usage",
+		Help:        "cluster cpu usage number.",
+		ConstLabels: prometheus.Labels{},
+	}, []string{"cluster"})
+)
@@ -46,6 +46,17 @@ var collectorGroup = []prometheus.Collector{
 	ApplicationReconcileTimeHistogram,
 	ApplyComponentTimeHistogram,
 	ResourceTrackerNumberGauge,
+	ClusterIsConnectedGauge,
+	ClusterWorkerNumberGauge,
+	ClusterMasterNumberGauge,
+	ClusterMemoryCapacityGauge,
+	ClusterCPUCapacityGauge,
+	ClusterPodCapacityGauge,
+	ClusterMemoryAllocatableGauge,
+	ClusterCPUAllocatableGauge,
+	ClusterPodAllocatableGauge,
+	ClusterMemoryUsageGauge,
+	ClusterCPUUsageGauge,
 }

 func init() {
@@ -18,6 +18,9 @@ package multicluster

 import (
 	"context"
+	"time"
+
+	"github.com/oam-dev/kubevela/pkg/monitor/metrics"

 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -28,7 +31,8 @@ var metricsMap map[string]*ClusterMetrics

 // ClusterMetricsMgr manage metrics of clusters
 type ClusterMetricsMgr struct {
-	kubeClient client.Client
+	kubeClient    client.Client
+	refreshPeriod time.Duration
 }

 // ClusterMetricsHelper is the interface that provides operations for cluster metrics
@@ -37,16 +41,17 @@ type ClusterMetricsHelper interface {
 }

 // NewClusterMetricsMgr will create a cluster metrics manager
-func NewClusterMetricsMgr(kubeClient client.Client) (*ClusterMetricsMgr, error) {
+func NewClusterMetricsMgr(ctx context.Context, kubeClient client.Client, refreshPeriod time.Duration) (*ClusterMetricsMgr, error) {
 	mgr := &ClusterMetricsMgr{
-		kubeClient: kubeClient,
+		kubeClient:    kubeClient,
+		refreshPeriod: refreshPeriod,
 	}
-	err := mgr.Refresh()
-	return mgr, err
+	go mgr.Start(ctx)
+	return mgr, nil
 }

 // Refresh will re-collect cluster metrics and refresh cache
-func (cmm *ClusterMetricsMgr) Refresh() error {
+func (cmm *ClusterMetricsMgr) Refresh() ([]VirtualCluster, error) {
 	clusters, _ := ListVirtualClusters(context.Background(), cmm.kubeClient)
 	m := make(map[string]*ClusterMetrics)

@@ -62,13 +67,58 @@ func (cmm *ClusterMetricsMgr) Refresh() error {
 		if err != nil {
 			klog.Warningf("failed to request metrics api of cluster-(%s)", cluster.Name)
 		}
-		metrics := &ClusterMetrics{
+		cm := &ClusterMetrics{
 			IsConnected:         isConnected,
 			ClusterInfo:         clusterInfo,
 			ClusterUsageMetrics: clusterUsageMetrics,
 		}
-		m[cluster.Name] = metrics
+		m[cluster.Name] = cm
+		cluster.Metrics = cm
 	}
 	metricsMap = m
-	return nil
+	return clusters, nil
+}
+
+// Start will start polling cluster api to collect metrics
+func (cmm *ClusterMetricsMgr) Start(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			klog.Warning("Stop cluster metrics polling loop.")
+			return
+		default:
+			clusters, _ := cmm.Refresh()
+			for _, cluster := range clusters {
+				exportMetrics(cluster.Metrics, cluster.Name)
+			}
+			time.Sleep(cmm.refreshPeriod)
+		}
+	}
+}
+
+// exportMetrics will report ClusterMetrics with a clusterName label
+func exportMetrics(m *ClusterMetrics, clusterName string) {
+	if m == nil {
+		return
+	}
+	metrics.ClusterIsConnectedGauge.WithLabelValues(clusterName).Set(func() float64 {
+		if m.IsConnected {
+			return 1
+		}
+		return 0
+	}())
+	if m.ClusterInfo != nil {
+		metrics.ClusterWorkerNumberGauge.WithLabelValues(clusterName).Set(float64(m.ClusterInfo.WorkerNumber))
+		metrics.ClusterMasterNumberGauge.WithLabelValues(clusterName).Set(float64(m.ClusterInfo.MasterNumber))
+		metrics.ClusterMemoryCapacityGauge.WithLabelValues(clusterName).Set(m.ClusterInfo.MemoryCapacity.AsApproximateFloat64())
+		metrics.ClusterCPUCapacityGauge.WithLabelValues(clusterName).Set(float64(m.ClusterInfo.CPUCapacity.MilliValue()))
+		metrics.ClusterPodCapacityGauge.WithLabelValues(clusterName).Set(m.ClusterInfo.PodCapacity.AsApproximateFloat64())
+		metrics.ClusterMemoryAllocatableGauge.WithLabelValues(clusterName).Set(m.ClusterInfo.MemoryAllocatable.AsApproximateFloat64())
+		metrics.ClusterCPUAllocatableGauge.WithLabelValues(clusterName).Set(float64(m.ClusterInfo.CPUAllocatable.MilliValue()))
+		metrics.ClusterPodAllocatableGauge.WithLabelValues(clusterName).Set(m.ClusterInfo.PodAllocatable.AsApproximateFloat64())
+	}
+	if m.ClusterUsageMetrics != nil {
+		metrics.ClusterMemoryUsageGauge.WithLabelValues(clusterName).Set(m.ClusterUsageMetrics.MemoryUsage.AsApproximateFloat64())
+		metrics.ClusterCPUUsageGauge.WithLabelValues(clusterName).Set(float64(m.ClusterUsageMetrics.CPUUsage.MilliValue()))
+	}
 }
@@ -21,6 +21,7 @@ import (
 	"errors"
 	"strconv"
 	"testing"
+	"time"

 	"gotest.tools/assert"

@@ -63,10 +64,10 @@ func TestRefresh(t *testing.T) {
 	fakeClient.AddCluster(NormalClusterName, normalCluster)
 	fakeClient.AddCluster(DisconnectedClusterName, disconnectedCluster)

-	mgr, err := NewClusterMetricsMgr(fakeClient)
+	mgr, err := NewClusterMetricsMgr(context.Background(), fakeClient, 15*time.Second)
 	assert.NilError(t, err)

-	err = mgr.Refresh()
+	_, err = mgr.Refresh()
 	assert.NilError(t, err)

 	clusters, err := ListVirtualClusters(context.Background(), fakeClient)
@@ -83,6 +84,9 @@ func TestRefresh(t *testing.T) {
 	norCluster, err := GetVirtualCluster(context.Background(), fakeClient, NormalClusterName)
 	assert.NilError(t, err)
 	assertClusterMetrics(t, norCluster)
+
+	exportMetrics(disCluster.Metrics, disCluster.Name)
+	exportMetrics(norCluster.Metrics, norCluster.Name)
 }

 func assertClusterMetrics(t *testing.T, cluster *VirtualCluster) {