Add no-space issue handler

internal/controller/etcdcluster_controller.go

@@ -39,6 +39,7 @@ import (
 
 	ecv1alpha1 "go.etcd.io/etcd-operator/api/v1alpha1"
 	"go.etcd.io/etcd-operator/internal/etcdutils"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 	etcdversions "go.etcd.io/etcd/api/v3/version"
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
@@ -418,6 +419,19 @@ func (r *EtcdClusterReconciler) refreshClusterState(ctx context.Context, s *reco
 	}
 }
 
+func getCurrentRevision(members map[string]etcdutils.EpHealth) int64 {
+	var revision int64
+	for _, m := range members {
+		if m.Status == nil {
+			continue
+		}
+		if rev := m.Status.Header.Revision; rev > revision {
+			revision = rev
+		}
+	}
+	return revision
+}
+
 // dispatch implements design doc §4.9's priority order, items 2-9:
 // at most one mutating action per reconcile. Each numbered step either
 // claims the loop (returns a non-zero Result or an error) or falls through
@@ -470,8 +484,45 @@ func (r *EtcdClusterReconciler) dispatch(ctx context.Context, s *reconcileState)
 	// this step just doesn't act on them yet.
 
 	// 5. NOSPACE alarm remediation (§4.7).
-	// TODO: §4.9 item 5 — compact/defragment/disarm cycle (M4). Same
-	// as above, s.health.Alarms already has the data.
+	if s.health != nil && s.health.Healthy && slices.ContainsFunc(s.health.Alarms,
+		func(a *etcdserverpb.AlarmMember) bool { return a.Alarm == etcdserverpb.AlarmType_NOSPACE }) {
+		currentRev := getCurrentRevision(s.health.Members)
+		if currentRev == 0 {
+			return ctrl.Result{}, fmt.Errorf("cannot determine the revision to compact at: no member reported a status")
+		}
+
+		cfg := etcdutils.ClientConfig{
+			Endpoints: clientEndpointsFromPods(s.cluster.Name, s.cluster.Namespace, s.pods, clusterTLSEnabled(s.cluster)),
+			TLS:       s.tlsConfig,
+		}
+		if err := etcdutils.Compact(cfg, currentRev); err != nil {
+			return ctrl.Result{}, err
+		}
+
+		endpoints := make([]string, 0, len(s.pods))
+		var leaderEp string
+		for _, m := range s.health.Members {
+			if m.Status == nil {
+				continue
+			}
+			if m.Status.Leader == m.Status.Header.MemberId {
+				leaderEp = m.Ep
+				continue
+			}
+			endpoints = append(endpoints, m.Ep)
+		}
+		endpoints = append(endpoints, leaderEp)
+
+		for _, ep := range endpoints {
+			if err := etcdutils.Defrag(cfg, ep); err != nil {
+				return ctrl.Result{}, err
+			}
+		}
+
+		if err := etcdutils.AlarmDisarm(cfg, s.health.Alarms); err != nil {
+			return ctrl.Result{}, err
+		}
+	}
 
 	// 6. Per-member repair: continue a member already Recreating, or start
 	// fixing exactly one newly-unhealthy Ready member (requirement 6).

internal/etcdutils/etcdutils.go

@@ -228,7 +228,7 @@ func MemberHealth(cfg ClientConfig) ([]EpHealth, error) {
 		name string
 	}
 
-	var cfgs = make([]epConfig, 0, len(cfg.Endpoints))
+	cfgs := make([]epConfig, 0, len(cfg.Endpoints))
 	for i, ep := range cfg.Endpoints {
 		epCfg := ClientConfig{Endpoints: []string{ep}, TLS: cfg.TLS}
 		built := epCfg.buildConfig()
@@ -285,7 +285,7 @@ func MemberHealth(cfg ClientConfig) ([]EpHealth, error) {
 	wg.Wait()
 	close(healthCh)
 
-	var healthList = make([]EpHealth, 0, len(healthCh))
+	healthList := make([]EpHealth, 0, len(healthCh))
 	for h := range healthCh {
 		healthList = append(healthList, h)
 	}
@@ -348,3 +348,33 @@ func MoveLeader(cfg ClientConfig, memberId uint64) error {
 	_, err = c.MoveLeader(ctx, memberId)
 	return err
 }
+
+func Compact(cfg ClientConfig, targetRev int64) error {
+	c, err := clientv3.New(cfg.buildConfig())
+	if err != nil {
+		return nil
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer func() { closeAndCancel(c, cancel) }()
+	if _, err := c.Compact(ctx, targetRev, clientv3.WithCompactPhysical()); err != nil {
+		if errors.Is(err, rpctypes.ErrCompacted) {
+			return nil
+		}
+		return err
+	}
+	return nil
+}
+
+func Defrag(cfg ClientConfig, endpoint string) error {
+	c, err := clientv3.New(cfg.buildConfig())
+	if err != nil {
+		return nil
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer func() { closeAndCancel(c, cancel) }()
+
+	_, err = c.Defragment(ctx, endpoint)
+	return err
+}

test/e2e/helpers_test.go

@@ -35,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/e2e-framework/klient"
 	"sigs.k8s.io/e2e-framework/klient/k8s/resources"
 	"sigs.k8s.io/e2e-framework/klient/wait"
@@ -290,6 +291,28 @@ func execInPod(
 	return stdout.String(), stderr.String(), err
 }
 
+func updateEtcdCluster(
+	ctx context.Context,
+	t *testing.T,
+	c *envconf.Config,
+	name string,
+	mutate func(*ecv1alpha1.EtcdCluster),
+) {
+	t.Helper()
+
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		var etcdCluster ecv1alpha1.EtcdCluster
+		if err := c.Client().Resources().Get(ctx, name, namespace, &etcdCluster); err != nil {
+			return err
+		}
+		mutate(&etcdCluster)
+		return c.Client().Resources().Update(ctx, &etcdCluster)
+	})
+	if err != nil {
+		t.Fatalf("Failed to update EtcdCluster %s: %v", name, err)
+	}
+}
+
 func scaleEtcdCluster(ctx context.Context, t *testing.T, c *envconf.Config, name string, size int) {
 	t.Helper()
 	var etcdCluster ecv1alpha1.EtcdCluster
@@ -544,7 +567,8 @@ func etcdctlCmd(podName, etcdClusterName, namespace string, tlsEnabled bool) []s
 	if tlsEnabled {
 		endpoint := fmt.Sprintf("https://%s.%s.%s.svc.cluster.local:2379",
 			podName, etcdClusterName, namespace)
-		args = append(args,
+		args = append(
+			args,
 			"--cacert="+etcdServerCAFile,
 			"--cert="+etcdServerCertFile,
 			"--key="+etcdServerKeyFile,