Merge pull request #469 from WangZzzhe/dev/add-rodan-metrics

feat(metric): add rodan metric provisioner

Author: nightmeng

cmd/katalyst-agent/app/options/metaserver/metaserver.go

@@ -56,6 +56,8 @@ const defaultCustomNodeResourceCacheTTL = 15 * time.Second
 
 const defaultCustomNodeConfigCacheTTL = 15 * time.Second
 
+const defaultRodanServerPort = 9102
+
 // MetaServerOptions holds all the configurations for metaserver.
 // we will not try to separate this structure into several individual
 // structures since it will not be used directly by other components; instead,
@@ -79,6 +81,7 @@ type MetaServerOptions struct {
 	// configurations for metric-fetcher
 	MetricInsurancePeriod time.Duration
 	MetricProvisions      []string
+	RodanServerPort       int
 
 	// configurations for pod-cache
 	KubeletPodCacheSyncPeriod    time.Duration
@@ -109,6 +112,7 @@ func NewMetaServerOptions() *MetaServerOptions {
 
 		MetricInsurancePeriod: defaultMetricInsurancePeriod,
 		MetricProvisions:      []string{metaserver.MetricProvisionerMalachite, metaserver.MetricProvisionerKubelet},
+		RodanServerPort:       defaultRodanServerPort,
 
 		KubeletPodCacheSyncPeriod:    defaultKubeletPodCacheSyncPeriod,
 		KubeletPodCacheSyncMaxRate:   defaultKubeletPodCacheSyncMaxRate,
@@ -150,6 +154,8 @@ func (o *MetaServerOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		"The meta server return metric data and MetricDataExpired if the update time of metric data is earlier than this period.")
 	fs.StringSliceVar(&o.MetricProvisions, "metric-provisioners", o.MetricProvisions,
 		"The provisioners that should be enabled by default")
+	fs.IntVar(&o.RodanServerPort, "rodan-server-port", o.RodanServerPort,
+		"The rodan metric provisioner server port")
 
 	fs.DurationVar(&o.KubeletPodCacheSyncPeriod, "kubelet-pod-cache-sync-period", o.KubeletPodCacheSyncPeriod,
 		"The period of meta server to sync pod from kubelet 10255 port")
@@ -183,6 +189,7 @@ func (o *MetaServerOptions) ApplyTo(c *metaserver.MetaServerConfiguration) error
 
 	c.MetricInsurancePeriod = o.MetricInsurancePeriod
 	c.MetricProvisions = o.MetricProvisions
+	c.RodanServerPort = o.RodanServerPort
 
 	c.KubeletPodCacheSyncPeriod = o.KubeletPodCacheSyncPeriod
 	c.KubeletPodCacheSyncMaxRate = rate.Limit(o.KubeletPodCacheSyncMaxRate)

pkg/config/agent/metaserver/agent.go

@@ -26,11 +26,17 @@ const (
 	MetricProvisionerMalachite = "malachite"
 	MetricProvisionerCgroup    = "cgroup"
 	MetricProvisionerKubelet   = "kubelet"
+	MetricProvisionerRodan     = "rodan"
 )
 
 type MetricConfiguration struct {
 	MetricInsurancePeriod time.Duration
 	MetricProvisions      []string
+	*RodanMetricConfiguration
+}
+
+type RodanMetricConfiguration struct {
+	RodanServerPort int
 }
 
 type PodConfiguration struct {
@@ -64,10 +70,12 @@ type AgentConfiguration struct {
 
 func NewAgentConfiguration() *AgentConfiguration {
 	return &AgentConfiguration{
-		MetricConfiguration: &MetricConfiguration{},
-		PodConfiguration:    &PodConfiguration{},
-		NodeConfiguration:   &NodeConfiguration{},
-		CNRConfiguration:    &CNRConfiguration{},
-		CNCConfiguration:    &CNCConfiguration{},
+		MetricConfiguration: &MetricConfiguration{
+			RodanMetricConfiguration: &RodanMetricConfiguration{},
+		},
+		PodConfiguration:  &PodConfiguration{},
+		NodeConfiguration: &NodeConfiguration{},
+		CNRConfiguration:  &CNRConfiguration{},
+		CNCConfiguration:  &CNCConfiguration{},
 	}
 }

pkg/metaserver/agent/metric/metric_impl.go

@@ -233,7 +233,7 @@ func NewMetricsFetcher(baseConf *global.BaseConfiguration, metricConf *metaserve
 	var enabledProvisioners []types.MetricsProvisioner
 	for _, name := range metricConf.MetricProvisions {
 		if f, ok := registeredProvisioners[name]; ok {
-			enabledProvisioners = append(enabledProvisioners, f(baseConf, emitter, podFetcher, metricStore))
+			enabledProvisioners = append(enabledProvisioners, f(baseConf, metricConf, emitter, podFetcher, metricStore))
 		}
 	}
 

pkg/metaserver/agent/metric/metric_provisoner.go

@@ -24,6 +24,7 @@ import (
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/cgroup"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/kubelet"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/malachite"
+	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/rodan"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/types"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/pod"
 	"github.com/kubewharf/katalyst-core/pkg/metrics"
@@ -34,9 +35,10 @@ func init() {
 	RegisterProvisioners(metaserver.MetricProvisionerMalachite, malachite.NewMalachiteMetricsProvisioner)
 	RegisterProvisioners(metaserver.MetricProvisionerKubelet, kubelet.NewKubeletSummaryProvisioner)
 	RegisterProvisioners(metaserver.MetricProvisionerCgroup, cgroup.NewCGroupMetricsProvisioner)
+	RegisterProvisioners(metaserver.MetricProvisionerRodan, rodan.NewRodanMetricsProvisioner)
 }
 
-type ProvisionerInitFunc func(baseConf *global.BaseConfiguration,
+type ProvisionerInitFunc func(baseConf *global.BaseConfiguration, metricConf *metaserver.MetricConfiguration,
 	emitter metrics.MetricEmitter, fetcher pod.PodFetcher, metricStore *utilmetric.MetricStore) types.MetricsProvisioner
 
 // provisioners stores the initializing function for each-provisioner

pkg/metaserver/agent/metric/provisioner/cgroup/provisioner.go

@@ -20,14 +20,15 @@ import (
 	"context"
 
 	"github.com/kubewharf/katalyst-core/pkg/config/agent/global"
+	"github.com/kubewharf/katalyst-core/pkg/config/agent/metaserver"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/types"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/pod"
 	"github.com/kubewharf/katalyst-core/pkg/metrics"
 	utilmetric "github.com/kubewharf/katalyst-core/pkg/util/metric"
 )
 
 // NewCGroupMetricsProvisioner returns the default implementation of CGroup.
-func NewCGroupMetricsProvisioner(baseConf *global.BaseConfiguration,
+func NewCGroupMetricsProvisioner(baseConf *global.BaseConfiguration, _ *metaserver.MetricConfiguration,
 	emitter metrics.MetricEmitter, _ pod.PodFetcher, metricStore *utilmetric.MetricStore) types.MetricsProvisioner {
 	return &CGroupMetricsProvisioner{
 		metricStore: metricStore,

pkg/metaserver/agent/metric/provisioner/kubelet/provisioner.go

@@ -24,6 +24,7 @@ import (
 	statsapi "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
 
 	"github.com/kubewharf/katalyst-core/pkg/config/agent/global"
+	"github.com/kubewharf/katalyst-core/pkg/config/agent/metaserver"
 	"github.com/kubewharf/katalyst-core/pkg/consts"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/kubelet/client"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/types"
@@ -36,7 +37,7 @@ const (
 	metricsNamKubeletSummaryUnHealthy = "kubelet_summary_unhealthy"
 )
 
-func NewKubeletSummaryProvisioner(baseConf *global.BaseConfiguration,
+func NewKubeletSummaryProvisioner(baseConf *global.BaseConfiguration, _ *metaserver.MetricConfiguration,
 	emitter metrics.MetricEmitter, _ pod.PodFetcher, metricStore *utilmetric.MetricStore) types.MetricsProvisioner {
 	return &KubeletSummaryProvisioner{
 		metricStore: metricStore,

pkg/metaserver/agent/metric/provisioner/malachite/provisioner.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"github.com/kubewharf/katalyst-core/pkg/config/agent/global"
+	"github.com/kubewharf/katalyst-core/pkg/config/agent/metaserver"
 	"github.com/kubewharf/katalyst-core/pkg/consts"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/malachite/client"
 	malachitetypes "github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/malachite/types"
@@ -53,7 +54,7 @@ const (
 )
 
 // NewMalachiteMetricsProvisioner returns the default implementation of MetricsFetcher.
-func NewMalachiteMetricsProvisioner(baseConf *global.BaseConfiguration,
+func NewMalachiteMetricsProvisioner(baseConf *global.BaseConfiguration, _ *metaserver.MetricConfiguration,
 	emitter metrics.MetricEmitter, fetcher pod.PodFetcher, metricStore *utilmetric.MetricStore) types.MetricsProvisioner {
 	return &MalachiteMetricsProvisioner{
 		malachiteClient: client.NewMalachiteClient(fetcher),

pkg/metaserver/agent/metric/provisioner/malachite/provisioner_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/kubewharf/katalyst-core/pkg/config/agent/global"
+	"github.com/kubewharf/katalyst-core/pkg/config/agent/metaserver"
 	malachitetypes "github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/malachite/types"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/pod"
 	"github.com/kubewharf/katalyst-core/pkg/metrics"
@@ -32,7 +33,7 @@ func Test_noneExistMetricsProvisioner(t *testing.T) {
 	store := utilmetric.NewMetricStore()
 
 	var err error
-	implement := NewMalachiteMetricsProvisioner(&global.BaseConfiguration{}, metrics.DummyMetrics{}, &pod.PodFetcherStub{}, store)
+	implement := NewMalachiteMetricsProvisioner(&global.BaseConfiguration{}, &metaserver.MetricConfiguration{}, metrics.DummyMetrics{}, &pod.PodFetcherStub{}, store)
 
 	fakeSystemCompute := &malachitetypes.SystemComputeData{
 		CPU: []malachitetypes.CPU{

pkg/metaserver/agent/metric/provisioner/rodan/client/client.go

@@ -0,0 +1,81 @@
+/*
+Copyright 2022 The Katalyst 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 client
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/rodan/types"
+	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/pod"
+)
+
+type RodanClient struct {
+	urls map[string]string
+
+	fetcher pod.PodFetcher
+
+	metricFunc MetricFunc
+}
+
+func NewRodanClient(fetcher pod.PodFetcher, metricFunc MetricFunc, port int) *RodanClient {
+	urls := make(map[string]string)
+	for path := range types.MetricsMap {
+		urls[path] = fmt.Sprintf("http://localhost:%d%s", port, path)
+	}
+
+	if metricFunc == nil {
+		metricFunc = getMetrics
+	}
+
+	return &RodanClient{
+		fetcher:    fetcher,
+		urls:       urls,
+		metricFunc: metricFunc,
+	}
+}
+
+type MetricFunc func(url string, params map[string]string) ([]byte, error)
+
+func getMetrics(url string, params map[string]string) ([]byte, error) {
+	if len(params) != 0 {
+		firstParam := true
+
+		for k, v := range params {
+			if firstParam {
+				url += "?"
+				firstParam = false
+			} else {
+				url += "&"
+			}
+			url = fmt.Sprintf("%s%s=%s", url, k, v)
+		}
+	}
+
+	rsp, err := http.Get(url)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get metrics, url: %v, err: %v", url, err)
+	}
+	defer func() { _ = rsp.Body.Close() }()
+
+	if rsp.StatusCode != 200 {
+		return nil, fmt.Errorf("invalid http response status code %d, status: %s, url: %s", rsp.StatusCode, rsp.Status, url)
+	}
+
+	return io.ReadAll(rsp.Body)
+}