Skip to content
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

Jump to bottom

Switch application_name back to default value #575

Merged
merged 16 commits into from
Aug 13, 2024
Merged

Switch application_name back to default value #575

Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
ea3d308
Double check config
eberlep Jul 12, 2024
59c5021
Logging
eberlep Jul 12, 2024
f1ca8db
Especially double check when bootstrapping
eberlep Jul 12, 2024
4ecb5f5
Check for new value
eberlep Jul 12, 2024
054ba69
Revert "Especially double check when bootstrapping"
eberlep Jul 12, 2024
5d3ca54
Especially double check when bootstrapping
eberlep Jul 12, 2024
7b86f4c
Revert "Check for new value"
eberlep Jul 12, 2024
fd79457
Fetch and use the actual postgres cluster name as application_name fo…
eberlep Jul 17, 2024
2c215f1
Always fetch application_name when needed
eberlep Jul 23, 2024
194b466
make linter happy
eberlep Jul 23, 2024
8b76805
Simplify
eberlep Jul 23, 2024
d9920c1
logging
eberlep Jul 23, 2024
827fe9f
Fix typos
eberlep Jul 24, 2024
715a61c
Merge branch 'sync_mode_reloaded_part_2' into sync_mode_reloaded_part_3
eberlep Jul 24, 2024
6575b87
Merge remote-tracking branch 'origin' into sync_mode_reloaded_part_3
eberlep Jul 24, 2024
030c0d8
Logging
eberlep Jul 25, 2024
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
70 changes: 52 additions & 18 deletions controllers/postgres_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1062,14 +1062,28 @@ func (r *PostgresReconciler) checkAndUpdatePatroniReplicationConfig(log logr.Log
return requeueAfterReconcile, nil
}
if instance.Spec.PostgresConnection.SynchronousReplication {
if resp.SynchronousNodesAdditional == nil || *resp.SynchronousNodesAdditional != instance.Spec.PostgresConnection.ConnectedPostgresID {
// fetch the sync standby to determine the correct application_name of the instance
log.V(debugLogLevel).Info("fetching the referenced sync standby")
var synchronousStandbyApplicationName *string
s := &pg.Postgres{}
ns := types.NamespacedName{
Name: instance.Spec.PostgresConnection.ConnectedPostgresID,
Namespace: instance.Namespace,
}
if err := r.CtrlClient.Get(ctx, ns, s); err != nil {
r.recorder.Eventf(s, "Warning", "Error", "failed to get referenced sync standby: %v", err)
synchronousStandbyApplicationName = nil
} else {
synchronousStandbyApplicationName = pointer.String(s.ToPeripheralResourceName())
}
if resp.SynchronousNodesAdditional == nil || *resp.SynchronousNodesAdditional != *synchronousStandbyApplicationName {
log.V(debugLogLevel).Info("synchronous_nodes_additional mismatch, updating and requeing", "response", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP, synchronousStandbyApplicationName)
}
} else {
if resp.SynchronousNodesAdditional != nil {
log.V(debugLogLevel).Info("synchronous_nodes_additional mismatch, updating and requeing", "response", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP, nil)
}
}

Expand All @@ -1078,25 +1092,25 @@ func (r *PostgresReconciler) checkAndUpdatePatroniReplicationConfig(log logr.Log
log.V(debugLogLevel).Info("standby_cluster mismatch, requeing", "response", resp)
return requeueAfterReconcile, nil
}
if resp.StandbyCluster.ApplicationName != instance.ObjectMeta.Name {
log.V(debugLogLevel).Info("application_name mismatch, updating and requeing", "response", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP)
}
if resp.SynchronousNodesAdditional != nil {
log.V(debugLogLevel).Info("synchronous_nodes_additional mismatch, updating and requeing", "response", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP)
}
if resp.StandbyCluster.CreateReplicaMethods == nil {
log.V(debugLogLevel).Info("create_replica_methods mismatch, updating and requeing", "response", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP, nil)
}
if resp.StandbyCluster.Host != instance.Spec.PostgresConnection.ConnectionIP {
log.V(debugLogLevel).Info("host mismatch, updating and requeing", "updating", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP, nil)
}
if resp.StandbyCluster.Port != int(instance.Spec.PostgresConnection.ConnectionPort) {
log.V(debugLogLevel).Info("port mismatch, updating and requeing", "updating", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP, nil)
}
if resp.StandbyCluster.ApplicationName != instance.ToPeripheralResourceName() {
log.V(debugLogLevel).Info("application_name mismatch, updating and requeing", "response", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP, nil)
}
if resp.SynchronousNodesAdditional != nil {
log.V(debugLogLevel).Info("synchronous_nodes_additional mismatch, updating and requeing", "response", resp)
return requeueAfterReconcile, r.httpPatchPatroni(log, ctx, instance, leaderIP, nil)
}
}

