add yurtadm join node in specified nodepool (#1402)

* add yurtadm join node in specified nodepool

charts/yurthub/templates/yurthub-cloud-yurtstaticset.yaml

@@ -42,6 +42,9 @@ spec:
             {{- if .Values.organizations }}
             - --hub-cert-organizations={{ .Values.organizations }}
             {{- end }}
+            {{- if .nodePoolName }}
+            - --nodepool-name={{ .Values.nodePoolName }}
+            {{- end }}
           livenessProbe:
             httpGet:
               host: {{ .Values.yurthubBindingAddr }}

charts/yurthub/templates/yurthub-yurtstaticset.yaml

@@ -42,6 +42,9 @@ spec:
             {{- if .Values.organizations }}
             - --hub-cert-organizations={{ .Values.organizations }}
             {{- end }}
+            {{- if .nodePoolName }}
+            - --nodepool-name={{ .Values.nodePoolName }}
+            {{- end }}
           livenessProbe:
             httpGet:
               host: {{ .Values.yurthubBindingAddr }}

pkg/util/kubeconfig/kubeconfig.go

@@ -24,6 +24,8 @@ import (
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
+
+	yurtclientset "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/client/clientset/versioned"
 )
 
 // CreateBasic creates a basic, general KubeConfig object that then can be extended
@@ -125,3 +127,18 @@ func GetAuthInfoFromKubeConfig(config *clientcmdapi.Config) *clientcmdapi.AuthIn
 	}
 	return nil
 }
+
+// ToYurtClientSet converts a KubeConfig object to a yurtClient
+func ToYurtClientSet(config *clientcmdapi.Config) (yurtclientset.Interface, error) {
+	overrides := clientcmd.ConfigOverrides{Timeout: "10s"}
+	clientConfig, err := clientcmd.NewDefaultClientConfig(*config, &overrides).ClientConfig()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create yurt client configuration from kubeconfig")
+	}
+
+	client, err := yurtclientset.NewForConfig(clientConfig)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create yurt client")
+	}
+	return client, nil
+}

pkg/util/kubeconfig/kubeconfig_test.go

@@ -183,6 +183,8 @@ func TestWriteKubeconfigToDisk(t *testing.T) {
 					newFile,
 				)
 			}
+			client, err := ToYurtClientSet(c)
+			t.Log(client, err)
 		})
 	}
 }

pkg/util/kubernetes/kubeadm/app/util/apiclient/idempotency.go

@@ -29,6 +29,8 @@ import (
 	clientsetretry "k8s.io/client-go/util/retry"
 
 	"github.com/openyurtio/openyurt/pkg/util/kubernetes/kubeadm/app/constants"
+	nodepoolv1alpha1 "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/apis/apps/v1alpha1"
+	yurtclientset "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/client/clientset/versioned"
 )
 
 // ConfigMapMutator is a function that mutates the given ConfigMap and optionally returns an error
@@ -131,3 +133,24 @@ func GetConfigMapWithRetry(client clientset.Interface, namespace, name string) (
 	}
 	return nil, lastError
 }
+
+func GetNodePoolInfoWithRetry(client yurtclientset.Interface, name string) (*nodepoolv1alpha1.NodePool, error) {
+	var np *nodepoolv1alpha1.NodePool
+	var lastError error
+	err := wait.ExponentialBackoff(clientsetretry.DefaultBackoff, func() (bool, error) {
+		var err error
+		np, err = client.AppsV1alpha1().NodePools().Get(context.TODO(), name, metav1.GetOptions{})
+		if err == nil {
+			return true, nil
+		}
+		if apierrors.IsNotFound(err) {
+			return true, nil
+		}
+		lastError = err
+		return false, nil
+	})
+	if err == nil {
+		return np, nil
+	}
+	return nil, lastError
+}

pkg/yurtadm/cmd/join/join.go

