support clusterfile add taint

pkg/cluster-runtime/installer.go

@@ -15,8 +15,11 @@
 package clusterruntime
 
 import (
+	"context"
 	"fmt"
+	"net"
 	"path/filepath"
+	"strings"
 
 	"github.com/sealerio/sealer/common"
 	containerruntime "github.com/sealerio/sealer/pkg/container-runtime"
@@ -30,6 +33,7 @@ import (
 	"github.com/sealerio/sealer/pkg/runtime/kubernetes"
 	"github.com/sealerio/sealer/pkg/runtime/kubernetes/kubeadm"
 	v1 "github.com/sealerio/sealer/types/api/v1"
+	k8sv1 "k8s.io/api/core/v1"
 )
 
 // RuntimeConfig for Installer
@@ -188,6 +192,15 @@ func (i *Installer) Install() error {
 		return err
 	}
 
+	runtimeDriver, err := kubeRuntimeInstaller.GetCurrentRuntimeDriver()
+	if err != nil {
+		return err
+	}
+
+	if err = i.setNodeTaints(all, runtimeDriver); err != nil {
+		return err
+	}
+
 	appInstaller := NewAppInstaller(i.infraDriver, i.Distributor, extension, i.RegistryConfig)
 
 	if err = appInstaller.LaunchClusterImage(master0, cmds); err != nil {
@@ -226,3 +239,40 @@ func (i *Installer) GetCurrentDriver() (registry.Driver, runtime.Driver, error)
 
 	return registryDriver, runtimeDriver, nil
 }
+
+func (i *Installer) setNodeTaints(hosts []net.IP, driver runtime.Driver) error {
+	var (
+		k8snode k8sv1.Node
+		ok      bool
+	)
+	nodeList := k8sv1.NodeList{}
+	if err := driver.List(context.TODO(), &nodeList); err != nil {
+		return fmt.Errorf("failed to list cluster nodes: %v", err)
+	}
+	nodeTaint := make(map[string]k8sv1.Node)
+	for _, node := range nodeList.Items {
+		nodeTaint[getAddress(node.Status.Addresses)] = node
+	}
+
+	for _, ip := range hosts {
+		taints := i.infraDriver.GetHostTaints(ip)
+		if taints != nil {
+			if k8snode, ok = nodeTaint[ip.String()]; ok {
+				k8snode.Spec.Taints = taints
+			}
+			if err := driver.Update(context.TODO(), &k8snode); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func getAddress(addresses []k8sv1.NodeAddress) string {
+	for _, v := range addresses {
+		if strings.EqualFold(string(v.Type), "InternalIP") {
+			return v.Address
+		}
+	}
+	return ""
+}

pkg/cluster-runtime/scale.go

@@ -90,6 +90,10 @@ func (i *Installer) ScaleUp(newMasters, newWorkers []net.IP) (registry.Driver, r
 		return nil, nil, err
 	}
 
+	if err = i.setNodeTaints(all, runtimeDriver); err != nil {
+		return nil, nil, err
+	}
+
 	return registryDriver, runtimeDriver, nil
 }
 

pkg/infradriver/infradriver.go

@@ -17,12 +17,17 @@ package infradriver
 import (
 	"net"
 
+	k8sv1 "k8s.io/api/core/v1"
+
 	v1 "github.com/sealerio/sealer/types/api/v1"
 )
 
 // InfraDriver treat the entire cluster as an operating system kernel,
 // interface function here is the target system call.
 type InfraDriver interface {
+	// GetHostTaints GetHostTaint return host taint
+	GetHostTaints(host net.IP) []k8sv1.Taint
+
 	GetHostIPList() []net.IP
 
 	GetHostIPListByRole(role string) []net.IP

pkg/infradriver/ssh_infradriver.go

@@ -29,12 +29,14 @@ import (
 	"github.com/sealerio/sealer/utils/shellcommand"
 	"github.com/sealerio/sealer/utils/ssh"
 	"golang.org/x/sync/errgroup"
+	k8sv1 "k8s.io/api/core/v1"
 	k8snet "k8s.io/utils/net"
 )
 
 type SSHInfraDriver struct {
 	sshConfigs         map[string]ssh.Interface
 	hosts              []net.IP
+	hostTaint          map[string][]k8sv1.Taint
 	roleHostsMap       map[string][]net.IP
 	hostEnvMap         map[string]map[string]interface{}
 	clusterEnv         map[string]interface{}
@@ -57,6 +59,18 @@ func mergeList(hostEnv, globalEnv map[string]interface{}) map[string]interface{}
 	return hostEnv
 }
 
+func convertTaints(taints []string) ([]k8sv1.Taint, error) {
+	var k8staints []k8sv1.Taint
+	for _, taint := range taints {
+		data, err := formatData(taint)
+		if err != nil {
+			return nil, err
+		}
+		k8staints = append(k8staints, data)
+	}
+	return k8staints, nil
+}
+
 func copyEnv(origin map[string]interface{}) map[string]interface{} {
 	if origin == nil {
 		return nil
@@ -105,6 +119,7 @@ func NewInfraDriver(cluster *v2.Cluster) (InfraDriver, error) {
 		roleHostsMap:      map[string][]net.IP{},
 		// todo need to separate env into app render data and sys render data
 		hostEnvMap:         map[string]map[string]interface{}{},
+		hostTaint:          map[string][]k8sv1.Taint{},
 		clusterHostAliases: cluster.Spec.HostAliases,
 	}
 
@@ -157,12 +172,21 @@ func NewInfraDriver(cluster *v2.Cluster) (InfraDriver, error) {
 	for _, host := range cluster.Spec.Hosts {
 		for _, ip := range host.IPS {
 			ret.hostEnvMap[ip.String()] = mergeList(ConvertEnv(host.Env), ret.clusterEnv)
+			taints, err := convertTaints(host.Taints)
+			if err != nil {
+				return nil, err
+			}
+			ret.hostTaint[ip.String()] = taints
 		}
 	}
 
 	return ret, err
 }
 
+func (d *SSHInfraDriver) GetHostTaints(host net.IP) []k8sv1.Taint {
+	return d.hostTaint[host.String()]
+}
+
 func (d *SSHInfraDriver) GetHostIPList() []net.IP {
 	return d.hosts
 }

pkg/infradriver/utils.go

@@ -0,0 +1,89 @@
+// Copyright © 2022 Alibaba Group Holding Ltd.
+//
+// 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 infradriver
+
+import (
+	"fmt"
+	"strings"
+
+	k8sv1 "k8s.io/api/core/v1"
+)
+
+const (
+	DelSymbol   = "-"
+	EqualSymbol = "="
+	ColonSymbol = ":"
+)
+
+// FormatData process data in the specified format
+//eg: key1=value1:NoSchedule;key1:NoSchedule;key1=value1:NoSchedule
+func formatData(data string) (k8sv1.Taint, error) {
+	var (
+		key, value, effect string
+		taint              k8sv1.Taint
+		TaintEffectValues  = []k8sv1.TaintEffect{k8sv1.TaintEffectNoSchedule, k8sv1.TaintEffectNoExecute, k8sv1.TaintEffectPreferNoSchedule}
+	)
+
+	data = strings.TrimSpace(data)
+	switch {
+	case strings.Contains(data, EqualSymbol) && !strings.Contains(data, EqualSymbol+ColonSymbol):
+		temps := strings.Split(data, EqualSymbol)
+		if len(temps) != 2 {
+			return k8sv1.Taint{}, fmt.Errorf("faild to split taint argument: %s", data)
+		}
+		key = temps[0]
+		taintArgs := strings.Split(temps[1], ColonSymbol)
+		if len(taintArgs) != 2 {
+			return k8sv1.Taint{}, fmt.Errorf("error: invalid taint data: %s", data)
+		}
+		value, effect = taintArgs[0], taintArgs[1]
+		effect = strings.TrimSuffix(effect, DelSymbol)
+
+	case !strings.Contains(data, EqualSymbol):
+		temps := strings.Split(data, ColonSymbol)
+		if len(temps) != 2 {
+			return k8sv1.Taint{}, fmt.Errorf("faild to split taint argument: %s", data)
+		}
+		key, value, effect = temps[0], "", temps[1]
+
+	case strings.Contains(data, EqualSymbol+ColonSymbol):
+		temps := strings.Split(data, EqualSymbol+ColonSymbol)
+		if len(temps) != 2 {
+			return k8sv1.Taint{}, fmt.Errorf("faild to split taint argument: %s", data)
+		}
+		key, value, effect = temps[0], "", temps[1]
+	}
+
+	effect = strings.TrimSuffix(effect, DelSymbol)
+	if notInEffect(k8sv1.TaintEffect(effect), TaintEffectValues) {
+		return k8sv1.Taint{}, fmt.Errorf("taint effect %s need in %v", data, TaintEffectValues)
+	}
+
+	taint = k8sv1.Taint{
+		Key:    key,
+		Value:  value,
+		Effect: k8sv1.TaintEffect(effect),
+	}
+	return taint, nil
+}
+
+func notInEffect(effect k8sv1.TaintEffect, effects []k8sv1.TaintEffect) bool {
+	for _, e := range effects {
+		if e == effect {
+			return false
+		}
+	}
+	return true
+}

types/api/v2/cluster_types.go

@@ -52,7 +52,8 @@ type Host struct {
 	//overwrite SSH config
 	SSH v1.SSH `json:"ssh,omitempty"`
 	//overwrite env
-	Env []string `json:"env,omitempty"`
+	Env    []string `json:"env,omitempty"`
+	Taints []string `json:"taints,omitempty"`
 }
 
 // HostAlias holds the mapping between IP and hostnames that will be injected as an entry in the