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: