Merge pull request #23 from Mirantis/PRODENG-2674

PRODENG-2674 Adding a new phase for validating worker counts

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/k0sproject/version v0.6.0
 	github.com/sirupsen/logrus v1.9.3
 	golang.org/x/crypto v0.24.0
+	golang.org/x/exp v0.0.0-20230809150735-7b3493d9a819
 	gopkg.in/yaml.v2 v2.4.0
 )
 
@@ -111,7 +112,6 @@ require (
 	github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/zclconf/go-cty v1.14.4 // indirect
-	golang.org/x/exp v0.0.0-20230809150735-7b3493d9a819 // indirect
 	golang.org/x/mod v0.18.0 // indirect
 	golang.org/x/net v0.26.0 // indirect
 	golang.org/x/sync v0.7.0 // indirect

internal/k0sctl/action/apply.go

@@ -0,0 +1,152 @@
+package action
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/k0sproject/k0sctl/analytics"
+	"github.com/k0sproject/k0sctl/phase"
+
+	provider_phase "github.com/mirantis/terraform-provider-k0sctl/internal/k0sctl/phase"
+
+	log "github.com/sirupsen/logrus"
+)
+
+type Apply struct {
+	// Manager is the phase manager
+	Manager *phase.Manager
+	// DisableDowngradeCheck skips the downgrade check
+	DisableDowngradeCheck bool
+	// Force allows forced installation in case of certain failures
+	Force bool
+	// NoWait skips waiting for the cluster to be ready
+	NoWait bool
+	// NoDrain skips draining worker nodes
+	NoDrain bool
+	// RestoreFrom is the path to a cluster backup archive to restore the state from
+	RestoreFrom string
+	// KubeconfigOut is a writer to write the kubeconfig to
+	KubeconfigOut io.Writer
+	// KubeconfigAPIAddress is the API address to use in the kubeconfig
+	KubeconfigAPIAddress string
+	// ConfigPath is the path to the configuration file (used for kubeconfig command tip on success)
+	ConfigPath string
+}
+
+func (a Apply) Run() error {
+	start := time.Now()
+
+	phase.NoWait = a.NoWait
+	phase.Force = a.Force
+
+	lockPhase := &phase.Lock{}
+
+	a.Manager.AddPhase(
+		&phase.DefaultK0sVersion{},
+		&phase.Connect{},
+		&phase.DetectOS{},
+		lockPhase,
+		&phase.PrepareHosts{},
+		&phase.GatherFacts{},
+		&phase.ValidateHosts{},
+		&phase.GatherK0sFacts{},
+		&phase.ValidateFacts{SkipDowngradeCheck: a.DisableDowngradeCheck},
+
+		// if UploadBinaries: true
+		&phase.DownloadBinaries{}, // downloads k0s binaries to local cache
+		&phase.UploadK0s{},        // uploads k0s binaries to hosts from cache
+
+		// if UploadBinaries: false
+		&phase.DownloadK0s{}, // downloads k0s binaries directly from hosts
+
+		&phase.UploadFiles{},
+		&phase.InstallBinaries{},
+		&phase.PrepareArm{},
+		&phase.ConfigureK0s{},
+		&phase.Restore{
+			RestoreFrom: a.RestoreFrom,
+		},
+		&phase.RunHooks{Stage: "before", Action: "apply"},
+		&phase.InitializeK0s{},
+		&phase.InstallControllers{},
+		&phase.InstallWorkers{},
+		&phase.UpgradeControllers{},
+		&phase.UpgradeWorkers{NoDrain: a.NoDrain},
+		&phase.ResetWorkers{NoDrain: a.NoDrain},
+		&phase.ResetControllers{NoDrain: a.NoDrain},
+		&provider_phase.ValidateHostsExtended{},
+		&phase.RunHooks{Stage: "after", Action: "apply"},
+	)
+
+	if a.KubeconfigOut != nil {
+		a.Manager.AddPhase(&phase.GetKubeconfig{APIAddress: a.KubeconfigAPIAddress})
+	}
+
+	a.Manager.AddPhase(
+		&phase.Unlock{Cancel: lockPhase.Cancel},
+		&phase.Disconnect{},
+	)
+
+	analytics.Client.Publish("apply-start", map[string]interface{}{})
+
+	var result error
+
+	if result = a.Manager.Run(); result != nil {
+		analytics.Client.Publish("apply-failure", map[string]interface{}{"clusterID": a.Manager.Config.Spec.K0s.Metadata.ClusterID})
+		log.Info(phase.Colorize.Red("==> Apply failed").String())
+		return result
+	}
+
+	analytics.Client.Publish("apply-success", map[string]interface{}{"duration": time.Since(start), "clusterID": a.Manager.Config.Spec.K0s.Metadata.ClusterID})
+	if a.KubeconfigOut != nil {
+		if _, err := a.KubeconfigOut.Write([]byte(a.Manager.Config.Metadata.Kubeconfig)); err != nil {
+			log.Warnf("failed to write kubeconfig to %s: %v", a.KubeconfigOut, err)
+		}
+	}
+
+	duration := time.Since(start).Truncate(time.Second)
+	text := fmt.Sprintf("==> Finished in %s", duration)
+	log.Infof(phase.Colorize.Green(text).String())
+
+	for _, host := range a.Manager.Config.Spec.Hosts {
+		if host.Reset {
+			log.Info("There were nodes that got uninstalled during the apply phase. Please remove them from your k0sctl config file")
+			break
+		}
+	}
+
+	if !a.Manager.DryRun {
+		log.Infof("k0s cluster version %s is now installed", a.Manager.Config.Spec.K0s.Version)
+	}
+
+	if a.KubeconfigOut == nil {
+		cmd := &strings.Builder{}
+		executable, err := os.Executable()
+		if err != nil {
+			executable = "k0sctl"
+		} else {
+			// check if the basename of executable is in the PATH, if so, just use the basename
+			if _, err := exec.LookPath(filepath.Base(executable)); err == nil {
+				executable = filepath.Base(executable)
+			}
+		}
+
+		cmd.WriteString(executable)
+		cmd.WriteString(" kubeconfig")
+
+		if a.ConfigPath != "" && a.ConfigPath != "-" && a.ConfigPath != "k0sctl.yaml" {
+			cmd.WriteString(" --config ")
+			cmd.WriteString(a.ConfigPath)
+		}
+
+		log.Infof("Tip: To access the cluster you can now fetch the admin kubeconfig using:")
+		log.Infof("     " + phase.Colorize.Cyan(cmd.String()).String())
+	}
+
+	return nil
+}

