From 38665e319db21ef4db59e0391e5a795150e548b8 Mon Sep 17 00:00:00 2001 From: Kunshuai Zhu Date: Tue, 15 Mar 2022 13:03:10 +0800 Subject: [PATCH] Feat: poll multi-cluster metrics and export to prometheus (#3429) * Feat: poll multi-cluster metrics and export to prometheus Signed-off-by: zhukunshuai * pass context to polling loop Signed-off-by: zhukunshuai * move metrics definition to montitor/metrics/multicluster.go Signed-off-by: zhukunshuai * remove pod usage metric and make reviewable Signed-off-by: zhukunshuai * revert the change of GetClusterMetricsFromMetricsAPI Signed-off-by: zhukunshuai * revert the change of GetClusterMetricsFromMetricsAPI Signed-off-by: zhukunshuai * Separate the polling logic into a function Signed-off-by: zhukunshuai * add start menber function Signed-off-by: zhukunshuai * make refreshPeriod a menber var Signed-off-by: zhukunshuai * fix typo Signed-off-by: zhukunshuai --- cmd/core/main.go | 15 ++- pkg/monitor/metrics/multicluster.go | 97 +++++++++++++++++++ pkg/monitor/metrics/workflow.go | 11 +++ .../cluster_metrics_management.go | 68 +++++++++++-- .../cluster_metrics_management_test.go | 8 +- 5 files changed, 187 insertions(+), 12 deletions(-) create mode 100644 pkg/monitor/metrics/multicluster.go diff --git a/cmd/core/main.go b/cmd/core/main.go index 8f9eadd7e..5c20c554d 100644 --- a/cmd/core/main.go +++ b/cmd/core/main.go @@ -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()) diff --git a/pkg/monitor/metrics/multicluster.go b/pkg/monitor/metrics/multicluster.go new file mode 100644 index 000000000..9765be236 --- /dev/null +++ b/pkg/monitor/metrics/multicluster.go @@ -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"}) +) diff --git a/pkg/monitor/metrics/workflow.go b/pkg/monitor/metrics/workflow.go index 4861dc0ce..1d405501b 100644 --- a/pkg/monitor/metrics/workflow.go +++ b/pkg/monitor/metrics/workflow.go @@ -46,6 +46,17 @@ var collectorGroup = []prometheus.Collector{ ApplicationReconcileTimeHistogram, ApplyComponentTimeHistogram, ResourceTrackerNumberGauge, + ClusterIsConnectedGauge, + ClusterWorkerNumberGauge, + ClusterMasterNumberGauge, + ClusterMemoryCapacityGauge, + ClusterCPUCapacityGauge, + ClusterPodCapacityGauge, + ClusterMemoryAllocatableGauge, + ClusterCPUAllocatableGauge, + ClusterPodAllocatableGauge, + ClusterMemoryUsageGauge, + ClusterCPUUsageGauge, } func init() { diff --git a/pkg/multicluster/cluster_metrics_management.go b/pkg/multicluster/cluster_metrics_management.go index 33085ad59..fddfb3f0b 100644 --- a/pkg/multicluster/cluster_metrics_management.go +++ b/pkg/multicluster/cluster_metrics_management.go @@ -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())) + } } diff --git a/pkg/multicluster/cluster_metrics_management_test.go b/pkg/multicluster/cluster_metrics_management_test.go index b2e8ec4ea..95c51eee3 100644 --- a/pkg/multicluster/cluster_metrics_management_test.go +++ b/pkg/multicluster/cluster_metrics_management_test.go @@ -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) {