yurtadm join: support deploying local mode yurthub in systemd

cmd/yurthub/app/config/config.go

@@ -279,7 +279,8 @@
 	var err error
 	// If yurthub is in local mode, create kubeconfig for host control plane to prepare	informerFactory.
 	if util.WorkingMode(options.WorkingMode) == util.WorkingModeLocal {
-		kubeConfig, err = clientcmd.BuildConfigFromFlags(fmt.Sprintf("http://%s", options.HostControlPlaneAddr), "")
+		// kubeConfig, err = clientcmd.BuildConfigFromFlags(fmt.Sprintf("http://%s", options.HostControlPlaneAddr), "")
+		kubeConfig, err = clientcmd.BuildConfigFromFlags("", "/root/ackkubeconfig")
 	} else {
 		kubeConfig, err = clientcmd.BuildConfigFromFlags(fmt.Sprintf("http://%s:%d", options.YurtHubProxyHost, options.YurtHubProxyPort), "")
 	}

pkg/yurtadm/cmd/join/join.go

@@ -20,6 +20,7 @@
 import (
 	"fmt"
 	"io"
+	"os"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -38,6 +39,7 @@
 	yurtconstants "github.com/openyurtio/openyurt/pkg/yurtadm/constants"
 	"github.com/openyurtio/openyurt/pkg/yurtadm/util/edgenode"
 	yurtadmutil "github.com/openyurtio/openyurt/pkg/yurtadm/util/kubernetes"
+	"github.com/openyurtio/openyurt/pkg/yurtadm/util/localnode"
 	"github.com/openyurtio/openyurt/pkg/yurtadm/util/yurthub"
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/yurtstaticset/util"
 )
@@ -52,6 +54,8 @@
 	organizations            string
 	pauseImage               string
 	yurthubImage             string
+	yurthubBinary            string
+	hostControlPlaneAddr     string // hostControlPlaneAddr is the address (ip:port) of host kubernetes cluster that used for yurthub local mode.
 	namespace                string
 	caCertHashes             []string
 	unsafeSkipCAVerification bool
@@ -124,7 +128,7 @@
 	)
 	flagSet.StringVar(
 		&joinOptions.nodeType, yurtconstants.NodeType, joinOptions.nodeType,
-		"Sets the node is edge or cloud",
+		"Sets the node is edge, cloud or local",
 	)
 	flagSet.StringVar(
 		&joinOptions.nodeName, yurtconstants.NodeName, joinOptions.nodeName,
@@ -154,6 +158,14 @@
 		&joinOptions.yurthubImage, yurtconstants.YurtHubImage, joinOptions.yurthubImage,
 		"Sets the image version of yurthub component",
 	)
+	flagSet.StringVar(
+		&joinOptions.yurthubBinary, yurtconstants.YurtHubBinary, joinOptions.yurthubBinary,
+		"Sets the binary path of yurthub, this is used for deploying local mode yurthub in systemd",
+	)
+	flagSet.StringVar(
+		&joinOptions.hostControlPlaneAddr, yurtconstants.HostControlPlaneAddr, joinOptions.hostControlPlaneAddr,
+		"Sets the address of hostControlPlaneAddr, which is the address (ip:port) of host kubernetes cluster that used for yurthub local mode",
+	)
 	flagSet.StringSliceVar(
 		&joinOptions.caCertHashes, yurtconstants.TokenDiscoveryCAHash, joinOptions.caCertHashes,
 		"For token-based discovery, validate that the root CA public key matches this hash (format: \"<type>:<value>\").",
@@ -227,6 +239,9 @@
 	organizations            string
 	pauseImage               string
 	yurthubImage             string
+	yurthubBinary            string
+	hostControlPlaneAddr     string
+	tenantApiServerEndpoints string
 	yurthubTemplate          string
 	yurthubManifest          string
 	kubernetesVersion        string
@@ -257,6 +272,25 @@
 		apiServerEndpoint = args[0]
 	}
 