internal/k0sctl/phase/validate_hosts_extended.go

@@ -0,0 +1,108 @@
+package phase
+
+import (
+	"fmt"
+	"strings"
+
+	k0sctl_phase "github.com/k0sproject/k0sctl/phase"
+
+	k0sctl_cluster "github.com/k0sproject/k0sctl/pkg/apis/k0sctl.k0sproject.io/v1beta1/cluster"
+	"github.com/k0sproject/rig/exec"
+	"github.com/sirupsen/logrus"
+	"golang.org/x/exp/slices"
+)
+
+type ValidateHostsExtended struct {
+	k0sctl_phase.GenericPhase
+}
+
+// Title for the phase.
+func (p *ValidateHostsExtended) Title() string {
+	return "Validate hosts"
+}
+
+// Run the phase.
+func (p *ValidateHostsExtended) Run() error {
+	return p.validateWorkerCount()
+}
+
+func (p *ValidateHostsExtended) validateWorkerCount() error {
+
+	configWorkerMachineIDs, err := p.getWorkerMachineIDs()
+	if err != nil {
+		return err
+	}
+
+	logrus.Debugf("Machine IDs of the hosts in the configuration: %s", configWorkerMachineIDs)
+
+	nodeNames, leader, err := p.getNodeNamesAndLeader()
+	if err != nil {
+		return err
+	}
+
+	logrus.Debugf("Node names: %s", nodeNames)
+
+	for _, node := range nodeNames {
+		err := p.validateAndDeleteNode(node, configWorkerMachineIDs, leader)
+		if err != nil {
+			logrus.Errorf("Error occurred while validating and deleting node: %s", err)
+		}
+	}
+
+	logrus.Debug("ValidateHostsExtended phase ran successfully")
+	return nil
+}
+
+func (p *ValidateHostsExtended) getWorkerMachineIDs() ([]string, error) {
+	var configWorkerMachineIDs []string
+
+	for _, h := range p.Config.Spec.Hosts.Workers() {
+		id, err := h.Configurer.MachineID(h)
+		if err != nil {
+			return nil, err
+		}
+		configWorkerMachineIDs = append(configWorkerMachineIDs, id)
+	}
+
+	return configWorkerMachineIDs, nil
+}
+
+func (p *ValidateHostsExtended) getNodeNamesAndLeader() ([]string, *k0sctl_cluster.Host, error) {
+	var nodeNames []string
+	var leader *k0sctl_cluster.Host
+
+	for i, h := range p.Config.Spec.Hosts.Controllers() {
+		output, err := h.ExecOutput(h.Configurer.KubectlCmdf(h, h.K0sDataDir(), "get nodes -o custom-columns=NAME:.metadata.name --no-headers"), exec.Sudo(h))
+		if err != nil {
+			if i < len(p.Config.Spec.Hosts.Controllers()) {
+				continue
+			} else {
+				logrus.Errorf("Could not retrieve the list of worker nodes. Error: %s", err)
+				return nil, nil, err
+			}
+		}
+		nodeNames = strings.Fields(output)
+		leader = h
+		break
+	}
+
+	return nodeNames, leader, nil
+}
+
+func (p *ValidateHostsExtended) validateAndDeleteNode(node string, configWorkerMachineIDs []string, leader *k0sctl_cluster.Host) error {
+	machineID, err := leader.ExecOutput(leader.Configurer.KubectlCmdf(leader, leader.K0sDataDir(), fmt.Sprintf("describe node %s | grep -i 'Machine ID:' | awk '{print $3}'", node)), exec.Sudo(leader))
+	if err != nil {
+		return err
+	}
+
+	if !slices.Contains(configWorkerMachineIDs, machineID) {
+		output, err := leader.ExecOutput(leader.Configurer.KubectlCmdf(leader, leader.K0sDataDir(), fmt.Sprintf("delete node %s", node)), exec.Sudo(leader))
+		if err != nil {
+			logrus.Errorf("Error occurred while deleting node: %s. Kubectl output: %s. Error: %s", node, output, err)
+			return err
+		}
+		logrus.Debugf("Node %s successfully deleted", node)
+	}
+
+	return nil
+}

internal/provider/k0s_config_resource.go

@@ -11,8 +11,12 @@ import (
 	"github.com/hashicorp/terraform-plugin-framework/types"
 	"github.com/hashicorp/terraform-plugin-log/tflog"
 
-	k0sctl_action "github.com/k0sproject/k0sctl/action"
 	k0sctl_phase "github.com/k0sproject/k0sctl/phase"
+
+	k0sctl_action "github.com/k0sproject/k0sctl/action"
+
+	provider_action "github.com/mirantis/terraform-provider-k0sctl/internal/k0sctl/action"
+
 	k0sctl_v1beta1 "github.com/k0sproject/k0sctl/pkg/apis/k0sctl.k0sproject.io/v1beta1"
 )
 
@@ -168,7 +172,7 @@ func (r *K0sctlConfigResource) Update(ctx context.Context, req resource.UpdateRe
 
 	kc = bytes.NewBuffer([]byte{})
 
-	aa := k0sctl_action.Apply{
+	aa := provider_action.Apply{
 		Force:         kcsm.Force.ValueBool(),
 		Manager:       pm,
 		KubeconfigOut: kc,