add forwardkubesvctraffic filter and adapt server certificate

cmd/yurthub/app/options/filters.go

@@ -19,6 +19,7 @@ package options
 import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/filter/base"
 	"github.com/openyurtio/openyurt/pkg/yurthub/filter/discardcloudservice"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter/forwardkubesvctraffic"
 	"github.com/openyurtio/openyurt/pkg/yurthub/filter/inclusterconfig"
 	"github.com/openyurtio/openyurt/pkg/yurthub/filter/masterservice"
 	"github.com/openyurtio/openyurt/pkg/yurthub/filter/nodeportisolation"
@@ -27,15 +28,16 @@ import (
 
 var (
 	// DisabledInCloudMode contains the filters that should be disabled when yurthub is working in cloud mode.
-	DisabledInCloudMode = []string{discardcloudservice.FilterName}
+	DisabledInCloudMode = []string{discardcloudservice.FilterName, forwardkubesvctraffic.FilterName}
 
 	// SupportedComponentsForFilter is used for specifying which components are supported by filters as default setting.
 	SupportedComponentsForFilter = map[string]string{
-		masterservice.FilterName:       "kubelet",
-		discardcloudservice.FilterName: "kube-proxy",
-		servicetopology.FilterName:     "kube-proxy, coredns, nginx-ingress-controller",
-		inclusterconfig.FilterName:     "kubelet",
-		nodeportisolation.FilterName:   "kube-proxy",
+		masterservice.FilterName:         "kubelet",
+		discardcloudservice.FilterName:   "kube-proxy",
+		servicetopology.FilterName:       "kube-proxy, coredns, nginx-ingress-controller",
+		inclusterconfig.FilterName:       "kubelet",
+		nodeportisolation.FilterName:     "kube-proxy",
+		forwardkubesvctraffic.FilterName: "kube-proxy",
 	}
 )
 
@@ -49,4 +51,5 @@ func RegisterAllFilters(filters *base.Filters) {
 	discardcloudservice.Register(filters)
 	inclusterconfig.Register(filters)
 	nodeportisolation.Register(filters)
+	forwardkubesvctraffic.Register(filters)
 }

pkg/yurthub/certificate/server/server.go

@@ -17,13 +17,15 @@
 package server
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
 	"net"
 	"time"
 
 	"github.com/pkg/errors"
 	certificatesv1 "k8s.io/api/certificates/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/authentication/user"
 	clientset "k8s.io/client-go/kubernetes"
@@ -41,6 +43,7 @@
 type hubServerCertificateManager struct {
 	hubServerCertManager certificate.Manager
 	hubServerCertStore   certificate.FileStore
+	kubeSvcClusterIP     net.IP
 }
 
 func NewHubServerCertificateManager(client clientset.Interface, clientCertManager hubCert.YurtClientCertificateManager, nodeName, pkiDir string, certIPs []net.IP) (hubCert.YurtServerCertificateManager, error) {
@@ -49,6 +52,10 @@
 		return nil, errors.Wrap(err, "couldn't new hub server cert store")
 	}
 
+	hscm := &hubServerCertificateManager{
+		hubServerCertStore: hubServerCertStore,
+	}
+
 	kubeClientFn := func(current *tls.Certificate) (clientset.Interface, error) {
 		// waiting for the certificate is generated
 		_ = wait.PollInfinite(5*time.Second, func() (bool, error) {
@@ -67,22 +74,61 @@
 		return kubeconfigutil.ClientSetFromFile(clientCertManager.GetHubConfFile())
 	}
 
+	serverIPsGetter := func() ([]net.IP, error) {
+		ips := []net.IP{}
+		if hscm.kubeSvcClusterIP != nil {
+			ips = append(ips, hscm.kubeSvcClusterIP)
+			ips = append(ips, certIPs...)
+			return ips, nil
+		}
+		if clientCertManager.GetAPIServerClientCert() == nil {
+			return ips, fmt.Errorf("client certificate(%s) of hub agent is not ready", clientCertManager.GetHubConfFile())
+		}
+
+		if !yurtutil.IsNil(client) {
+			return certIPs, nil
+		}
+
+		kubeClient, err := kubeconfigutil.ClientSetFromFile(clientCertManager.GetHubConfFile())
+		if err != nil {
+			return ips, err
+		}
+
+		kubeSvc, err := kubeClient.CoreV1().Services("default").Get(context.Background(), "kubernetes", metav1.GetOptions{})
+		if err != nil {
+			return ips, err
+		} else if kubeSvc == nil {
+			return ips, fmt.Errorf("couldn't get default/kubernetes service")
+		} else {
+			ip := net.ParseIP(kubeSvc.Spec.ClusterIP)
+			if ip == nil {
+				return ips, fmt.Errorf("couldn't get clusterIP from default/kubernetes service")
+			}
+			klog.Infof("completed to prepare default/kubernetes service clusterIP(%s) for server certificate", ip.String())
+			hscm.kubeSvcClusterIP = ip
+			ips = append(ips, ip)
+		}
+
+		ips = append(ips, certIPs...)
+
+		return ips, nil
+	}
+
 	hubServerCertManager, sErr := certfactory.NewCertManagerFactoryWithFnAndStore(kubeClientFn, hubServerCertStore).New(&certfactory.CertManagerConfig{
 		ComponentName:  fmt.Sprintf("%s-server", projectinfo.GetHubName()),
 		SignerName:     certificatesv1.KubeletServingSignerName,
 		ForServerUsage: true,
 		CommonName:     fmt.Sprintf("system:node:%s", nodeName),
 		Organizations:  []string{user.NodesGroup},
-		IPs:            certIPs,
+		IPGetter:       serverIPsGetter,
+		DNSNames:       []string{"kubernetes", "kubernetes.default", "kubernetes.default.svc", "kubernetes.default.svc.cluster.local"},
 	})
 	if sErr != nil {
 		return nil, sErr
 	}
 
-	return &hubServerCertificateManager{
-		hubServerCertManager: hubServerCertManager,
-		hubServerCertStore:   hubServerCertStore,
-	}, nil
+	hscm.hubServerCertManager = hubServerCertManager
+	return hscm, nil
 }
 
 func (hcm *hubServerCertificateManager) Start() {

pkg/yurthub/filter/forwardkubesvctraffic/filter.go

@@ -0,0 +1,121 @@
+/*
+Copyright 2024 The OpenYurt 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 forwardkubesvctraffic
+
+import (
+	"strconv"
+
+	discovery "k8s.io/api/discovery/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2"
+	utilnet "k8s.io/utils/net"
+
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter"
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter/base"
+)
+
+const (
+	// FilterName filter is used to mutate the default/kubernetes endpointslices
+	// in order to make pods on edge nodes can access kube-apiserver directly by default/kubernetes service.
+	FilterName = "forwardkubesvctraffic"
+
+	KubeSVCNamespace = "default"
+	KubeSVCName      = "kubernetes"
+	KubeSVCPortName  = "https"
+)
+
+// Register registers a filter
+func Register(filters *base.Filters) {
+	filters.Register(FilterName, func() (filter.ObjectFilter, error) {
+		return NewForwardKubeSVCTrafficFilter()
+	})
+}
+
+func NewForwardKubeSVCTrafficFilter() (*forwardKubeSVCTrafficFilter, error) {
+	return &forwardKubeSVCTrafficFilter{addressType: discovery.AddressTypeIPv4}, nil
+}
+
+type forwardKubeSVCTrafficFilter struct {
+	addressType discovery.AddressType
+	host        string
+	port        int32
+}
+
+func (fkst *forwardKubeSVCTrafficFilter) Name() string {
+	return FilterName
+}
+
+func (fkst *forwardKubeSVCTrafficFilter) SupportedResourceAndVerbs() map[string]sets.String {
+	return map[string]sets.String{
+		"endpointslices": sets.NewString("list", "watch"),
+	}
+}
+
+func (fkst *forwardKubeSVCTrafficFilter) SetMasterServiceHost(host string) error {
+	fkst.host = host
+	if utilnet.IsIPv6String(host) {
+		fkst.addressType = discovery.AddressTypeIPv6
+	}
+	return nil
+
+}
+
+func (fkst *forwardKubeSVCTrafficFilter) SetMasterServicePort(portStr string) error {
+	port, err := strconv.ParseInt(portStr, 10, 32)
+	if err != nil {
+		return err
+	}
+	fkst.port = int32(port)
+	return nil
+}
+
+func (fkst *forwardKubeSVCTrafficFilter) Filter(obj runtime.Object, stopCh <-chan struct{}) runtime.Object {
+	switch v := obj.(type) {
+	case *discovery.EndpointSlice:
+		fkst.mutateDefaultKubernetesEps(v)
+		return v
+	default:
+		return obj
+	}
+}
+
+func (fkst *forwardKubeSVCTrafficFilter) mutateDefaultKubernetesEps(eps *discovery.EndpointSlice) {
+	trueCondition := true
+	if eps.Namespace == KubeSVCNamespace && eps.Name == KubeSVCName {
+		if eps.AddressType != fkst.addressType {
+			klog.Warningf("address type of default/kubernetes endpoinstlice(%s) and hub server is different(%s), hub server address type need to be configured", eps.AddressType, fkst.addressType)
+			return
+		}
+		for j := range eps.Ports {
+			if eps.Ports[j].Name != nil && *eps.Ports[j].Name == KubeSVCPortName {
+				eps.Ports[j].Port = &fkst.port
+				break
+			}
+		}
+		eps.Endpoints = []discovery.Endpoint{
+			{
+				Addresses: []string{fkst.host},
+				Conditions: discovery.EndpointConditions{
+					Ready: &trueCondition,
+				},
+			},
+		}
+		klog.V(2).Infof("mutate default/kubernetes endpointslice to %v in forwardkubesvctraffic filter", *eps)
+	}
+	return
+}