Merge pull request #265 from Danil-Grigorev/reconcile-etcd-members-sc…

…ale-down

🐛 Reconcile etcd members on control plane scale down

Makefile

@@ -48,6 +48,8 @@ E2E_DATA_DIR ?= $(ROOT_DIR)/test/e2e/data
 E2E_CONF_FILE ?= $(ROOT_DIR)/test/e2e/config/e2e_conf.yaml
 
 export PATH := $(abspath $(TOOLS_BIN_DIR)):$(PATH)
+export KREW_ROOT := $(abspath $(TOOLS_BIN_DIR))
+export PATH := $(KREW_ROOT)/bin:$(PATH)
 
 # Set --output-base for conversion-gen if we are not within GOPATH
 ifneq ($(abspath $(ROOT_DIR)),$(shell go env GOPATH)/src/github.com/rancher-sandbox/cluster-api-provider-rke2)
@@ -99,7 +101,7 @@ GOLANGCI_LINT_VER := v1.55.1
 GOLANGCI_LINT_BIN := golangci-lint
 GOLANGCI_LINT := $(abspath $(TOOLS_BIN_DIR)/$(GOLANGCI_LINT_BIN))
 
-GINKGO_VER := v2.14.0
+GINKGO_VER := v2.16.0
 GINKGO_BIN := ginkgo
 GINKGO := $(abspath $(TOOLS_BIN_DIR)/$(GINKGO_BIN)-$(GINKGO_VER))
 GINKGO_PKG := github.com/onsi/ginkgo/v2/ginkgo
@@ -363,6 +365,12 @@ kind-cluster: ## Create a new kind cluster designed for development with Tilt
 tilt-up: kind-cluster ## Start tilt and build kind cluster if needed.
 	tilt up
 
+.PHONY: kubectl
+kubectl: # Download kubectl cli into tools bin folder
+	hack/ensure-kubectl.sh \
+		-b $(TOOLS_BIN_DIR) \
+		$(KUBECTL_VERSION)
+
 ## --------------------------------------
 ## E2E
 ## --------------------------------------
@@ -376,15 +384,15 @@ GINKGO_NODES ?= 1
 GINKGO_NOCOLOR ?= false
 GINKGO_ARGS ?=
 GINKGO_TIMEOUT ?= 2h
-GINKGO_POLL_PROGRESS_AFTER ?= 10m
-GINKGO_POLL_PROGRESS_INTERVAL ?= 1m
+GINKGO_POLL_PROGRESS_AFTER ?= 25m
+GINKGO_POLL_PROGRESS_INTERVAL ?= 2m
 ARTIFACTS ?= $(ROOT_DIR)/_artifacts
 SKIP_CLEANUP ?= false
 SKIP_CREATE_MGMT_CLUSTER ?= false
 
 .PHONY: test-e2e-run
-test-e2e-run: $(GINKGO) $(KUSTOMIZE) e2e-image inotify-check ## Run the end-to-end tests
-	CAPI_KUSTOMIZE_PATH="$(KUSTOMIZE)" time $(GINKGO) -v --trace -poll-progress-after=$(GINKGO_POLL_PROGRESS_AFTER) -poll-progress-interval=$(GINKGO_POLL_PROGRESS_INTERVAL) \
+test-e2e-run: $(GINKGO) $(KUSTOMIZE) kubectl e2e-image inotify-check ## Run the end-to-end tests
+	CAPI_KUSTOMIZE_PATH="$(KUSTOMIZE)" time $(GINKGO) -v -poll-progress-after=$(GINKGO_POLL_PROGRESS_AFTER) -poll-progress-interval=$(GINKGO_POLL_PROGRESS_INTERVAL) \
 	--tags=e2e --focus="$(GINKGO_FOCUS)" -skip="$(GINKGO_SKIP)" --nodes=$(GINKGO_NODES) --no-color=$(GINKGO_NOCOLOR) \
 	--timeout=$(GINKGO_TIMEOUT) --output-dir="$(ARTIFACTS)" --junit-report="junit.e2e_suite.1.xml" $(GINKGO_ARGS) ./test/e2e -- \
 		-e2e.artifacts-folder="$(ARTIFACTS)" \

