CNTRLPLANE-3460: Refactor defrag controller to lower impact of defrag

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 \

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 {

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 {

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)