feat: add a CustomMetricsProvider to the metrics-adapter

Signed-off-by: chaunceyjiang <chaunceyjiang@gmail.com>

artifacts/deploy/karmada-metrics-adapter-apiservice.yaml

@@ -15,6 +15,34 @@ spec:
   version: v1beta1
   versionPriority: 10
 ---
+apiVersion: apiregistration.k8s.io/v1
+kind: APIService
+metadata:
+  name: v1beta2.custom.metrics.k8s.io
+spec:
+  service:
+    name: karmada-metrics-adapter
+    namespace:  karmada-system
+  group: custom.metrics.k8s.io
+  version: v1beta2
+  insecureSkipTLSVerify: true
+  groupPriorityMinimum: 100
+  versionPriority: 200
+---
+apiVersion: apiregistration.k8s.io/v1
+kind: APIService
+metadata:
+  name: v1beta1.custom.metrics.k8s.io
+spec:
+  service:
+    name: karmada-metrics-adapter
+    namespace:  karmada-system
+  group: custom.metrics.k8s.io
+  version: v1beta1
+  insecureSkipTLSVerify: true
+  groupPriorityMinimum: 100
+  versionPriority: 200
+---
 apiVersion: v1
 kind: Service
 metadata:

pkg/metricsadapter/adapter.go

