From a6957347ebc1023bb6332e8a634ca7dfad523d24 Mon Sep 17 00:00:00 2001 From: Said Boudjelda Date: Sat, 17 Oct 2026 19:23:18 +0200 Subject: [PATCH] fix(migrate): up and run refuse a lab whose mirrored topics already exist on the target, and a run refused at its prerequisites removes nothing Nothing checked that a lab's mirrored topic names were new on the target. Under the default identity policy, --topics orders mirrors into the primary's own orders. When krafter already had orders, made by a Kates test through the AdminClient (so with no KafkaTopic) or by an app, `kates migrate run --from 2.8.2 --topics orders --yes` had MirrorMaker 2 append the 2.8.2 corpus to it. Then run's teardown adopted orders as a lab topic and deleted it from krafter, with the data it held before the lab. The prerequisites of up and run now list the target's topics as kates-mm2, from the kates-migrate-target client pod that `kates migrate target topics` uses, and delete the pod afterwards. A lab is refused when any of its replicated names, after the policy, is already listed. The refusal names each topic and the ways out: --topics with new names, or --policy default. The check fails closed: a lab whose names cannot be listed is not created. A re-run, whose own KafkaMirrorMaker2 already exists, may keep only the topics that mirror already writes, read from its spec in the list the overlap check already fetches; it skips the listing when it asks for nothing more. New --topics or a new --policy under the same lab name are checked like a new lab, so `--topics orders` on a lab that mirrored kates.orders cannot take over the primary's orders. run used to tear the lab down after any failure, including one in the prerequisites, so this refusal would have been followed by the deletion of the very topic it refused. The existing refusal of a lab whose topics another lab already mirrors had that bug too: the teardown then deleted the other lab's topics. A run that fails in its prerequisites now removes nothing, since it has created nothing yet. The MirrorMaker 2 internal topics a teardown also deletes are not checked: every single-source lab writes the same source.checkpoints.internal, so checking them would forbid concurrent labs. Labs created before this change are not protected. --- cli/cmd/migrate_fake_test.go | 8 +- cli/cmd/migrate_run_test.go | 36 +++++++++ cli/cmd/migrate_up.go | 92 +++++++++++++++++++---- cli/cmd/migrate_up_test.go | 138 +++++++++++++++++++++++++++++++++- cli/cmd/migrate_verify.go | 9 ++- docs/book/10-cli-reference.md | 2 + 6 files changed, 262 insertions(+), 23 deletions(-) diff --git a/cli/cmd/migrate_fake_test.go b/cli/cmd/migrate_fake_test.go index dcaa2db4..b4078a5a 100644 --- a/cli/cmd/migrate_fake_test.go +++ b/cli/cmd/migrate_fake_test.go @@ -186,15 +186,21 @@ func scriptCluster(f *fakeProc, kind bool) { f.on("kubectl get daemonset kindnet -n kube-system --ignore-not-found --no-headers", kindnet) } -// scriptPrerequisites scripts the scripts' phase 1. +// scriptPrerequisites scripts the scripts' phase 1, with a target that has +// none of the lab's topics yet. func scriptPrerequisites(f *fakeProc) { f.on("kubectl cluster-info", "Kubernetes control plane is running") f.on("kubectl config current-context", "kind-panda") f.on("kubectl get crd kafkamirrormaker2s.kafka.strimzi.io -o name", "customresourcedefinition.apiextensions.k8s.io/kafkamirrormaker2s.kafka.strimzi.io") f.on("kubectl -n kafka wait kafka/krafter --for=condition=Ready", "kafka.kafka.strimzi.io/krafter condition met") f.on("kubectl -n kafka get kafkamirrormaker2 -l kates.io/lab -o json", `{"items":[]}`) + scriptPods(f) + f.on(targetTopicsList, "__consumer_offsets\nkates-events") } +// targetTopicsList is the topic listing the prerequisites run on the target. +const targetTopicsList = "kubectl -n kafka exec kates-migrate-target -c client -- /opt/kafka/bin/kafka-topics.sh --bootstrap-server krafter-kafka-bootstrap.kafka.svc.cluster.local:9092 --command-config /tmp/client.properties --list" + // mirrorCRJSON renders a KafkaMirrorMaker2 object with the given connector // states. func mirrorCRJSON(name, sourceState, checkpointState string) string { diff --git a/cli/cmd/migrate_run_test.go b/cli/cmd/migrate_run_test.go index ed4306f2..604b58ac 100644 --- a/cli/cmd/migrate_run_test.go +++ b/cli/cmd/migrate_run_test.go @@ -1,6 +1,7 @@ package cmd import ( + "context" "encoding/json" "os" "strings" @@ -402,6 +403,41 @@ func TestMigrateRunKeepLeavesTheLab(t *testing.T) { } } +func TestMigrateRunRefusedAtPrerequisitesRemovesNothing(t *testing.T) { + for _, tc := range []struct { + name, mirrors, topics, want string + }{ + {"the topic is on the target", `{"items":[]}`, "__consumer_offsets\nkates.orders", "kates.orders"}, + {"another lab mirrors it", `{"items":[{"metadata":{"name":"mm2-other-mirror-maker2","labels":{"kates.io/lab":"other"}}, + "spec":{"mirrors":[{"topicsPattern":"kates\\.orders"}]}}]}`, "__consumer_offsets\nkates.orders", "lab other"}, + } { + t.Run(tc.name, func(t *testing.T) { + migrateTestRepo(t) + f := newFakeProc(t) + scriptCluster(f, true) + scriptPrerequisites(f) + f.on("kubectl -n kafka get kafkamirrormaker2 -l kates.io/lab -o json", tc.mirrors) + f.on(targetTopicsList, tc.topics) + captureOutput(t, "json") + + // --yes, as newPairFlags sets: nothing but the refusal stands + // between the run and its teardown. + err := runMigrateRun(context.Background(), newPairFlags("2.8.2")) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("want the refusal naming %s, got %v", tc.want, err) + } + if f.hasLine("kubectl -n kafka delete kafkatopic") || f.hasLine("helm uninstall") { + t.Fatalf("a refused run must remove nothing:\n %s", strings.Join(f.lines(), "\n ")) + } + for _, a := range f.find("kubectl -n kafka apply -f -") { + if strings.Contains(a.stdin, "kind: KafkaTopic") { + t.Fatalf("a refused run adopted a topic:\n%s", a.stdin) + } + } + }) + } +} + func TestMigrateRunStopsWhenTheTopicNeverAppears(t *testing.T) { migrateTestRepo(t) f := newFakeProc(t) diff --git a/cli/cmd/migrate_up.go b/cli/cmd/migrate_up.go index a9a52ae4..840904c7 100644 --- a/cli/cmd/migrate_up.go +++ b/cli/cmd/migrate_up.go @@ -12,6 +12,7 @@ import ( "github.com/spf13/cobra" + "github.com/bmscomp/kates/cli/internal/podrun" "github.com/bmscomp/kates/cli/output" "github.com/bmscomp/kates/cli/pkg/kafkaversion" "github.com/bmscomp/kates/cli/pkg/migrate" @@ -218,51 +219,110 @@ func (r *labRun) phasePrerequisites(ctx context.Context, s *migrate.State) error return fmt.Errorf("Secret %s not found in %s. The kafka-cluster chart provisions the KafkaUser %q", migrate.TargetUser, l.TargetNamespace, migrate.TargetUser) } rep.Pass(migrate.RowTargetCredentials, "secret "+migrate.TargetUser) - return r.refuseOverlappingLab(ctx) + written, err := r.refuseOverlappingLab(ctx) + if err != nil { + return err + } + return r.refuseExistingTopics(ctx, written) } // refuseOverlappingLab refuses a second lab against the same target whose // topic set overlaps this one's: under the identity policy two mirrors -// writing one topic name is data corruption, not a test. -func (r *labRun) refuseOverlappingLab(ctx context.Context) error { +// writing one topic name is data corruption, not a test. written holds the +// target-side names this lab's own KafkaMirrorMaker2, if there is one +// already, writes: on a re-run those topics are the lab's own. +func (r *labRun) refuseOverlappingLab(ctx context.Context) (written map[string]bool, err error) { l := r.lab out, err := defaultRunner.Run(ctx, "kubectl", "-n", l.MirrorNamespace, "get", "kafkamirrormaker2", "-l", migrate.LabelLab, "-o", "json") if err != nil { - return nil // best effort: the mirror's own install still fails on a real collision + return nil, nil // best effort: the mirror's own install still fails on a real collision } var list struct { Items []struct { Metadata struct { Labels map[string]string `json:"labels"` } `json:"metadata"` - Spec struct { - Mirrors []struct { - TopicsPattern string `json:"topicsPattern"` - } `json:"mirrors"` - } `json:"spec"` + Spec json.RawMessage `json:"spec"` } `json:"items"` } if err := json.Unmarshal([]byte(out), &list); err != nil { - return nil + return nil, nil } mine := map[string]bool{} for _, t := range l.Topics { mine[t] = true } + written = map[string]bool{} for _, item := range list.Items { other := item.Metadata.Labels[migrate.LabelLab] - if other == "" || other == l.Name { + if other == "" { continue } - for _, m := range item.Spec.Mirrors { - for _, t := range topicsFromPattern(m.TopicsPattern) { - if mine[t] { - return fmt.Errorf("lab %s already mirrors topic %s onto %s in %s — two mirrors writing one topic name is corruption, not a test; pass --topics with names of your own, or kates migrate down --name %s", - other, t, l.TargetCluster, l.MirrorNamespace, other) + d := &labDiscovery{} + d.readMirrorSpec(item.Spec) + if other == l.Name { + for _, leg := range d.legs { + for _, t := range leg.Topics { + written[d.replicatedTopicOf(leg.Alias, t)] = true } } + continue + } + for _, t := range d.Topics { + if mine[t] { + return nil, fmt.Errorf("lab %s already mirrors topic %s onto %s in %s — two mirrors writing one topic name is corruption, not a test; pass --topics with names of your own, or kates migrate down --name %s", + other, t, l.TargetCluster, l.MirrorNamespace, other) + } + } + } + return written, nil +} + +// refuseExistingTopics refuses a lab whose mirrored topic names the target +// already has: removing a lab deletes the topics it mirrored into, so such +// a topic would go with it, data and all. It asks the brokers, because a +// topic made through the AdminClient has no KafkaTopic, and it fails +// closed: a lab whose names cannot be checked is not created. Names in +// written are the lab's own mirror's, so a re-run keeps them. +func (r *labRun) refuseExistingTopics(ctx context.Context, written map[string]bool) error { + l := r.lab + var check []string + for _, t := range l.ReplicatedTopics() { + if !written[t] { + check = append(check, t) } } + if len(check) == 0 { + return nil + } + c, err := openTargetClient(ctx, migrateTargetOptions{Namespace: l.TargetNamespace, Cluster: l.TargetCluster, User: migrate.TargetUser, Timeout: int(r.timeout / time.Second)}) + if err != nil { + return fmt.Errorf("cannot list the topics on %s to check that the lab's are new: %w", l.TargetCluster, err) + } + // The pod carries no lab label, so no teardown would remove it. + defer func() { + if err := podrun.Delete(context.Background(), defaultRunner, c.pod); err != nil { + migrateWarn(err.Error()) + } + }() + out, err := podrun.Exec(ctx, defaultRunner, c.pod, c.tools.ListTopics(c.bootstrap, c.cfg)...) + if err != nil { + return fmt.Errorf("cannot list the topics on %s to check that the lab's are new: %w", l.TargetCluster, err) + } + have := map[string]bool{} + for _, t := range strings.Fields(out) { + have[t] = true + } + var taken []string + for _, t := range check { + if have[t] { + taken = append(taken, t) + } + } + if len(taken) > 0 { + return fmt.Errorf("topic(s) %s already exist on %s in %s — removing a lab deletes the topics it mirrored into, so their data would go with it; pass --topics with names the target does not have, or --policy default (each source's topics land as .)", + strings.Join(taken, ", "), l.TargetCluster, l.TargetNamespace) + } return nil } diff --git a/cli/cmd/migrate_up_test.go b/cli/cmd/migrate_up_test.go index 42fd5ff8..4ae0852b 100644 --- a/cli/cmd/migrate_up_test.go +++ b/cli/cmd/migrate_up_test.go @@ -1,6 +1,7 @@ package cmd import ( + "context" "os" "strings" "testing" @@ -41,6 +42,7 @@ func TestMigrateUpPhasesAndHelmArgv(t *testing.T) { "kubectl get crd kafkamirrormaker2s.kafka.strimzi.io -o name", "kubectl -n kafka wait kafka/krafter --for=condition=Ready --timeout=300s", "kubectl -n kafka get secret kates-mm2 --ignore-not-found", + targetTopicsList, "kubectl create namespace kafka-m282-430-src", "kubectl label namespace kafka-m282-430-src --overwrite kates.io/lab=m282-430 kates.io/lab-role=source", "helm upgrade --install m282-430-src charts/legacy-kafka -n kafka-m282-430-src -f .build/migrate/m282-430/source-values.yaml --wait --timeout 600s", @@ -224,11 +226,17 @@ func TestMigrateUpSASLWritesTheCredentialOnStdin(t *testing.T) { if err := runMigrateUp(nil, flags); err != nil { t.Fatalf("up: %v", err) } - applies := f.find("kubectl -n kafka apply -f -") - if len(applies) != 1 { - t.Fatalf("want one Secret apply, got %d", len(applies)) + // The prerequisites' client pod is applied in kafka too. + var secrets []fakeCall + for _, a := range f.find("kubectl -n kafka apply -f -") { + if strings.Contains(a.stdin, "kind: Secret") { + secrets = append(secrets, a) + } + } + if len(secrets) != 1 { + t.Fatalf("want one Secret apply, got %d", len(secrets)) } - manifest := applies[0].stdin + manifest := secrets[0].stdin if !strings.Contains(manifest, "name: m282-430-src-mm2reader") || !strings.Contains(manifest, "password:") || !strings.Contains(manifest, "kates.io/lab: m282-430") { t.Fatalf("secret manifest:\n%s", manifest) } @@ -303,6 +311,128 @@ func TestMigrateUpRefusesOverlappingLab(t *testing.T) { } } +func TestMigrateUpRefusesATopicTheTargetHas(t *testing.T) { + migrateTestRepo(t) + f := newFakeProc(t) + scriptCluster(f, true) + scriptPrerequisites(f) + // kates.orders is already on the target, made by something else. + f.on(targetTopicsList, "__consumer_offsets\nkates-events\nkates.orders") + captureOutput(t, "table") + + flags := newPairFlags("2.8.2") + flags.SkipBuild = true + err := runMigrateUp(context.Background(), flags) + if err == nil || !strings.Contains(err.Error(), "kates.orders") || !strings.Contains(err.Error(), "--topics") { + t.Fatalf("want the refusal naming kates.orders and --topics, got %v", err) + } + if f.hasLine("kubectl create namespace") { + t.Fatal("nothing may be created after the refusal") + } + if !f.hasLine("kubectl -n kafka delete pod kates-migrate-target") { + t.Fatal("the check's client pod was left behind") + } +} + +func TestMigrateUpReRunSkipsTheTopicCheck(t *testing.T) { + migrateTestRepo(t) + f := newFakeProc(t) + scriptCluster(f, true) + scriptPrerequisites(f) + scriptUp(f) + // The lab's own mirror is there, and so is the topic it mirrored. + f.on("kubectl -n kafka get kafkamirrormaker2 -l kates.io/lab -o json", `{"items":[{"metadata":{"name":"mm2-m282-430-mirror-maker2","labels":{"kates.io/lab":"m282-430"}}, + "spec":{"mirrors":[{"topicsPattern":"kates\\.orders"}]}}]}`) + f.on(targetTopicsList, "kates.orders") + captureOutput(t, "table") + + flags := newPairFlags("2.8.2") + flags.SkipBuild = true + if err := runMigrateUp(context.Background(), flags); err != nil { + t.Fatalf("a re-run of the same lab must pass: %v", err) + } + if f.hasLine(targetTopicsList) { + t.Fatal("a re-run must not list the target's topics") + } +} + +func TestMigrateUpReRunWithNewTopicsIsChecked(t *testing.T) { + migrateTestRepo(t) + f := newFakeProc(t) + scriptCluster(f, true) + scriptPrerequisites(f) + scriptUp(f) + // The lab's own mirror writes kates.orders; the re-run asks for orders, + // which the target has and the mirror never wrote. + f.on("kubectl -n kafka get kafkamirrormaker2 -l kates.io/lab -o json", `{"items":[{"metadata":{"name":"mm2-m282-430-mirror-maker2","labels":{"kates.io/lab":"m282-430"}}, + "spec":{"mirrors":[{"topicsPattern":"kates\\.orders"}]}}]}`) + f.on(targetTopicsList, "__consumer_offsets\nkates.orders\norders") + captureOutput(t, "table") + + flags := newPairFlags("2.8.2") + flags.SkipBuild = true + flags.Topics = "orders" + err := runMigrateUp(context.Background(), flags) + if err == nil || !strings.Contains(err.Error(), "topic(s) orders already exist") { + t.Fatalf("want the refusal naming orders, got %v", err) + } + if f.hasLine("helm upgrade") { + t.Fatalf("nothing may be upgraded after the refusal:\n %s", strings.Join(f.lines(), "\n ")) + } +} + +func TestMigrateUpRefusesWhenTheTargetCannotBeListed(t *testing.T) { + for _, tc := range []struct { + name string + fail string + }{ + {"the listing fails", targetTopicsList}, + {"the client pod is never Ready", "kubectl -n kafka wait --for=condition=Ready pod/kates-migrate-target"}, + } { + t.Run(tc.name, func(t *testing.T) { + migrateTestRepo(t) + f := newFakeProc(t) + scriptCluster(f, true) + scriptPrerequisites(f) + scriptUp(f) + f.fail(tc.fail, "boom") + captureOutput(t, "table") + + flags := newPairFlags("2.8.2") + flags.SkipBuild = true + err := runMigrateUp(context.Background(), flags) + if err == nil || !strings.Contains(err.Error(), "cannot list the topics") { + t.Fatalf("want a refusal when the target cannot be listed, got %v", err) + } + if f.hasLine("kubectl create namespace") { + t.Fatal("nothing may be created when the check cannot run") + } + }) + } +} + +func TestMigrateUpChecksTheNameOnTheTarget(t *testing.T) { + migrateTestRepo(t) + f := newFakeProc(t) + scriptCluster(f, true) + scriptPrerequisites(f) + scriptUp(f) + f.on(targetTopicsList, "__consumer_offsets\nkates.orders") + captureOutput(t, "table") + + // Under the default policy the lab writes source.kates.orders, which the + // target does not have. + flags := newPairFlags("2.8.2") + flags.SkipBuild = true + flags.Policy = migrate.PolicyDefault + if err := runMigrateUp(context.Background(), flags); err != nil { + t.Fatalf("up --policy default: %v", err) + } + if !f.hasLine(targetTopicsList) { + t.Fatal("the target's topics were not checked") + } +} + func TestMigrateUpBuildsTheImageForABuiltLine(t *testing.T) { migrateTestRepo(t) f := newFakeProc(t) diff --git a/cli/cmd/migrate_verify.go b/cli/cmd/migrate_verify.go index 5aa57ee8..2bd1616b 100644 --- a/cli/cmd/migrate_verify.go +++ b/cli/cmd/migrate_verify.go @@ -1049,12 +1049,17 @@ func runMigrateRun(ctx context.Context, f *migratePairFlags) error { cleanupCtx, cancel := context.WithTimeout(context.Background(), 2*r.timeout) defer cancel() - if f.Keep { + var phaseErr *migrate.PhaseError + switch { + case errors.As(runErr, &phaseErr) && phaseErr.Phase == "prerequisites": + // Nothing exists yet, and what the target holds is not the lab's: a + // teardown would delete the very topics a refusal protects. + case f.Keep: if r.clients != nil { r.deleteClients(cleanupCtx) } printKeepBlock(r.lab) - } else { + default: r.deleteClients(cleanupCtx) migrateHint("cleaning up...") if err := labDown(cleanupCtx, discoveryFromLab(r.lab, r.createdNamespace), r.timeout); err != nil { diff --git a/docs/book/10-cli-reference.md b/docs/book/10-cli-reference.md index 756d053b..8d552c7e 100644 --- a/docs/book/10-cli-reference.md +++ b/docs/book/10-cli-reference.md @@ -1449,6 +1449,8 @@ Two combinations the chart refuses at render time are refused by the CLI first, - **the same alias twice** (`--from 2.8.2 --from 2.8.2`) — an alias names the replicated topic prefix, the offset-syncs topic and the checkpoints topic, so two sources sharing one interleave into a single set of target topics without any error; - **an identity fan-in over one topic name** (`--from 2.8.2 --from 3.9.1 --topics orders`) — under the identity policy topic names are kept, so both legs would write `orders` on the target. The refusal names the three ways out: `--policy default` (each source's topics land as `.`), dropping `--topics` so the lab gives each source its own corpus, or one lab per source. +`up` and `run` also refuse a lab whose topic names on the target, after the policy, are already there. A re-run, whose own KafkaMirrorMaker2 exists, may keep only the topics that mirror already writes: new `--topics` or a new `--policy` under the same lab name are checked like a new lab. Removing a lab deletes the topics it mirrored into, so with the identity policy `--topics orders` would take the primary's own `orders` with it, data and all. The CLI lists the target's topics as `kates-mm2` from a client pod it deletes afterwards, and refuses when it cannot. The refusal names each topic and the ways out: `--topics` with names the target does not have, or `--policy default`. A `run` refused at its prerequisites, by this check or because another lab already mirrors one of its topics, removes nothing. + #### Read-only sources `--read-only-source` (on `plan`, `up`, `run` and `migrate mirror deploy`) sets `readOnlySource: true` on every mirror, which the chart turns into `offset-syncs.topic.location: target` (KIP-716 [@kip716]) on both connectors and into the matching target ACL. It is **off by default**, and the difference is what the source principal must be granted: