diff --git a/bindata/etcd/pod.gotpl.yaml b/bindata/etcd/pod.gotpl.yaml index 2dbf1a7ddc..f5955660d7 100644 --- a/bindata/etcd/pod.gotpl.yaml +++ b/bindata/etcd/pod.gotpl.yaml @@ -170,7 +170,7 @@ spec: exec nice -n -19 ionice -c2 -n0 etcd \ --logger=zap \ --log-level={{.LogLevel}} \ - --feature-gates=InitialCorruptCheck=true \ + --feature-gates=InitialCorruptCheck=true,StopGRPCServiceOnDefrag=true,NonBlockingDefrag=true \ --initial-advertise-peer-urls=https://${NODE_NODE_ENVVAR_NAME_IP}:2380 \ --cert-file=/etc/kubernetes/static-pod-certs/secrets/etcd-all-certs/etcd-serving-NODE_NAME.crt \ --key-file=/etc/kubernetes/static-pod-certs/secrets/etcd-all-certs/etcd-serving-NODE_NAME.key \ diff --git a/pkg/etcdcli/etcdcli.go b/pkg/etcdcli/etcdcli.go index f067c02aae..ad752f9044 100644 --- a/pkg/etcdcli/etcdcli.go +++ b/pkg/etcdcli/etcdcli.go @@ -246,6 +246,27 @@ func (g *etcdClientGetter) MemberUpdatePeerURL(ctx context.Context, id uint64, p return err } +// MoveLeader creates a new client connected directly to the given leader member +// and issues the MoveLeader RPC. The MoveLeader API requires the request to be +// sent to the current leader; using the client pool can route to a follower and +// return "etcdserver: not leader". +func (g *etcdClientGetter) MoveLeader(ctx context.Context, leader *etcdserverpb.Member, toMember uint64) error { + cli, err := newEtcdClientWithClientOpts([]string{leader.ClientURLs[0]}, false) + if err != nil { + return fmt.Errorf("failed to create client to leader %s for MoveLeader: %w", leader.ClientURLs[0], err) + } + defer func() { + if err := cli.Close(); err != nil { + klog.Errorf("error closing leader client for MoveLeader: %v", err) + } + }() + + ctx, cancel := context.WithTimeout(ctx, DefaultClientTimeout) + defer cancel() + _, err = cli.MoveLeader(ctx, toMember) + return err +} + func (g *etcdClientGetter) MemberRemove(ctx context.Context, memberID uint64) error { cli, err := g.clientPool.Get() if err != nil { diff --git a/pkg/etcdcli/helpers.go b/pkg/etcdcli/helpers.go index e16f068779..f20888366a 100644 --- a/pkg/etcdcli/helpers.go +++ b/pkg/etcdcli/helpers.go @@ -22,6 +22,12 @@ func (f *fakeEtcdClient) Defragment(ctx context.Context, member *etcdserverpb.Me f.opts.defragErrors = f.opts.defragErrors[1:] return nil, err } + for _, status := range f.opts.status { + if status.Header.MemberId == member.ID { + status.DbSize = status.DbSizeInUse + break + } + } // dramatic simplification f.opts.dbSize = f.opts.dbSizeInUse return nil, nil @@ -82,6 +88,13 @@ func (f *fakeEtcdClient) VotingMemberList(ctx context.Context) ([]*etcdserverpb. return filterVotingMembers(members), nil } +func (f *fakeEtcdClient) MoveLeader(ctx context.Context, leader *etcdserverpb.Member, toMember uint64) error { + for _, status := range f.opts.status { + status.Leader = toMember + } + return nil +} + func (f *fakeEtcdClient) MemberRemove(ctx context.Context, memberID uint64) error { var memberExists bool for _, m := range f.members { diff --git a/pkg/etcdcli/interfaces.go b/pkg/etcdcli/interfaces.go index 6a31a9c620..3cfe7c896b 100644 --- a/pkg/etcdcli/interfaces.go +++ b/pkg/etcdcli/interfaces.go @@ -25,6 +25,7 @@ type EtcdClient interface { HealthyMemberLister UnhealthyMemberLister MemberStatusChecker + LeaderMover Status GetMember(ctx context.Context, name string) (*etcdserverpb.Member, error) @@ -64,6 +65,10 @@ type MemberRemover interface { MemberRemove(ctx context.Context, memberID uint64) error } +type LeaderMover interface { + MoveLeader(ctx context.Context, leader *etcdserverpb.Member, toMember uint64) error +} + type MemberLister interface { // MemberList lists all members in a cluster MemberList(ctx context.Context) ([]*etcdserverpb.Member, error) diff --git a/pkg/operator/defragcontroller/defragcontroller.go b/pkg/operator/defragcontroller/defragcontroller.go index 4438dd6fbc..9ff301f2cb 100644 --- a/pkg/operator/defragcontroller/defragcontroller.go +++ b/pkg/operator/defragcontroller/defragcontroller.go @@ -1,9 +1,11 @@ package defragcontroller import ( + "cmp" "context" "fmt" "math" + "slices" "time" configv1 "github.com/openshift/api/config/v1" @@ -16,7 +18,6 @@ import ( "go.etcd.io/etcd/api/v3/etcdserverpb" clientv3 "go.etcd.io/etcd/client/v3" k8serror "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/util/wait" corev1listers "k8s.io/client-go/listers/core/v1" "k8s.io/klog/v2" @@ -27,13 +28,14 @@ import ( const ( minDefragBytes int64 = 100 * 1024 * 1024 // 100MB - minDefragWaitDuration = 36 * time.Second maxFragmentedPercentage float64 = 45 - pollWaitDuration = 2 * time.Second - pollTimeoutDuration = 60 * time.Second compactionInterval = 10 * time.Minute maxDefragFailuresBeforeDegrade = 3 + // defragSettleTime is the minimum time to wait after a leader transfer + // to allow the cluster to stabilize before proceeding with defrag. + defragSettleTime = 10 * time.Second + defragDisabledCondition = "DefragControllerDisabled" defragDisableConfigmapName = "etcd-disable-defrag" @@ -47,11 +49,13 @@ type DefragController struct { memberLister etcdcli.AllMemberLister defragClient etcdcli.Defragment statusClient etcdcli.Status + leaderMover etcdcli.LeaderMover infrastructureLister configv1listers.InfrastructureLister configmapLister corev1listers.ConfigMapLister - numDefragFailures int - defragWaitDuration time.Duration + numDefragFailures int + // defragTargets tracks the ids of members that need to be defragged during the current cycle + defragTargets []uint64 } func NewDefragController( @@ -60,6 +64,7 @@ func NewDefragController( memberLister etcdcli.AllMemberLister, defragClient etcdcli.Defragment, statusClient etcdcli.Status, + leaderMover etcdcli.LeaderMover, infrastructureLister configv1listers.InfrastructureLister, eventRecorder events.Recorder, kubeInformers v1helpers.KubeInformersForNamespaces) factory.Controller { @@ -68,14 +73,14 @@ func NewDefragController( memberLister: memberLister, defragClient: defragClient, statusClient: statusClient, + leaderMover: leaderMover, infrastructureLister: infrastructureLister, configmapLister: kubeInformers.ConfigMapLister(), - defragWaitDuration: minDefragWaitDuration, } syncer := health.NewCheckingSyncWrapper(c.sync, 3*compactionInterval+1*time.Minute) livenessChecker.Add("DefragController", syncer) - return factory.New().ResyncEvery(compactionInterval+1*time.Minute).WithInformers( // attempt to sync outside of etcd compaction interval to ensure maximum gain by defragmentation. + return factory.New().ResyncEvery(compactionInterval+1*time.Minute).WithBareInformers( // attempt to sync outside of etcd compaction interval to ensure maximum gain by defragmentation. operatorClient.Informer(), ).WithSync(syncer.Sync).ToController("DefragController", eventRecorder.WithComponentSuffix("defrag-controller")) } @@ -90,7 +95,7 @@ func (c *DefragController) sync(ctx context.Context, syncCtx factory.SyncContext return nil } - return c.runDefrag(ctx, syncCtx.Recorder()) + return c.runDefrag(ctx, syncCtx) } func (c *DefragController) checkDefragEnabled(ctx context.Context, recorder events.Recorder) (bool, error) { @@ -124,7 +129,17 @@ func (c *DefragController) checkDefragEnabled(ctx context.Context, recorder even return true, nil } -func (c *DefragController) runDefrag(ctx context.Context, recorder events.Recorder) error { +type StatusMember struct { + Status *clientv3.StatusResponse + Member *etcdserverpb.Member +} + +func (sm StatusMember) IsLeader() bool { + return sm.Status.Leader == sm.Member.ID +} + +func (c *DefragController) runDefrag(ctx context.Context, syncCtx factory.SyncContext) error { + recorder := syncCtx.Recorder() // Do not defrag if any of the cluster members are unhealthy. memberHealth, err := c.memberLister.MemberHealth(ctx) if err != nil { @@ -139,120 +154,130 @@ func (c *DefragController) runDefrag(ctx context.Context, recorder events.Record return err } - // filter out learner members since they don't support the defragment API call - var etcdMembers []*etcdserverpb.Member - for _, m := range members { - if !m.IsLearner { - etcdMembers = append(etcdMembers, m) - } - } - - var endpointStatus []*clientv3.StatusResponse - var leader *clientv3.StatusResponse - for _, member := range etcdMembers { - if len(member.ClientURLs) == 0 { - // skip unstarted member + var ( + isNewCycle = len(c.defragTargets) == 0 + statusMembers = make(map[uint64]StatusMember) + ) + for _, member := range members { + // filter out learner members since they don't support the defragment API call + // and filter out unstarted members + if member.IsLearner || len(member.ClientURLs) == 0 { continue } + status, err := c.statusClient.Status(ctx, member.ClientURLs[0]) if err != nil { return err + } else if status == nil { + return fmt.Errorf("endpoint status returned nil for member %q (%s)", member.Name, member.ClientURLs[0]) } - if leader == nil && status.Leader == member.ID { - leader = status - continue + + sm := StatusMember{ + Status: status, + Member: member, } - endpointStatus = append(endpointStatus, status) - } - // Leader last if possible. - if leader != nil { - klog.V(4).Infof("Appending leader last, ID: %x", leader.Header.MemberId) - endpointStatus = append(endpointStatus, leader) + statusMembers[member.ID] = sm + + if isNewCycle && isEndpointBackendFragmented(member, status) { + c.defragTargets = append(c.defragTargets, member.ID) + } } - successfulDefrags := 0 - var errors []error - for _, status := range endpointStatus { - member, err := getMemberFromStatus(etcdMembers, status) - if err != nil { - errors = append(errors, err) - continue + c.defragTargets = slices.DeleteFunc(c.defragTargets, func(i uint64) bool { + statusMember, has := statusMembers[i] + if !has { + // Remove any defrag targets that we don't have a status for. + return true } - // Check each member's status which includes the db size on disk "DbSize" and the db size in use "DbSizeInUse" - // compare the % difference and if that difference is over the max diff threshold and also above the minimum - // db size we defrag the members state file. In the case where this command only partially completed controller - // can clean that up on the next sync. Having the db sizes slightly different is not a problem in itself. - if isEndpointBackendFragmented(member, status) { - recorder.Eventf("DefragControllerDefragmentAttempt", "Attempting defrag on member: %s, memberID: %x, dbSize: %d, dbInUse: %d, leader ID: %d", member.Name, member.ID, status.DbSize, status.DbSizeInUse, status.Leader) - if _, err := c.defragClient.Defragment(ctx, member); err != nil { - // Defrag can timeout if defragmentation takes longer than etcdcli.DefragDialTimeout. - errMsg := fmt.Sprintf("failed defrag on member: %s, memberID: %x: %v", member.Name, member.ID, err) - recorder.Eventf("DefragControllerDefragmentFailed", errMsg) - errors = append(errors, fmt.Errorf("%s", errMsg)) + // Remove any members that no longer meet the conditions for defrag. + return !isEndpointBackendFragmented(statusMember.Member, statusMember.Status) + }) + + if len(c.defragTargets) == 0 { + klog.V(4).Info("Defrag skipped: no etcd members meet the conditions for defragmentation") + return nil + } + + // Sort fragmented members so we defragment the most fragmented member first while defragging the leader last + slices.SortFunc(c.defragTargets, func(a, b uint64) int { + aIsLeader, bIsLeader := statusMembers[a].IsLeader(), statusMembers[b].IsLeader() + if aIsLeader && !bIsLeader { + return 1 + } else if !aIsLeader && bIsLeader { + return -1 + } + return sortByMostFragmented(statusMembers, a, b) + }) + + defragTarget := statusMembers[c.defragTargets[0]] + defragTargetStatus, defragTargetMember := defragTarget.Status, defragTarget.Member + + // Preemptively attempt to move the leadership away from the current defrag target to another valid follower. + // We try this to avoid multiple leader elections in the case where defragging the leader causes leadership + // to move to a member we've yet to defrag, which could in turn lose leadership, etc. causing a lot of churn. + // We record any error that occurs while attempting this, but we do not halt defrag if the move fails. + if defragTarget.IsLeader() && len(statusMembers) > 1 { + followers := make([]StatusMember, 0, len(statusMembers)) + for id, member := range statusMembers { + if defragTargetMember.ID == id { continue } + followers = append(followers, member) + } + + slices.SortFunc(followers, sortByLeastFragmented) - recorder.Eventf("DefragControllerDefragmentSuccess", "etcd member has been defragmented: %s, memberID: %d", member.Name, member.ID) - successfulDefrags++ - - // Give cluster time to recover before we move to the next member. - if err := wait.Poll( - pollWaitDuration, - pollTimeoutDuration, - func() (bool, error) { - // Ensure defragmentation attempts have clear observable signal. - klog.V(4).Infof("Sleeping to allow cluster to recover before defrag next member: %v", c.defragWaitDuration) - time.Sleep(c.defragWaitDuration) - - memberHealth, err := c.memberLister.MemberHealth(ctx) - if err != nil { - klog.Warningf("failed checking member health: %v", err) - return false, nil - } - if !etcdcli.IsClusterHealthy(memberHealth) { - klog.Warningf("cluster is unhealthy: %s", memberHealth.Status()) - return false, nil - } - return true, nil - }); err != nil { - errors = append(errors, fmt.Errorf("timeout waiting for cluster to stabilize after defrag: %w", err)) + for _, newLeader := range followers { + err := c.leaderMover.MoveLeader(ctx, defragTargetMember, newLeader.Member.ID) + if err != nil { + recorder.Warningf("DefragControllerLeaderTransferAttemptFailed", "Failed to move leader away from member %s to member %s before defrag: %v", defragTargetMember.Name, newLeader.Member.Name, err) + continue } - } else { - // no fragmentation needed is also a success - successfulDefrags++ + + recorder.Eventf("DefragControllerLeaderTransferSuccess", "Moved leadership away from member %s (memberID: %x) to member %s (memberID: %x) before defrag, requeueing to allow etcd to settle", defragTargetMember.Name, defragTargetMember.ID, newLeader.Member.Name, newLeader.Member.ID) + syncCtx.Queue().AddAfter(syncCtx.QueueKey(), defragSettleTime) + return nil } + recorder.Warningf("DefragControllerLeaderTransferFailed", "Failed to move leader away from member %s, continuing with blocking leader defrag", defragTargetMember.Name) } - if successfulDefrags != len(endpointStatus) { + recorder.Eventf("DefragControllerDefragmentAttempt", "Attempting defrag on member: %s, memberID: %x, dbSize: %d, dbInUse: %d, leader ID: %d", defragTargetMember.Name, defragTargetMember.ID, defragTargetStatus.DbSize, defragTargetStatus.DbSizeInUse, defragTargetStatus.Leader) + if _, err := c.defragClient.Defragment(ctx, defragTargetMember); err != nil { + errMsg := fmt.Sprintf("failed defrag on member: %s, memberID: %x: %v", defragTargetMember.Name, defragTargetMember.ID, err) + recorder.Warningf("DefragControllerDefragmentFailed", errMsg) + klog.Errorf("%s", errMsg) c.numDefragFailures++ - recorder.Eventf("DefragControllerDefragmentPartialFailure", - "only %d/%d members were successfully defragmented, %d tries left before controller degrades", - successfulDefrags, len(endpointStatus), maxDefragFailuresBeforeDegrade-c.numDefragFailures) - if c.numDefragFailures >= maxDefragFailuresBeforeDegrade { - _, _, updateErr := v1helpers.UpdateStatus(ctx, c.operatorClient, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{ - Type: defragDegradedCondition, - Status: operatorv1.ConditionTrue, - Reason: "Error", - Message: fmt.Sprintf("degraded after %d attempts at defragmenting all etcd members", c.numDefragFailures), - })) - if updateErr != nil { - recorder.Warning("DefragControllerUpdatingStatus", updateErr.Error()) - } + c.setDegraded(ctx, recorder) } - - // return all errors here for the sync loop to retry immediately - return v1helpers.NewMultiLineAggregate(errors) + syncCtx.Queue().AddAfter(syncCtx.QueueKey(), defragSettleTime) + return nil } - if len(errors) > 0 { - klog.Warningf("found errors even though all members have been successfully defragmented: %s", - v1helpers.NewMultiLineAggregate(errors).Error()) + recorder.Eventf("DefragControllerDefragmentSuccess", "etcd member has been defragmented: %s, memberID: %d", defragTargetMember.Name, defragTargetMember.ID) + c.numDefragFailures = 0 + c.clearDegraded(ctx, recorder) + + c.defragTargets = c.defragTargets[1:] + syncCtx.Queue().AddAfter(syncCtx.QueueKey(), defragSettleTime) + return nil +} + +func (c *DefragController) setDegraded(ctx context.Context, recorder events.Recorder) { + _, _, updateErr := v1helpers.UpdateStatus(ctx, c.operatorClient, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{ + Type: defragDegradedCondition, + Status: operatorv1.ConditionTrue, + Reason: "Error", + Message: fmt.Sprintf("degraded after %d attempts at defragmenting etcd members", c.numDefragFailures), + })) + if updateErr != nil { + recorder.Warning("DefragControllerUpdatingStatus", updateErr.Error()) } +} - c.numDefragFailures = 0 +func (c *DefragController) clearDegraded(ctx context.Context, recorder events.Recorder) { _, _, updateErr := v1helpers.UpdateStatus(ctx, c.operatorClient, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{ Type: defragDegradedCondition, @@ -262,8 +287,6 @@ func (c *DefragController) runDefrag(ctx context.Context, recorder events.Record if updateErr != nil { recorder.Warning("DefragControllerUpdatingStatus", updateErr.Error()) } - - return updateErr } func (c *DefragController) ensureControllerDisabledCondition(ctx context.Context, desiredStatus operatorv1.ConditionStatus, recorder events.Recorder) error { @@ -293,10 +316,6 @@ func (c *DefragController) ensureControllerDisabledCondition(ctx context.Context // This can happen if the operator starts defrag of the cluster but then loses leader status and is rescheduled before // the operator can defrag all members. func isEndpointBackendFragmented(member *etcdserverpb.Member, endpointStatus *clientv3.StatusResponse) bool { - if endpointStatus == nil { - klog.Errorf("endpoint status validation failed: %v", endpointStatus) - return false - } fragmentedPercentage := checkFragmentationPercentage(endpointStatus.DbSize, endpointStatus.DbSizeInUse) if fragmentedPercentage > 0.00 { klog.Infof("etcd member %q backend store fragmented: %.2f %%, dbSize: %d", member.Name, fragmentedPercentage, endpointStatus.DbSize) @@ -310,14 +329,17 @@ func checkFragmentationPercentage(ondisk, inuse int64) float64 { return math.Round(fragmentedPercentage*100) / 100 } -func getMemberFromStatus(members []*etcdserverpb.Member, endpointStatus *clientv3.StatusResponse) (*etcdserverpb.Member, error) { - if endpointStatus == nil { - return nil, fmt.Errorf("endpoint status validation failed: %v", endpointStatus) - } - for _, member := range members { - if member.ID == endpointStatus.Header.MemberId { - return member, nil - } - } - return nil, fmt.Errorf("no member found in MemberList matching ID: %v", endpointStatus.Header.MemberId) +func sortByLeastFragmented(a, b StatusMember) int { + return cmp.Compare( + checkFragmentationPercentage(a.Status.DbSize, a.Status.DbSizeInUse), + checkFragmentationPercentage(b.Status.DbSize, b.Status.DbSizeInUse), + ) +} + +func sortByMostFragmented(statusMembers map[uint64]StatusMember, a, b uint64) int { + aStatus, bStatus := statusMembers[a].Status, statusMembers[b].Status + return cmp.Compare( + checkFragmentationPercentage(bStatus.DbSize, bStatus.DbSizeInUse), + checkFragmentationPercentage(aStatus.DbSize, aStatus.DbSizeInUse), + ) } diff --git a/pkg/operator/defragcontroller/defragcontroller_test.go b/pkg/operator/defragcontroller/defragcontroller_test.go index c4b00314d1..d94b20d97f 100644 --- a/pkg/operator/defragcontroller/defragcontroller_test.go +++ b/pkg/operator/defragcontroller/defragcontroller_test.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "regexp" "strings" "testing" "time" @@ -69,6 +68,7 @@ func TestNewDefragController(t *testing.T) { staticPodStatus *operatorv1.StaticPodOperatorStatus objects []runtime.Object clusterSize int + syncLoops int memberHealth *etcdcli.FakeMemberHealth dbInUse int64 dbSize int64 @@ -84,6 +84,7 @@ func TestNewDefragController(t *testing.T) { dbInUse: minDefragBytes / 2, // 500MB defragSuccessEvents: 3, clusterSize: 3, + syncLoops: 4, // 2 non-leader defrags + 1 leader transfer + 1 former-leader defrag memberHealth: &etcdcli.FakeMemberHealth{Healthy: 3}, objects: []runtime.Object{ u.FakeInfrastructureTopology(configv1.HighlyAvailableTopologyMode), @@ -97,6 +98,7 @@ func TestNewDefragController(t *testing.T) { dbInUse: minDefragBytes / 2, // 500MB defragSuccessEvents: 2, clusterSize: 2, + syncLoops: 3, // 1 non-leader defrag + 1 leader transfer + 1 former-leader defrag memberHealth: &etcdcli.FakeMemberHealth{Healthy: 2}, objects: []runtime.Object{ u.FakeInfrastructureTopology(configv1.DualReplicaTopologyMode), @@ -110,6 +112,7 @@ func TestNewDefragController(t *testing.T) { dbInUse: minDefragBytes / 2, // 500MB defragSuccessEvents: 3, clusterSize: 3, + syncLoops: 4, // 2 non-leader defrags + 1 leader transfer + 1 former-leader defrag memberHealth: &etcdcli.FakeMemberHealth{Healthy: 3}, objects: []runtime.Object{ u.FakeInfrastructureTopology(configv1.HighlyAvailableArbiterMode), @@ -222,38 +225,35 @@ func TestNewDefragController(t *testing.T) { memberLister: fakeEtcdClient, defragClient: fakeEtcdClient, statusClient: fakeEtcdClient, + leaderMover: fakeEtcdClient, infrastructureLister: configv1listers.NewInfrastructureLister(indexer), configmapLister: corev1listers.NewConfigMapLister(indexer), - // to speed the tests up, in real life we use minDefragWaitDuration - defragWaitDuration: 1 * time.Second, } - err := controller.sync(context.TODO(), factory.NewSyncContext("defrag-controller", eventRecorder)) - if err != nil && !scenario.wantErr { - t.Fatalf("unexepected error %v", err) + syncLoops := scenario.syncLoops + if syncLoops == 0 { + syncLoops = 1 + } + var err error + for i := 0; i < syncLoops; i++ { + err = controller.sync(context.TODO(), factory.NewSyncContext("defrag-controller", eventRecorder)) + if err != nil && !scenario.wantErr { + t.Fatalf("unexpected error on sync %d: %v", i, err) + } } if err == nil && scenario.wantErr { t.Fatal("expected error got nil") } if err != nil && scenario.wantErr { if !strings.HasPrefix(err.Error(), scenario.wantErrMsg) { - t.Fatalf("unexepected error prefix want: %q got: %q", scenario.wantErrMsg, err.Error()) + t.Fatalf("unexpected error prefix want: %q got: %q", scenario.wantErrMsg, err.Error()) } } var defragSuccessEvents int - lastEvent := len(eventRecorder.Events()) - 1 - for i, event := range eventRecorder.Events() { + for _, event := range eventRecorder.Events() { if strings.HasPrefix(event.Message, "etcd member has been defragmented") { defragSuccessEvents++ } - // ensure the leader was defragged last - if i == lastEvent { - // last event will print leader ID - regex := regexp.MustCompile(fmt.Sprint(status[0].Leader)) - if len(regex.FindAll([]byte(event.Message), -1)) != 1 { - t.Fatalf("expected leader defrag event to be last got %q", event.Message) - } - } } if defragSuccessEvents != scenario.defragSuccessEvents { t.Fatalf("defragSuccessEvents invalid want %d got %d", scenario.defragSuccessEvents, defragSuccessEvents) @@ -304,13 +304,13 @@ func TestNewDefragControllerMultiSyncs(t *testing.T) { defragSuccessEvents: 0, clusterSize: 3, syncLoops: maxDefragFailuresBeforeDegrade, - errSyncLoops: maxDefragFailuresBeforeDegrade, + errSyncLoops: 0, memberHealth: &etcdcli.FakeMemberHealth{Healthy: 3}, objects: []runtime.Object{ u.FakeInfrastructureTopology(configv1.HighlyAvailableTopologyMode), }, fakeClientOpts: []etcdcli.FakeClientOption{ - etcdcli.WithFakeDefragErrors(generateErrors(maxDefragFailuresBeforeDegrade * 3)), + etcdcli.WithFakeDefragErrors(generateErrors(maxDefragFailuresBeforeDegrade)), }, wantDisabledCondition: operatorv1.ConditionFalse, wantDegradedCondition: operatorv1.ConditionTrue, @@ -322,16 +322,15 @@ func TestNewDefragControllerMultiSyncs(t *testing.T) { dbInUse: minDefragBytes / 2, defragSuccessEvents: 3, clusterSize: 3, - syncLoops: maxDefragFailuresBeforeDegrade + 1, - errSyncLoops: maxDefragFailuresBeforeDegrade, + syncLoops: maxDefragFailuresBeforeDegrade + 4, // +1 for leader transfer requeue + errSyncLoops: 0, memberHealth: &etcdcli.FakeMemberHealth{Healthy: 3}, objects: []runtime.Object{ u.FakeInfrastructureTopology(configv1.HighlyAvailableTopologyMode), }, fakeClientOpts: []etcdcli.FakeClientOption{ - etcdcli.WithFakeDefragErrors(generateErrors(maxDefragFailuresBeforeDegrade * 3)), + etcdcli.WithFakeDefragErrors(generateErrors(maxDefragFailuresBeforeDegrade)), }, - // ignoring errors here since the first maxDefragFailuresBeforeDegrade invocations will return an error, the ones after won't wantDisabledCondition: operatorv1.ConditionFalse, wantDegradedCondition: operatorv1.ConditionFalse, }, @@ -378,10 +377,9 @@ func TestNewDefragControllerMultiSyncs(t *testing.T) { memberLister: fakeEtcdClient, statusClient: fakeEtcdClient, defragClient: fakeEtcdClient, + leaderMover: fakeEtcdClient, infrastructureLister: configv1listers.NewInfrastructureLister(indexer), configmapLister: corev1listers.NewConfigMapLister(indexer), - // to speed the tests up, in real life we use minDefragWaitDuration - defragWaitDuration: 1 * time.Second, } numSyncErr := 0 @@ -413,6 +411,181 @@ func TestNewDefragControllerMultiSyncs(t *testing.T) { } } +func TestDefragMovesLeadershipBeforeDefrag(t *testing.T) { + fakeOperatorClient := v1helpers.NewFakeStaticPodOperatorClient( + &operatorv1.StaticPodOperatorSpec{ + OperatorSpec: operatorv1.OperatorSpec{ + ManagementState: operatorv1.Managed, + }, + }, + u.StaticPodOperatorStatus(), + nil, + nil, + ) + + integration.BeforeTestExternal(t) + testServer := integration.NewCluster(t, &integration.ClusterConfig{Size: 3}) + defer testServer.Terminate(t) + + etcdMembers := waitForMembersWithClientURLs(t, testServer) + + var status []*clientv3.StatusResponse + var leaderID uint64 + for _, member := range testServer.Members { + statusResp, err := testServer.Client(0).Status(context.TODO(), member.GRPCURL) + require.NoError(t, err) + if leaderID == 0 { + leaderID = statusResp.Leader + } + // Only the leader is fragmented, so it's the sole defrag target + // and must trigger a leader transfer. + statusResp.DbSize = minDefragBytes / 2 + statusResp.DbSizeInUse = minDefragBytes / 2 + for _, m := range etcdMembers { + if m.ID == leaderID && statusResp.Header.MemberId == leaderID { + statusResp.DbSize = minDefragBytes + statusResp.DbSizeInUse = minDefragBytes / 2 + } + } + status = append(status, statusResp) + } + + fakeEtcdClient, _ := etcdcli.NewFakeEtcdClient( + etcdMembers, + etcdcli.WithFakeClusterHealth(&etcdcli.FakeMemberHealth{Healthy: 3}), + etcdcli.WithFakeStatus(status), + ) + eventRecorder := events.NewInMemoryRecorder(t.Name(), clock.RealClock{}) + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + require.NoError(t, indexer.Add(u.FakeInfrastructureTopology(configv1.HighlyAvailableTopologyMode))) + + controller := &DefragController{ + operatorClient: fakeOperatorClient, + memberLister: fakeEtcdClient, + defragClient: fakeEtcdClient, + statusClient: fakeEtcdClient, + leaderMover: fakeEtcdClient, + infrastructureLister: configv1listers.NewInfrastructureLister(indexer), + configmapLister: corev1listers.NewConfigMapLister(indexer), + } + + syncCtx := factory.NewSyncContext("defrag-controller", eventRecorder) + + // First sync: leader is the most fragmented, so leadership is transferred and sync requeues. + err := controller.sync(context.TODO(), syncCtx) + require.NoError(t, err) + + var leaderTransferEvents int + for _, event := range eventRecorder.Events() { + if strings.HasPrefix(event.Message, "Moved leadership away from member") { + leaderTransferEvents++ + } + } + assert.Equal(t, 1, leaderTransferEvents, "expected exactly one leader transfer event") + + // Second sync: the former leader is now a follower and gets defragged. + err = controller.sync(context.TODO(), syncCtx) + require.NoError(t, err) + + var defragSuccessEvents int + for _, event := range eventRecorder.Events() { + if strings.HasPrefix(event.Message, "etcd member has been defragmented") { + defragSuccessEvents++ + } + } + assert.Equal(t, 1, defragSuccessEvents, "expected exactly one defrag success event") +} + +// TestDefragLeaderNotStarvedInHighChurn verifies that in a high-churn environment +// where non-leader members re-fragment between sync cycles, the leader is eventually +// defragged rather than being perpetually skipped. +func TestDefragLeaderNotStarvedInHighChurn(t *testing.T) { + fakeOperatorClient := v1helpers.NewFakeStaticPodOperatorClient( + &operatorv1.StaticPodOperatorSpec{ + OperatorSpec: operatorv1.OperatorSpec{ + ManagementState: operatorv1.Managed, + }, + }, + u.StaticPodOperatorStatus(), + nil, + nil, + ) + + integration.BeforeTestExternal(t) + testServer := integration.NewCluster(t, &integration.ClusterConfig{Size: 3}) + defer testServer.Terminate(t) + + etcdMembers := waitForMembersWithClientURLs(t, testServer) + + var status []*clientv3.StatusResponse + for _, member := range testServer.Members { + statusResp, err := testServer.Client(0).Status(context.TODO(), member.GRPCURL) + require.NoError(t, err) + // All members are fragmented. + statusResp.DbSize = minDefragBytes + statusResp.DbSizeInUse = minDefragBytes / 2 + status = append(status, statusResp) + } + + fakeEtcdClient, _ := etcdcli.NewFakeEtcdClient( + etcdMembers, + etcdcli.WithFakeClusterHealth(&etcdcli.FakeMemberHealth{Healthy: 3}), + etcdcli.WithFakeStatus(status), + ) + + eventRecorder := events.NewInMemoryRecorder(t.Name(), clock.RealClock{}) + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + require.NoError(t, indexer.Add(u.FakeInfrastructureTopology(configv1.HighlyAvailableTopologyMode))) + + controller := &DefragController{ + operatorClient: fakeOperatorClient, + memberLister: fakeEtcdClient, + defragClient: fakeEtcdClient, + statusClient: fakeEtcdClient, + leaderMover: fakeEtcdClient, + infrastructureLister: configv1listers.NewInfrastructureLister(indexer), + configmapLister: corev1listers.NewConfigMapLister(indexer), + } + + syncCtx := factory.NewSyncContext("defrag-controller", eventRecorder) + + // Simulate high churn: after each sync, re-fragment all members. + // Without the defraggedMembers tracking, the leader would be starved + // because len(defragTargets) > 1 is always true. + // + // Expected sequence for a 3-node cluster: + // sync 1: defrag non-leader A + // sync 2: defrag non-leader B (A re-fragmented but already tracked) + // sync 3: leader transfer (all non-leaders defragged this cycle) + // sync 4: defrag former leader + maxSyncs := 10 + var leaderDefragged bool + for i := 0; i < maxSyncs; i++ { + // Re-fragment all members to simulate high churn. + for _, s := range status { + s.DbSize = minDefragBytes + s.DbSizeInUse = minDefragBytes / 2 + } + + err := controller.sync(context.TODO(), syncCtx) + require.NoError(t, err) + + // Check if the leader was defragged by looking for a defrag attempt + // on the leader member. + for _, event := range eventRecorder.Events() { + if strings.HasPrefix(event.Message, "Moved leadership away from member") { + leaderDefragged = true + break + } + } + if leaderDefragged { + break + } + } + + assert.True(t, leaderDefragged, "leader should have been selected for defrag (via leader transfer) within %d syncs", maxSyncs) +} + func Test_isEndpointBackendFragmented(t *testing.T) { scenarios := []struct { name string diff --git a/pkg/operator/starter.go b/pkg/operator/starter.go index 026ec2c2ee..7ab623affc 100644 --- a/pkg/operator/starter.go +++ b/pkg/operator/starter.go @@ -495,9 +495,10 @@ func RunOperator(ctx context.Context, controllerContext *controllercmd.Controlle defragController := defragcontroller.NewDefragController( AlivenessChecker, operatorClient, - cachedMemberClient, // for cached List/Health calls - etcdClient, // for status calls - etcdClient, // for defrag calls + etcdClient, // for member list/health calls + etcdClient, // for defrag calls + etcdClient, // for status calls + etcdClient, // for leader transfer before defrag configInformers.Config().V1().Infrastructures().Lister(), controllerContext.EventRecorder, kubeInformersForNamespaces,