@@ -1,6 +1,10 @@
 package metricsadapter
 
 import (
+	"context"
+
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
 	basecmd "sigs.k8s.io/custom-metrics-apiserver/pkg/cmd"
 	"sigs.k8s.io/custom-metrics-apiserver/pkg/cmd/options"
 
@@ -18,9 +22,13 @@ type MetricsAdapter struct {
 func NewMetricsAdapter(controller *MetricsController, customMetricsAdapterServerOptions *options.CustomMetricsAdapterServerOptions) *MetricsAdapter {
 	adapter := &MetricsAdapter{}
 	adapter.CustomMetricsAdapterServerOptions = customMetricsAdapterServerOptions
-
+	kubeClient := kubernetes.NewForConfigOrDie(controller.restConfig)
+	kubeFactory := informers.NewSharedInformerFactory(kubeClient, 0)
+	kubeFactory.Core().V1().Secrets().Informer()
+	kubeFactory.Start(context.TODO().Done())
+	kubeFactory.WaitForCacheSync(context.TODO().Done())
 	adapter.ResourceMetricsProvider = provider.NewResourceMetricsProvider(controller.ClusterLister, controller.InformerManager)
-	customProvider := provider.MakeCustomMetricsProvider()
+	customProvider := provider.MakeCustomMetricsProvider(controller.ClusterLister, kubeFactory.Core().V1().Secrets().Lister())
 	externalProvider := provider.MakeExternalMetricsProvider()
 	adapter.WithCustomMetrics(customProvider)
 	adapter.WithExternalMetrics(externalProvider)

pkg/metricsadapter/provider/custommetrics.go

@@ -3,33 +3,285 @@ package provider
 import (
 	"context"
 	"fmt"
+	"reflect"
+	"strings"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/discovery"
+	listcorev1 "k8s.io/client-go/listers/core/v1"
+	"k8s.io/klog/v2"
 	"k8s.io/metrics/pkg/apis/custom_metrics"
+	custommetricsv1beta2 "k8s.io/metrics/pkg/apis/custom_metrics/v1beta2"
+	custommetricsclient "k8s.io/metrics/pkg/client/custom_metrics"
+	custommetricsschema "k8s.io/metrics/pkg/client/custom_metrics/scheme"
 	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
+
+	clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1"
+	clusterlister "github.com/karmada-io/karmada/pkg/generated/listers/cluster/v1alpha1"
+	"github.com/karmada-io/karmada/pkg/util"
+)
+
+var (
+	versionConverter = custommetricsclient.NewMetricConverter()
 )
 
 // CustomMetricsProvider is a custom metrics provider
 type CustomMetricsProvider struct {
+	// newDiscoveryClientFunc returns a discovery client for member cluster apiserver
+	newDiscoveryClientFunc func(string) (*discovery.DiscoveryClient, error)
+	clusterLister          clusterlister.ClusterLister
 }
 
 // MakeCustomMetricsProvider creates a new custom metrics provider
-func MakeCustomMetricsProvider() *CustomMetricsProvider {
-	return &CustomMetricsProvider{}
+func MakeCustomMetricsProvider(clusterLister clusterlister.ClusterLister, secretLister listcorev1.SecretLister) *CustomMetricsProvider {
+	return &CustomMetricsProvider{
+		clusterLister:          clusterLister,
+		newDiscoveryClientFunc: discoveryClientForClusterFunc(clusterLister, secretLister),
+	}
 }
 
 // GetMetricByName will query metrics by name from member clusters and return the result
 func (c *CustomMetricsProvider) GetMetricByName(ctx context.Context, name types.NamespacedName, info provider.CustomMetricInfo, metricSelector labels.Selector) (*custom_metrics.MetricValue, error) {
-	return nil, fmt.Errorf("karmada-metrics-adapter still not implement it")
+	clusters, err := c.clusterLister.List(labels.Everything())
+	if err != nil {
+		klog.Errorf("Failed to list clusters: %v", err)
+		return nil, err
+	}
+	metricValueList := &custom_metrics.MetricValueList{}
+	for _, cluster := range clusters {
+		metrics, err := c.getMetricByName(ctx, cluster.Name, name, info, metricSelector)
+		if err != nil {
+			klog.Warningf("query %s's %s metric from cluster %s failed, err: %+v", info.GroupResource.String(), info.Metric, cluster.Name, err)
+			continue
+		}
+		metricValueList.Items = append(metricValueList.Items, metrics.Items...)
+	}
+	if len(metricValueList.Items) != 1 {
+		return nil, fmt.Errorf("the custom metrics API server returned %v results when we asked for exactly one", len(metricValueList.Items))
+	}
+	return &metricValueList.Items[0], nil
 }
 
 // GetMetricBySelector will query metrics by selector from member clusters and return the result
 func (c *CustomMetricsProvider) GetMetricBySelector(ctx context.Context, namespace string, selector labels.Selector, info provider.CustomMetricInfo, metricSelector labels.Selector) (*custom_metrics.MetricValueList, error) {
-	return nil, fmt.Errorf("karmada-metrics-adapter still not implement it")
+	clusters, err := c.clusterLister.List(labels.Everything())
+	if err != nil {
+		klog.Errorf("Failed to list clusters: %v", err)
+		return nil, err
+	}
+	metricValueList := &custom_metrics.MetricValueList{}
+	for _, cluster := range clusters {
+		metrics, err := c.getMetricBySelector(ctx, cluster.Name, namespace, selector, info, metricSelector)
+		if err != nil {
+			klog.Warningf("query %s's %s metric from cluster %s failed", info.GroupResource.String(), info.Metric, cluster.Name)
+			continue
+		}
+		metricValueList.Items = append(metricValueList.Items, metrics.Items...)
+	}
+	return metricValueList, nil
+}
+
+func (c *CustomMetricsProvider) getPreferredVersion(discoveryClient *discovery.DiscoveryClient) (schema.GroupVersion, error) {
+	apiGroups, err := discoveryClient.ServerGroups()
+	if err != nil {
+		return schema.GroupVersion{}, err
+	}
+	if gv, support := supportedMetricsAPIVersionAvailable(apiGroups); support {
+		return gv, nil
+	}
+	return schema.GroupVersion{}, fmt.Errorf("custom.metrics.k8s.io not found")
+}
+
+func (c *CustomMetricsProvider) getMetricByName(ctx context.Context, clusterName string, name types.NamespacedName, info provider.CustomMetricInfo, metricSelector labels.Selector) (*custom_metrics.MetricValueList, error) {
+	// handle namespace separately
+	if info.GroupResource.Resource == "namespaces" && info.GroupResource.Group == "" {
+		return c.getForNamespace(ctx, clusterName, name.Name, info.Metric, metricSelector)
+	}
+	discoveryClient, err := c.newDiscoveryClientFunc(clusterName)
+	if err != nil {
+		klog.Errorf("Failed to build discovery client for cluster(%s): %+v", clusterName, err)
+		return nil, err
+	}
+	version, err := c.getPreferredVersion(discoveryClient)
+	if err != nil {
+		return nil, err
+	}
+	params, err := versionConverter.ConvertListOptionsToVersion(&custom_metrics.MetricListOptions{
+		MetricLabelSelector: metricSelector.String(),
+	}, version)
+	if err != nil {
+		return nil, err
+	}
+	req := discoveryClient.RESTClient().Get().Prefix("/apis/" + version.String()).Resource(info.GroupResource.String())
+	if info.Namespaced {
+		req = req.Namespace(name.Namespace)
+	}
+	result := req.Name(name.Name).SubResource(info.Metric).SpecificallyVersionedParams(params, custommetricsschema.ParameterCodec, version).Do(ctx)
+	metricObj, err := versionConverter.ConvertResultToVersion(result, custommetricsv1beta2.SchemeGroupVersion)
+	if err != nil {
+		return nil, err
+	}
+	return metricsConvertCustomMetricsV1beta2ToInternalCustomMetrics(metricObj)
+}
+
+func (c *CustomMetricsProvider) getMetricBySelector(ctx context.Context, clusterName, namespace string, selector labels.Selector, info provider.CustomMetricInfo, metricSelector labels.Selector) (*custom_metrics.MetricValueList, error) {
+	// handle namespace separately
+	if info.GroupResource.Resource == "namespaces" && info.GroupResource.Group == "" {
+		return c.getForNamespace(ctx, clusterName, custommetricsv1beta2.AllObjects, info.Metric, metricSelector)
+	}
+	discoveryClient, err := c.newDiscoveryClientFunc(clusterName)
+	if err != nil {
+		klog.Errorf("Failed to build discovery client for cluster(%s): %+v", clusterName, err)
+		return nil, err
+	}
+	version, err := c.getPreferredVersion(discoveryClient)
+	if err != nil {
+		return nil, err
+	}
+	params, err := versionConverter.ConvertListOptionsToVersion(&custom_metrics.MetricListOptions{
+		MetricLabelSelector: metricSelector.String(),
+		LabelSelector:       selector.String(),
+	}, version)
+	if err != nil {
+		return nil, err
+	}
+	req := discoveryClient.RESTClient().Get().Prefix("/apis/" + version.String()).Resource(info.GroupResource.String())
+	if info.Namespaced {
+		req = req.Namespace(namespace)
+	}
+	result := req.Name(custommetricsv1beta2.AllObjects).SubResource(info.Metric).
+		SpecificallyVersionedParams(params, custommetricsschema.ParameterCodec, version).
+		Do(ctx)
+	metricObj, err := versionConverter.ConvertResultToVersion(result, custommetricsv1beta2.SchemeGroupVersion)
+	if err != nil {
+		return nil, err
+	}
+	return metricsConvertCustomMetricsV1beta2ToInternalCustomMetrics(metricObj)
+}
+func (c *CustomMetricsProvider) getForNamespace(ctx context.Context, clusterName, namespace string, metricName string, metricSelector labels.Selector) (*custom_metrics.MetricValueList, error) {
+	discoveryClient, err := c.newDiscoveryClientFunc(clusterName)
+	if err != nil {
+		klog.Errorf("Failed to build discovery client for cluster(%s): %+v", clusterName, err)
+		return nil, err
+	}
+	version, err := c.getPreferredVersion(discoveryClient)
+	if err != nil {
+		return nil, err
+	}
+	params, err := versionConverter.ConvertListOptionsToVersion(&custom_metrics.MetricListOptions{
+		MetricLabelSelector: metricSelector.String(),
+	}, version)
+	if err != nil {
+		return nil, err
+	}
+	result := discoveryClient.RESTClient().Get().Prefix("/apis/"+version.String()).
+		Resource("metrics").
+		Namespace(namespace).
+		Name(metricName).
+		SpecificallyVersionedParams(params, custommetricsschema.ParameterCodec, version).
+		Do(ctx)
+	metricObj, err := versionConverter.ConvertResultToVersion(result, custommetricsv1beta2.SchemeGroupVersion)
+	if err != nil {
+		return nil, err
+	}
+	return metricsConvertCustomMetricsV1beta2ToInternalCustomMetrics(metricObj)
 }
 
 // ListAllMetrics returns all metrics in all member clusters
 func (c *CustomMetricsProvider) ListAllMetrics() []provider.CustomMetricInfo {
-	return []provider.CustomMetricInfo{}
+	clusters, err := c.clusterLister.List(labels.Everything())
+	if err != nil {
+		klog.Errorf("Failed to list clusters: %v", err)
+		return []provider.CustomMetricInfo{}
+	}
+	var customMetricInfos []provider.CustomMetricInfo
+	for _, cluster := range clusters {
+		discoveryClient, err := c.newDiscoveryClientFunc(cluster.Name)
+		if err != nil {
+			klog.Errorf("Failed to build discovery client for cluster(%s): %+v", cluster.Name, err)
+			continue
+		}
+		apiGroups, err := discoveryClient.ServerGroups()
+		if err != nil {
+			klog.Errorf("Failed to query resource in cluster(%s): %+v", cluster.Name, err)
+			continue
+		}
+		supportGroupVersion, support := supportedMetricsAPIVersionAvailable(apiGroups)
+		if !support {
+			klog.Warningf("custom.metrics.k8s.io not found in cluster(%s)", cluster.Name)
+			continue
+		}
+		resources, err := discoveryClient.ServerResourcesForGroupVersion(supportGroupVersion.String())
+		if err != nil {
+			klog.Warningf("Failed to query custom.metrics.k8s.io/v1beta2 resource in cluster(%s): %+v", cluster.Name, err)
+			continue
+		}
+		for _, resource := range resources.APIResources {
+			groupResourceAndMetricName := strings.SplitN(resource.Name, "/", 2)
+			if len(groupResourceAndMetricName) != 2 {
+				klog.Warningf("Failed to query custom.metrics.k8s.io/v1beta2 resource in cluster(%s): %+v", cluster.Name, err)
+				continue
+			}
+			customMetricInfos = append(customMetricInfos, provider.CustomMetricInfo{
+				GroupResource: schema.ParseGroupResource(groupResourceAndMetricName[0]),
+				Namespaced:    resource.Namespaced,
+				Metric:        groupResourceAndMetricName[1],
+			})
+		}
+	}
+	return customMetricInfos
+}
+
+func discoveryClientForClusterFunc(clusterLister clusterlister.ClusterLister,
+	secretLister listcorev1.SecretLister) func(string) (*discovery.DiscoveryClient, error) {
+	clusterGetter := func(cluster string) (*clusterv1alpha1.Cluster, error) {
+		return clusterLister.Get(cluster)
+	}
+	secretGetter := func(namespace string, name string) (*corev1.Secret, error) {
+		return secretLister.Secrets(namespace).Get(name)
+	}
+	return func(clusterName string) (*discovery.DiscoveryClient, error) {
+		clusterConfig, err := util.BuildClusterConfig(clusterName, clusterGetter, secretGetter)
+		if err != nil {
+			return nil, err
+		}
+		return discovery.NewDiscoveryClientForConfigOrDie(clusterConfig), nil
+	}
+}
+
+func supportedMetricsAPIVersionAvailable(discoveredAPIGroups *metav1.APIGroupList) (schema.GroupVersion, bool) {
+	supportedVersionSet := sets.New[string]()
+	for _, discoveredAPIGroup := range discoveredAPIGroups.Groups {
+		if discoveredAPIGroup.Name != custom_metrics.GroupName {
+			continue
+		}
+		for _, version := range discoveredAPIGroup.Versions {
+			supportedVersionSet.Insert(version.Version)
+		}
+	}
+	for _, supportedVersion := range custommetricsclient.MetricVersions {
+		if supportedVersionSet.Has(supportedVersion.Version) {
+			return supportedVersion, true
+		}
+	}
+	return schema.GroupVersion{}, false
+}
+
+func metricsConvertCustomMetricsV1beta2ToInternalCustomMetrics(obj runtime.Object) (*custom_metrics.MetricValueList, error) {
+	var tmp *custommetricsv1beta2.MetricValueList
+	var ok bool
+	if tmp, ok = obj.(*custommetricsv1beta2.MetricValueList); !ok {
+		return nil, fmt.Errorf("the custom metrics API server didn't return MetricValueList, the type is %v", reflect.TypeOf(obj))
+	}
+	res := &custom_metrics.MetricValueList{}
+	if err := custommetricsv1beta2.Convert_v1beta2_MetricValueList_To_custom_metrics_MetricValueList(tmp, res, nil); err != nil {
+		return nil, err
+	}
+	return res, nil
 }