ETCD-328: automatic replacement of an unhealthy member machine

pkg/etcdcli/health.go

@@ -240,7 +240,11 @@ func HasStarted(member *etcdserverpb.Member) bool {
 // loss of a single etcd member. Such loss is common during new static pod revision.
 func IsQuorumFaultTolerant(memberHealth []healthCheck) bool {
 	totalMembers := len(memberHealth)
-	quorum := totalMembers/2 + 1
+	quorum, err := MinimumTolerableQuorum(totalMembers)
+	if err != nil {
+		klog.Errorf("etcd cluster could not determine minimum quorum required. total number of members is %v. minimum quorum required is %v: %w", totalMembers, quorum, err)
+		return false
+	}
 	healthyMembers := len(GetHealthyMemberNames(memberHealth))
 	switch {
 	case totalMembers-quorum < 1:
@@ -256,7 +260,10 @@ func IsQuorumFaultTolerant(memberHealth []healthCheck) bool {
 // IsQuorumFaultTolerantErr is the same as IsQuorumFaultTolerant but with an error return instead of the log
 func IsQuorumFaultTolerantErr(memberHealth []healthCheck) error {
 	totalMembers := len(memberHealth)
-	quorum := totalMembers/2 + 1
+	quorum, err := MinimumTolerableQuorum(totalMembers)
+	if err != nil {
+		return fmt.Errorf("etcd cluster could not determine minimum quorum required. total number of members is %v. minimum quorum required is %v: %w", totalMembers, quorum, err)
+	}
 	healthyMembers := len(GetHealthyMemberNames(memberHealth))
 	switch {
 	case totalMembers-quorum < 1:
@@ -320,3 +327,10 @@ func (c *raftTermsCollector) Collect(ch chan<- prometheus.Metric) {
 		)
 	}
 }
+
+func MinimumTolerableQuorum(members int) (int, error) {
+	if members <= 0 {
+		return 0, fmt.Errorf("invalid etcd member length: %v", members)
+	}
+	return (members / 2) + 1, nil
+}

pkg/etcdcli/health_test.go

@@ -2,6 +2,7 @@ package etcdcli
 
 import (
 	"fmt"
+	"github.com/stretchr/testify/require"
 	"reflect"
 	"testing"
 
@@ -260,6 +261,51 @@ func healthyMember(member int) healthCheck {
 	}
 }
 
+func TestMinimumTolerableQuorum(t *testing.T) {
+
+	scenarios := []struct {
+		name   string
+		input  int
+		expErr error
+		exp    int
+	}{
+		{
+			name:   "valid input `3`",
+			input:  3,
+			expErr: nil,
+			exp:    2,
+		},
+		{
+			name:   "valid input `5`",
+			input:  5,
+			expErr: nil,
+			exp:    3,
+		},
+		{
+			name:   "invalid input `0`",
+			input:  0,
+			expErr: fmt.Errorf("invalid etcd member length: %v", 0),
+			exp:    0,
+		},
+		{
+			name:   "invalid input `-10`",
+			input:  -10,
+			expErr: fmt.Errorf("invalid etcd member length: %v", -10),
+			exp:    0,
+		},
+	}
+
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			// act
+			actual, err := MinimumTolerableQuorum(scenario.input)
+			// assert
+			require.Equal(t, scenario.expErr, err)
+			require.Equal(t, scenario.exp, actual)
+		})
+	}
+}
+
 func unHealthyMember(member int) healthCheck {
 	return healthCheck{
 		Member: &etcdserverpb.Member{

pkg/operator/ceohelpers/common.go

@@ -1,23 +1,24 @@
 package ceohelpers
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/openshift/cluster-etcd-operator/pkg/dnshelpers"
+	"github.com/openshift/cluster-etcd-operator/pkg/etcdcli"
 	"net"
 	"net/url"
 
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 
 	machinev1beta1 "github.com/openshift/api/machine/v1beta1"
+	machinelistersv1beta1 "github.com/openshift/client-go/machine/listers/machine/v1beta1"
+	"github.com/openshift/library-go/pkg/operator/resource/resourcemerge"
+	"github.com/openshift/library-go/pkg/operator/v1helpers"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
-	corev1listers "k8s.io/client-go/listers/core/v1"
-
-	machinelistersv1beta1 "github.com/openshift/client-go/machine/listers/machine/v1beta1"
-	"github.com/openshift/library-go/pkg/operator/resource/resourcemerge"
-	"github.com/openshift/library-go/pkg/operator/v1helpers"
 
 	"github.com/openshift/cluster-etcd-operator/pkg/operator/configobservation/controlplanereplicascount"
 )
@@ -178,14 +179,20 @@ func memberToURL(member *etcdserverpb.Member) (string, error) {
 	return member.PeerURLs[0], nil
 }
 
-func VotingMemberIPListSet(configMapLister corev1listers.ConfigMapNamespaceLister) (sets.String, error) {
-	etcdEndpointsConfigMap, err := configMapLister.Get("etcd-endpoints")
+func VotingMemberIPListSet(ctx context.Context, cli etcdcli.EtcdClient) (sets.String, error) {
+	members, err := cli.VotingMemberList(ctx)
 	if err != nil {
 		return sets.NewString(), err // should not happen
 	}
 	currentVotingMemberIPListSet := sets.NewString()
-	for _, votingMemberIP := range etcdEndpointsConfigMap.Data {
-		currentVotingMemberIPListSet.Insert(votingMemberIP)
+
+	for _, member := range members {
+		// Use of PeerURL is expected here because it is a mandatory field, and it will mirror ClientURL.
+		ip, err := dnshelpers.GetIPFromAddress(member.PeerURLs[0])
+		if err != nil {
+			return sets.NewString(), err
+		}
+		currentVotingMemberIPListSet.Insert(ip)
 	}
 
 	return currentVotingMemberIPListSet, nil

pkg/operator/clustermembercontroller/clustermembercontroller.go

@@ -372,7 +372,7 @@ func (c *ClusterMemberController) isEtcdContainerRunningNotReady(node *corev1.No
 // The voting members are read from the etcd-endpoints configmap
 func (c *ClusterMemberController) allNodesMapToVotingMembers(nodes []*corev1.Node) ([]*corev1.Node, error) {
 	var nonVotingMemberNodes []*corev1.Node
-	currentVotingMemberIPListSet, err := ceohelpers.VotingMemberIPListSet(c.configMapLister)
+	currentVotingMemberIPListSet, err := ceohelpers.VotingMemberIPListSet(context.Background(), c.etcdClient)
 	if err != nil {
 		return nonVotingMemberNodes, fmt.Errorf("failed to get the set of voting members: %v", err)
 	}

pkg/operator/clustermemberremovalcontroller/clustermemberremovalcontroller.go

@@ -113,7 +113,7 @@ func (c *clusterMemberRemovalController) sync(ctx context.Context, syncCtx facto
 // attemptToScaleDown attempts to remove a voting member only once we have identified that
 // a Machine resource is being deleted and a replacement member has been created
 func (c *clusterMemberRemovalController) attemptToScaleDown(ctx context.Context, recorder events.Recorder) error {
-	currentVotingMemberIPListSet, err := ceohelpers.VotingMemberIPListSet(c.configMapListerForTargetNamespace)
+	currentVotingMemberIPListSet, err := ceohelpers.VotingMemberIPListSet(ctx, c.etcdClient)
 	if err != nil {
 		return err
 	}
@@ -125,15 +125,21 @@ func (c *clusterMemberRemovalController) attemptToScaleDown(ctx context.Context,
 	if desiredControlPlaneReplicasCount == 0 {
 		return fmt.Errorf("desired control plane replicas count cannot be empty")
 	}
-	if currentVotingMemberIPListSet.Len() <= desiredControlPlaneReplicasCount {
-		klog.V(4).Infof("Ignoring scale-down since the number of etcd voting members (%d) < desired number of control-plane replicas (%d) ", currentVotingMemberIPListSet.Len(), desiredControlPlaneReplicasCount)
+
+	liveVotingMembers, err := c.getAllVotingMembers(ctx)
+	if err != nil {
+		return fmt.Errorf("could not list etcd members: %w", err)
+	}
+
+	if len(liveVotingMembers) < desiredControlPlaneReplicasCount {
+		klog.V(2).Infof("Ignoring scale-down since the number of etcd members (%d) < desired number of control-plane replicas (%d) ", len(liveVotingMembers), desiredControlPlaneReplicasCount)
 		return nil
 	}
 
 	// machines with master role and deletion hook
 	memberMachines, err := ceohelpers.CurrentMemberMachinesWithDeletionHooks(c.masterMachineSelector, c.masterMachineLister)
 	if err != nil {
-		return err
+		return fmt.Errorf("could not find master machines with deletion hook: %w", err)
 	}
 
 	var votingMembersMachines []*machinev1beta1.Machine
@@ -151,14 +157,19 @@ func (c *clusterMemberRemovalController) attemptToScaleDown(ctx context.Context,
 	// do not trust data in the cache, compare with the current state
 	healthyLiveVotingMembers, err := c.getHealthyVotingMembers(ctx)
 	if err != nil {
-		return err
+		return fmt.Errorf("could not list etcd healthy members: %w", err)
 	}
 
 	// scaling down invariant
-	if len(healthyLiveVotingMembers) < desiredControlPlaneReplicasCount {
-		klog.V(2).Infof("Ignoring scale down since the number of healthy live etcd voting members (%d) < desired number of control-plane replicas (%d) ", len(healthyLiveVotingMembers), desiredControlPlaneReplicasCount)
+	minTolerableQuorum, err := etcdcli.MinimumTolerableQuorum(desiredControlPlaneReplicasCount)
+	if err != nil {
+		klog.V(2).Infof("etcd cluster could not determine minimum quorum required. desiredControlPlaneReplicasCount is %v. minimum quorum required is %v: %w", desiredControlPlaneReplicasCount, minTolerableQuorum, err)
+	}
+
+	if len(healthyLiveVotingMembers) < minTolerableQuorum {
+		klog.V(2).Infof("ignoring scale down since the number of healthy live etcd members (%d) < minimum required to maintain quorum (%d) ", len(healthyLiveVotingMembers), minTolerableQuorum)
 		if time.Now().After(c.lastTimeScaleDownEventWasSent.Add(5 * time.Minute)) {
-			recorder.Eventf("ScaleDown", "Ignoring scale down since the number of healthy live etcd voting members (%d) < desired number of control-plane replicas (%d) ", len(healthyLiveVotingMembers), desiredControlPlaneReplicasCount)
+			recorder.Eventf("ScaleDown", "Ignoring scale down since the number of healthy live etcd members (%d) < minimum required to maintain quorum (%d) ", len(healthyLiveVotingMembers), minTolerableQuorum)
 			c.lastTimeScaleDownEventWasSent = time.Now()
 		}
 		return nil
@@ -192,25 +203,19 @@ func (c *clusterMemberRemovalController) attemptToScaleDown(ctx context.Context,
 				unhealthyMembersURLs = append(unhealthyMembersURLs, unhealthyMember.Name)
 			}
 		}
-		if len(unhealthyVotingMemberMachinesPendingDeletion) > 0 {
-			klog.V(4).Infof("found unhealthy voting members with machine pending deletion: %v", unhealthyVotingMemberMachinesPendingDeletion)
-			klog.V(4).Infof("unhealthy members found: %v", unhealthyMembersURLs)
-		} else {
-			return fmt.Errorf("cannot proceed with scaling down, unhealthy voting members found: %v, none are pending deletion", unhealthyMembersURLs)
+		if len(unhealthyVotingMemberMachinesPendingDeletion) == 0 {
+			klog.V(2).Infof("cannot proceed with scaling down, unhealthy voting etcd members found: %v but none are pending deletion", unhealthyMembersURLs)
+			return fmt.Errorf("cannot proceed with scaling down, unhealthy voting etcd members found: %v but none are pending deletion", unhealthyMembersURLs)
 		}
 	}
 
 	// remove the unhealthy machine pending deletion first
 	// if no unhealthy machine pending deletion found, then attempt to scale down the healthy machines pending deletion
 	if len(unhealthyVotingMemberMachinesPendingDeletion) > 0 {
+		klog.V(2).Infof("found unhealthy voting etcd members with machine pending deletion: %v", unhealthyVotingMemberMachinesPendingDeletion)
 		votingMembersMachinesPendingDeletion = append(unhealthyVotingMemberMachinesPendingDeletion, votingMembersMachinesPendingDeletion...)
 	}
 
-	liveVotingMembers, err := c.getAllVotingMembers(ctx)
-	if err != nil {
-		return err
-	}
-
 	var allErrs []error
 	for _, votingMemberMachinePendingDeletion := range votingMembersMachinesPendingDeletion {
 		removed, errs := c.attemptToRemoveMemberFor(ctx, liveVotingMembers, votingMemberMachinePendingDeletion, recorder)
@@ -290,13 +295,13 @@ func (c *clusterMemberRemovalController) removeMemberWithoutMachine(ctx context.
 
 // attemptToRemoveLearningMember attempts to remove a learning member pending deletion regardless of whether a replacement member has been found
 func (c *clusterMemberRemovalController) attemptToRemoveLearningMember(ctx context.Context, recorder events.Recorder) error {
-	currentVotingMemberIPListSet, err := ceohelpers.VotingMemberIPListSet(c.configMapListerForTargetNamespace)
+	currentVotingMemberIPListSet, err := ceohelpers.VotingMemberIPListSet(ctx, c.etcdClient)
 	if err != nil {
 		return err
 	}
 	memberMachines, err := ceohelpers.CurrentMemberMachinesWithDeletionHooks(c.masterMachineSelector, c.masterMachineLister)
 	if err != nil {
-		return err
+		return fmt.Errorf("could not find master machines with deletion hook: %w", err)
 	}
 	var learningMachines []*machinev1beta1.Machine
 	for memberMachineIP, memberMachine := range ceohelpers.IndexMachinesByNodeInternalIP(memberMachines) {