Skip to content

feat: fail-fast preflight for schema upgrade path (multi-minor jumps) #444

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
WentingWu666666 wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat: fail-fast preflight for schema upgrade path (multi-minor jumps) #444

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 69 additions & 0 deletions operator/src/internal/controller/documentdb_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -769,6 +769,46 @@ func parseExtensionVersionsFromOutput(output string) (defaultVersion, installedV
return defaultVersion, installedVersion, true
}

// updatePathSentinelYes / updatePathSentinelNo are the tokens the preflight
// query emits so the result can be parsed out of psql's aligned table output
// without depending on column layout.
const (
updatePathSentinelYes = "HAS_UPDATE_PATH"
updatePathSentinelNo = "NO_UPDATE_PATH"
)

// extensionUpdatePathExists reports whether PostgreSQL can resolve an
// ALTER EXTENSION update path for the documentdb extension from source to
// target. Both versions are in pg_available_extensions format (e.g.
// "0.109-0"). It queries pg_extension_update_paths, which exposes the same
// version graph ALTER EXTENSION UPDATE walks internally: a row exists for a
// resolvable (source, target) pair with a non-NULL path, and the path is NULL
// when no chain of update scripts connects them.
//
// The query returns a unique sentinel token rather than the raw path so the
// answer survives psql's aligned output formatting.
func (r *DocumentDBReconciler) extensionUpdatePathExists(
ctx context.Context,
cluster *cnpgv1.Cluster,
source, target string,
) (bool, error) {
// A no-op update (source == target) has no path row but is trivially safe.
if source == target {
return true, nil
}

query := fmt.Sprintf(
"SELECT CASE WHEN EXISTS (SELECT 1 FROM pg_extension_update_paths('documentdb') "+
"WHERE source = '%s' AND target = '%s' AND path IS NOT NULL) THEN '%s' ELSE '%s' END",
source, target, updatePathSentinelYes, updatePathSentinelNo)

output, err := r.SQLExecutor(ctx, cluster, query)
if err != nil {
return false, fmt.Errorf("query pg_extension_update_paths: %w", err)
}
return strings.Contains(output, updatePathSentinelYes), nil
}

// handleExtensionUpgrade handles the ALTER EXTENSION lifecycle after images have been synced
// by SyncCnpgCluster. It:
// 1. Updates DocumentDB status with the current images from the CNPG cluster
Expand Down Expand Up @@ -866,6 +906,35 @@ func (r *DocumentDBReconciler) handleExtensionUpgrade(ctx context.Context, curre
return nil
}

// Preflight: confirm PostgreSQL can resolve an update-script path from the
// installed schema to the target before attempting ALTER EXTENSION. The
// documentdb extension ships minors frequently, so a user may jump more
// than one minor in a single bump; that only works if the extension
// packages the intermediate documentdb--A--B.sql scripts so PostgreSQL can
// chain them. When the chain is broken, ALTER EXTENSION UPDATE fails at
// execution time with a raw "no update path" error every reconcile. Detect
// it here and surface an actionable event instead, leaving the schema and
// data untouched.
pathExists, err := r.extensionUpdatePathExists(ctx, currentCluster, installedVersion, schemaTarget)
if err != nil {
return fmt.Errorf("failed to check extension update path from %s to %s: %w",
installedVersion, schemaTarget, err)
}
if !pathExists {
msg := fmt.Sprintf(
"No supported schema upgrade path from %s to %s. The documentdb extension does not "+
"ship the update scripts needed to bridge this gap in one step. Upgrade to an "+
"intermediate version first, or report a missing migration script for this release. "+
"ALTER EXTENSION UPDATE was skipped; the schema and data are unchanged.",
util.ExtensionVersionToSemver(installedVersion),
util.ExtensionVersionToSemver(schemaTarget))
logger.Info(msg)
if r.Recorder != nil {
r.Recorder.Event(documentdb, corev1.EventTypeWarning, "SchemaUpgradePathMissing", msg)
}
return nil
}

// Run ALTER EXTENSION to upgrade
logger.Info("Upgrading DocumentDB extension",
"fromVersion", installedVersion,
Expand Down
193 changes: 176 additions & 17 deletions operator/src/internal/controller/documentdb_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -838,29 +838,116 @@ var _ = Describe("DocumentDB Controller", func() {
Recorder: recorder,
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) {
sqlCalls = append(sqlCalls, sql)
if len(sqlCalls) == 1 {
// First call: version check — installed 0.109-0, default 0.110-0
if strings.Contains(sql, "pg_available_extensions") {
// Version check — installed 0.109-0, default 0.110-0
return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil
}
// Second call: ALTER EXTENSION
if strings.Contains(sql, "pg_extension_update_paths") {
// Preflight — a valid update path exists
return "HAS_UPDATE_PATH", nil
}
// ALTER EXTENSION
return "ALTER EXTENSION", nil
},
}

err := reconciler.handleExtensionUpgrade(ctx, cluster, documentdb)
Expect(err).ToNot(HaveOccurred())

// Verify both SQL calls were made
Expect(sqlCalls).To(HaveLen(2))
// Verify version check, update-path preflight, and ALTER were called
Expect(sqlCalls).To(HaveLen(3))
Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions"))
Expect(sqlCalls[1]).To(Equal("ALTER EXTENSION documentdb UPDATE"))
Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths"))
Expect(sqlCalls[2]).To(Equal("ALTER EXTENSION documentdb UPDATE"))

// Status should reflect the upgraded version (default version as semver)
updatedDB := &dbpreview.DocumentDB{}
Expect(fakeClient.Get(ctx, types.NamespacedName{Name: "test-documentdb", Namespace: clusterNamespace}, updatedDB)).To(Succeed())
Expect(updatedDB.Status.SchemaVersion).To(Equal("0.110.0"))
})

