Fix the issue of missing workqueue metrics in the Karmada controller

pkg/resourceinterpreter/customized/webhook/customized.go

@@ -31,7 +31,6 @@ import (
 	webhookutil "k8s.io/apiserver/pkg/util/webhook"
 	corev1 "k8s.io/client-go/listers/core/v1"
 	"k8s.io/klog/v2"
-	"k8s.io/kube-aggregator/pkg/apiserver"
 	utiltrace "k8s.io/utils/trace"
 
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
@@ -70,7 +69,7 @@ func NewCustomizedInterpreter(informer genericmanager.SingleClusterInformerManag
 	}
 
 	cm.SetAuthenticationInfoResolver(authInfoResolver)
-	cm.SetServiceResolver(apiserver.NewClusterIPServiceResolver(serviceLister))
+	cm.SetServiceResolver(NewServiceResolver(serviceLister))
 
 	return &CustomizedInterpreter{
 		hookManager:   configmanager.NewExploreConfigManager(informer),

pkg/resourceinterpreter/customized/webhook/serviceresolvers.go

@@ -0,0 +1,77 @@
+package webhook
+
+import (
+	"fmt"
+	"net"
+	"net/url"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	webhookutil "k8s.io/apiserver/pkg/util/webhook"
+	corev1lister "k8s.io/client-go/listers/core/v1"
+)
+
+// ServiceResolver knows how to convert a service reference into an actual location.
+type ServiceResolver interface {
+	ResolveEndpoint(namespace, name string, port int32) (*url.URL, error)
+}
+
+// NewServiceResolver returns a ServiceResolver that parses service first,
+// if service not exist, constructs a service URL from a given namespace and name.
+func NewServiceResolver(services corev1lister.ServiceLister) ServiceResolver {
+	return &serviceResolver{
+		services:        services,
+		defaultResolver: webhookutil.NewDefaultServiceResolver(),
+	}
+}
+
+type serviceResolver struct {
+	services        corev1lister.ServiceLister
+	defaultResolver ServiceResolver
+}
+
+func (r *serviceResolver) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
+	svc, err := r.services.Services(namespace).Get(name)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return r.defaultResolver.ResolveEndpoint(namespace, name, port)
+		}
+		return nil, err
+	}
+	return resolveCluster(svc, port)
+}
+
+// resolveCluster parses Service resource to url.
+func resolveCluster(svc *corev1.Service, port int32) (*url.URL, error) {
+	switch {
+	case svc.Spec.Type == corev1.ServiceTypeClusterIP && svc.Spec.ClusterIP == corev1.ClusterIPNone:
+		return nil, fmt.Errorf(`cannot route to service with ClusterIP "None"`)
+	// use IP from a clusterIP for these service types
+	case svc.Spec.Type == corev1.ServiceTypeClusterIP, svc.Spec.Type == corev1.ServiceTypeLoadBalancer, svc.Spec.Type == corev1.ServiceTypeNodePort:
+		svcPort, err := findServicePort(svc, port)
+		if err != nil {
+			return nil, err
+		}
+		return &url.URL{
+			Scheme: "https",
+			Host:   net.JoinHostPort(svc.Spec.ClusterIP, fmt.Sprintf("%d", svcPort.Port)),
+		}, nil
+	case svc.Spec.Type == corev1.ServiceTypeExternalName:
+		return &url.URL{
+			Scheme: "https",
+			Host:   net.JoinHostPort(svc.Spec.ExternalName, fmt.Sprintf("%d", port)),
+		}, nil
+	default:
+		return nil, fmt.Errorf("unsupported service type %q", svc.Spec.Type)
+	}
+}
+
+// findServicePort finds the service port by name or numerically.
+func findServicePort(svc *corev1.Service, port int32) (*corev1.ServicePort, error) {
+	for _, svcPort := range svc.Spec.Ports {
+		if svcPort.Port == port {
+			return &svcPort, nil
+		}
+	}
+	return nil, apierrors.NewServiceUnavailable(fmt.Sprintf("no service port %d found for service %q", port, svc.Name))
+}

pkg/resourceinterpreter/customized/webhook/serviceresolvers_test.go

@@ -0,0 +1,153 @@
+package webhook
+
+import (
+	"fmt"
+	"net"
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+func Test_resolveCluster(t *testing.T) {
+	type args struct {
+		svc  *corev1.Service
+		port int32
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    *url.URL
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			name: "ClusterIP service without expect port, can not be resolved",
+			args: args{
+				svc: &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeClusterIP,
+						ClusterIP: "10.10.10.10",
+						Ports: []corev1.ServicePort{
+							{Port: 1234, TargetPort: intstr.FromInt32(1234)},
+						}}},
+				port: 443,
+			},
+			wantErr: assert.Error,
+		},
+		{
+			name: "ClusterIP service, can be resolved",
+			args: args{
+				svc: &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeClusterIP,
+						ClusterIP: "10.10.10.10",
+						Ports: []corev1.ServicePort{
+							{Name: "https", Port: 443, TargetPort: intstr.FromInt32(1443)},
+							{Port: 1234, TargetPort: intstr.FromInt32(1234)},
+						}}},
+				port: 443,
+			},
+			want: &url.URL{
+				Scheme: "https",
+				Host:   net.JoinHostPort("10.10.10.10", "443"),
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "headless service, can not be resolved",
+			args: args{
+				svc: &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeClusterIP,
+						ClusterIP: corev1.ClusterIPNone,
+					}},
+				port: 443,
+			},
+			wantErr: assert.Error,
+		},
+		{
+			name: "LoadBalancer service, can be resolved",
+			args: args{
+				svc: &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeLoadBalancer,
+						ClusterIP: "10.10.10.10",
+						Ports: []corev1.ServicePort{
+							{Name: "https", Port: 443, TargetPort: intstr.FromInt32(1443)},
+							{Port: 1234, TargetPort: intstr.FromInt32(1234)},
+						}}},
+				port: 443,
+			},
+			want: &url.URL{
+				Scheme: "https",
+				Host:   net.JoinHostPort("10.10.10.10", "443"),
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "NodePort service, can be resolved",
+			args: args{
+				svc: &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"},
+					Spec: corev1.ServiceSpec{
+						Type:      corev1.ServiceTypeLoadBalancer,
+						ClusterIP: "10.10.10.10",
+						Ports: []corev1.ServicePort{
+							{Name: "https", Port: 443, TargetPort: intstr.FromInt32(1443)},
+							{Port: 1234, TargetPort: intstr.FromInt32(1234)},
+						}}},
+				port: 443,
+			},
+			want: &url.URL{
+				Scheme: "https",
+				Host:   net.JoinHostPort("10.10.10.10", "443"),
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "ExternalName service, can be resolved",
+			args: args{
+				svc: &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"},
+					Spec: corev1.ServiceSpec{
+						Type:         corev1.ServiceTypeExternalName,
+						ExternalName: "foo.bar.com",
+					}},
+				port: 443,
+			},
+			want: &url.URL{
+				Scheme: "https",
+				Host:   net.JoinHostPort("foo.bar.com", "443"),
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "unsupported type service, can not be resolved",
+			args: args{
+				svc: &corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"},
+					Spec: corev1.ServiceSpec{
+						Type: "unsupported service",
+					}},
+				port: 443,
+			},
+			wantErr: assert.Error,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := resolveCluster(tt.args.svc, tt.args.port)
+			if !tt.wantErr(t, err, fmt.Sprintf("resolveCluster(%v, %v)", tt.args.svc, tt.args.port)) {
+				return
+			}
+			assert.Equalf(t, tt.want, got, "resolveCluster(%v, %v)", tt.args.svc, tt.args.port)
+		})
+	}
+}