rke/hosts/hosts.go

package hosts

import (
	"context"
	"fmt"
	"path"
	"strings"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/container"
	"github.com/sirupsen/logrus"

	"github.com/docker/docker/client"
	"github.com/rancher/rke/docker"
	"github.com/rancher/rke/k8s"
	"github.com/rancher/rke/log"
	v3 "github.com/rancher/rke/types"
	"github.com/rancher/rke/util"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/client-go/kubernetes"
)

type Host struct {
	v3.RKEConfigNode
	DClient             *client.Client
	LocalConnPort       int
	IsControl           bool
	IsWorker            bool
	IsEtcd              bool
	IgnoreDockerVersion bool
	ToAddEtcdMember     bool
	ExistingEtcdCluster bool
	SavedKeyPhrase      string
	ToAddLabels         map[string]string
	ToDelLabels         map[string]string
	ToAddTaints         []string
	ToDelTaints         []string
	DockerInfo          types.Info
	UpdateWorker        bool
	PrefixPath          string
	BastionHost         v3.BastionHost
}

const (
	ToCleanEtcdDir          = "/var/lib/etcd/"
	ToCleanSSLDir           = "/etc/kubernetes/"
	ToCleanCNIConf          = "/etc/cni/"
	ToCleanCNIBin           = "/opt/cni/"
	ToCleanCNILib           = "/var/lib/cni/"
	ToCleanCalicoRun        = "/var/run/calico/"
	ToCleanTempCertPath     = "/etc/kubernetes/.tmp/"
	CleanerContainerName    = "kube-cleaner"
	LogCleanerContainerName = "rke-log-cleaner"
	RKELogsPath             = "/var/lib/rancher/rke/log"
	SELinuxLabel            = "label=type:rke_container_t"

	B2DOS               = "Boot2Docker"
	B2DPrefixPath       = "/mnt/sda1/rke"
	ROS                 = "RancherOS"
	ROSPrefixPath       = "/opt/rke"
	CoreOS              = "CoreOS"
	CoreOSPrefixPath    = "/opt/rke"
	FlatcarOS           = "Flatcar"
	FlatcarOSPrefixPath = "/opt/rke"
	WindowsOS           = "Windows"
	WindowsPrefixPath   = "c:/"
)

func (h *Host) CleanUpAll(ctx context.Context, cleanerImage string, prsMap map[string]v3.PrivateRegistry, externalEtcd bool, k8sVersion string) error {
	log.Infof(ctx, "[hosts] Cleaning up host [%s]", h.Address)
	toCleanPaths := []string{
		path.Join(h.PrefixPath, ToCleanSSLDir),
		ToCleanCNIConf,
		ToCleanCNIBin,
		ToCleanCalicoRun,
		path.Join(h.PrefixPath, ToCleanTempCertPath),
		path.Join(h.PrefixPath, ToCleanCNILib),
	}

	if !externalEtcd {
		toCleanPaths = append(toCleanPaths, path.Join(h.PrefixPath, ToCleanEtcdDir))
	}
	return h.CleanUp(ctx, toCleanPaths, cleanerImage, prsMap, k8sVersion)
}

func (h *Host) CleanUpWorkerHost(ctx context.Context, cleanerImage string, prsMap map[string]v3.PrivateRegistry, k8sVersion string) error {
	if h.IsControl || h.IsEtcd {
		log.Infof(ctx, "[hosts] Host [%s] is already a controlplane or etcd host, skipping cleanup.", h.Address)
		return nil
	}
	toCleanPaths := []string{
		path.Join(h.PrefixPath, ToCleanSSLDir),
		ToCleanCNIConf,
		ToCleanCNIBin,
		ToCleanCalicoRun,
		path.Join(h.PrefixPath, ToCleanCNILib),
	}
	return h.CleanUp(ctx, toCleanPaths, cleanerImage, prsMap, k8sVersion)
}

func (h *Host) CleanUpControlHost(ctx context.Context, cleanerImage string, prsMap map[string]v3.PrivateRegistry, k8sVersion string) error {
	if h.IsWorker || h.IsEtcd {
		log.Infof(ctx, "[hosts] Host [%s] is already a worker or etcd host, skipping cleanup.", h.Address)
		return nil
	}
	toCleanPaths := []string{
		path.Join(h.PrefixPath, ToCleanSSLDir),
		ToCleanCNIConf,
		ToCleanCNIBin,
		ToCleanCalicoRun,
		path.Join(h.PrefixPath, ToCleanCNILib),
	}
	return h.CleanUp(ctx, toCleanPaths, cleanerImage, prsMap, k8sVersion)
}

