Skip to content

Commit dfd8518

Browse files
armrugbartolini
armru
and
gbartolini
authored
fix(replica clusters): apply hot-standby parameter changes (#9952)
In replica clusters, decreasing hot-standby sensitive parameters (e.g., max_connections) never triggered a restart because: 1. pg_controldata values come from the external source primary via WAL, not from the local cluster's designated primary, making the pg_controldata comparison permanently unsatisfiable for replicas. 2. The designated primary was not recognized as "primary-like" for restart purposes since IsPrimary is always false in replica clusters. Skip the pg_controldata check for replica clusters and treat the designated primary as primary-like for in-place restart triggering. Closes #9855 Signed-off-by: Armando Ruocco <[email protected]> Signed-off-by: Gabriele Bartolini <[email protected]> Co-authored-by: Gabriele Bartolini <[email protected]>
1 parent 9e82398 commit dfd8518

4 files changed

Lines changed: 221 additions & 7 deletions

File tree

‎internal/management/controller/instance_controller.go‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1034,12 +1034,22 @@ func (r *InstanceReconciler) processConfigReloadAndManageRestart(ctx context.Con
10341034
return nil
10351035
}
10361036

1037-
// if there is a pending restart, the instance is a primary and
1038-
// the restart is due to a decrease of sensible parameters,
1039-
// we will need to restart the primary instance in place
1037+
// if there is a pending restart, the instance is a primary (or the designated
1038+
// primary in a replica cluster) and the restart is due to a decrease of sensible
1039+
// parameters, we will need to restart the primary instance in place
10401040
phase := apiv1.PhaseApplyingConfiguration
10411041
phaseReason := "PostgreSQL configuration changed"
1042-
if status.IsPrimary && status.PendingRestartForDecrease {
1042+
1043+
// In a replica cluster, the designated primary acts as the local primary.
1044+
// It should handle hot-standby sensitive parameter decreases the same way
1045+
// a real primary would, triggering an in-place restart.
1046+
isDesignatedPrimary := cluster.IsReplica() &&
1047+
cluster.Status.CurrentPrimary == r.instance.GetPodName()
1048+
// Determine the local acting primary instance in cluster, which is either
1049+
// the real primary or a designated primary
1050+
isActingPrimary := status.IsPrimary || isDesignatedPrimary
1051+
1052+
if isActingPrimary && status.PendingRestartForDecrease {
10431053
if cluster.GetPrimaryUpdateStrategy() == apiv1.PrimaryUpdateStrategyUnsupervised {
10441054
return r.triggerRestartForDecrease(ctx, cluster)
10451055
}
@@ -1052,7 +1062,7 @@ func (r *InstanceReconciler) processConfigReloadAndManageRestart(ctx context.Con
10521062
}
10531063
if phase == apiv1.PhaseApplyingConfiguration &&
10541064
(cluster.Status.Phase == apiv1.PhaseApplyingConfiguration ||
1055-
(status.IsPrimary && cluster.Spec.Instances > 1)) {
1065+
(isActingPrimary && cluster.Spec.Instances > 1)) {
10561066
// I'm not the first instance spotting the configuration
10571067
// change, everything is fine and there is no need to signal
10581068
// the operator again.

‎pkg/management/postgres/probes.go‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -149,8 +149,19 @@ func updateResultForDecrease(
149149
// mark the pending restart as due to a decrease
150150
result.PendingRestartForDecrease = true
151151
if !result.IsPrimary {
152-
// in case of hot standby parameters being decreased,
153-
// followers need to wait for the new value to be present in the PGDATA before being restarted.
152+
// In a replica cluster, pg_controldata values are received from the external
153+
// source primary through the WAL stream, not from this cluster's designated
154+
// primary. Comparing decreased parameter values against pg_controldata is
155+
// incorrect for replica clusters because those values reflect the source
156+
// primary's configuration, not the local cluster's. We skip this check and
157+
// keep PendingRestart as-is, allowing the restart to proceed.
158+
cluster := instance.GetClusterOrDefault()
159+
if cluster.IsReplica() {
160+
return nil
161+
}
162+
163+
// In non-replica clusters, followers need to wait for the new value to be
164+
// present in the PGDATA before being restarted.
154165
pgControldataParams, err := LoadEnforcedParametersFromPgControldata(instance.PgData)
155166
if err != nil {
156167
return err

‎pkg/management/postgres/probes_test.go‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,136 @@ import (
2525

2626
"github.com/DATA-DOG/go-sqlmock"
2727
"github.com/blang/semver"
28+
"k8s.io/utils/ptr"
2829

30+
apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
2931
"github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
3032

3133
. "github.com/onsi/ginkgo/v2"
3234
. "github.com/onsi/gomega"
3335
)
3436

37+
var _ = Describe("areAllParamsUpdated", func() {
38+
It("should return true when all params match", func() {
39+
decreased := map[string]int{"max_connections": 90, "max_worker_processes": 4}
40+
controldata := map[string]int{"max_connections": 90, "max_worker_processes": 4}
41+
Expect(areAllParamsUpdated(decreased, controldata)).To(BeTrue())
42+
})
43+
44+
It("should return false when a param doesn't match", func() {
45+
decreased := map[string]int{"max_connections": 90}
46+
controldata := map[string]int{"max_connections": 100}
47+
Expect(areAllParamsUpdated(decreased, controldata)).To(BeFalse())
48+
})
49+
50+
It("should return false when a param is missing from controldata", func() {
51+
decreased := map[string]int{"max_connections": 90}
52+
controldata := map[string]int{}
53+
Expect(areAllParamsUpdated(decreased, controldata)).To(BeFalse())
54+
})
55+
56+
It("should return true for empty decreased values", func() {
57+
decreased := map[string]int{}
58+
controldata := map[string]int{"max_connections": 100}
59+
Expect(areAllParamsUpdated(decreased, controldata)).To(BeTrue())
60+
})
61+
})
62+
63+
var _ = Describe("updateResultForDecrease", func() {
64+
// decreasedSettingsQuery matches the SQL used by GetDecreasedSensibleSettings
65+
decreasedSettingsQuery := regexp.QuoteMeta(
66+
`SELECT pending_settings.name, CAST(coalesce(new_setting,default_setting) AS INTEGER) as new_setting`)
67+
68+
Context("when there are no decreased standby-sensitive settings", func() {
69+
It("should not modify the result", func() {
70+
db, mock, err := sqlmock.New()
71+
Expect(err).ToNot(HaveOccurred())
72+
73+
mock.ExpectQuery(decreasedSettingsQuery).
74+
WillReturnRows(sqlmock.NewRows([]string{"name", "new_setting"}))
75+
76+
instance := &Instance{}
77+
result := &postgres.PostgresqlStatus{
78+
IsPrimary: false,
79+
PendingRestart: true,
80+
}
81+
82+
err = updateResultForDecrease(instance, db, result)
83+
Expect(err).ToNot(HaveOccurred())
84+
Expect(result.PendingRestart).To(BeTrue())
85+
Expect(result.PendingRestartForDecrease).To(BeFalse())
86+
Expect(mock.ExpectationsWereMet()).To(Succeed())
87+
})
88+
})
89+
90+
Context("when this is a primary instance", func() {
91+
It("should keep PendingRestart true and set PendingRestartForDecrease", func() {
92+
db, mock, err := sqlmock.New()
93+
Expect(err).ToNot(HaveOccurred())
94+
95+
mock.ExpectQuery(decreasedSettingsQuery).
96+
WillReturnRows(sqlmock.NewRows([]string{"name", "new_setting"}).
97+
AddRow("max_connections", 90))
98+
99+
instance := &Instance{}
100+
result := &postgres.PostgresqlStatus{
101+
IsPrimary: true,
102+
PendingRestart: true,
103+
}
104+
105+
err = updateResultForDecrease(instance, db, result)
106+
Expect(err).ToNot(HaveOccurred())
107+
Expect(result.PendingRestart).To(BeTrue())
108+
Expect(result.PendingRestartForDecrease).To(BeTrue())
109+
Expect(mock.ExpectationsWereMet()).To(Succeed())
110+
})
111+
})
112+
113+
Context("when this is a replica cluster instance", func() {
114+
DescribeTable("should keep PendingRestart true regardless of pod role",
115+
func(podName string, currentPrimary string) {
116+
db, mock, err := sqlmock.New()
117+
Expect(err).ToNot(HaveOccurred())
118+
119+
mock.ExpectQuery(decreasedSettingsQuery).
120+
WillReturnRows(sqlmock.NewRows([]string{"name", "new_setting"}).
121+
AddRow("max_connections", 95))
122+
123+
instance := (&Instance{}).WithPodName(podName)
124+
instance.SetCluster(&apiv1.Cluster{
125+
Spec: apiv1.ClusterSpec{
126+
ReplicaCluster: &apiv1.ReplicaClusterConfiguration{
127+
Enabled: ptr.To(true),
128+
Source: "cluster-example",
129+
},
130+
},
131+
Status: apiv1.ClusterStatus{
132+
CurrentPrimary: currentPrimary,
133+
},
134+
})
135+
136+
result := &postgres.PostgresqlStatus{
137+
IsPrimary: false,
138+
PendingRestart: true,
139+
}
140+
141+
err = updateResultForDecrease(instance, db, result)
142+
Expect(err).ToNot(HaveOccurred())
143+
// In a replica cluster, PendingRestart should remain true
144+
// because pg_controldata values come from the external source
145+
// primary, not from this cluster
146+
Expect(result.PendingRestart).To(BeTrue())
147+
Expect(result.PendingRestartForDecrease).To(BeTrue())
148+
Expect(mock.ExpectationsWereMet()).To(Succeed())
149+
},
150+
Entry("designated primary",
151+
"cluster-replica-tls-1", "cluster-replica-tls-1"),
152+
Entry("non-designated-primary standby",
153+
"cluster-replica-tls-2", "cluster-replica-tls-1"),
154+
)
155+
})
156+
})
157+
35158
var _ = Describe("probes", func() {
36159
It("fillWalStatus should properly handle errors", func() {
37160
instance := &Instance{}

‎tests/e2e/replica_mode_cluster_test.go‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import (
3131
corev1 "k8s.io/api/core/v1"
3232
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3333
"k8s.io/apimachinery/pkg/types"
34+
"k8s.io/client-go/util/retry"
3435
"k8s.io/utils/ptr"
3536
k8client "sigs.k8s.io/controller-runtime/pkg/client"
3637

@@ -105,6 +106,75 @@ var _ = Describe("Replica Mode", Label(tests.LabelReplication), func() {
105106
AssertSwitchoverOnReplica(replicaNamespace, replicaName, env)
106107

107108
assertReplicaClusterTopology(replicaNamespace, replicaName)
109+
110+
By("increasing max_connections to 120 on the replica cluster", func() {
111+
err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
112+
cluster, err := clusterutils.Get(env.Ctx, env.Client, replicaNamespace, replicaName)
113+
if err != nil {
114+
return err
115+
}
116+
if cluster.Spec.PostgresConfiguration.Parameters == nil {
117+
cluster.Spec.PostgresConfiguration.Parameters = map[string]string{}
118+
}
119+
cluster.Spec.PostgresConfiguration.Parameters["max_connections"] = "120"
120+
return env.Client.Update(env.Ctx, cluster)
121+
})
122+
Expect(err).ToNot(HaveOccurred())
123+
})
124+
125+
By("waiting for the replica cluster to apply the increased max_connections", func() {
126+
AssertClusterEventuallyReachesPhase(replicaNamespace, replicaName,
127+
[]string{
128+
apiv1.PhaseApplyingConfiguration, apiv1.PhaseUpgrade,
129+
apiv1.PhaseWaitingForInstancesToBeActive,
130+
}, 30)
131+
AssertClusterIsReady(replicaNamespace, replicaName,
132+
testTimeouts[timeouts.ClusterIsReadyQuick], env)
133+
})
134+
135+
By("verifying max_connections is 120 on all pods", func() {
136+
podList, err := clusterutils.ListPods(env.Ctx, env.Client, replicaNamespace, replicaName)
137+
Expect(err).ToNot(HaveOccurred())
138+
for idx := range podList.Items {
139+
pod := &podList.Items[idx]
140+
Eventually(QueryMatchExpectationPredicate(
141+
pod, postgres.PostgresDBName, "SHOW max_connections", "120"),
142+
30).Should(Succeed())
143+
}
144+
})
145+
146+
By("decreasing max_connections to 110 on the replica cluster", func() {
147+
err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
148+
cluster, err := clusterutils.Get(env.Ctx, env.Client, replicaNamespace, replicaName)
149+
if err != nil {
150+
return err
151+
}
152+
cluster.Spec.PostgresConfiguration.Parameters["max_connections"] = "110"
153+
return env.Client.Update(env.Ctx, cluster)
154+
})
155+
Expect(err).ToNot(HaveOccurred())
156+
})
157+
158+
By("waiting for the replica cluster to apply the decreased max_connections", func() {
159+
AssertClusterEventuallyReachesPhase(replicaNamespace, replicaName,
160+
[]string{
161+
apiv1.PhaseApplyingConfiguration, apiv1.PhaseUpgrade,
162+
apiv1.PhaseWaitingForInstancesToBeActive,
163+
}, 30)
164+
AssertClusterIsReady(replicaNamespace, replicaName,
165+
testTimeouts[timeouts.ClusterIsReadyQuick], env)
166+
})
167+
168+
By("verifying max_connections is 110 on all pods", func() {
169+
podList, err := clusterutils.ListPods(env.Ctx, env.Client, replicaNamespace, replicaName)
170+
Expect(err).ToNot(HaveOccurred())
171+
for idx := range podList.Items {
172+
pod := &podList.Items[idx]
173+
Eventually(QueryMatchExpectationPredicate(
174+
pod, postgres.PostgresDBName, "SHOW max_connections", "110"),
175+
30).Should(Succeed())
176+
}
177+
})
108178
})
109179
})
110180

0 commit comments

Comments
 (0)