Merge pull request #4208 from ntnn/fix-etcd-parsing

Fix skipping etcd keys to delete during migration

Author: kcp-ci-bot

go.mod

@@ -33,6 +33,7 @@ require (
 	github.com/spf13/pflag v1.0.10
 	github.com/stretchr/testify v1.11.1
 	github.com/xrstf/mockoidc v0.0.0-20250721141841-711cc4e835f6
+	go.etcd.io/etcd/api/v3 v3.6.8
 	go.etcd.io/etcd/client/pkg/v3 v3.6.8
 	go.etcd.io/etcd/client/v3 v3.6.8
 	go.uber.org/goleak v1.3.1-0.20251210191316-2b7fd8a0d244
@@ -180,7 +181,6 @@ require (
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/xiang90/probing v0.0.0-20221125231312-a49e3df8f510 // indirect
 	go.etcd.io/bbolt v1.4.3 // indirect
-	go.etcd.io/etcd/api/v3 v3.6.8 // indirect
 	go.etcd.io/etcd/pkg/v3 v3.6.8 // indirect
 	go.etcd.io/etcd/server/v3 v3.6.8 // indirect
 	go.etcd.io/raft/v3 v3.6.0 // indirect

pkg/reconciler/migration/logicalclustermigration/datacleanup.go

@@ -47,8 +47,7 @@ func (c *Controller) deleteOriginData(ctx context.Context, lcName logicalcluster
 	errCh := make(chan error, 1)
 
 	go func() {
-		defer close(errCh)
-		errCh <- c.scanEtcdKeys(ctx, prefix, lcName, prefixes)
+		errCh <- scanEtcdKeys(ctx, c.etcdClient, prefix, lcName, kcpetcd.ScanPageSize, prefixes)
 	}()
 
 	var deleteErrors []error
@@ -70,24 +69,20 @@ func (c *Controller) deleteOriginData(ctx context.Context, lcName logicalcluster
 	return nil
 }
 
-// scanEtcdKeys scans etcd and emits per-cluster deletion prefixes for all keys belonging to targetLogicalCluster.
-func (c *Controller) scanEtcdKeys(ctx context.Context, prefix string, targetLogicalCluster logicalcluster.Name, out chan<- string) error {
+// scanEtcdKeys scans etcd under prefix and emits one cluster-scoped prefix
+// per (group, resource[, segment]) family that contains keys belonging to
+// targetLogicalCluster.
+func scanEtcdKeys(ctx context.Context, kv clientv3.KV, prefix string, targetLogicalCluster logicalcluster.Name, pageSize int64, out chan<- string) error {
 	defer close(out)
 
+	seen := make(map[string]struct{})
+
 	key := prefix
 	for {
-		// While this logic is pulling batches of keys it only processes
-		// a batch until it found a matching key; then the prefix for
-		// this "tree" is written to the channel for the consumer to
-		// delete the key.
-		// That means in small trees we might retrieve the same keys
-		// a few times.
-		// TODO(ntnn): Arguably it would be better to process the
-		// full batch but this is simpler for now.
-		resp, err := c.etcdClient.Get(ctx, key,
+		resp, err := kv.Get(ctx, key,
 			clientv3.WithRange(clientv3.GetPrefixRangeEnd(prefix)),
 			clientv3.WithKeysOnly(),
-			clientv3.WithLimit(kcpetcd.ScanPageSize),
+			clientv3.WithLimit(pageSize),
 		)
 		if err != nil {
 			return fmt.Errorf("failed to list etcd keys: %w", err)
@@ -98,35 +93,25 @@ func (c *Controller) scanEtcdKeys(ctx context.Context, prefix string, targetLogi
 				return err
 			}
 
-			k := string(kv.Key)
-			split, ok := kcpetcd.SplitKey(prefix, k, targetLogicalCluster)
+			split, ok := kcpetcd.SplitKey(prefix, string(kv.Key), targetLogicalCluster)
 			if !ok {
-				// not a relevant etcd key
+				continue
+			}
+			if split.Cluster != targetLogicalCluster {
 				continue
 			}
 
 			builtPrefix := split.ClusterPrefix(prefix)
-
-			if split.Cluster == targetLogicalCluster {
-				// key belongs to this cluster, pass the prefix into the channel
-				out <- builtPrefix
+			if _, dup := seen[builtPrefix]; dup {
+				continue
 			}
-
-			// skip to the next subtree
-			key = split.ClusterPrefix(builtPrefix) + "\x00"
-			break
+			seen[builtPrefix] = struct{}{}
+			out <- builtPrefix
 		}
 
 		if !resp.More {
 			return nil
 		}
-
-		if !strings.HasSuffix(key, "\x00") {
-			// Didn't instruct to skip any keys, continue from last key
-			// in the batch
-			key = string(resp.Kvs[len(resp.Kvs)-1].Key) + "\x00"
-		}
-
-		continue
+		key = string(resp.Kvs[len(resp.Kvs)-1].Key) + "\x00"
 	}
 }

pkg/reconciler/migration/logicalclustermigration/datacleanup_test.go

@@ -0,0 +1,209 @@
+/*
+Copyright 2026 The kcp Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package logicalclustermigration
+
+import (
+	"context"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/kcp-dev/logicalcluster/v3"
+)
+
+func TestScanEtcdKeys_singlePageEmitsAllPrefixFamilies(t *testing.T) {
+	t.Parallel()
+
+	prefix := "/registry/"
+	target := logicalcluster.Name("root:ws")
+
+	// All keys fit in one page. Three distinct (group, resource) families
+	// for the target, plus an unrelated cluster's data that must NOT be
+	// emitted.
+	kv := newFakeKV(map[string]string{
+		"/registry/apps/deployments/root:ws/default/foo": "v",
+		"/registry/apps/deployments/root:ws/default/bar": "v",
+		"/registry/core/configmaps/root:ws/default/cm1":  "v",
+		"/registry/core/secrets/root:ws/default/s1":      "v",
+		// other cluster, must not be emitted
+		"/registry/apps/deployments/root:other/default/x": "v",
+		"/registry/core/configmaps/root:other/default/y":  "v",
+	})
+
+	got, err := collectScan(t, kv, prefix, target, 1000)
+	require.NoError(t, err)
+
+	want := []string{
+		"/registry/apps/deployments/root:ws",
+		"/registry/core/configmaps/root:ws",
+		"/registry/core/secrets/root:ws",
+	}
+	sort.Strings(got)
+	sort.Strings(want)
+	require.Equal(t, want, got)
+}
+
+func TestScanEtcdKeys_multiPageEmitsAllPrefixFamilies(t *testing.T) {
+	t.Parallel()
+
+	prefix := "/registry/"
+	target := logicalcluster.Name("root:ws")
+
+	// 5 distinct prefix families for the target. With pageSize=2, the
+	// scanner must paginate at least 3 times and still emit every family
+	// exactly once.
+	kv := newFakeKV(map[string]string{
+		"/registry/apps/deployments/root:ws/default/a":                "v",
+		"/registry/apps/deployments/root:ws/default/b":                "v",
+		"/registry/apps/replicasets/root:ws/default/a":                "v",
+		"/registry/core/configmaps/root:ws/default/a":                 "v",
+		"/registry/core/secrets/root:ws/default/a":                    "v",
+		"/registry/rbac.authorization.k8s.io/roles/root:ws/default/a": "v",
+		// noise from another cluster
+		"/registry/apps/deployments/root:other/default/x": "v",
+		"/registry/core/secrets/root:other/default/y":     "v",
+	})
+
+	got, err := collectScan(t, kv, prefix, target, 2)
+	require.NoError(t, err)
+
+	want := []string{
+		"/registry/apps/deployments/root:ws",
+		"/registry/apps/replicasets/root:ws",
+		"/registry/core/configmaps/root:ws",
+		"/registry/core/secrets/root:ws",
+		"/registry/rbac.authorization.k8s.io/roles/root:ws",
+	}
+	sort.Strings(got)
+	sort.Strings(want)
+	require.Equal(t, want, got)
+}
+
+func TestScanEtcdKeys_noTargetKeysEmitsNothing(t *testing.T) {
+	t.Parallel()
+
+	prefix := "/registry/"
+	target := logicalcluster.Name("root:ws")
+
+	kv := newFakeKV(map[string]string{
+		"/registry/apps/deployments/root:other/default/a": "v",
+		"/registry/core/configmaps/root:other/default/b":  "v",
+	})
+
+	got, err := collectScan(t, kv, prefix, target, 1000)
+	require.NoError(t, err)
+	require.Empty(t, got)
+}
+
+func TestScanEtcdKeys_crdAndIdentityResources(t *testing.T) {
+	t.Parallel()
+
+	prefix := "/registry/"
+	target := logicalcluster.Name("root:ws")
+
+	kv := newFakeKV(map[string]string{
+		"/registry/widgets.example.io/widgets/customresources/root:ws/default/w1": "v",
+		"/registry/widgets.example.io/widgets/customresources/root:ws/default/w2": "v",
+		"/registry/things.example.io/things/abc123def/root:ws/default/t1":         "v",
+		"/registry/apps/deployments/root:ws/default/d1":                           "v",
+	})
+
+	got, err := collectScan(t, kv, prefix, target, 1000)
+	require.NoError(t, err)
+
+	want := []string{
+		"/registry/apps/deployments/root:ws",
+		"/registry/things.example.io/things/abc123def/root:ws",
+		"/registry/widgets.example.io/widgets/customresources/root:ws",
+	}
+	sort.Strings(got)
+	sort.Strings(want)
+	require.Equal(t, want, got)
+}
+
+// collectScan invokes scanEtcdKeys with the given page size and collects
+// every prefix it emits. It exists so each test reads as a single
+// require.Equal.
+func collectScan(t *testing.T, kv *fakeKV, prefix string, target logicalcluster.Name, pageSize int64) ([]string, error) {
+	t.Helper()
+
+	out := make(chan string)
+	errCh := make(chan error, 1)
+
+	go func() {
+		errCh <- scanEtcdKeys(context.Background(), kv, prefix, target, pageSize, out)
+	}()
+
+	var got []string
+	for p := range out {
+		got = append(got, p)
+	}
+	return got, <-errCh
+}
+
+// fakeKV is a minimal in-memory fake of clientv3.KV for unit-testing range
+// scans. It supports only the operations scanEtcdKeys uses: Get with a
+// range end, optional limit, and optional keys-only. Other methods are not
+// implemented and panic if called.
+type fakeKV struct {
+	kvs map[string]string
+}
+
+func newFakeKV(kvs map[string]string) *fakeKV { return &fakeKV{kvs: kvs} }
+
+func (f *fakeKV) Get(_ context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
+	op := clientv3.OpGet(key, opts...)
+	rangeEnd := string(op.RangeBytes())
+	limit := op.Limit()
+
+	keys := make([]string, 0, len(f.kvs))
+	for k := range f.kvs {
+		if k >= key && (rangeEnd == "" || k < rangeEnd) {
+			keys = append(keys, k)
+		}
+	}
+	sort.Strings(keys)
+
+	more := false
+	if limit > 0 && int64(len(keys)) > limit {
+		keys = keys[:limit]
+		more = true
+	}
+
+	resp := &clientv3.GetResponse{More: more}
+	for _, k := range keys {
+		resp.Kvs = append(resp.Kvs, &mvccpb.KeyValue{Key: []byte(k), Value: []byte(f.kvs[k])})
+	}
+	return resp, nil
+}
+
+func (f *fakeKV) Put(context.Context, string, string, ...clientv3.OpOption) (*clientv3.PutResponse, error) {
+	panic("not implemented")
+}
+func (f *fakeKV) Delete(context.Context, string, ...clientv3.OpOption) (*clientv3.DeleteResponse, error) {
+	panic("not implemented")
+}
+func (f *fakeKV) Compact(context.Context, int64, ...clientv3.CompactOption) (*clientv3.CompactResponse, error) {
+	panic("not implemented")
+}
+func (f *fakeKV) Do(context.Context, clientv3.Op) (clientv3.OpResponse, error) {
+	panic("not implemented")
+}
+func (f *fakeKV) Txn(context.Context) clientv3.Txn { panic("not implemented") }