func (h *Host) CleanUpEtcdHost(ctx context.Context, cleanerImage string, prsMap map[string]v3.PrivateRegistry, k8sVersion string) error {
	toCleanPaths := []string{
		path.Join(h.PrefixPath, ToCleanEtcdDir),
		path.Join(h.PrefixPath, ToCleanSSLDir),
	}
	if h.IsWorker || h.IsControl {
		log.Infof(ctx, "[hosts] Host [%s] is already a worker or control host, skipping cleanup certs.", h.Address)
		toCleanPaths = []string{
			path.Join(h.PrefixPath, ToCleanEtcdDir),
		}
	}
	return h.CleanUp(ctx, toCleanPaths, cleanerImage, prsMap, k8sVersion)
}

func (h *Host) CleanUp(ctx context.Context, toCleanPaths []string, cleanerImage string, prsMap map[string]v3.PrivateRegistry, k8sVersion string) error {
	log.Infof(ctx, "[hosts] Cleaning up host [%s]", h.Address)
	imageCfg, hostCfg := buildCleanerConfig(h, toCleanPaths, cleanerImage, k8sVersion)
	log.Infof(ctx, "[hosts] Running cleaner container on host [%s]", h.Address)
	if err := docker.DoRunContainer(ctx, h.DClient, imageCfg, hostCfg, CleanerContainerName, h.Address, CleanerContainerName, prsMap); err != nil {
		return err
	}

	if _, err := docker.WaitForContainer(ctx, h.DClient, h.Address, CleanerContainerName, true); err != nil {
		return err
	}

	log.Infof(ctx, "[hosts] Removing cleaner container on host [%s]", h.Address)
	if err := docker.RemoveContainer(ctx, h.DClient, h.Address, CleanerContainerName); err != nil {
		return err
	}
	log.Infof(ctx, "[hosts] Removing dead container logs on host [%s]", h.Address)
	if err := DoRunLogCleaner(ctx, h, cleanerImage, prsMap); err != nil {
		return err
	}
	log.Infof(ctx, "[hosts] Successfully cleaned up host [%s]", h.Address)
	return nil
}

func (h *Host) OS() string {
	return h.DockerInfo.OSType
}

func (h *Host) IsWindows() bool {
	return h.DockerInfo.OSType == "windows"
}

func (h *Host) IsLinux() bool {
	return h.DockerInfo.OSType == "linux"
}

func (h *Host) ProcessFilter(processes map[string]v3.Process) map[string]v3.Process {
	if h.IsWindows() {
		for name, process := range processes {
			// doesn't support host network on windows
			if process.NetworkMode == "host" {
				process.NetworkMode = ""
			}

			// doesn't support PID on windows
			if process.PidMode != "" {
				process.PidMode = ""
			}

			// doesn't support privileged mode on windows
			if process.Privileged {
				process.Privileged = false
			}

			// doesn't execute health check
			process.HealthCheck = v3.HealthCheck{}

			processes[name] = process
		}
	}

	return processes
}

func DeleteNode(ctx context.Context, toDeleteHost *Host, kubeClient *kubernetes.Clientset, hasAnotherRole bool, cloudProvider string) error {
	if hasAnotherRole {
		log.Infof(ctx, "[hosts] host [%s] has another role, skipping delete from kubernetes cluster", toDeleteHost.Address)
		return nil
	}
	log.Infof(ctx, "[hosts] Cordoning host [%s]", toDeleteHost.Address)
	if _, err := k8s.GetNode(kubeClient, toDeleteHost.HostnameOverride); err != nil {
		if apierrors.IsNotFound(err) {
			log.Warnf(ctx, "[hosts] Can't find node by name [%s]", toDeleteHost.Address)
			return nil
		}
		return err

	}
	if err := k8s.CordonUncordon(kubeClient, toDeleteHost.HostnameOverride, true); err != nil {
		return err
	}
	log.Infof(ctx, "[hosts] Deleting host [%s] from the cluster", toDeleteHost.Address)
	if err := k8s.DeleteNode(kubeClient, toDeleteHost.HostnameOverride, cloudProvider); err != nil {
		return err
	}
	log.Infof(ctx, "[hosts] Successfully deleted host [%s] from the cluster", toDeleteHost.Address)
	return nil
}

func RemoveTaintFromHost(ctx context.Context, host *Host, taintKey string, kubeClient *kubernetes.Clientset) error {
	log.Infof(ctx, "[hosts] removing taint [%s] from host [%s]", taintKey, host.Address)
	if err := k8s.RemoveTaintFromNodeByKey(kubeClient, host.HostnameOverride, taintKey); err != nil {
		return err
	}
	log.Infof(ctx, "[hosts] Successfully deleted taint [%s] from host [%s]", taintKey, host.Address)
	return nil
}

func GetToDeleteHosts(currentHosts, configHosts, inactiveHosts []*Host, includeInactive bool) []*Host {
	toDeleteHosts := []*Host{}
	for _, currentHost := range currentHosts {
		found := false
		for _, newHost := range configHosts {
			if currentHost.Address == newHost.Address {
				found = true
			}
		}
		if !found {
			inactive := false
			for _, inactiveHost := range inactiveHosts {
				if inactiveHost.Address == currentHost.Address {
					inactive = true
					break
				}
			}
			if (inactive && includeInactive) || !inactive {
				toDeleteHosts = append(toDeleteHosts, currentHost)
			}
		}
	}
	return toDeleteHosts
}

func GetToAddHosts(currentHosts, configHosts []*Host) []*Host {
	toAddHosts := []*Host{}
	for _, configHost := range configHosts {
		found := false
		for _, currentHost := range currentHosts {
			if currentHost.Address == configHost.Address {
				found = true
				break
			}
		}
		if !found {
			toAddHosts = append(toAddHosts, configHost)
		}
	}
	return toAddHosts
}

func IsHostListChanged(currentHosts, configHosts []*Host) bool {
	changed := false
	for _, host := range currentHosts {
		found := false
		for _, configHost := range configHosts {
			if host.Address == configHost.Address {
				found = true
				break
			}
		}
		if !found {
			return true
		}
	}
	for _, host := range configHosts {
		found := false
		for _, currentHost := range currentHosts {
			if host.Address == currentHost.Address {
				found = true
				break
			}
		}
		if !found {
			return true
		}
	}
	return changed
}

func buildCleanerConfig(host *Host, toCleanDirs []string, cleanerImage, k8sVersion string) (*container.Config, *container.HostConfig) {
	cmd := []string{
		"sh",
		"-c",
		fmt.Sprintf("find %s -mindepth 1 -delete", strings.Join(toCleanDirs, " ")),
	}
	imageCfg := &container.Config{
		Image: cleanerImage,
		Cmd:   cmd,
	}
	binds := []string{}
	for _, vol := range toCleanDirs {
		binds = append(binds, fmt.Sprintf("%s:%s:z", vol, vol))
	}
	hostCfg := &container.HostConfig{}

	matchedRange, err := util.SemVerMatchRange(k8sVersion, util.SemVerK8sVersion122OrHigher)
	if err != nil {
		return nil, nil
	}

	if matchedRange {
		binds = util.RemoveZFromBinds(binds)

		if IsDockerSELinuxEnabled(host) {
			hostCfg.SecurityOpt = append(hostCfg.SecurityOpt, SELinuxLabel)
		}
	}

	hostCfg.Binds = binds
	return imageCfg, hostCfg
}

func NodesToHosts(rkeNodes []v3.RKEConfigNode, nodeRole string) []*Host {
	hostList := make([]*Host, 0)
	// Return all nodes if there is no noderole passed to the function
	if nodeRole == "" {
		for _, node := range rkeNodes {
			newHost := Host{
				RKEConfigNode: node,
			}
			hostList = append(hostList, &newHost)
		}
		return hostList
	}
	for _, node := range rkeNodes {
		for _, role := range node.Role {
			if role == nodeRole {
				newHost := Host{
					RKEConfigNode: node,
				}
				hostList = append(hostList, &newHost)
				break
			}
		}
	}
	return hostList
}

func GetUniqueHostList(etcdHosts, cpHosts, workerHosts []*Host) []*Host {
	hostList := []*Host{}
	hostList = append(hostList, etcdHosts...)
	hostList = append(hostList, cpHosts...)
	hostList = append(hostList, workerHosts...)
	// little trick to get a unique host list
	uniqHostMap := make(map[*Host]bool)
	for _, host := range hostList {
		uniqHostMap[host] = true
	}
	uniqHostList := []*Host{}
	for host := range uniqHostMap {
		uniqHostList = append(uniqHostList, host)
	}
	return uniqHostList
}

func (h *Host) SetPrefixPath(clusterPrefixPath string) {
	var prefixPath string
	switch {
	case clusterPrefixPath != "/":
		prefixPath = clusterPrefixPath
	case strings.Contains(h.DockerInfo.OperatingSystem, B2DOS):
		prefixPath = B2DPrefixPath
	case strings.Contains(h.DockerInfo.OperatingSystem, ROS):
		prefixPath = ROSPrefixPath
	case strings.Contains(h.DockerInfo.OperatingSystem, CoreOS):
		prefixPath = CoreOSPrefixPath
	case strings.Contains(h.DockerInfo.OperatingSystem, FlatcarOS):
		prefixPath = FlatcarOSPrefixPath
	case strings.Contains(h.DockerInfo.OperatingSystem, WindowsOS):
		prefixPath = WindowsPrefixPath
	default:
		prefixPath = clusterPrefixPath
	}

	h.PrefixPath = prefixPath
}

func (h *Host) GetExtraBinds(service v3.BaseService) []string {
	switch {
	case h.OS() == "windows" && len(service.WindowsExtraBinds) > 0:
		return service.WindowsExtraBinds
	default:
		return service.ExtraBinds
	}
}

func (h *Host) GetExtraEnv(service v3.BaseService) []string {
	switch {
	case h.OS() == "windows" && len(service.WindowsExtraEnv) > 0:
		return service.WindowsExtraEnv
	default:
		return service.ExtraEnv
	}
}
func (h *Host) GetExtraArgs(service v3.BaseService) map[string]string {
	switch {
	case h.OS() == "windows" && len(service.WindowsExtraArgs) > 0:
		return service.WindowsExtraArgs
	default:
		return service.ExtraArgs
	}
}

func (h *Host) GetExtraArgsArray(service v3.BaseService) map[string][]string {
	if h.OS() == "windows" && len(service.WindowsExtraArgsArray) > 0 {
		return service.WindowsExtraArgsArray
	}
	return service.ExtraArgsArray
}

func DoRunLogCleaner(ctx context.Context, host *Host, alpineImage string, prsMap map[string]v3.PrivateRegistry) error {
	logrus.Debugf("[cleanup] Starting log link cleanup on host [%s]", host.Address)
	imageCfg := &container.Config{
		Image: alpineImage,
		Tty:   true,
		Cmd: []string{
			"sh",
			"-c",
			fmt.Sprintf("find %s -type l ! -exec test -e {} \\; -print -delete", RKELogsPath),
		},
	}
	hostCfg := &container.HostConfig{
		Binds: []string{
			host.DockerInfo.DockerRootDir + ":" + host.DockerInfo.DockerRootDir,
			"/var/lib:/var/lib",
		},
		Privileged: true,
	}
	if err := docker.DoRemoveContainer(ctx, host.DClient, LogCleanerContainerName, host.Address); err != nil {
		return err
	}
	if err := docker.DoRunContainer(ctx, host.DClient, imageCfg, hostCfg, LogCleanerContainerName, host.Address, "cleanup", prsMap); err != nil {
		return err
	}
	if err := docker.DoRemoveContainer(ctx, host.DClient, LogCleanerContainerName, host.Address); err != nil {
		return err
	}
	logrus.Debugf("[cleanup] Successfully cleaned up log links on host [%s]", host.Address)
	return nil
}

func IsNodeInList(host *Host, hostList []*Host) bool {
	for _, h := range hostList {
		if h.HostnameOverride == host.HostnameOverride {
			return true
		}
	}
	return false
}

func GetHostListIntersect(a []*Host, b []*Host) []*Host {
	s := []*Host{}
	hash := map[string]*Host{}
	for _, h := range a {
		hash[h.Address] = h
	}
	for _, h := range b {
		if _, ok := hash[h.Address]; ok {
			s = append(s, h)
		}
	}
	return s
}

func GetInternalAddressForHosts(hostList []*Host) []string {
	hostAddresses := []string{}
	for _, host := range hostList {
		hostAddresses = append(hostAddresses, host.InternalAddress)
	}
	return hostAddresses
}

func IsDockerSELinuxEnabled(host *Host) bool {
	for _, securityOpt := range host.DockerInfo.SecurityOptions {
		logrus.Tracef("IsDockerSELinuxEnabled: securityOpt found: [%s]", securityOpt)
		// name=selinux was the value returned after removing statically set Docker API version 1.24
		if securityOpt == "selinux" || securityOpt == "name=selinux" {
			logrus.Debugf("Host [%s] has SELinux enabled in Docker", host.Address)
			return true
		}
	}
	return false
}

func IsEnterpriseLinuxHost(host *Host) bool {
	operatingSystem := strings.ToLower(host.DockerInfo.OperatingSystem)
	if strings.Contains(operatingSystem, "centos") || strings.Contains(operatingSystem, "enterprise linux") || strings.Contains(operatingSystem, "oracle linux") {
		logrus.Debugf("Host [%s] with OperatingSystem [%s] is Enterprise Linux", host.Address, operatingSystem)
		return true
	}
	return false
}

func IsEnterpriseLinuxDocker(host *Host) bool {
	dockerInitBinary := host.DockerInfo.InitBinary
	// Init binary for Enterprise Linux Docker (not upstream) is /usr/libexec/docker/docker-init-current
	// Init binary for upstream Docker is docker-init
	if strings.EqualFold(dockerInitBinary, "/usr/libexec/docker/docker-init-current") {
		return true
	}
	return false
}