bootstrap/internal/cloudinit/cloudinit_test.go

@@ -48,7 +48,7 @@ runcmd:
   - 'INSTALL_RKE2_ARTIFACT_PATH=/opt/rke2-artifacts INSTALL_RKE2_TYPE="agent" sh /opt/install.sh'
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - 'echo success > /run/cluster-api/bootstrap-success.complete'
 `))
 	})
@@ -82,7 +82,7 @@ runcmd:
   - 'curl -sfL https://get.rke2.io | INSTALL_RKE2_VERSION=v1.25.6+rke2r1 INSTALL_RKE2_TYPE="agent" sh -s -'
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - 'echo success > /run/cluster-api/bootstrap-success.complete'
 `))
 	})
@@ -118,7 +118,7 @@ runcmd:
   - 'curl -sfL https://get.rke2.io | INSTALL_RKE2_VERSION= INSTALL_RKE2_TYPE="agent" sh -s -'
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - 'echo success > /run/cluster-api/bootstrap-success.complete'
 `))
 	})
@@ -154,7 +154,7 @@ runcmd:
   - '/opt/rke2-cis-script.sh'
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - 'echo success > /run/cluster-api/bootstrap-success.complete'
 `))
 	})
@@ -175,7 +175,7 @@ runcmd:
   - 'curl -sfL https://get.rke2.io | INSTALL_RKE2_VERSION= INSTALL_RKE2_TYPE=\"agent\" sh -s -'
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - 'echo success > /run/cluster-api/bootstrap-success.complete'
 `
 
@@ -263,7 +263,7 @@ runcmd:
   - '/opt/rke2-cis-script.sh'
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - 'echo success > /run/cluster-api/bootstrap-success.complete'
 `))
 	})
@@ -295,7 +295,7 @@ runcmd:
   - '/opt/rke2-cis-script.sh'
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - 'echo success > /run/cluster-api/bootstrap-success.complete'
 device_aliases:
   ephemeral0: /dev/vdb

bootstrap/internal/cloudinit/controlplane_init.go

@@ -35,7 +35,8 @@ runcmd:
   - '/opt/rke2-cis-script.sh'{{ end }}
   - 'systemctl enable rke2-server.service'
   - 'systemctl start rke2-server.service'
-  - 'mkdir /run/cluster-api' 
+  - 'kubectl create secret tls cluster-etcd -o yaml --dry-run=client -n kube-system --cert=/var/lib/rancher/rke2/server/tls/etcd/server-ca.crt --key=/var/lib/rancher/rke2/server/tls/etcd/server-ca.key --kubeconfig /etc/rancher/rke2/rke2.yaml | kubectl apply -f- --kubeconfig /etc/rancher/rke2/rke2.yaml'
+  - 'mkdir -p /run/cluster-api'
   - '{{ .SentinelFileCommand }}'
 {{- template "commands" .PostRKE2Commands }}
 {{ .AdditionalCloudInit -}}

bootstrap/internal/cloudinit/worker_join.go

@@ -33,7 +33,7 @@ runcmd:
   - '/opt/rke2-cis-script.sh'{{ end }}
   - 'systemctl enable rke2-agent.service'
   - 'systemctl start rke2-agent.service'
-  - 'mkdir /run/cluster-api' 
+  - 'mkdir -p /run/cluster-api'
   - '{{ .SentinelFileCommand }}'
 {{- template "commands" .PostRKE2Commands }}
 {{ .AdditionalCloudInit -}}

bootstrap/internal/ignition/ignition.go

@@ -36,6 +36,10 @@ var (
 		"setenforce 0",
 		"systemctl enable rke2-server.service",
 		"systemctl start rke2-server.service",
+		"kubectl create secret tls cluster-etcd -o yaml --dry-run=client -n kube-system " +
+			"--cert=/var/lib/rancher/rke2/server/tls/etcd/server-ca.crt --key=/var/lib/rancher/rke2/server/tls/etcd/server-ca.key " +
+			"--kubeconfig /etc/rancher/rke2/rke2.yaml |" +
+			" kubectl apply -f- --kubeconfig /etc/rancher/rke2/rke2.yaml",
 		"restorecon /etc/systemd/system/rke2-server.service",
 		"mkdir -p /run/cluster-api && echo success > /run/cluster-api/bootstrap-success.complete",
 		"setenforce 1",

bootstrap/internal/ignition/ignition_test.go

@@ -207,15 +207,15 @@ var _ = Describe("getControlPlaneRKE2Commands", func() {
 	It("should return slice of control plane commands", func() {
 		commands, err := getControlPlaneRKE2Commands(baseUserData)
 		Expect(err).ToNot(HaveOccurred())
-		Expect(commands).To(HaveLen(8))
+		Expect(commands).To(HaveLen(9))
 		Expect(commands).To(ContainElements(fmt.Sprintf(controlPlaneCommand, baseUserData.RKE2Version), serverDeployCommands[0], serverDeployCommands[1]))
 	})
 
 	It("should return slice of control plane commands with air gapped", func() {
 		baseUserData.AirGapped = true
 		commands, err := getControlPlaneRKE2Commands(baseUserData)
 		Expect(err).ToNot(HaveOccurred())
-		Expect(commands).To(HaveLen(8))
+		Expect(commands).To(HaveLen(9))
 		Expect(commands).To(ContainElements(airGappedControlPlaneCommand, serverDeployCommands[0], serverDeployCommands[1]))
 	})
 

controlplane/internal/controllers/rke2controlplane_controller.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/blang/semver/v4"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
@@ -39,6 +40,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	"sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/annotations"
 	"sigs.k8s.io/cluster-api/util/collections"
@@ -65,11 +67,15 @@ const (
 type RKE2ControlPlaneReconciler struct {
 	Log logr.Logger
 	client.Client
-	Scheme                    *runtime.Scheme
+	Scheme *runtime.Scheme
+
+	SecretCachingClient client.Client
+
 	managementClusterUncached rke2.ManagementCluster
 	managementCluster         rke2.ManagementCluster
 	recorder                  record.EventRecorder
 	controller                controller.Controller
+	workloadCluster           rke2.WorkloadCluster
 }
 
 //nolint:lll
@@ -231,12 +237,42 @@ func (r *RKE2ControlPlaneReconciler) SetupWithManager(ctx context.Context, mgr c
 	r.controller = c
 	r.recorder = mgr.GetEventRecorderFor("rke2-control-plane-controller")
 
+	// Set up a ClusterCacheTracker and ClusterCacheReconciler to provide to controllers
+	// requiring a connection to a remote cluster
+	tracker, err := remote.NewClusterCacheTracker(
+		mgr,
+		remote.ClusterCacheTrackerOptions{
+			SecretCachingClient: r.SecretCachingClient,
+			ControllerName:      "rke2-control-plane-controller",
+			Log:                 &ctrl.Log,
+			Indexes:             []remote.Index{},
+			ClientUncachedObjects: []client.Object{
+				&corev1.ConfigMap{},
+				&corev1.Secret{},
+			},
+		},
+	)
+	if err != nil {
+		return errors.Wrap(err, "unable to create cluster cache tracker")
+	}
+
+	if err := (&remote.ClusterCacheReconciler{
+		Client:  mgr.GetClient(),
+		Tracker: tracker,
+	}).SetupWithManager(ctx, mgr, controller.Options{}); err != nil {
+		return errors.Wrap(err, "unable to create controller")
+	}
+
 	if r.managementCluster == nil {
-		r.managementCluster = &rke2.Management{Client: r.Client}
+		r.managementCluster = &rke2.Management{
+			Client:              r.Client,
+			SecretCachingClient: r.SecretCachingClient,
+			Tracker:             tracker,
+		}
 	}
 
 	if r.managementClusterUncached == nil {
-		r.managementClusterUncached = &rke2.Management{Client: mgr.GetAPIReader()}
+		r.managementClusterUncached = &rke2.Management{Client: mgr.GetClient()}
 	}
 
 	return nil
@@ -462,6 +498,12 @@ func (r *RKE2ControlPlaneReconciler) reconcileNormal(
 		return result, err
 	}
 
+	// Ensures the number of etcd members is in sync with the number of machines/nodes.
+	// NOTE: This is usually required after a machine deletion.
+	if err := r.reconcileEtcdMembers(ctx, controlPlane); err != nil {
+		return ctrl.Result{}, err
+	}
+
 	// Control plane machines rollout due to configuration changes (e.g. upgrades) takes precedence over other operations.
 	needRollout := controlPlane.MachinesNeedingRollout()
 
@@ -518,6 +560,72 @@ func (r *RKE2ControlPlaneReconciler) reconcileNormal(
 	return ctrl.Result{}, nil
 }
 
+// GetWorkloadCluster builds a cluster object.
+// The cluster comes with an etcd client generator to connect to any etcd pod living on a managed machine.
+func (r *RKE2ControlPlaneReconciler) GetWorkloadCluster(ctx context.Context, controlPlane *rke2.ControlPlane) (rke2.WorkloadCluster, error) {
+	workloadCluster, err := r.managementCluster.GetWorkloadCluster(ctx, client.ObjectKeyFromObject(controlPlane.Cluster))
+	if err != nil {
+		return nil, err
+	}
+
+	r.workloadCluster = workloadCluster
+
+	return r.workloadCluster, nil
+}
+
+// reconcileEtcdMembers ensures the number of etcd members is in sync with the number of machines/nodes.
+// This is usually required after a machine deletion.
+//
+// NOTE: this func uses KCP conditions, it is required to call reconcileControlPlaneConditions before this.
+func (r *RKE2ControlPlaneReconciler) reconcileEtcdMembers(ctx context.Context, controlPlane *rke2.ControlPlane) error {
+	log := ctrl.LoggerFrom(ctx)
+
+	// If there is no RKE-owned control-plane machines, then control-plane has not been initialized yet.
+	if controlPlane.Machines.Len() == 0 {
+		return nil
+	}
+
+	// Collect all the node names.
+	nodeNames := []string{}
+
+	for _, machine := range controlPlane.Machines {
+		if machine.Status.NodeRef == nil {
+			// If there are provisioning machines (machines without a node yet), return.
+			return nil
+		}
+
+		nodeNames = append(nodeNames, machine.Status.NodeRef.Name)
+	}
+
+	// Potential inconsistencies between the list of members and the list of machines/nodes are
+	// surfaced using the EtcdClusterHealthyCondition; if this condition is true, meaning no inconsistencies exists, return early.
+	if conditions.IsTrue(controlPlane.RCP, controlplanev1.EtcdClusterHealthyCondition) {
+		return nil
+	}
+
+	workloadCluster, err := r.GetWorkloadCluster(ctx, controlPlane)
+	if err != nil {
+		// Failing at connecting to the workload cluster can mean workload cluster is unhealthy for a variety of reasons such as etcd quorum loss.
+		return errors.Wrap(err, "cannot get remote client to workload cluster")
+	}
+
+	parsedVersion, err := semver.ParseTolerant(controlPlane.RCP.Spec.AgentConfig.Version)
+	if err != nil {
+		return errors.Wrapf(err, "failed to parse kubernetes version %q", controlPlane.RCP.Spec.AgentConfig.Version)
+	}
+
+	removedMembers, err := workloadCluster.ReconcileEtcdMembers(ctx, nodeNames, parsedVersion)
+	if err != nil {
+		return errors.Wrap(err, "failed attempt to reconcile etcd members")
+	}
+
+	if len(removedMembers) > 0 {
+		log.Info("Etcd members without nodes removed from the cluster", "members", removedMembers)
+	}
+
+	return nil
+}
+
 func (r *RKE2ControlPlaneReconciler) reconcileDelete(ctx context.Context,
 	cluster *clusterv1.Cluster,
 	rcp *controlplanev1.RKE2ControlPlane,

controlplane/internal/controllers/rke2controlplane_controller_test.go

@@ -230,7 +230,7 @@ var _ = Describe("Reconclie control plane conditions", func() {
 		r := &RKE2ControlPlaneReconciler{
 			Client:                    testEnv.GetClient(),
 			Scheme:                    testEnv.GetScheme(),
-			managementCluster:         &rke2.Management{Client: testEnv.GetClient()},
+			managementCluster:         &rke2.Management{Client: testEnv.GetClient(), SecretCachingClient: testEnv.GetClient()},
 			managementClusterUncached: &rke2.Management{Client: testEnv.GetClient()},
 		}
 		_, err := r.reconcileControlPlaneConditions(ctx, cp)

controlplane/internal/controllers/scale.go

@@ -153,6 +153,20 @@ func (r *RKE2ControlPlaneReconciler) scaleDownControlPlane(
 		return ctrl.Result{}, errors.New("failed to pick control plane Machine to delete")
 	}
 
+	// If etcd leadership is on machine that is about to be deleted, move it to the newest member available.
+	etcdLeaderCandidate := controlPlane.Machines.Newest()
+	if err := r.workloadCluster.ForwardEtcdLeadership(ctx, machineToDelete, etcdLeaderCandidate); err != nil {
+		logger.Error(err, "Failed to move leadership to candidate machine", "candidate", etcdLeaderCandidate.Name)
+
+		return ctrl.Result{}, err
+	}
+
+	if err := r.workloadCluster.RemoveEtcdMemberForMachine(ctx, machineToDelete); err != nil {
+		logger.Error(err, "Failed to remove etcd member for machine")
+
+		return ctrl.Result{}, err
+	}
+
 	logger = logger.WithValues("machine", machineToDelete)
 	if err := r.Client.Delete(ctx, machineToDelete); err != nil && !apierrors.IsNotFound(err) {
 		logger.Error(err, "Failed to delete control plane machine")
@@ -198,7 +212,10 @@ func (r *RKE2ControlPlaneReconciler) preflightChecks(
 	}
 
 	// Check machine health conditions; if there are conditions with False or Unknown, then wait.
-	allMachineHealthConditions := []clusterv1.ConditionType{controlplanev1.MachineAgentHealthyCondition}
+	allMachineHealthConditions := []clusterv1.ConditionType{
+		controlplanev1.MachineAgentHealthyCondition,
+		controlplanev1.MachineEtcdMemberHealthyCondition,
+	}
 	machineErrors := []error{}
 
 loopmachines:

controlplane/main.go

@@ -204,9 +204,21 @@ func setupChecks(mgr ctrl.Manager) {
 }
 
 func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
+	secretCachingClient, err := client.New(mgr.GetConfig(), client.Options{
+		HTTPClient: mgr.GetHTTPClient(),
+		Cache: &client.CacheOptions{
+			Reader: mgr.GetCache(),
+		},
+	})
+	if err != nil {
+		setupLog.Error(err, "unable to create secret caching client")
+		os.Exit(1)
+	}
+
 	if err := (&controllers.RKE2ControlPlaneReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
+		Client:              mgr.GetClient(),
+		Scheme:              mgr.GetScheme(),
+		SecretCachingClient: secretCachingClient,
 	}).SetupWithManager(ctx, mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "RKE2ControlPlane")
 		os.Exit(1)