test(e2e): schema-upgrade coverage — auto, rollback, HA, unpullable version

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}

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
+}

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")
+		})
+	})