From 378ddb0511afd4cf8a4e90bc6b7879dc84f0dd67 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Thu, 13 Aug 2026 12:20:12 -0400 Subject: [PATCH 1/3] test(e2e): add schema-upgrade auto-mode and safe-window rollback specs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers two of the gaps tracked in #439 that PR #426 deferred: - schemaVersion: "auto" single-step path — a single spec.documentDBVersion bump migrates both the binary and the extension schema in one step (operator runs ALTER EXTENSION automatically, no finalize patch). - Rollback within the safe window — in two-phase mode the installed schema stays at the old version, so reverting spec.documentDBVersion before finalize is admitted by the image-rollback webhook; asserts data and schema integrity across the down-hop. Adds a reusable schema_version mixin so specs can set spec.schemaVersion at creation. Both specs reuse the existing upgrade helpers and the E2E_UPGRADE_OLD/NEW_DOCUMENTDB_VERSION env pair. Refs #439. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d53b8fec-0585-4888-89da-d9a6847597a5 Signed-off-by: Wenting Wu --- .../mixins/schema_version.yaml.template | 12 ++ .../tests/upgrade/upgrade_schema_auto_test.go | 171 ++++++++++++++++ .../upgrade/upgrade_schema_rollback_test.go | 190 ++++++++++++++++++ 3 files changed, 373 insertions(+) create mode 100644 test/e2e/manifests/mixins/schema_version.yaml.template create mode 100644 test/e2e/tests/upgrade/upgrade_schema_auto_test.go create mode 100644 test/e2e/tests/upgrade/upgrade_schema_rollback_test.go diff --git a/test/e2e/manifests/mixins/schema_version.yaml.template b/test/e2e/manifests/mixins/schema_version.yaml.template new file mode 100644 index 000000000..81f3e5db4 --- /dev/null +++ b/test/e2e/manifests/mixins/schema_version.yaml.template @@ -0,0 +1,12 @@ +apiVersion: documentdb.io/preview +kind: DocumentDB +metadata: + name: ${NAME} + namespace: ${NAMESPACE} +spec: + # Desired extension schema version. Set to "auto" to let the operator run + # ALTER EXTENSION UPDATE automatically whenever the binary is upgraded + # (single-step path), or to an explicit semver to finalize a specific + # version. Left empty (line dropped by envsubst) the base stays in + # two-phase mode. + schemaVersion: ${SCHEMA_VERSION} diff --git a/test/e2e/tests/upgrade/upgrade_schema_auto_test.go b/test/e2e/tests/upgrade/upgrade_schema_auto_test.go new file mode 100644 index 000000000..b1e11c4c5 --- /dev/null +++ b/test/e2e/tests/upgrade/upgrade_schema_auto_test.go @@ -0,0 +1,171 @@ +package upgrade + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "go.mongodb.org/mongo-driver/v2/bson" + "k8s.io/apimachinery/pkg/types" + + previewv1 "github.com/documentdb/documentdb-operator/api/preview" + "github.com/documentdb/documentdb-operator/test/e2e" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/assertions" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/documentdb" + e2emongo "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/mongo" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/namespaces" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/seed" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/timeouts" + shareddb "github.com/documentdb/documentdb-operator/test/shared/documentdb" + sharedmongo "github.com/documentdb/documentdb-operator/test/shared/mongo" +) + +// DocumentDB upgrade — schema, "auto" mode (single-step migration). +// +// This spec is the counterpart to upgrade_schema_test.go (two-phase mode). +// It exercises the other documented schemaVersion contract +// (documentdb_types.go): with spec.schemaVersion set to "auto", a single +// spec.documentDBVersion bump upgrades BOTH the binary and the extension +// schema in one step — the operator runs ALTER EXTENSION documentdb UPDATE +// automatically, with no separate finalize patch. +// +// The flow: +// +// 1. Create a DocumentDB pinned to the OLD version with spec.schemaVersion +// set to "auto" and seed data. status.schemaVersion settles on OLD (the +// binary and schema already agree, so "auto" is a no-op at first). +// 2. Upgrade the binary by patching spec.documentDBVersion to NEW. Because +// schemaVersion is "auto", the operator must migrate the schema in the +// same reconcile cycle — status.schemaVersion advances to NEW WITHOUT any +// schemaVersion patch. Seeded data is retained. +// +// The absence of a separate finalize step (present in the two-phase spec) is +// the point of this test: it asserts the single-step path works end-to-end. +// +// Old/new versions come from the same env vars and defaults as the two-phase +// spec (E2E_UPGRADE_OLD_DOCUMENTDB_VERSION / _NEW_). +var _ = Describe("DocumentDB upgrade — schema (auto mode)", + Label(e2e.UpgradeLabel, e2e.DisruptiveLabel, e2e.SlowLabel), + e2e.HighLevelLabel, + Serial, Ordered, func() { + const ( + ddName = "upgrade-schema-auto" + dbName = "upgrade_schema_auto" + collName = "seed" + ) + var ( + oldVersion string + newVersion string + ctx context.Context + cancel context.CancelFunc + ) + + BeforeAll(func() { + skipUnlessUpgradeEnabled() + oldVersion = envOr(envOldDocumentDBVersion, defaultOldDocumentDBVersion) + newVersion = envOr(envNewDocumentDBVersion, defaultNewDocumentDBVersion) + if oldVersion == newVersion { + Skip(envOldDocumentDBVersion + " and " + envNewDocumentDBVersion + " are identical; nothing to upgrade") + } + }) + + BeforeEach(func() { + e2e.SkipUnlessLevel(e2e.High) + ctx, cancel = context.WithTimeout(context.Background(), imageRolloutTimeout) + DeferCleanup(func() { cancel() }) + }) + + It("migrates schema and binary in a single step when schemaVersion is auto, retaining data", func() { + env := e2e.SuiteEnv() + Expect(env).NotTo(BeNil(), "SuiteEnv must be initialized by SetupSuite") + Expect(ctx).NotTo(BeNil(), "BeforeEach must have populated the spec context") + c := env.Client + + By("creating a DocumentDB pinned to the old version with schemaVersion=auto") + ns := namespaces.NamespaceForSpec(e2e.UpgradeLabel) + createNamespace(ctx, c, ns) + createCredentialSecret(ctx, c, ns) + + vars := baseVars(ddName, ns, "2Gi") + // Drive the version via documentDBVersion, not raw images. + vars["DOCUMENTDB_IMAGE"] = "" + vars["GATEWAY_IMAGE"] = "" + vars["DOCUMENTDB_VERSION"] = oldVersion + vars["SCHEMA_VERSION"] = "auto" + + dd, err := documentdb.Create(ctx, c, ns, ddName, documentdb.CreateOptions{ + Base: "documentdb", + Mixins: []string{"documentdb_version", "schema_version"}, + Vars: vars, + ManifestsRoot: manifestsRoot(), + }) + Expect(err).NotTo(HaveOccurred(), "create DocumentDB %s/%s", ns, ddName) + DeferCleanup(func(ctx SpecContext) { + _ = shareddb.Delete(ctx, c, dd, 3*time.Minute) + }) + + key := types.NamespacedName{Namespace: ns, Name: ddName} + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Succeed(), "DocumentDB did not reach Ready on oldVersion=%s", oldVersion) + + // Single reused schema-version poller (caches last good read so a + // transient API error can't fail a Consistently window). + schemaVersion := schemaVersionGetter(ctx, c, key) + + By("waiting for status.schemaVersion to settle on the old version") + Eventually(schemaVersion, + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Equal(oldVersion), "initial schema version should be %s", oldVersion) + + By("seeding data on the old schema") + docs := seed.SmallDataset() + handle, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "connect to DocumentDB gateway on oldVersion") + inserted, err := sharedmongo.Seed(ctx, handle.Client(), dbName, collName, docs) + Expect(err).NotTo(HaveOccurred(), "seed %s.%s", dbName, collName) + Expect(inserted).To(Equal(seed.SmallDatasetSize)) + Expect(handle.Close(ctx)).To(Succeed()) + + By("upgrading the binary via spec.documentDBVersion (schemaVersion stays auto)") + fresh, err := shareddb.Get(ctx, c, key) + Expect(err).NotTo(HaveOccurred(), "re-fetch DocumentDB before version patch") + Expect(shareddb.PatchSpec(ctx, c, fresh, func(s *previewv1.DocumentDBSpec) { + s.DocumentDBVersion = newVersion + })).To(Succeed(), "patch DocumentDBVersion from %s to %s", oldVersion, newVersion) + + By("waiting for the operator to apply the new version and DocumentDB to be Ready") + Eventually(statusDocumentDBImageGetter(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(ContainSubstring(newVersion), "status.documentDBImage did not advance to version %s", newVersion) + + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Succeed(), "DocumentDB did not reach Ready on newVersion=%s", newVersion) + + By("verifying auto mode migrated the schema in a single step (no finalize patch)") + // This is the single-step assertion: because schemaVersion is + // "auto", the schema must advance to newVersion on its own — we + // never set spec.schemaVersion. Contrast with the two-phase spec, + // which requires an explicit finalize. + Eventually(schemaVersion, + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Equal(newVersion), "auto mode should migrate schema to %s in a single step", newVersion) + + By("verifying seeded data survived the single-step upgrade") + handle2, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "reconnect to DocumentDB gateway after single-step upgrade") + DeferCleanup(func(ctx SpecContext) { _ = handle2.Close(ctx) }) + n, err := sharedmongo.Count(ctx, handle2.Client(), dbName, collName, bson.M{}) + Expect(err).NotTo(HaveOccurred(), "count %s.%s after single-step upgrade", dbName, collName) + Expect(n).To(Equal(int64(seed.SmallDatasetSize)), + "seeded document count changed across single-step upgrade") + }) + }) diff --git a/test/e2e/tests/upgrade/upgrade_schema_rollback_test.go b/test/e2e/tests/upgrade/upgrade_schema_rollback_test.go new file mode 100644 index 000000000..b1871701a --- /dev/null +++ b/test/e2e/tests/upgrade/upgrade_schema_rollback_test.go @@ -0,0 +1,190 @@ +package upgrade + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "go.mongodb.org/mongo-driver/v2/bson" + "k8s.io/apimachinery/pkg/types" + + previewv1 "github.com/documentdb/documentdb-operator/api/preview" + "github.com/documentdb/documentdb-operator/test/e2e" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/assertions" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/documentdb" + e2emongo "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/mongo" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/namespaces" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/seed" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/timeouts" + shareddb "github.com/documentdb/documentdb-operator/test/shared/documentdb" + sharedmongo "github.com/documentdb/documentdb-operator/test/shared/mongo" +) + +// DocumentDB upgrade — schema, rollback within the safe window. +// +// Two-phase mode (spec.schemaVersion unset) exists precisely to give a +// rollback-safe window: after the binary is bumped but BEFORE the schema is +// finalized, the installed extension schema still sits at the OLD version, so +// the binary can be reverted without hitting the irreversible ALTER EXTENSION +// migration. The image-rollback webhook only blocks downgrades below the +// *installed schema* (documentdb_webhook.go: validateImageRollback), so a +// downgrade back to the old binary is permitted while the schema is still old. +// +// The flow: +// +// 1. Create a DocumentDB pinned to the OLD version with spec.schemaVersion +// unset (two-phase) and seed data. status.schemaVersion settles on OLD. +// 2. Upgrade the binary to NEW via spec.documentDBVersion. The schema stays +// at OLD (two-phase); data is retained. +// 3. Roll the binary back to OLD via spec.documentDBVersion — BEFORE any +// finalize. Because the installed schema is still OLD, the webhook admits +// the downgrade. Assert DocumentDB returns to Ready on the old binary, the +// schema is still OLD, and the seeded data survived the down-hop. +// +// This is the safe-window guarantee documented on spec.schemaVersion; the +// negative case (rollback blocked AFTER finalize) is covered by the webhook +// unit tests. +var _ = Describe("DocumentDB upgrade — schema (rollback within safe window)", + Label(e2e.UpgradeLabel, e2e.DisruptiveLabel, e2e.SlowLabel), + e2e.HighLevelLabel, + Serial, Ordered, func() { + const ( + ddName = "upgrade-schema-rollback" + dbName = "upgrade_schema_rollback" + collName = "seed" + ) + var ( + oldVersion string + newVersion string + ctx context.Context + cancel context.CancelFunc + ) + + BeforeAll(func() { + skipUnlessUpgradeEnabled() + oldVersion = envOr(envOldDocumentDBVersion, defaultOldDocumentDBVersion) + newVersion = envOr(envNewDocumentDBVersion, defaultNewDocumentDBVersion) + if oldVersion == newVersion { + Skip(envOldDocumentDBVersion + " and " + envNewDocumentDBVersion + " are identical; nothing to upgrade") + } + }) + + BeforeEach(func() { + e2e.SkipUnlessLevel(e2e.High) + ctx, cancel = context.WithTimeout(context.Background(), imageRolloutTimeout) + DeferCleanup(func() { cancel() }) + }) + + It("allows rolling the binary back to the old version before finalize, retaining data", func() { + env := e2e.SuiteEnv() + Expect(env).NotTo(BeNil(), "SuiteEnv must be initialized by SetupSuite") + Expect(ctx).NotTo(BeNil(), "BeforeEach must have populated the spec context") + c := env.Client + + By("creating a DocumentDB pinned to the old version (schemaVersion unset → two-phase)") + ns := namespaces.NamespaceForSpec(e2e.UpgradeLabel) + createNamespace(ctx, c, ns) + createCredentialSecret(ctx, c, ns) + + vars := baseVars(ddName, ns, "2Gi") + vars["DOCUMENTDB_IMAGE"] = "" + vars["GATEWAY_IMAGE"] = "" + vars["DOCUMENTDB_VERSION"] = oldVersion + + dd, err := documentdb.Create(ctx, c, ns, ddName, documentdb.CreateOptions{ + Base: "documentdb", + Mixins: []string{"documentdb_version"}, + Vars: vars, + ManifestsRoot: manifestsRoot(), + }) + Expect(err).NotTo(HaveOccurred(), "create DocumentDB %s/%s", ns, ddName) + DeferCleanup(func(ctx SpecContext) { + _ = shareddb.Delete(ctx, c, dd, 3*time.Minute) + }) + + key := types.NamespacedName{Namespace: ns, Name: ddName} + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Succeed(), "DocumentDB did not reach Ready on oldVersion=%s", oldVersion) + + schemaVersion := schemaVersionGetter(ctx, c, key) + + By("waiting for status.schemaVersion to settle on the old version") + Eventually(schemaVersion, + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Equal(oldVersion), "initial schema version should be %s", oldVersion) + + By("seeding data on the old schema") + docs := seed.SmallDataset() + handle, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "connect to DocumentDB gateway on oldVersion") + inserted, err := sharedmongo.Seed(ctx, handle.Client(), dbName, collName, docs) + Expect(err).NotTo(HaveOccurred(), "seed %s.%s", dbName, collName) + Expect(inserted).To(Equal(seed.SmallDatasetSize)) + Expect(handle.Close(ctx)).To(Succeed()) + + By("upgrading the binary via spec.documentDBVersion without setting schemaVersion") + fresh, err := shareddb.Get(ctx, c, key) + Expect(err).NotTo(HaveOccurred(), "re-fetch DocumentDB before version patch") + Expect(shareddb.PatchSpec(ctx, c, fresh, func(s *previewv1.DocumentDBSpec) { + s.DocumentDBVersion = newVersion + })).To(Succeed(), "patch DocumentDBVersion from %s to %s", oldVersion, newVersion) + + By("waiting for the new binary to be applied and DocumentDB to be Ready") + Eventually(statusDocumentDBImageGetter(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(ContainSubstring(newVersion), "status.documentDBImage did not advance to version %s", newVersion) + + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Succeed(), "DocumentDB did not reach Ready on newVersion=%s", newVersion) + + By("confirming two-phase mode kept the schema at the old version (rollback window open)") + Consistently(schemaVersion, + 30*time.Second, 5*time.Second, + ).Should(Equal(oldVersion), + "schema must remain at %s while schemaVersion is unset (rollback-safe window)", oldVersion) + + By("rolling the binary back to the old version before finalize") + // The installed schema is still oldVersion, so the image-rollback + // webhook admits this downgrade. Reverting after a finalize would + // instead be rejected (covered by webhook unit tests). + fresh2, err := shareddb.Get(ctx, c, key) + Expect(err).NotTo(HaveOccurred(), "re-fetch DocumentDB before rollback patch") + Expect(shareddb.PatchSpec(ctx, c, fresh2, func(s *previewv1.DocumentDBSpec) { + s.DocumentDBVersion = oldVersion + })).To(Succeed(), "rollback DocumentDBVersion from %s back to %s", newVersion, oldVersion) + + By("waiting for the old binary to be re-applied and DocumentDB to be Ready") + Eventually(statusDocumentDBImageGetter(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(ContainSubstring(oldVersion), "status.documentDBImage did not roll back to version %s", oldVersion) + + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Succeed(), "DocumentDB did not reach Ready after rollback to oldVersion=%s", oldVersion) + + By("verifying the schema is still at the old version after rollback") + Consistently(schemaVersion, + 30*time.Second, 5*time.Second, + ).Should(Equal(oldVersion), + "schema must still be %s after rolling the binary back", oldVersion) + + By("verifying seeded data survived the binary rollback") + handle2, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "reconnect to DocumentDB gateway after rollback") + DeferCleanup(func(ctx SpecContext) { _ = handle2.Close(ctx) }) + n, err := sharedmongo.Count(ctx, handle2.Client(), dbName, collName, bson.M{}) + Expect(err).NotTo(HaveOccurred(), "count %s.%s after rollback", dbName, collName) + Expect(n).To(Equal(int64(seed.SmallDatasetSize)), + "seeded document count changed across binary rollback") + }) + }) From f62069df7baa4cae705311122a4241f721bb8d7b Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Thu, 13 Aug 2026 13:12:20 -0400 Subject: [PATCH 2/3] test(e2e): add HA schema-upgrade and unpullable-version specs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers two more of the gaps tracked in #439: - Multi-instance / HA rollout — new upgrade_schema_ha_test.go runs the two-phase schema upgrade on a 3-instance cluster, asserts the cluster returns to 3 ready instances after both the binary roll and the schema finalize, and execs psql on a replica pod to confirm the migrated extension version propagated via WAL streaming replication (the operator computes status.schemaVersion from the primary only, so the replica is verified directly). - Invalid / unpullable documentDBVersion — added as a second ordered spec in upgrade_schema_test.go that reuses the existing cluster (no extra DocumentDB). It patches to a version above the installed schema (0.999.0, so the image-rollback webhook admits it but no image exists), asserts the schema does not advance and seeded data stays readable while pods can't pull, then rolls back to the valid version and asserts recovery. Adds a replicaInstalledSchemaVersion helper (reads pg_extension.extversion on a replica via CNPG EventuallyExecCommand) and hoists the shared cluster creation into BeforeAll so both ordered specs share one cluster. Refs #439. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d53b8fec-0585-4888-89da-d9a6847597a5 Signed-off-by: Wenting Wu --- test/e2e/tests/upgrade/helpers_test.go | 57 +++++ .../tests/upgrade/upgrade_schema_ha_test.go | 203 ++++++++++++++++++ test/e2e/tests/upgrade/upgrade_schema_test.go | 115 ++++++++-- 3 files changed, 360 insertions(+), 15 deletions(-) create mode 100644 test/e2e/tests/upgrade/upgrade_schema_ha_test.go diff --git a/test/e2e/tests/upgrade/helpers_test.go b/test/e2e/tests/upgrade/helpers_test.go index 7cd43cb47..8093ec8d7 100644 --- a/test/e2e/tests/upgrade/helpers_test.go +++ b/test/e2e/tests/upgrade/helpers_test.go @@ -2,14 +2,17 @@ package upgrade import ( "context" + "fmt" "os" "os/exec" "path/filepath" "runtime" + "strings" "time" . "github.com/onsi/ginkgo/v2" //nolint:revive + "github.com/cloudnative-pg/cloudnative-pg/tests/utils/environment" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -186,3 +189,57 @@ func createCredentialSecret(ctx context.Context, c client.Client, ns string) { Fail("create credential secret " + ns + "/" + credentialSecretName + ": " + err.Error()) } } + +// replicaInstalledSchemaVersion execs psql on a replica pod of the CNPG +// cluster backing the DocumentDB and returns the installed documentdb +// extension version, normalized to semver (e.g. "0.110.0"). +// +// This exists because the operator computes status.schemaVersion by +// querying the PRIMARY only (see executeSQLCommand in the controller), +// so the CR status does not independently prove that a schema migration +// propagated to replicas. The extension schema (an ALTER EXTENSION +// catalog change) reaches replicas via WAL streaming replication; this +// helper reads pg_extension.extversion directly on a replica to confirm +// that convergence. +// +// The extension reports its version in "Major.Minor-Patch" form (e.g. +// "0.110-0"); replacing the final "-" with "." yields the semver used +// throughout the upgrade specs. clusterName is the CNPG cluster name, +// which for a single-cluster DocumentDB equals the DocumentDB name. +func replicaInstalledSchemaVersion( + ctx context.Context, + env *environment.TestingEnvironment, + ns, clusterName string, +) (string, error) { + var pods corev1.PodList + if err := env.Client.List(ctx, &pods, + client.InNamespace(ns), + client.MatchingLabels{ + "cnpg.io/cluster": clusterName, + "cnpg.io/instanceRole": "replica", + }, + ); err != nil { + return "", fmt.Errorf("list replica pods for cluster %s/%s: %w", ns, clusterName, err) + } + if len(pods.Items) == 0 { + return "", fmt.Errorf("no replica pods found for cluster %s/%s", ns, clusterName) + } + + timeout := time.Minute + stdout, stderr, err := env.EventuallyExecCommand(ctx, pods.Items[0], "postgres", &timeout, + "psql", "-U", "postgres", "-d", "postgres", "-tAc", + "SELECT extversion FROM pg_extension WHERE extname='documentdb'") + if err != nil { + return "", fmt.Errorf("exec psql on replica %s: %w (stderr: %s)", pods.Items[0].Name, err, stderr) + } + + raw := strings.TrimSpace(stdout) + if raw == "" { + return "", fmt.Errorf("documentdb extension not installed on replica %s", pods.Items[0].Name) + } + // "0.110-0" -> "0.110.0"; a value already in semver form is unchanged. + if i := strings.LastIndex(raw, "-"); i >= 0 { + raw = raw[:i] + "." + raw[i+1:] + } + return raw, nil +} diff --git a/test/e2e/tests/upgrade/upgrade_schema_ha_test.go b/test/e2e/tests/upgrade/upgrade_schema_ha_test.go new file mode 100644 index 000000000..f54418a4a --- /dev/null +++ b/test/e2e/tests/upgrade/upgrade_schema_ha_test.go @@ -0,0 +1,203 @@ +package upgrade + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "go.mongodb.org/mongo-driver/v2/bson" + "k8s.io/apimachinery/pkg/types" + + previewv1 "github.com/documentdb/documentdb-operator/api/preview" + "github.com/documentdb/documentdb-operator/test/e2e" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/assertions" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/documentdb" + e2emongo "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/mongo" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/namespaces" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/seed" + "github.com/documentdb/documentdb-operator/test/e2e/pkg/e2eutils/timeouts" + shareddb "github.com/documentdb/documentdb-operator/test/shared/documentdb" + sharedmongo "github.com/documentdb/documentdb-operator/test/shared/mongo" +) + +// DocumentDB upgrade — schema on a multi-instance (HA) cluster. +// +// The single-instance schema spec (upgrade_schema_test.go) pins +// INSTANCES=1, so it cannot observe how the two-phase migration behaves +// across a primary + replicas. This spec runs the same two-phase flow on +// a 3-instance cluster and adds two HA-specific guarantees: +// +// 1. Rollout health — the cluster returns to 3 ready instances after both +// the binary roll and the schema finalize (CNPG rolls replicas first, +// then the primary; we assert the observable end-state via +// AssertInstanceCount rather than reimplementing CNPG's internal +// rollout ordering). +// 2. Replica schema convergence — the operator computes +// status.schemaVersion from the PRIMARY only, so this spec additionally +// execs psql on a REPLICA pod to confirm the migrated extension version +// propagated via WAL streaming replication. +// +// Old/new versions come from the same env vars/defaults as the other +// schema specs (E2E_UPGRADE_OLD_DOCUMENTDB_VERSION / _NEW_). +var _ = Describe("DocumentDB upgrade — schema (multi-instance HA)", + Label(e2e.UpgradeLabel, e2e.DisruptiveLabel, e2e.SlowLabel), + e2e.HighestLevelLabel, + Serial, Ordered, func() { + const ( + ddName = "upgrade-schema-ha" + dbName = "upgrade_schema_ha" + collName = "seed" + instances = 3 + ) + var ( + oldVersion string + newVersion string + ctx context.Context + cancel context.CancelFunc + ) + + BeforeAll(func() { + skipUnlessUpgradeEnabled() + oldVersion = envOr(envOldDocumentDBVersion, defaultOldDocumentDBVersion) + newVersion = envOr(envNewDocumentDBVersion, defaultNewDocumentDBVersion) + if oldVersion == newVersion { + Skip(envOldDocumentDBVersion + " and " + envNewDocumentDBVersion + " are identical; nothing to upgrade") + } + }) + + BeforeEach(func() { + e2e.SkipUnlessLevel(e2e.Highest) + ctx, cancel = context.WithTimeout(context.Background(), imageRolloutTimeout) + DeferCleanup(func() { cancel() }) + }) + + It("migrates the schema across a 3-instance cluster and propagates it to replicas", func() { + env := e2e.SuiteEnv() + Expect(env).NotTo(BeNil(), "SuiteEnv must be initialized by SetupSuite") + Expect(ctx).NotTo(BeNil(), "BeforeEach must have populated the spec context") + c := env.Client + + By("creating a 3-instance DocumentDB pinned to the old version (two-phase)") + ns := namespaces.NamespaceForSpec(e2e.UpgradeLabel) + createNamespace(ctx, c, ns) + createCredentialSecret(ctx, c, ns) + + vars := baseVars(ddName, ns, "2Gi") + vars["DOCUMENTDB_IMAGE"] = "" + vars["GATEWAY_IMAGE"] = "" + vars["DOCUMENTDB_VERSION"] = oldVersion + vars["INSTANCES"] = "3" + + dd, err := documentdb.Create(ctx, c, ns, ddName, documentdb.CreateOptions{ + Base: "documentdb", + Mixins: []string{"documentdb_version"}, + Vars: vars, + ManifestsRoot: manifestsRoot(), + }) + Expect(err).NotTo(HaveOccurred(), "create DocumentDB %s/%s", ns, ddName) + DeferCleanup(func(ctx SpecContext) { + _ = shareddb.Delete(ctx, c, dd, 3*time.Minute) + }) + + key := types.NamespacedName{Namespace: ns, Name: ddName} + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Succeed(), "DocumentDB did not reach Ready on oldVersion=%s", oldVersion) + + By("waiting for all 3 instances to become ready") + Eventually(assertions.AssertInstanceCount(ctx, c, key, instances), + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Succeed(), "cluster did not reach %d ready instances", instances) + + schemaVersion := schemaVersionGetter(ctx, c, key) + + By("waiting for status.schemaVersion to settle on the old version") + Eventually(schemaVersion, + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Equal(oldVersion), "initial schema version should be %s", oldVersion) + + By("confirming a replica also reports the old schema version before upgrade") + Eventually(func() (string, error) { + return replicaInstalledSchemaVersion(ctx, env, ns, ddName) + }, timeouts.For(timeouts.DocumentDBReady), timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Equal(oldVersion), "replica should start at schema %s", oldVersion) + + By("seeding data on the old schema") + docs := seed.SmallDataset() + handle, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "connect to DocumentDB gateway on oldVersion") + inserted, err := sharedmongo.Seed(ctx, handle.Client(), dbName, collName, docs) + Expect(err).NotTo(HaveOccurred(), "seed %s.%s", dbName, collName) + Expect(inserted).To(Equal(seed.SmallDatasetSize)) + Expect(handle.Close(ctx)).To(Succeed()) + + By("upgrading the binary via spec.documentDBVersion without setting schemaVersion") + fresh, err := shareddb.Get(ctx, c, key) + Expect(err).NotTo(HaveOccurred(), "re-fetch DocumentDB before version patch") + Expect(shareddb.PatchSpec(ctx, c, fresh, func(s *previewv1.DocumentDBSpec) { + s.DocumentDBVersion = newVersion + })).To(Succeed(), "patch DocumentDBVersion from %s to %s", oldVersion, newVersion) + + By("waiting for the new binary to roll out and all 3 instances to be ready again") + Eventually(statusDocumentDBImageGetter(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(ContainSubstring(newVersion), "status.documentDBImage did not advance to version %s", newVersion) + + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Succeed(), "DocumentDB did not reach Ready on newVersion=%s", newVersion) + Eventually(assertions.AssertInstanceCount(ctx, c, key, instances), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Succeed(), "cluster did not return to %d ready instances after binary roll", instances) + + By("verifying two-phase mode kept the schema at the old version across the HA roll") + Consistently(schemaVersion, + 30*time.Second, 5*time.Second, + ).Should(Equal(oldVersion), + "schema must remain at %s until spec.schemaVersion is set (two-phase)", oldVersion) + + By("finalizing the schema by setting spec.schemaVersion to the new version") + fresh2, err := shareddb.Get(ctx, c, key) + Expect(err).NotTo(HaveOccurred(), "re-fetch DocumentDB before schema finalize") + Expect(shareddb.PatchSpec(ctx, c, fresh2, func(s *previewv1.DocumentDBSpec) { + s.SchemaVersion = newVersion + })).To(Succeed(), "patch DocumentDB schemaVersion to %s", newVersion) + + By("waiting for status.schemaVersion (primary) to advance to the new version") + Eventually(schemaVersion, + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Equal(newVersion), "schema did not migrate to %s after finalize", newVersion) + + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Succeed(), "DocumentDB not Ready after schema migration to %s", newVersion) + + By("verifying the migrated schema propagated to a replica via streaming replication") + // The operator only runs ALTER EXTENSION on the primary and only + // reads status.schemaVersion from the primary. Confirm the catalog + // change reached a replica by reading pg_extension.extversion there. + Eventually(func() (string, error) { + return replicaInstalledSchemaVersion(ctx, env, ns, ddName) + }, timeouts.For(timeouts.DocumentDBUpgrade), timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Equal(newVersion), "replica schema did not converge to %s after finalize", newVersion) + + By("verifying seeded data survived the HA schema migration") + handle2, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "reconnect to DocumentDB gateway after HA schema migration") + DeferCleanup(func(ctx SpecContext) { _ = handle2.Close(ctx) }) + n, err := sharedmongo.Count(ctx, handle2.Client(), dbName, collName, bson.M{}) + Expect(err).NotTo(HaveOccurred(), "count %s.%s after HA schema migration", dbName, collName) + Expect(n).To(Equal(int64(seed.SmallDatasetSize)), + "seeded document count changed across HA schema migration") + }) + }) diff --git a/test/e2e/tests/upgrade/upgrade_schema_test.go b/test/e2e/tests/upgrade/upgrade_schema_test.go index a1d6dcf17..187d05696 100644 --- a/test/e2e/tests/upgrade/upgrade_schema_test.go +++ b/test/e2e/tests/upgrade/upgrade_schema_test.go @@ -69,8 +69,14 @@ var _ = Describe("DocumentDB upgrade — schema", newVersion string ctx context.Context cancel context.CancelFunc + ns string + key types.NamespacedName ) + // The cluster is created once in BeforeAll and shared by the ordered + // specs below: the happy-path two-phase upgrade, then the + // unpullable-version recovery check. Both run against the same + // DocumentDB so we don't pay for a second cluster. BeforeAll(func() { skipUnlessUpgradeEnabled() oldVersion = envOr(envOldDocumentDBVersion, defaultOldDocumentDBVersion) @@ -78,24 +84,18 @@ var _ = Describe("DocumentDB upgrade — schema", if oldVersion == newVersion { Skip(envOldDocumentDBVersion + " and " + envNewDocumentDBVersion + " are identical; nothing to upgrade") } - }) - - BeforeEach(func() { - e2e.SkipUnlessLevel(e2e.High) - ctx, cancel = context.WithTimeout(context.Background(), imageRolloutTimeout) - DeferCleanup(func() { cancel() }) - }) - It("keeps the schema at the old version until finalized, then migrates and retains data", func() { env := e2e.SuiteEnv() Expect(env).NotTo(BeNil(), "SuiteEnv must be initialized by SetupSuite") - Expect(ctx).NotTo(BeNil(), "BeforeEach must have populated the spec context") c := env.Client + setupCtx, setupCancel := context.WithTimeout(context.Background(), imageRolloutTimeout) + DeferCleanup(func() { setupCancel() }) + By("creating a DocumentDB pinned to the old version (schemaVersion unset → two-phase)") - ns := namespaces.NamespaceForSpec(e2e.UpgradeLabel) - createNamespace(ctx, c, ns) - createCredentialSecret(ctx, c, ns) + ns = namespaces.NamespaceForSpec(e2e.UpgradeLabel) + createNamespace(setupCtx, c, ns) + createCredentialSecret(setupCtx, c, ns) vars := baseVars(ddName, ns, "2Gi") // Drive the version via documentDBVersion, not raw images: the @@ -105,7 +105,7 @@ var _ = Describe("DocumentDB upgrade — schema", vars["GATEWAY_IMAGE"] = "" vars["DOCUMENTDB_VERSION"] = oldVersion - dd, err := documentdb.Create(ctx, c, ns, ddName, documentdb.CreateOptions{ + dd, err := documentdb.Create(setupCtx, c, ns, ddName, documentdb.CreateOptions{ Base: "documentdb", Mixins: []string{"documentdb_version"}, Vars: vars, @@ -116,11 +116,24 @@ var _ = Describe("DocumentDB upgrade — schema", _ = shareddb.Delete(ctx, c, dd, 3*time.Minute) }) - key := types.NamespacedName{Namespace: ns, Name: ddName} - Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + key = types.NamespacedName{Namespace: ns, Name: ddName} + Eventually(assertions.AssertDocumentDBReady(setupCtx, c, key), timeouts.For(timeouts.DocumentDBReady), timeouts.PollInterval(timeouts.DocumentDBReady), ).Should(Succeed(), "DocumentDB did not reach Ready on oldVersion=%s", oldVersion) + }) + + BeforeEach(func() { + e2e.SkipUnlessLevel(e2e.High) + ctx, cancel = context.WithTimeout(context.Background(), imageRolloutTimeout) + DeferCleanup(func() { cancel() }) + }) + + It("keeps the schema at the old version until finalized, then migrates and retains data", func() { + env := e2e.SuiteEnv() + Expect(env).NotTo(BeNil(), "SuiteEnv must be initialized by SetupSuite") + Expect(ctx).NotTo(BeNil(), "BeforeEach must have populated the spec context") + c := env.Client // Single schema-version poller reused across every Eventually/ // Consistently below: it caches the last good read so a transient @@ -210,6 +223,78 @@ var _ = Describe("DocumentDB upgrade — schema", Expect(n2).To(Equal(int64(seed.SmallDatasetSize)), "seeded document count changed across schema migration") }) + + It("keeps serving on the current version and recovers when patched to an unpullable version", func() { + env := e2e.SuiteEnv() + Expect(env).NotTo(BeNil(), "SuiteEnv must be initialized by SetupSuite") + Expect(ctx).NotTo(BeNil(), "BeforeEach must have populated the spec context") + c := env.Client + + // Runs after the happy-path spec (Ordered), so the shared cluster + // is at newVersion with the schema finalized. We deliberately use a + // version ABOVE the installed schema (0.999.0) so the image-rollback + // webhook admits the patch — a version below the installed schema + // would instead be rejected at admission (a different guard). The + // bogus tag has no published image, so its pods can never pull. + const bogusVersion = "0.999.0" + + schemaVersion := schemaVersionGetter(ctx, c, key) + + By("confirming the cluster starts this spec Ready at the new version") + Eventually(schemaVersion, + timeouts.For(timeouts.DocumentDBReady), + timeouts.PollInterval(timeouts.DocumentDBReady), + ).Should(Equal(newVersion), "expected shared cluster to be at schema %s", newVersion) + + By("patching spec.documentDBVersion to an unpullable version") + fresh, err := shareddb.Get(ctx, c, key) + Expect(err).NotTo(HaveOccurred(), "re-fetch DocumentDB before bad-version patch") + Expect(shareddb.PatchSpec(ctx, c, fresh, func(s *previewv1.DocumentDBSpec) { + s.DocumentDBVersion = bogusVersion + })).To(Succeed(), "patch DocumentDBVersion to unpullable %s", bogusVersion) + + By("verifying the installed schema does not change while the new image cannot be pulled") + // The operator cannot run ALTER EXTENSION against a binary that + // never starts, so the installed schema must stay at newVersion. + Consistently(schemaVersion, + 30*time.Second, 5*time.Second, + ).Should(Equal(newVersion), + "schema must not advance past %s when the target image is unpullable", newVersion) + + By("verifying seeded data is still readable despite the failed image pull") + handle, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "connect to DocumentDB gateway during failed upgrade") + n, err := sharedmongo.Count(ctx, handle.Client(), dbName, collName, bson.M{}) + Expect(err).NotTo(HaveOccurred(), "count %s.%s during failed upgrade", dbName, collName) + Expect(n).To(Equal(int64(seed.SmallDatasetSize)), + "seeded document count changed while the upgrade was failing") + Expect(handle.Close(ctx)).To(Succeed()) + + By("rolling the version back to the valid new version and recovering") + fresh2, err := shareddb.Get(ctx, c, key) + Expect(err).NotTo(HaveOccurred(), "re-fetch DocumentDB before recovery patch") + Expect(shareddb.PatchSpec(ctx, c, fresh2, func(s *previewv1.DocumentDBSpec) { + s.DocumentDBVersion = newVersion + })).To(Succeed(), "restore DocumentDBVersion to %s", newVersion) + + Eventually(assertions.AssertDocumentDBReady(ctx, c, key), + timeouts.For(timeouts.DocumentDBUpgrade), + timeouts.PollInterval(timeouts.DocumentDBUpgrade), + ).Should(Succeed(), "DocumentDB did not recover to Ready on %s after the bad-version patch", newVersion) + + By("verifying the schema is still the new version and data survived after recovery") + Consistently(schemaVersion, + 15*time.Second, 5*time.Second, + ).Should(Equal(newVersion), "schema should remain %s after recovery", newVersion) + + handle2, err := e2emongo.NewFromDocumentDB(ctx, env, ns, ddName) + Expect(err).NotTo(HaveOccurred(), "reconnect to DocumentDB gateway after recovery") + DeferCleanup(func(ctx SpecContext) { _ = handle2.Close(ctx) }) + n2, err := sharedmongo.Count(ctx, handle2.Client(), dbName, collName, bson.M{}) + Expect(err).NotTo(HaveOccurred(), "count %s.%s after recovery", dbName, collName) + Expect(n2).To(Equal(int64(seed.SmallDatasetSize)), + "seeded document count changed across the failed-upgrade/recovery cycle") + }) }) // schemaVersionGetter returns a poll function reporting the DocumentDB's From 7dd9c9a4a9d3540682778269aff7c2db65471598 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Thu, 13 Aug 2026 13:22:21 -0400 Subject: [PATCH 3/3] test(e2e): verify schema version on all replicas, not just the first The HA schema-upgrade spec's replica check only exec'd psql on the first replica pod, so a lagging or not-yet-rolled second replica could pass unnoticed. replicaInstalledSchemaVersion now takes an expected replica count, requires exactly that many replica pods to be present, reads pg_extension.extversion on every one, and only returns a version when all replicas agree (erroring otherwise so Eventually keeps polling until the whole cluster converges). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d53b8fec-0585-4888-89da-d9a6847597a5 Signed-off-by: Wenting Wu --- test/e2e/tests/upgrade/helpers_test.go | 53 +++++++++++++++---- .../tests/upgrade/upgrade_schema_ha_test.go | 11 ++-- 2 files changed, 49 insertions(+), 15 deletions(-) diff --git a/test/e2e/tests/upgrade/helpers_test.go b/test/e2e/tests/upgrade/helpers_test.go index 8093ec8d7..c399adcc5 100644 --- a/test/e2e/tests/upgrade/helpers_test.go +++ b/test/e2e/tests/upgrade/helpers_test.go @@ -190,26 +190,33 @@ func createCredentialSecret(ctx context.Context, c client.Client, ns string) { } } -// replicaInstalledSchemaVersion execs psql on a replica pod of the CNPG -// cluster backing the DocumentDB and returns the installed documentdb -// extension version, normalized to semver (e.g. "0.110.0"). +// replicaInstalledSchemaVersion execs psql on every replica pod of the +// CNPG cluster backing the DocumentDB and returns their agreed installed +// documentdb extension version, normalized to semver (e.g. "0.110.0"). // // This exists because the operator computes status.schemaVersion by // querying the PRIMARY only (see executeSQLCommand in the controller), // so the CR status does not independently prove that a schema migration // propagated to replicas. The extension schema (an ALTER EXTENSION // catalog change) reaches replicas via WAL streaming replication; this -// helper reads pg_extension.extversion directly on a replica to confirm -// that convergence. +// helper reads pg_extension.extversion directly on each replica to +// confirm that convergence across all of them. // // The extension reports its version in "Major.Minor-Patch" form (e.g. // "0.110-0"); replacing the final "-" with "." yields the semver used // throughout the upgrade specs. clusterName is the CNPG cluster name, // which for a single-cluster DocumentDB equals the DocumentDB name. +// +// wantReplicas is the number of replica pods the caller expects (instances +// minus the primary). The helper errors until exactly that many replicas +// are present AND they all report the same installed version, so a lagging +// or not-yet-rolled replica keeps an Eventually polling rather than passing +// on the first replica alone. func replicaInstalledSchemaVersion( ctx context.Context, env *environment.TestingEnvironment, ns, clusterName string, + wantReplicas int, ) (string, error) { var pods corev1.PodList if err := env.Client.List(ctx, &pods, @@ -221,21 +228,47 @@ func replicaInstalledSchemaVersion( ); err != nil { return "", fmt.Errorf("list replica pods for cluster %s/%s: %w", ns, clusterName, err) } - if len(pods.Items) == 0 { - return "", fmt.Errorf("no replica pods found for cluster %s/%s", ns, clusterName) + if len(pods.Items) != wantReplicas { + return "", fmt.Errorf("expected %d replica pods for cluster %s/%s, found %d", + wantReplicas, ns, clusterName, len(pods.Items)) + } + + var agreed string + for i := range pods.Items { + pod := pods.Items[i] + v, err := podInstalledSchemaVersion(ctx, env, pod) + if err != nil { + return "", err + } + switch { + case agreed == "": + agreed = v + case agreed != v: + return "", fmt.Errorf("replicas disagree on installed schema version: %s vs %s (%s)", + agreed, v, pod.Name) + } } + return agreed, nil +} +// podInstalledSchemaVersion execs psql on a single pod and returns the +// installed documentdb extension version normalized to semver. +func podInstalledSchemaVersion( + ctx context.Context, + env *environment.TestingEnvironment, + pod corev1.Pod, +) (string, error) { timeout := time.Minute - stdout, stderr, err := env.EventuallyExecCommand(ctx, pods.Items[0], "postgres", &timeout, + stdout, stderr, err := env.EventuallyExecCommand(ctx, pod, "postgres", &timeout, "psql", "-U", "postgres", "-d", "postgres", "-tAc", "SELECT extversion FROM pg_extension WHERE extname='documentdb'") if err != nil { - return "", fmt.Errorf("exec psql on replica %s: %w (stderr: %s)", pods.Items[0].Name, err, stderr) + return "", fmt.Errorf("exec psql on pod %s: %w (stderr: %s)", pod.Name, err, stderr) } raw := strings.TrimSpace(stdout) if raw == "" { - return "", fmt.Errorf("documentdb extension not installed on replica %s", pods.Items[0].Name) + return "", fmt.Errorf("documentdb extension not installed on pod %s", pod.Name) } // "0.110-0" -> "0.110.0"; a value already in semver form is unchanged. if i := strings.LastIndex(raw, "-"); i >= 0 { diff --git a/test/e2e/tests/upgrade/upgrade_schema_ha_test.go b/test/e2e/tests/upgrade/upgrade_schema_ha_test.go index f54418a4a..715c7d319 100644 --- a/test/e2e/tests/upgrade/upgrade_schema_ha_test.go +++ b/test/e2e/tests/upgrade/upgrade_schema_ha_test.go @@ -121,9 +121,9 @@ var _ = Describe("DocumentDB upgrade — schema (multi-instance HA)", timeouts.PollInterval(timeouts.DocumentDBReady), ).Should(Equal(oldVersion), "initial schema version should be %s", oldVersion) - By("confirming a replica also reports the old schema version before upgrade") + By("confirming all replicas also report the old schema version before upgrade") Eventually(func() (string, error) { - return replicaInstalledSchemaVersion(ctx, env, ns, ddName) + return replicaInstalledSchemaVersion(ctx, env, ns, ddName, instances-1) }, timeouts.For(timeouts.DocumentDBReady), timeouts.PollInterval(timeouts.DocumentDBReady), ).Should(Equal(oldVersion), "replica should start at schema %s", oldVersion) @@ -182,12 +182,13 @@ var _ = Describe("DocumentDB upgrade — schema (multi-instance HA)", timeouts.PollInterval(timeouts.DocumentDBUpgrade), ).Should(Succeed(), "DocumentDB not Ready after schema migration to %s", newVersion) - By("verifying the migrated schema propagated to a replica via streaming replication") + By("verifying the migrated schema propagated to all replicas via streaming replication") // The operator only runs ALTER EXTENSION on the primary and only // reads status.schemaVersion from the primary. Confirm the catalog - // change reached a replica by reading pg_extension.extversion there. + // change reached every replica by reading pg_extension.extversion + // on each of them. Eventually(func() (string, error) { - return replicaInstalledSchemaVersion(ctx, env, ns, ddName) + return replicaInstalledSchemaVersion(ctx, env, ns, ddName, instances-1) }, timeouts.For(timeouts.DocumentDBUpgrade), timeouts.PollInterval(timeouts.DocumentDBUpgrade), ).Should(Equal(newVersion), "replica schema did not converge to %s after finalize", newVersion)