feat: anp support tcp mode

Signed-off-by: wangdepeng <[email protected]>
(cherry picked from commit c6b3fc4)

cmd/kubenest/operator/app/operator.go

@@ -89,11 +89,12 @@ func run(ctx context.Context, opts *options.Options) error {
 	}
 
 	VirtualClusterInitController := controller.VirtualClusterInitController{
-		Client:        mgr.GetClient(),
-		Config:        mgr.GetConfig(),
-		EventRecorder: mgr.GetEventRecorderFor(constants.InitControllerName),
-		RootClientSet: hostKubeClient,
-		KosmosClient:  kosmosClient,
+		Client:          mgr.GetClient(),
+		Config:          mgr.GetConfig(),
+		EventRecorder:   mgr.GetEventRecorderFor(constants.InitControllerName),
+		RootClientSet:   hostKubeClient,
+		KosmosClient:    kosmosClient,
+		KubeNestOptions: &opts.KubeNestOptions,
 	}
 	if err = VirtualClusterInitController.SetupWithManager(mgr); err != nil {
 		return fmt.Errorf("error starting %s: %v", constants.InitControllerName, err)

cmd/kubenest/operator/app/options/options.go

@@ -25,6 +25,7 @@ type KubernetesOptions struct {
 
 type KubeNestOptions struct {
 	ForceDestroy bool
+	AnpMode      string
 }
 
 func NewOptions() *Options {
@@ -53,4 +54,5 @@ func (o *Options) AddFlags(flags *pflag.FlagSet) {
 	flags.BoolVar(&o.AllowNodeOwnbyMulticluster, "multiowner", false, "Allow node own by multicluster or not.")
 	flags.BoolVar(&o.KosmosJoinController, "kosmos-join-controller", false, "Turn on or off kosmos-join-controller.")
 	flags.BoolVar(&o.KubeNestOptions.ForceDestroy, "kube-nest-force-destroy", false, "Force destroy the node.If it set true.If set to true, Kubernetes will not evict the existing nodes on the node when joining nodes to the tenant's control plane, but will instead force destroy.")
+	flags.StringVar(&o.KubeNestOptions.AnpMode, "kube-nest-anp-mode", "tcp", "kube-apiserver network proxy mode, must be set to tcp or uds. uds mode the replicas for apiserver should be one, and tcp for multi apiserver replicas.")
 }

pkg/kubenest/constants/constant.go

@@ -36,6 +36,7 @@ const (
 	EtcdClientCertAndKeyName       = "etcd-client"
 	FrontProxyCaCertAndKeyName     = "front-proxy-ca"
 	FrontProxyClientCertAndKeyName = "front-proxy-client"
+	ProxyServerCertAndKeyName      = "proxy-server"
 
 	//controlplane apiserver
 	ApiServer                     = "apiserver"

pkg/kubenest/controller/virtualcluster_execute_controller.go

@@ -8,6 +8,7 @@ import (
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	ko "github.com/kosmos.io/kosmos/cmd/kubenest/operator/app/options"
 	"github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
 	"github.com/kosmos.io/kosmos/pkg/kubenest"
 	"github.com/kosmos.io/kosmos/pkg/kubenest/constants"
@@ -21,12 +22,13 @@ type Executor struct {
 	config         *rest.Config
 }
 
-func NewExecutor(virtualCluster *v1alpha1.VirtualCluster, c client.Client, config *rest.Config) (*Executor, error) {
+func NewExecutor(virtualCluster *v1alpha1.VirtualCluster, c client.Client, config *rest.Config, kubeNestOptions *ko.KubeNestOptions) (*Executor, error) {
 	var phase *workflow.Phase
 
 	opts := []kubenest.InitOpt{
 		kubenest.NewInitOptWithVirtualCluster(virtualCluster),
 		kubenest.NewInitOptWithKubeconfig(config),
+		kubenest.NewInitOptWithKubeNestOptions(kubeNestOptions),
 	}
 	options := kubenest.NewPhaseInitOptions(opts...)
 	action := recognizeActionFor(virtualCluster)

pkg/kubenest/controller/virtualcluster_init_controller.go

@@ -27,6 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/kosmos.io/kosmos/cmd/kubenest/operator/app/options"
 	"github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
 	"github.com/kosmos.io/kosmos/pkg/generated/clientset/versioned"
 	"github.com/kosmos.io/kosmos/pkg/kubenest/constants"
@@ -35,11 +36,12 @@ import (
 
 type VirtualClusterInitController struct {
 	client.Client
-	Config        *rest.Config
-	EventRecorder record.EventRecorder
-	RootClientSet kubernetes.Interface
-	KosmosClient  versioned.Interface
-	lock          sync.Mutex
+	Config          *rest.Config
+	EventRecorder   record.EventRecorder
+	RootClientSet   kubernetes.Interface
+	KosmosClient    versioned.Interface
+	lock            sync.Mutex
+	KubeNestOptions *options.KubeNestOptions
 }
 
 type NodePool struct {
@@ -118,7 +120,7 @@ func (c *VirtualClusterInitController) Reconcile(ctx context.Context, request re
 			return reconcile.Result{RequeueAfter: RequeueTime}, nil
 		}
 		updatedCluster := originalCluster.DeepCopy()
-		err = c.createVirtualCluster(updatedCluster)
+		err = c.createVirtualCluster(updatedCluster, c.KubeNestOptions)
 		if err != nil {
 			klog.Errorf("Failed to create virtualcluster %s. err: %s", updatedCluster.Name, err.Error())
 			updatedCluster.Status.Reason = err.Error()
@@ -230,7 +232,7 @@ func (c *VirtualClusterInitController) removeFinalizer(virtualCluster *v1alpha1.
 }
 
 // createVirtualCluster assign work nodes, create control plane and create compoennts from manifests
-func (c *VirtualClusterInitController) createVirtualCluster(virtualCluster *v1alpha1.VirtualCluster) error {
+func (c *VirtualClusterInitController) createVirtualCluster(virtualCluster *v1alpha1.VirtualCluster, kubeNestOptions *options.KubeNestOptions) error {
 	klog.V(2).Infof("Reconciling virtual cluster", "name", virtualCluster.Name)
 
 	//Assign host port
@@ -239,7 +241,7 @@ func (c *VirtualClusterInitController) createVirtualCluster(virtualCluster *v1al
 		return errors.Wrap(err, "Error in assign host port!")
 	}
 
-	executer, err := NewExecutor(virtualCluster, c.Client, c.Config)
+	executer, err := NewExecutor(virtualCluster, c.Client, c.Config, kubeNestOptions)
 	if err != nil {
 		return err
 	}
@@ -267,7 +269,7 @@ func (c *VirtualClusterInitController) createVirtualCluster(virtualCluster *v1al
 
 func (c *VirtualClusterInitController) destroyVirtualCluster(virtualCluster *v1alpha1.VirtualCluster) error {
 	klog.V(2).Infof("Destroying virtual cluster %s", virtualCluster.Name)
-	execute, err := NewExecutor(virtualCluster, c.Client, c.Config)
+	execute, err := NewExecutor(virtualCluster, c.Client, c.Config, c.KubeNestOptions)
 	if err != nil {
 		return err
 	}

pkg/kubenest/controlplane/service.go

@@ -18,8 +18,8 @@ import (
 	"github.com/kosmos.io/kosmos/pkg/kubenest/util"
 )
 
-func EnsureVirtualClusterService(client clientset.Interface, name, namespace string, port int32) error {
-	if err := createServerService(client, name, namespace, port); err != nil {
+func EnsureVirtualClusterService(client clientset.Interface, name, namespace string, portMap map[string]int32) error {
+	if err := createServerService(client, name, namespace, portMap); err != nil {
 		return fmt.Errorf("failed to create virtual cluster apiserver-service, err: %w", err)
 	}
 	return nil
@@ -46,19 +46,30 @@ func DeleteVirtualClusterService(client clientset.Interface, name, namespace str
 	return nil
 }
 
-func createServerService(client clientset.Interface, name, namespace string, port int32) error {
+func createServerService(client clientset.Interface, name, namespace string, portMap map[string]int32) error {
 	apiserverServiceBytes, err := util.ParseTemplate(apiserver.ApiserverService, struct {
 		ServiceName, Namespace, ServiceType string
 		ServicePort                         int32
 	}{
 		ServiceName: fmt.Sprintf("%s-%s", name, "apiserver"),
 		Namespace:   namespace,
 		ServiceType: constants.ApiServerServiceType,
-		ServicePort: port,
+		ServicePort: portMap[constants.ApiServerPortKey],
 	})
 	if err != nil {
 		return fmt.Errorf("error when parsing virtualClusterApiserver serive template: %w", err)
 	}
+	anpServiceBytes, err := util.ParseTemplate(apiserver.ApiserverAnpService, struct {
+		ServiceName, Namespace string
+		ProxyServerPort        int32
+	}{
+		ServiceName:     fmt.Sprintf("%s-%s", name, "konnectivity-server"),
+		Namespace:       namespace,
+		ProxyServerPort: portMap[constants.ApiServerNetworkProxyServerPortKey],
+	})
+	if err != nil {
+		return fmt.Errorf("error when parsing virtualClusterApiserver anp service template: %w", err)
+	}
 
 	apiserverService := &corev1.Service{}
 	if err := yaml.Unmarshal([]byte(apiserverServiceBytes), apiserverService); err != nil {
@@ -68,6 +79,14 @@ func createServerService(client clientset.Interface, name, namespace string, por
 		return fmt.Errorf("err when creating virtual cluster apiserver service for %s, err: %w", apiserverService.Name, err)
 	}
 
+	anpService := &corev1.Service{}
+	if err := yaml.Unmarshal([]byte(anpServiceBytes), anpService); err != nil {
+		return fmt.Errorf("error when decoding virtual cluster anp service: %w", err)
+	}
+	if err := createOrUpdateService(client, anpService); err != nil {
+		return fmt.Errorf("err when creating virtual cluster anp service for %s, err: %w", anpService.Name, err)
+	}
+
 	etcdServicePeerBytes, err := util.ParseTemplate(etcd.EtcdPeerService, struct {
 		ServiceName, Namespace                   string
 		EtcdListenClientPort, EtcdListenPeerPort int32

pkg/kubenest/init.go

@@ -10,6 +10,7 @@ import (
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
+	ko "github.com/kosmos.io/kosmos/cmd/kubenest/operator/app/options"
 	"github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
 	"github.com/kosmos.io/kosmos/pkg/generated/clientset/versioned"
 	"github.com/kosmos.io/kosmos/pkg/kubenest/tasks"
@@ -36,6 +37,7 @@ type initData struct {
 	externalIP            string
 	hostPort              int32
 	hostPortMap           map[string]int32
+	kubeNestOptions       *ko.KubeNestOptions
 }
 
 type InitOptions struct {
@@ -45,6 +47,7 @@ type InitOptions struct {
 	virtualClusterVersion string
 	virtualClusterDataDir string
 	virtualCluster        *v1alpha1.VirtualCluster
+	KubeNestOptions       *ko.KubeNestOptions
 }
 
 func NewInitPhase(opts *InitOptions) *workflow.Phase {
@@ -122,6 +125,12 @@ func NewInitOptWithKubeconfig(config *rest.Config) InitOpt {
 	}
 }
 
+func NewInitOptWithKubeNestOptions(options *ko.KubeNestOptions) InitOpt {
+	return func(o *InitOptions) {
+		o.KubeNestOptions = options
+	}
+}
+
 func newRunData(opt *InitOptions) (*initData, error) {
 	if err := opt.Validate(); err != nil {
 		return nil, err
@@ -174,6 +183,7 @@ func newRunData(opt *InitOptions) (*initData, error) {
 		externalIP:            opt.virtualCluster.Spec.ExternalIP,
 		hostPort:              opt.virtualCluster.Status.Port,
 		hostPortMap:           opt.virtualCluster.Status.PortMap,
+		kubeNestOptions:       opt.KubeNestOptions,
 	}, nil
 }
 
@@ -244,3 +254,7 @@ func (i initData) HostPortMap() map[string]int32 {
 func (i initData) DynamicClient() *dynamic.DynamicClient {
 	return i.dynamicClient
 }
+
+func (i initData) KubeNestOpt() *ko.KubeNestOptions {
+	return i.kubeNestOptions
+}

pkg/kubenest/manifest/controlplane/apiserver/mainfests_deployment.go

@@ -22,6 +22,7 @@ spec:
     spec:
       automountServiceAccountToken: false
       hostNetwork: true
+      dnsPolicy: ClusterFirstWithHostNet
       tolerations:
       - key: "node-role.kubernetes.io/control-plane"
         operator: "Exists"
@@ -150,6 +151,7 @@ spec:
     spec:
       automountServiceAccountToken: false
       hostNetwork: true
+      dnsPolicy: ClusterFirstWithHostNet
       tolerations:
       - key: "node-role.kubernetes.io/control-plane"
         operator: "Exists"
@@ -265,19 +267,27 @@ spec:
           runAsUser: 0
         command: [ "/proxy-server"]
         args: [
-          "--log-file=/var/log/konnectivity-server.log",
+          "--log-file=/var/log/{{ .Namespace }}/{{ .Name }}/konnectivity-server.log",
           "--logtostderr=true",
           "--log-file-max-size=0",
-          "--uds-name=/etc/kubernetes/konnectivity-server/{{ .Namespace }}/{{ .Name }}/konnectivity-server.socket",
-          "--delete-existing-uds-file",
           "--cluster-cert=/etc/virtualcluster/pki/apiserver.crt",
           "--cluster-key=/etc/virtualcluster/pki/apiserver.key",
+          {{ if eq .AnpMode "uds" }}
           "--server-port=0",
+          "--mode=grpc",
+          "--uds-name=/etc/kubernetes/konnectivity-server/{{ .Namespace }}/{{ .Name }}/konnectivity-server.socket",
+          "--delete-existing-uds-file",
+          {{ else }}
+          "--server-port={{ .ServerPort }}",
+          "--mode=http-connect",
+          "--server-cert=/etc/virtualcluster/pki/proxy-server.crt",
+          "--server-ca-cert=/etc/virtualcluster/pki/ca.crt",
+          "--server-key=/etc/virtualcluster/pki/proxy-server.key",
+          {{ end }}
           "--agent-port={{ .AgentPort }}",
           "--health-port={{ .HealthPort }}",
           "--admin-port={{ .AdminPort }}",
           "--keepalive-time=1h",
-          "--mode=grpc",
           "--agent-namespace=kube-system",
           "--agent-service-account=konnectivity-agent",
           "--kubeconfig=/etc/apiserver/kubeconfig",
@@ -309,7 +319,7 @@ spec:
           name: apiserver-cert
           readOnly: true
         - name: varlogkonnectivityserver
-          mountPath: /var/log/konnectivity-server.log
+          mountPath: /var/log/{{ .Namespace }}/{{ .Name }}
           readOnly: false
         - name: konnectivity-home
           mountPath: /etc/kubernetes/konnectivity-server/{{ .Namespace }}/{{ .Name }}
@@ -324,8 +334,8 @@ spec:
           secretName: {{ .KubeconfigSecret }}
       - name: varlogkonnectivityserver
         hostPath:
-          path: /var/log/{{ .Namespace }}/{{ .Name }}/konnectivity-server.log
-          type: FileOrCreate
+          path: /var/log/{{ .Namespace }}/{{ .Name }}
+          type: DirectoryOrCreate
       - name: konnectivity-home
         hostPath:
           path: /etc/kubernetes/konnectivity-server/{{ .Namespace }}/{{ .Name }}
@@ -413,7 +423,11 @@ spec:
             "--sync-interval-cap=30s",
             "--probe-interval=5s",
             "--service-account-token-path=/var/run/secrets/tokens/konnectivity-agent-token",
-            "--agent-identifiers=ipv4=$(HOST_IP)"
+            "--agent-identifiers=ipv4=$(HOST_IP)",
+            {{ if ne .AnpMode "uds" }}
+            "--agent-cert=/etc/virtualcluster/pki/apiserver.crt",
+            "--agent-key=/etc/virtualcluster/pki/apiserver.key",
+            {{ end }}
           ]
           env:
             - name: POD_NAME
@@ -443,10 +457,15 @@ spec:
             initialDelaySeconds: 15
             timeoutSeconds: 15
           volumeMounts:
+            - name: agent-cert
+              mountPath: /etc/virtualcluster/pki
             - mountPath: /var/run/secrets/tokens
               name: konnectivity-agent-token
       serviceAccountName: konnectivity-agent
       volumes:
+        - name: agent-cert
+          secret:
+            secretName: {{ .AgentCertName }}
         - name: konnectivity-agent-token
           projected:
             sources:

pkg/kubenest/manifest/controlplane/apiserver/mainfests_service.go

@@ -19,5 +19,24 @@ spec:
   selector:
     virtualCluster-app: apiserver
   type: {{ .ServiceType }}
+`
+	ApiserverAnpService = `
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    virtualCluster-app: apiserver
+    app.kubernetes.io/managed-by: virtual-cluster-controller
+  name: {{ .ServiceName }}
+  namespace: {{ .Namespace }}
+spec:
+  ports:
+  - name: proxy-server
+    port: {{ .ProxyServerPort }}
+    protocol: TCP
+    targetPort: {{ .ProxyServerPort }}
+  selector:
+    virtualCluster-app: apiserver
+  type: ClusterIP
 `
 )

pkg/kubenest/manifest/controlplane/apiserver/manifeats_configmap.go

@@ -10,10 +10,19 @@ data:
     egressSelections:
     - name: cluster
       connection:
-        proxyProtocol: GRPC
+        proxyProtocol: {{ if eq .AnpMode "uds" }}GRPC{{ else }}HTTPConnect{{ end }}
         transport:
+          {{ if eq .AnpMode "uds" }}
           uds:
             udsName: /etc/kubernetes/konnectivity-server/{{ .Namespace }}/{{ .Name }}/konnectivity-server.socket
+          {{ else }}
+          tcp:
+            url: https://{{ .SvcName }}:{{ .ProxyServerPort }}
+            tlsConfig:
+              caBundle: /etc/virtualcluster/pki/ca.crt
+              clientKey: /etc/virtualcluster/pki/proxy-server.key
+              clientCert: /etc/virtualcluster/pki/proxy-server.crt
+          {{ end }}
     - name: master
       connection:
         proxyProtocol: Direct