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

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 {

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)

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 <alias>.<topic>)",
+			strings.Join(taken, ", "), l.TargetCluster, l.TargetNamespace)
+	}
 	return nil
 }
 

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)

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 {