+	if opt.nodeType == yurtconstants.LocalNode {
+		// in local mode, it is necessary to prepare yurthub binary file for deploying systemd yurthub.
+		if len(opt.yurthubBinary) == 0 {
+			return nil, errors.New("yurthub binary filepath is empty, so unable to run systemd yurthub in local mode.")
+		}
+		_, err := os.Stat(opt.yurthubBinary)
+		if err != nil {
+			if os.IsNotExist(err) {
+				return nil, errors.New("yurthub binary file does not exist.")
+			}
+			return nil, errors.Wrapf(err, "stat yurthub binary file %s fail", opt.yurthubBinary)
+		}
+
+		// in local mode, hostControlPlaneAddr is needed for systemd yurthub accessing host kubernetes cluster.
+		if len(opt.hostControlPlaneAddr) == 0 {
+			return nil, errors.New("host control plane address is empty, so unable to run systemd yurthub in local mode.")
+		}
+	}
+
 	if len(opt.token) == 0 {
 		return nil, errors.New("join token is empty, so unable to bootstrap worker node.")
 	}
@@ -265,8 +299,8 @@
 		return nil, errors.Errorf("the bootstrap token %s was not of the form %s", opt.token, yurtconstants.BootstrapTokenPattern)
 	}
 
-	if opt.nodeType != yurtconstants.EdgeNode && opt.nodeType != yurtconstants.CloudNode {
-		return nil, errors.Errorf("node type(%s) is invalid, only \"edge and cloud\" are supported", opt.nodeType)
+	if opt.nodeType != yurtconstants.EdgeNode && opt.nodeType != yurtconstants.CloudNode && opt.nodeType != yurtconstants.LocalNode {
+		return nil, errors.Errorf("node type(%s) is invalid, only \"edge, cloud and local\" are supported", opt.nodeType)
 	}
 
 	if opt.unsafeSkipCAVerification && len(opt.caCertHashes) != 0 {
@@ -298,6 +332,8 @@
 		ignorePreflightErrors: ignoreErrors,
 		pauseImage:            opt.pauseImage,
 		yurthubImage:          opt.yurthubImage,
+		yurthubBinary:         opt.yurthubBinary,
+		hostControlPlaneAddr:  opt.hostControlPlaneAddr,
 		yurthubServer:         opt.yurthubServer,
 		caCertHashes:          opt.caCertHashes,
 		organizations:         opt.organizations,
@@ -327,13 +363,37 @@
 		}
 	}
 
+	// if the node type is local, before get tls bootstrap config, we need to deploy systemd yurthub for maintaining iptables rules
+	if opt.nodeType == yurtconstants.LocalNode {
+		// deploy systemd yurthub
+		if err := localnode.DeployYurthubInSystemd(data.HostControlPlaneAddr(), data.ServerAddr(), data.YurtHubBinary(), data.NodeRegistration().Name); err != nil {
+			klog.Errorf("could not deploy local yurthub in systemd, %v", err)
+			return nil, err
+		}
+
+		// check systemd yurthub is ready or not
+		if err := localnode.CheckYurthubStatus(); err != nil {
+			return nil, err
+		}
+		klog.V(1).Infof("systemd yurthub agent is ready")
+
+		tenantApiServerEndpoints, err := localnode.GetTenantApiServerEndpoints()
+		if err != nil {
+			klog.Errorf("could not get tenantApiServerEndpoints, %v", err)
+			return nil, err
+		}
+		data.tenantApiServerEndpoints = tenantApiServerEndpoints
+		klog.V(1).Infof("get tenantApiServerEndpoints: %s", tenantApiServerEndpoints)
+	}
+
 	// get tls bootstrap config
 	cfg, err := yurtadmutil.RetrieveBootstrapConfig(data)
 	if err != nil {
 		klog.Errorf("could not retrieve bootstrap config, %v", err)
 		return nil, err
 	}
 	data.tlsBootstrapCfg = cfg
+	klog.Infof("RetrieveBootstrapConfig: %#+v", *cfg)
 
 	// get kubernetes version
 	client, err := kubeconfigutil.ToClientSet(cfg)
@@ -342,6 +402,7 @@
 		return nil, err
 	}
 	data.clientSet = client
+	klog.Infof("ToClientSet: %#+v", *client)
 
 	k8sVersion, err := yurtadmutil.GetKubernetesVersionFromCluster(client)
 	if err != nil {
@@ -350,6 +411,9 @@
 	}
 	data.kubernetesVersion = k8sVersion
 
+	// test for get k8s version
+	klog.Infof("GetKubernetesVersionFromCluster: %s", k8sVersion)
+
 	// check whether specified nodePool exists
 	if len(opt.nodePoolName) != 0 {
 		np, err := apiclient.GetNodePoolInfoWithRetry(cfg, opt.nodePoolName)
@@ -439,6 +503,19 @@
 	return j.yurthubImage
 }
 
+// YurtHubBinary returns the YurtHub binary.
+func (j *joinData) YurtHubBinary() string {
+	return j.yurthubBinary
+}
+
+func (j *joinData) HostControlPlaneAddr() string {
+	return j.hostControlPlaneAddr
+}
+
+func (j *joinData) TenantApiServerEndpoints() string {
+	return j.tenantApiServerEndpoints
+}
+
 // YurtHubServer returns the YurtHub server addr.
 func (j *joinData) YurtHubServer() string {
 	return j.yurthubServer

pkg/yurtadm/cmd/join/joindata/data.go

@@ -36,6 +36,9 @@ type YurtJoinData interface {
 	JoinToken() string
 	PauseImage() string
 	YurtHubImage() string
+	YurtHubBinary() string
+	HostControlPlaneAddr() string
+	TenantApiServerEndpoints() string
 	YurtHubServer() string
 	YurtHubTemplate() string
 	YurtHubManifest() string

pkg/yurtadm/cmd/join/phases/postcheck.go

@@ -20,6 +20,7 @@
 	"k8s.io/klog/v2"
 
 	"github.com/openyurtio/openyurt/pkg/yurtadm/cmd/join/joindata"
+	"github.com/openyurtio/openyurt/pkg/yurtadm/constants"
 	"github.com/openyurtio/openyurt/pkg/yurtadm/util/kubernetes"
 	"github.com/openyurtio/openyurt/pkg/yurtadm/util/yurthub"
 )
@@ -33,15 +34,19 @@
 	klog.V(1).Infof("kubelet service is active")
 
 	klog.V(1).Infof("waiting hub agent ready.")
-	if err := yurthub.CheckYurthubHealthz(data.YurtHubServer()); err != nil {
-		return err
-	}
-	klog.V(1).Infof("hub agent is ready")
 
-	if err := yurthub.CleanHubBootstrapConfig(); err != nil {
-		return err
+	// check staticpod yurthub for edge node and cloud node
+	if data.NodeRegistration().WorkingMode != constants.LocalNode {
+		if err := yurthub.CheckYurthubHealthz(data.YurtHubServer()); err != nil {
+			return err
+		}
+		klog.V(1).Infof("staticpod yurthub agent is ready")
+
+		if err := yurthub.CleanHubBootstrapConfig(); err != nil {
+			return err
+		}
+		klog.V(1).Infof("clean yurthub bootstrap config file success")
 	}
-	klog.V(1).Infof("clean yurthub bootstrap config file success")
 
 	return nil
 }

pkg/yurtadm/cmd/join/phases/prepare.go

@@ -65,23 +65,25 @@
 	if err := yurtadmutil.SetKubeletUnitConfig(); err != nil {
 		return err
 	}
-	if err := yurtadmutil.SetKubeletConfigForNode(); err != nil {
-		return err
-	}
-	if err := yurthub.SetHubBootstrapConfig(data.ServerAddr(), data.JoinToken(), data.CaCertHashes()); err != nil {
-		return err
-	}
-	if err := yurthub.AddYurthubStaticYaml(data, constants.StaticPodPath); err != nil {
-		return err
-	}
-	if len(data.StaticPodTemplateList()) != 0 {
-		// deploy user specified static pods
-		if err := edgenode.DeployStaticYaml(data.StaticPodManifestList(), data.StaticPodTemplateList(), constants.StaticPodPath); err != nil {
+	if data.NodeRegistration().WorkingMode != constants.LocalNode {
+		if err := yurtadmutil.SetKubeletConfigForNode(); err != nil {
+			return err
+		}
+		if err := yurthub.SetHubBootstrapConfig(data.ServerAddr(), data.JoinToken(), data.CaCertHashes()); err != nil {
+			return err
+		}
+		if err := yurthub.AddYurthubStaticYaml(data, constants.StaticPodPath); err != nil {
+			return err
+		}
+		if len(data.StaticPodTemplateList()) != 0 {
+			// deploy user specified static pods
+			if err := edgenode.DeployStaticYaml(data.StaticPodManifestList(), data.StaticPodTemplateList(), constants.StaticPodPath); err != nil {
+				return err
+			}
+		}
+		if err := yurtadmutil.SetDiscoveryConfig(data); err != nil {
 			return err
 		}
-	}
-	if err := yurtadmutil.SetDiscoveryConfig(data); err != nil {
-		return err
 	}
 	if data.CfgPath() == "" {
 		if err := yurtadmutil.SetKubeadmJoinConfig(data); err != nil {

pkg/yurtadm/constants/constants.go

@@ -35,6 +35,8 @@ const (
 	PauseImagePath                = "registry.cn-hangzhou.aliyuncs.com/google_containers/pause:3.2"
 	DefaultCertificatesDir        = "/etc/kubernetes/pki"
 	DefaultDockerCRISocket        = "/var/run/dockershim.sock"
+	YurthubServiceFilepath        = "/etc/systemd/system/yurthub.service"
+	YurthubEnvironmentFilePath    = "/etc/systemd/system/yurthub.default"
 	YurthubYamlName               = "yurthub.yaml"
 	YurthubStaticPodManifest      = "yurthub"
 	YurthubNamespace              = "kube-system"
@@ -72,6 +74,7 @@ const (
 
 	EdgeNode  = "edge"
 	CloudNode = "cloud"
+	LocalNode = "local"
 
 	// CertificatesDir
 	CertificatesDir = "cert-dir"
@@ -107,6 +110,10 @@ const (
 	Namespace = "namespace"
 	// YurtHubImage flag sets the yurthub image for worker node.
 	YurtHubImage = "yurthub-image"
+	// YurtHubBinary flag sets the yurthub Binary for worker node.
+	YurtHubBinary = "yurthub-binary"
+	// HostControlPlaneAddr flag sets the address of host kubernetes cluster
+	HostControlPlaneAddr = "host-control-plane-addr"
 	// YurtHubServerAddr flag set the address of yurthub server (not proxy server!)
 	YurtHubServerAddr = "yurthub-server-addr"
 	// ServerAddr flag set the address of kubernetes kube-apiserver
@@ -272,5 +279,21 @@ spec:
   hostNetwork: true
   priorityClassName: system-node-critical
   priority: 2000001000
+`
+
+	YurthubSyetmdServiceContent = `
+[Unit]
+Description=local mode yurthub is deployed in systemd
+Documentation=https://github.com/openyurtio/openyurt/pull/2124
+
+[Service]
+EnvironmentFile=/etc/systemd/system/yurthub.default
+ExecStart=/usr/bin/yurthub --working-mode ${WORKINGMODE} --node-name ${NODENAME} --server-addr ${SERVERADDR} --host-control-plane-address ${HOSTCONTROLPLANEADDRESS}
+Restart=always
+StartLimitInterval=0
+RestartSec=10
+
+[Install]
+WantedBy=multi-user.target
 `
 )

pkg/yurtadm/util/initsystem/initsystem.go

@@ -27,4 +27,7 @@ type InitSystem interface {
 
 	// ServiceIsActive ensures the service is running, or attempting to run. (crash looping in the case of kubelet)
 	ServiceIsActive(service string) bool
+
+	// ServiceToStart tries to start a specific service
+	ServiceStart(service string) error
 }

pkg/yurtadm/util/initsystem/initsystem_unix.go

@@ -52,6 +52,12 @@
 	return !strings.Contains(outStr, "stopped") && !strings.Contains(outStr, "does not exist")
 }
 
+// ServiceStart tries to start a specific service
+func (openrc OpenRCInitSystem) ServiceStart(service string) error {
+	args := []string{service, "start"}
+	return exec.Command("rc-service", args...).Run()
+}
+
 // SystemdInitSystem defines systemd
 type SystemdInitSystem struct{}
 
@@ -94,6 +100,16 @@
 	return false
 }
 
+// ServiceStart tries to start a specific service
+func (sysd SystemdInitSystem) ServiceStart(service string) error {
+	// Before we try to start any service, make sure that systemd is ready
+	if err := sysd.reloadSystemd(); err != nil {
+		return err
+	}
+	args := []string{"start", service}
+	return exec.Command("systemctl", args...).Run()
+}
+
 // GetInitSystem returns an InitSystem for the current system, or nil
 // if we cannot detect a supported init system.
 // This indicates we will skip init system checks, not an error.