Expand Down Expand Up @@ -1142,7 +1156,7 @@ func (r *PostgresReconciler) updatePatroniReplicationConfigOnAllPods(log logr.Lo
for _, pod := range pods.Items {
pod := pod // pin!
podIP := pod.Status.PodIP
if err := r.httpPatchPatroni(log, ctx, instance, podIP); err != nil {
if err := r.httpPatchPatroni(log, ctx, instance, podIP, nil); err != nil {
lastErr = err
log.Info("failed to update pod")
}
Expand All @@ -1155,7 +1169,7 @@ func (r *PostgresReconciler) updatePatroniReplicationConfigOnAllPods(log logr.Lo
return nil
}

func (r *PostgresReconciler) httpPatchPatroni(log logr.Logger, ctx context.Context, instance *pg.Postgres, podIP string) error {
func (r *PostgresReconciler) httpPatchPatroni(log logr.Logger, ctx context.Context, instance *pg.Postgres, podIP string, synchronousStandbyApplicationName *string) error {
if podIP == "" {
return errors.New("podIP must not be empty")
}
Expand All @@ -1170,8 +1184,23 @@ func (r *PostgresReconciler) httpPatchPatroni(log logr.Logger, ctx context.Conte
StandbyCluster: nil,
}
if instance.Spec.PostgresConnection.SynchronousReplication {
if synchronousStandbyApplicationName == nil {
// fetch the sync standby to determine the correct application_name of the instance
log.V(debugLogLevel).Info("unexpectetly having to fetch the referenced sync standby")
s := &pg.Postgres{}
ns := types.NamespacedName{
Name: instance.Spec.PostgresConnection.ConnectedPostgresID,
Namespace: instance.Namespace,
}
if err := r.CtrlClient.Get(ctx, ns, s); err != nil {
r.recorder.Eventf(s, "Warning", "Error", "failed to get referenced sync standby: %v", err)
synchronousStandbyApplicationName = nil
} else {
synchronousStandbyApplicationName = pointer.String(s.ToPeripheralResourceName())
}
}
// enable sync replication
request.SynchronousNodesAdditional = pointer.String(instance.Spec.PostgresConnection.ConnectedPostgresID)
request.SynchronousNodesAdditional = synchronousStandbyApplicationName
} else {
// disable sync replication
request.SynchronousNodesAdditional = nil
Expand All @@ -1183,7 +1212,7 @@ func (r *PostgresReconciler) httpPatchPatroni(log logr.Logger, ctx context.Conte
CreateReplicaMethods: []string{"basebackup_fast_xlog"},
Host: instance.Spec.PostgresConnection.ConnectionIP,
Port: int(instance.Spec.PostgresConnection.ConnectionPort),
ApplicationName: instance.ObjectMeta.Name,
ApplicationName: instance.ToPeripheralResourceName(),
},
SynchronousNodesAdditional: nil,
}
Expand Down Expand Up @@ -1212,6 +1241,11 @@ func (r *PostgresReconciler) httpPatchPatroni(log logr.Logger, ctx context.Conte
}
defer resp.Body.Close()

// fake error when standbyApplicationName is required but not provided
if instance.IsReplicationPrimary() && instance.Spec.PostgresConnection.SynchronousReplication && synchronousStandbyApplicationName == nil {
return fmt.Errorf("missing application_name of synchronous standby, disable synchronous replication")
}

return nil
}

Expand Down
Loading