It("should skip ALTER EXTENSION and emit an event when no update path exists", func() {
cluster := &cnpgv1.Cluster{
ObjectMeta: metav1.ObjectMeta{
Name: clusterName,
Namespace: clusterNamespace,
},
Spec: cnpgv1.ClusterSpec{
PostgresConfiguration: cnpgv1.PostgresConfiguration{
Extensions: []cnpgv1.ExtensionConfiguration{
{
Name: "documentdb",
ImageVolumeSource: corev1.ImageVolumeSource{
Reference: "documentdb/documentdb:v1.0.0",
},
},
},
},
},
Status: cnpgv1.ClusterStatus{
CurrentPrimary: "test-cluster-1",
InstancesStatus: map[cnpgv1.PodStatus][]string{
cnpgv1.PodHealthy: {"test-cluster-1"},
},
},
}

documentdb := &dbpreview.DocumentDB{
ObjectMeta: metav1.ObjectMeta{
Name: "test-documentdb",
Namespace: clusterNamespace,
},
Spec: dbpreview.DocumentDBSpec{
SchemaVersion: "auto",
},
}

fakeClient := fake.NewClientBuilder().
WithScheme(scheme).
WithObjects(cluster, documentdb).
WithStatusSubresource(&dbpreview.DocumentDB{}).
Build()

sqlCalls := []string{}
reconciler := &DocumentDBReconciler{
Client: fakeClient,
Scheme: scheme,
Recorder: recorder,
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) {
sqlCalls = append(sqlCalls, sql)
if strings.Contains(sql, "pg_available_extensions") {
// Version check — installed 0.109-0, default 0.110-0
return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil
}
if strings.Contains(sql, "pg_extension_update_paths") {
// Preflight — no update path bridges installed → target
return "NO_UPDATE_PATH", nil
}
// ALTER EXTENSION must not be reached
return "ALTER EXTENSION", nil
},
}

err := reconciler.handleExtensionUpgrade(ctx, cluster, documentdb)
Expect(err).ToNot(HaveOccurred())

// Only version check and preflight ran — ALTER EXTENSION was skipped
Expect(sqlCalls).To(HaveLen(2))
Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions"))
Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths"))

// A Warning event should surface the missing upgrade path
Expect(recorder.Events).To(HaveLen(1))
event := <-recorder.Events
Expect(event).To(ContainSubstring("SchemaUpgradePathMissing"))

// Schema version must stay at the installed version (no premature
// bump to the target since the migration was skipped)
updatedDB := &dbpreview.DocumentDB{}
Expect(fakeClient.Get(ctx, types.NamespacedName{Name: "test-documentdb", Namespace: clusterNamespace}, updatedDB)).To(Succeed())
Expect(updatedDB.Status.SchemaVersion).ToNot(Equal("0.110.0"))
})

