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/helpers_test.go b/test/e2e/tests/upgrade/helpers_test.go index 7cd43cb47..c399adcc5 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,90 @@ func createCredentialSecret(ctx context.Context, c client.Client, ns string) { Fail("create credential secret " + ns + "/" + credentialSecretName + ": " + err.Error()) } } + +// 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 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, + 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) != 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, 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 pod %s: %w (stderr: %s)", pod.Name, err, stderr) + } + + raw := strings.TrimSpace(stdout) + if raw == "" { + 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 { + raw = raw[:i] + "." + raw[i+1:] + } + return raw, nil +} 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_ha_test.go b/test/e2e/tests/upgrade/upgrade_schema_ha_test.go new file mode 100644 index 000000000..715c7d319 --- /dev/null +++ b/test/e2e/tests/upgrade/upgrade_schema_ha_test.go @@ -0,0 +1,204 @@ +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 all replicas also report the old schema version before upgrade") + Eventually(func() (string, error) { + 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) + + 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 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 every replica by reading pg_extension.extversion + // on each of them. + Eventually(func() (string, error) { + 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) + + 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_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") + }) + }) 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