@@ -32,17 +32,20 @@ import (
 
 	"github.com/openyurtio/openyurt/pkg/controller/yurtstaticset/util"
 	kubeconfigutil "github.com/openyurtio/openyurt/pkg/util/kubeconfig"
+	"github.com/openyurtio/openyurt/pkg/util/kubernetes/kubeadm/app/util/apiclient"
 	"github.com/openyurtio/openyurt/pkg/yurtadm/cmd/join/joindata"
 	yurtphases "github.com/openyurtio/openyurt/pkg/yurtadm/cmd/join/phases"
 	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"
+	nodepoolv1alpha1 "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/apis/apps/v1alpha1"
 )
 
 type joinOptions struct {
 	token                    string
 	nodeType                 string
 	nodeName                 string
+	nodePoolName             string
 	criSocket                string
 	organizations            string
 	pauseImage               string
@@ -125,6 +128,10 @@ func addJoinConfigFlags(flagSet *flag.FlagSet, joinOptions *joinOptions) {
 		&joinOptions.namespace, yurtconstants.Namespace, joinOptions.namespace,
 		`Specify the namespace of the yurthub staticpod configmap, if not specified, the namespace will be default.`,
 	)
+	flagSet.StringVar(
+		&joinOptions.nodePoolName, yurtconstants.NodePoolName, joinOptions.nodePoolName,
+		`Specify the nodePool name. if specified, that will add node into specified nodePool.`,
+	)
 	flagSet.StringVar(
 		&joinOptions.criSocket, yurtconstants.NodeCRISocket, joinOptions.criSocket,
 		"Path to the CRI socket to connect",
@@ -276,6 +283,7 @@ func newJoinData(args []string, opt *joinOptions) (*joinData, error) {
 		nodeLabels:            make(map[string]string),
 		joinNodeData: &joindata.NodeRegistration{
 			Name:          name,
+			NodePoolName:  opt.nodePoolName,
 			WorkingMode:   opt.nodeType,
 			CRISocket:     opt.criSocket,
 			Organizations: opt.organizations,
@@ -320,6 +328,23 @@ func newJoinData(args []string, opt *joinOptions) (*joinData, error) {
 		return nil, err
 	}
 	data.kubernetesVersion = k8sVersion
+
+	// check whether specified nodePool exists
+	if len(opt.nodePoolName) != 0 {
+		yurtClient, err := kubeconfigutil.ToYurtClientSet(cfg)
+		if err != nil {
+			klog.Errorf("failed to create yurt client, %v", err)
+			return nil, err
+		}
+
+		np, err := apiclient.GetNodePoolInfoWithRetry(yurtClient, opt.nodePoolName)
+		if err != nil || np == nil {
+			// the specified nodePool not exist, return
+			return nil, errors.Errorf("when --nodepool-name is specified, the specified nodePool should be exist.")
+		}
+		// add nodePool label for node by kubelet
+		data.nodeLabels[nodepoolv1alpha1.LabelDesiredNodePool] = opt.nodePoolName
+	}
 	klog.Infof("node join data info: %#+v", *data)
 
 	// get the yurthub template from the staticpod cr

pkg/yurtadm/cmd/join/join_test.go

@@ -235,6 +235,10 @@ func TestRun(t *testing.T) {
 
 func TestNewJoinData(t *testing.T) {
 	jo := newJoinOptions()
+	jo2 := newJoinOptions()
+	jo2.token = "v22u0b.17490yh3xp8azpr0"
+	jo2.unsafeSkipCAVerification = true
+	jo2.nodePoolName = "nodePool2"
 
 	tests := []struct {
 		name   string
@@ -244,10 +248,16 @@ func TestNewJoinData(t *testing.T) {
 	}{
 		{
 			"normal",
-			[]string{},
+			[]string{"localhost:8080"},
 			jo,
 			nil,
 		},
+		{
+			"norma2",
+			[]string{"localhost:8080"},
+			jo2,
+			nil,
+		},
 	}
 
 	for _, tt := range tests {

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

@@ -24,6 +24,7 @@ import (
 
 type NodeRegistration struct {
 	Name          string
+	NodePoolName  string
 	CRISocket     string
 	WorkingMode   string
 	Organizations string

pkg/yurtadm/constants/constants.go

@@ -84,6 +84,8 @@ const (
 	NodeLabels = "node-labels"
 	// NodeName flag sets the node name.
 	NodeName = "node-name"
+	// NodePoolName flag sets the nodePool name.
+	NodePoolName = "nodepool-name"
 	// NodeType flag sets the type of worker node to edge or cloud.
 	NodeType = "node-type"
 	// Organizations flag sets the extra organizations of hub agent client certificate.
@@ -234,6 +236,9 @@ spec:
       {{if .organizations }}
     - --hub-cert-organizations={{.organizations}}
       {{end}}
+      {{if .nodePoolName }}
+    - --nodepool-name={{.nodePoolName}}
+      {{end}}
     livenessProbe:
       httpGet:
         host: {{.yurthubBindingAddr}}

pkg/yurtadm/util/yurthub/yurthub.go

@@ -70,6 +70,9 @@ func AddYurthubStaticYaml(data joindata.YurtJoinData, podManifestPath string) er
 		"namespace":            data.Namespace(),
 		"image":                data.YurtHubImage(),
 	}
+	if len(data.NodeRegistration().NodePoolName) != 0 {
+		ctx["nodePoolName"] = data.NodeRegistration().NodePoolName
+	}
 
 	yurthubTemplate, err := templates.SubsituteTemplate(data.YurtHubTemplate(), ctx)
 	if err != nil {

pkg/yurtadm/util/yurthub/yurthub_test.go

@@ -20,6 +20,11 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/util/sets"
+	clientset "k8s.io/client-go/kubernetes"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
+
+	"github.com/openyurtio/openyurt/pkg/yurtadm/cmd/join/joindata"
 )
 
 var (
@@ -225,3 +230,107 @@ func Test_useRealServerAddr(t *testing.T) {
 		})
 	}
 }
+
+type testData struct {
+	joinNodeData *joindata.NodeRegistration
+}
+
+func (j *testData) ServerAddr() string {
+	return ""
+}
+
+func (j *testData) JoinToken() string {
+	return ""
+}
+
+func (j *testData) PauseImage() string {
+	return ""
+}
+
+func (j *testData) YurtHubImage() string {
+	return ""
+}
+
+func (j *testData) YurtHubServer() string {
+	return ""
+}
+
+func (j *testData) YurtHubTemplate() string {
+	return ""
+}
+
+func (j *testData) YurtHubManifest() string {
+	return ""
+}
+
+func (j *testData) KubernetesVersion() string {
+	return ""
+}
+
+func (j *testData) TLSBootstrapCfg() *clientcmdapi.Config {
+	return nil
+}
+
+func (j *testData) BootstrapClient() *clientset.Clientset {
+	return nil
+}
+
+func (j *testData) NodeRegistration() *joindata.NodeRegistration {
+	return j.joinNodeData
+}
+
+func (j *testData) IgnorePreflightErrors() sets.String {
+	return nil
+}
+
+func (j *testData) CaCertHashes() []string {
+	return nil
+}
+
+func (j *testData) NodeLabels() map[string]string {
+	return nil
+}
+
+func (j *testData) KubernetesResourceServer() string {
+	return ""
+}
+
+func (j *testData) ReuseCNIBin() bool {
+	return false
+}
+
+func (j *testData) Namespace() string {
+	return ""
+}
+
+func TestAddYurthubStaticYaml(t *testing.T) {
+	xdata := testData{
+		joinNodeData: &joindata.NodeRegistration{
+			Name:          "name1",
+			NodePoolName:  "nodePool1",
+			CRISocket:     "",
+			WorkingMode:   "edge",
+			Organizations: "",
+		}}
+
+	tests := []struct {
+		name            string
+		data            testData
+		podManifestPath string
+		wantErr         bool
+	}{
+		{
+			name:            "test",
+			data:            xdata,
+			podManifestPath: "/tmp",
+			wantErr:         false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := AddYurthubStaticYaml(&tt.data, tt.podManifestPath); (err != nil) != tt.wantErr {
+				t.Errorf("AddYurthubStaticYaml() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}