It("should return error when ALTER EXTENSION fails", func() {
cluster := &cnpgv1.Cluster{
ObjectMeta: metav1.ObjectMeta{
Expand Down Expand Up @@ -903,17 +990,19 @@ var _ = Describe("DocumentDB Controller", func() {
WithStatusSubresource(&dbpreview.DocumentDB{}).
Build()

callCount := 0
reconciler := &DocumentDBReconciler{
Client: fakeClient,
Scheme: scheme,
Recorder: recorder,
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) {
callCount++
if callCount == 1 {
if strings.Contains(sql, "pg_available_extensions") {
// Version check: upgrade needed
return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil
}
if strings.Contains(sql, "pg_extension_update_paths") {
// Preflight — a valid update path exists, so ALTER is attempted
return "HAS_UPDATE_PATH", nil
}
// ALTER EXTENSION fails
return "", fmt.Errorf("permission denied")
},
Expand Down Expand Up @@ -1212,18 +1301,80 @@ var _ = Describe("DocumentDB Controller", func() {
Recorder: recorder,
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) {
sqlCalls = append(sqlCalls, sql)
if len(sqlCalls) == 1 {
if strings.Contains(sql, "pg_available_extensions") {
// default > installed → triggers ALTER EXTENSION
return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil
}
if strings.Contains(sql, "pg_extension_update_paths") {
return "HAS_UPDATE_PATH", nil
}
return "ALTER EXTENSION", nil
},
}

err := reconciler.handleExtensionUpgrade(ctx, cluster, documentdb)
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("failed to update DocumentDB status after schema upgrade"))
Expect(sqlCalls).To(HaveLen(2))
Expect(sqlCalls).To(HaveLen(3))
})
})

Describe("extensionUpdatePathExists", func() {
It("returns true without querying when source equals target", func() {
called := false
reconciler := &DocumentDBReconciler{
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, _ string) (string, error) {
called = true
return "", nil
},
}

exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.110-0", "0.110-0")
Expect(err).ToNot(HaveOccurred())
Expect(exists).To(BeTrue())
Expect(called).To(BeFalse())
})

It("returns true when the preflight query reports a path", func() {
var gotSQL string
reconciler := &DocumentDBReconciler{
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) {
gotSQL = sql
return " case \n--------\n HAS_UPDATE_PATH\n", nil
},
}

exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.109-0", "0.110-0")
Expect(err).ToNot(HaveOccurred())
Expect(exists).To(BeTrue())
Expect(gotSQL).To(ContainSubstring("pg_extension_update_paths"))
Expect(gotSQL).To(ContainSubstring("source = '0.109-0'"))
Expect(gotSQL).To(ContainSubstring("target = '0.110-0'"))
})

It("returns false when the preflight query reports no path", func() {
reconciler := &DocumentDBReconciler{
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, _ string) (string, error) {
return " case \n--------\n NO_UPDATE_PATH\n", nil
},
}

exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.109-0", "0.130-0")
Expect(err).ToNot(HaveOccurred())
Expect(exists).To(BeFalse())
})

It("propagates SQL execution errors", func() {
reconciler := &DocumentDBReconciler{
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, _ string) (string, error) {
return "", fmt.Errorf("connection refused")
},
}

exists, err := reconciler.extensionUpdatePathExists(ctx, &cnpgv1.Cluster{}, "0.109-0", "0.110-0")
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("query pg_extension_update_paths"))
Expect(exists).To(BeFalse())
})
})

Expand Down Expand Up @@ -1347,9 +1498,12 @@ var _ = Describe("DocumentDB Controller", func() {
Recorder: recorder,
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) {
sqlCalls = append(sqlCalls, sql)
if len(sqlCalls) == 1 {
if strings.Contains(sql, "pg_available_extensions") {
return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil
}
if strings.Contains(sql, "pg_extension_update_paths") {
return "HAS_UPDATE_PATH", nil
}
return "ALTER EXTENSION", nil
},
}
Expand All @@ -1358,9 +1512,10 @@ var _ = Describe("DocumentDB Controller", func() {
Expect(err).ToNot(HaveOccurred())

// Both version-check and ALTER EXTENSION should have been called
Expect(sqlCalls).To(HaveLen(2))
Expect(sqlCalls).To(HaveLen(3))
Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions"))
Expect(sqlCalls[1]).To(Equal("ALTER EXTENSION documentdb UPDATE"))
Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths"))
Expect(sqlCalls[2]).To(Equal("ALTER EXTENSION documentdb UPDATE"))

// Status should reflect the upgraded version
updatedDB := &dbpreview.DocumentDB{}
Expand Down Expand Up @@ -1418,10 +1573,13 @@ var _ = Describe("DocumentDB Controller", func() {
Recorder: recorder,
SQLExecutor: func(_ context.Context, _ *cnpgv1.Cluster, sql string) (string, error) {
sqlCalls = append(sqlCalls, sql)
if len(sqlCalls) == 1 {
if strings.Contains(sql, "pg_available_extensions") {
// Binary is 0.110-0, installed is 0.109-0
return " default_version | installed_version \n-----------------+-------------------\n 0.110-0 | 0.109-0 \n", nil
}
if strings.Contains(sql, "pg_extension_update_paths") {
return "HAS_UPDATE_PATH", nil
}
return "ALTER EXTENSION", nil
},
}
Expand All @@ -1430,9 +1588,10 @@ var _ = Describe("DocumentDB Controller", func() {
Expect(err).ToNot(HaveOccurred())

// Should run ALTER EXTENSION UPDATE TO specific version
Expect(sqlCalls).To(HaveLen(2))
Expect(sqlCalls).To(HaveLen(3))
Expect(sqlCalls[0]).To(ContainSubstring("pg_available_extensions"))
Expect(sqlCalls[1]).To(Equal("ALTER EXTENSION documentdb UPDATE TO '0.110-0'"))
Expect(sqlCalls[1]).To(ContainSubstring("pg_extension_update_paths"))
Expect(sqlCalls[2]).To(Equal("ALTER EXTENSION documentdb UPDATE TO '0.110-0'"))

// Status should reflect the explicit version
updatedDB := &dbpreview.DocumentDB{}
Expand Down
Loading