diff --git a/README.md b/README.md index 9e5bc886b..d80739716 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ pipelines with no access to Kubernetes API directly, promoting infrastructure as * Restore and cloning Postgres clusters on AWS, GCS and Azure * Additionally logical backups to S3 or GCS bucket can be configured * Standby cluster from S3 or GCS WAL archive +* Hibernate and wake-up of Postgres clusters (scale StatefulSet and pooler to 0 while preserving data, then restore) * Configurable for non-cloud environments * Basic credential and user management on K8s, eases application deployments * Support for custom TLS certificates diff --git a/docs/reference/cluster_manifest.md b/docs/reference/cluster_manifest.md index 63d400934..9274123c2 100644 --- a/docs/reference/cluster_manifest.md +++ b/docs/reference/cluster_manifest.md @@ -486,6 +486,27 @@ Note that `s3_wal_path` and `gs_wal_path` are mutually exclusive. from a remote primary. See the Patroni documentation [here](https://patroni.readthedocs.io/en/latest/standby_cluster.html) for more details. Optional. +## Lifecycle configuration + +Parameters to control cluster hibernate/wake-up behavior. + +* **phase** + Set to `"stopped"` to hibernate the cluster. When this field is set on a + running cluster, the operator will: + * Store the current number of instances in the status + * Scale down the StatefulSet to 0 replicas + * Scale down the connection pooler to 0 replicas + * Suspend the logical backup CronJob (if enabled) + * Set the cluster status to "Stopping", then "Stopped" + + When this field is removed from a stopped cluster, the operator will: + * Restore the number of instances from the stored value + * Scale up the StatefulSet and connection pooler + * Resume the logical backup CronJob (if it was suspended) + * Set the cluster status to "Updating", then "Running" + + This field is optional. When not set, the cluster operates normally. + ## Volume properties Those parameters are grouped under the `volume` top-level key and define the @@ -714,3 +735,21 @@ can have the following properties: * **memory** memory requests to be set as an annotation on the stream resource. Optional. + +## Status fields + +The operator reports the cluster state through the `status` sub-resource. These +fields are managed by the operator and should not be set manually. + +* **PostgresClusterStatus** + Current state of the cluster. One of: Creating, Updating, Running, + UpdateFailed, SyncFailed, CreateFailed, Invalid, Stopping, Stopped. + +* **previousNumberOfInstances** + The number of instances the cluster had before hibernation. Used to restore + the cluster to its previous size when waking up. Cleared after wake-up. + +* **previousPoolerInstances** + A map of connection pooler role to its replica count before hibernation. + The keys are "master" and "replica". Used to restore the pooler when waking + up. Cleared after wake-up. diff --git a/docs/user.md b/docs/user.md index 1c530a48c..5cca700a3 100644 --- a/docs/user.md +++ b/docs/user.md @@ -930,6 +930,107 @@ When you apply this manifest, the operator will: The process is asynchronous. You can monitor the operator logs and the state of the `postgresql` resource to follow the progress. Once the new cluster is up and running, your applications can reconnect. +## Hibernate and Wake-up a Cluster + +The operator supports hibernating a PostgreSQL cluster to save resources when it's +not needed, and waking it up again when required. This feature: + +* Scales down the PostgreSQL StatefulSet to 0 replicas (stops all pods) +* Scales down the connection pooler to 0 replicas +* Preserves the cluster configuration and data (PVCs are retained) +* Stores the previous replica counts for automatic restoration + +### Initiating Hibernate + +To hibernate a running cluster, set the `lifecycle.phase` field to `"stopped"`: + +```yaml +apiVersion: "acid.zalan.do/v1" +kind: postgresql +metadata: + name: acid-test-cluster +spec: + teamId: "test-team" + # ... other cluster parameters + numberOfInstances: 3 + lifecycle: + phase: "stopped" +``` + +When you apply this manifest, the operator will: + +* Store the current `numberOfInstances` in `status.previousNumberOfInstances` +* Store the connection pooler replica counts in `status.previousPoolerInstances` +* Set `spec.numberOfInstances` to 0 +* Scale down the StatefulSet to 0 replicas +* Scale down the connection pooler deployments to 0 replicas +* Suspend the logical backup CronJob (if enabled) +* Set `status.PostgresClusterStatus` to "Stopping", then "Stopped" + +### Waking up a Cluster + +To wake up a hibernated cluster, remove the `lifecycle.phase` field or set it to +an empty value: + +```yaml +apiVersion: "acid.zalan.do/v1" +kind: postgresql +metadata: + name: acid-test-cluster +spec: + teamId: "test-team" + # ... other cluster parameters + # lifecycle.phase is not set or is removed +``` + +When you apply this manifest, the operator will: + +* Restore `numberOfInstances` from `status.previousNumberOfInstances` +* Restore the connection pooler replica counts from `status.previousPoolerInstances` +* Resume the logical backup CronJob (if enabled) +* Scale up the StatefulSet to the previous replica count +* Scale up the connection pooler deployments to the previous replica counts +* Set `status.PostgresClusterStatus` to "Updating", then "Running" +* Clear `status.previousNumberOfInstances` and `status.previousPoolerInstances` + +### Cluster Status During Lifecycle Transitions + +| Status | Meaning | +|--------|---------| +| Running | Cluster is running normally | +| Stopping | Cluster is transitioning to stopped state (pods terminating) | +| Stopped | All pods have been terminated, cluster is hibernated | + +### Restrictions During Hibernate + +* **During Stopping**: All spec changes are blocked. You must wait for the cluster + to reach the Stopped state before making changes. + +* **During Stopped**: Spec changes are blocked unless you remove `lifecycle.phase` + to wake up the cluster. This prevents accidental modifications to a hibernated + cluster. + +### Connection Pooler Behavior + +The connection pooler is automatically scaled alongside the cluster: + +* When the cluster hibernates, the pooler is scaled to 0 replicas +* When the cluster wakes up, the pooler is restored to its previous replica count +* The previous replica counts are stored in `status.previousPoolerInstances` + +Note: If the connection pooler was already at 0 replicas before hibernate, it +will remain at 0 after wake-up. + +### Logical Backup Behavior + +The logical backup CronJob is automatically suspended during hibernate: + +* When the cluster hibernates, the backup CronJob is suspended (`.spec.suspend: true`) +* When the cluster wakes up, the backup CronJob is automatically resumed +* The backup schedule is preserved and resumes from its normal schedule + +This prevents failed backup jobs from running when the database is unavailable. + ## Setting up a standby cluster Standby cluster is a [Patroni feature](https://github.com/zalando/patroni/blob/master/docs/replica_bootstrap.rst#standby-cluster) diff --git a/manifests/postgresql.crd.yaml b/manifests/postgresql.crd.yaml index 39811824e..10e952cd2 100644 --- a/manifests/postgresql.crd.yaml +++ b/manifests/postgresql.crd.yaml @@ -3246,6 +3246,13 @@ spec: - name type: object type: array + lifecycle: + description: LifecycleSpec describes the lifecycle state of a Postgres + cluster. + properties: + phase: + type: string + type: object logicalBackupRetention: type: string logicalBackupSchedule: @@ -4197,6 +4204,14 @@ spec: properties: PostgresClusterStatus: type: string + previousNumberOfInstances: + format: int32 + type: integer + previousPoolerInstances: + type: object + additionalProperties: + format: int32 + type: integer required: - PostgresClusterStatus type: object diff --git a/pkg/apis/acid.zalan.do/v1/const.go b/pkg/apis/acid.zalan.do/v1/const.go index 4102ea3d3..69012427a 100644 --- a/pkg/apis/acid.zalan.do/v1/const.go +++ b/pkg/apis/acid.zalan.do/v1/const.go @@ -9,6 +9,8 @@ const ( ClusterStatusSyncFailed = "SyncFailed" ClusterStatusAddFailed = "CreateFailed" ClusterStatusRunning = "Running" + ClusterStatusStopping = "Stopping" + ClusterStatusStopped = "Stopped" ClusterStatusInvalid = "Invalid" ) diff --git a/pkg/apis/acid.zalan.do/v1/postgresql.crd.yaml b/pkg/apis/acid.zalan.do/v1/postgresql.crd.yaml index 39811824e..10e952cd2 100644 --- a/pkg/apis/acid.zalan.do/v1/postgresql.crd.yaml +++ b/pkg/apis/acid.zalan.do/v1/postgresql.crd.yaml @@ -3246,6 +3246,13 @@ spec: - name type: object type: array + lifecycle: + description: LifecycleSpec describes the lifecycle state of a Postgres + cluster. + properties: + phase: + type: string + type: object logicalBackupRetention: type: string logicalBackupSchedule: @@ -4197,6 +4204,14 @@ spec: properties: PostgresClusterStatus: type: string + previousNumberOfInstances: + format: int32 + type: integer + previousPoolerInstances: + type: object + additionalProperties: + format: int32 + type: integer required: - PostgresClusterStatus type: object diff --git a/pkg/apis/acid.zalan.do/v1/postgresql_type.go b/pkg/apis/acid.zalan.do/v1/postgresql_type.go index 1dadfd06c..23c8a286a 100644 --- a/pkg/apis/acid.zalan.do/v1/postgresql_type.go +++ b/pkg/apis/acid.zalan.do/v1/postgresql_type.go @@ -115,6 +115,7 @@ type PostgresSpec struct { TLS *TLSDescription `json:"tls,omitempty"` AdditionalVolumes []AdditionalVolume `json:"additionalVolumes,omitempty"` Streams []Stream `json:"streams,omitempty"` + Lifecycle *LifecycleSpec `json:"lifecycle,omitempty"` Env []v1.EnvVar `json:"env,omitempty"` // deprecated @@ -257,6 +258,11 @@ type StandbyDescription struct { StandbyPrimarySlotName string `json:"standby_primary_slot_name,omitempty"` } +// LifecycleSpec describes the lifecycle state of a Postgres cluster. +type LifecycleSpec struct { + Phase string `json:"phase,omitempty"` +} + // TLSDescription specs TLS properties type TLSDescription struct { // +required @@ -302,7 +308,9 @@ type UserFlags []string // PostgresStatus contains status of the PostgreSQL cluster (running, creation failed etc.) type PostgresStatus struct { - PostgresClusterStatus string `json:"PostgresClusterStatus"` + PostgresClusterStatus string `json:"PostgresClusterStatus"` + PreviousNumberOfInstances int32 `json:"previousNumberOfInstances,omitempty"` + PreviousPoolerInstances map[string]int32 `json:"previousPoolerInstances,omitempty"` } // ConnectionPooler Options for connection pooler diff --git a/pkg/apis/acid.zalan.do/v1/util.go b/pkg/apis/acid.zalan.do/v1/util.go index 719defe93..7bbdc0bbf 100644 --- a/pkg/apis/acid.zalan.do/v1/util.go +++ b/pkg/apis/acid.zalan.do/v1/util.go @@ -101,6 +101,16 @@ func (postgresStatus PostgresStatus) Creating() bool { return postgresStatus.PostgresClusterStatus == ClusterStatusCreating } +// Stopping status of cluster +func (postgresStatus PostgresStatus) Stopping() bool { + return postgresStatus.PostgresClusterStatus == ClusterStatusStopping +} + +// Stopped status of cluster +func (postgresStatus PostgresStatus) Stopped() bool { + return postgresStatus.PostgresClusterStatus == ClusterStatusStopped +} + func (postgresStatus PostgresStatus) String() string { return postgresStatus.PostgresClusterStatus } diff --git a/pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go b/pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go index 159a87f35..692a6fe30 100644 --- a/pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go +++ b/pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go @@ -310,6 +310,22 @@ func (in *KubernetesMetaConfiguration) DeepCopy() *KubernetesMetaConfiguration { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *LifecycleSpec) DeepCopyInto(out *LifecycleSpec) { + *out = *in + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new LifecycleSpec. +func (in *LifecycleSpec) DeepCopy() *LifecycleSpec { + if in == nil { + return nil + } + out := new(LifecycleSpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *LoadBalancerConfiguration) DeepCopyInto(out *LoadBalancerConfiguration) { *out = *in @@ -874,6 +890,11 @@ func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.Lifecycle != nil { + in, out := &in.Lifecycle, &out.Lifecycle + *out = new(LifecycleSpec) + **out = **in + } if in.Env != nil { in, out := &in.Env, &out.Env *out = make([]corev1.EnvVar, len(*in)) diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go index 04c974f4c..15ffe0761 100644 --- a/pkg/cluster/cluster.go +++ b/pkg/cluster/cluster.go @@ -1008,20 +1008,64 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error { c.mu.Lock() defer c.mu.Unlock() - newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusUpdating + // Block all spec changes when cluster is stopped or stopping + blocked, err := c.blockLifecycleUpdate(newSpec) + if err != nil { + return err + } + if blocked { + return nil + } - newSpec, err := c.KubeClient.SetPostgresCRDStatus(c.clusterName(), newSpec) + // Handle lifecycle transitions (hibernate/wake-up) + handled, err := c.handleHibernateAndWakeUp(newSpec) if err != nil { - return fmt.Errorf("could not set cluster status to updating: %w", err) + return err + } + if handled { + return nil + } + + // If a previous Update already set a lifecycle status do not clobber it. + // The watch event that triggered this Update is a follow-up from the operator's + // write, so the status subresource is already at the correct value, and writing again + // would cause a 409 Conflict. + isLifecycleActive := newSpec.Status.PostgresClusterStatus == acidv1.ClusterStatusUpdating || + newSpec.Status.PostgresClusterStatus == acidv1.ClusterStatusStopping || + newSpec.Status.PostgresClusterStatus == acidv1.ClusterStatusStopped + + if !isLifecycleActive { + newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusUpdating + + newSpec, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), newSpec) + if err != nil { + return fmt.Errorf("could not set cluster status to updating: %w", err) + } } if !c.isInMaintenanceWindow(newSpec.Spec.MaintenanceWindows) { // do not apply any major version related changes yet newSpec.Spec.PostgresqlParam.PgVersion = oldSpec.Spec.PostgresqlParam.PgVersion } - c.setSpec(newSpec) + + // When isLifecycleActive, the watch delivered newSpec with a status from + // BEFORE our latest status write (e.g. status=Updating while c.Status is + // already Running). Clobbering c.Status with the stale value would break + // the next Update's blockLifecycleUpdate check — preserve the authoritative + // c.Status, only refresh the spec fields. + if !isLifecycleActive { + c.setSpec(newSpec) + } else { + cached, _ := c.GetSpec() + cached.Spec = newSpec.Spec + c.setSpec(cached) + } defer func() { + if isLifecycleActive { + // Status was set by a previous Update; sync's defer will take care + return + } currentStatus := newSpec.Status.DeepCopy() newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusRunning @@ -1220,6 +1264,34 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error { return nil } +// blockLifecycleUpdate checks if an update should be blocked due to lifecycle state. +// Returns (blocked bool, err error): +// - (true, nil) if update is blocked and caller should return early +// - (false, nil) if update can proceed +// - (false, error) on error +func (c *Cluster) blockLifecycleUpdate(newSpec *acidv1.Postgresql) (bool, error) { + if !c.Status.Stopped() && !c.Status.Stopping() { + return false, nil + } + + lifecyclePhase := "" + if newSpec.Spec.Lifecycle != nil { + lifecyclePhase = newSpec.Spec.Lifecycle.Phase + } + + // During Stopping: block ALL spec changes (no cancellation allowed) + if c.Status.Stopping() { + return true, fmt.Errorf("cannot update cluster while it is stopping. Wait for it to fully stop first") + } + + // During Stopped: only block if keeping lifecycle.phase="stopped" + if lifecyclePhase == "stopped" { + return true, fmt.Errorf("cannot update cluster while stopped. Remove lifecycle.phase to wake up the cluster") + } + + return false, nil +} + func syncResources(a, b *v1.ResourceRequirements) bool { for _, res := range []v1.ResourceName{ v1.ResourceCPU, diff --git a/pkg/cluster/lifecycle.go b/pkg/cluster/lifecycle.go new file mode 100644 index 000000000..8f4d0915a --- /dev/null +++ b/pkg/cluster/lifecycle.go @@ -0,0 +1,420 @@ +package cluster + +import ( + "context" + "fmt" + + k8serrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + + acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1" +) + +// LifecycleAction represents the detected lifecycle transition for a cluster. +// Used by both Sync (manageHibernateState) and Update (handleHibernateAndWakeUp) paths +// to determine what action to take regarding cluster hibernate/wake-up. +type LifecycleAction int + +const ( + LifecycleActionNone LifecycleAction = iota + LifecycleActionHibernate // Running -> Stopping (initiate hibernate) + LifecycleActionStoppingCompleted // Stopping -> Stopped (pods fully terminated) + LifecycleActionWakeUp // Stopped -> Updating (initiate wake-up) +) + +// detectLifecycleTransition is a pure function that examines the current and proposed specs +// and determines what lifecycle action (if any) should be taken. +// +// Detection logic: +// - Hibernate: lifecycle.phase="stopped" + status not Stopping or Stopped +// - Wake-up: (status == Stopped OR lifecycle cleared + has previousNumberOfInstances +// - numberOfInstances == 0) AND new lifecycle is cleared (or nil) +func detectLifecycleTransition( + currentStatus *acidv1.PostgresStatus, + newSpecLifecycle *acidv1.LifecycleSpec, + newSpecNumberOfInstances int32, + newSpecPreviousNumberOfInstances int32, +) LifecycleAction { + isWakingUpSimple := newSpecLifecycle == nil || newSpecLifecycle.Phase != "stopped" + hasPreviousInstances := newSpecPreviousNumberOfInstances > 0 + needsRestore := newSpecNumberOfInstances == 0 + + isWakingUp := isWakingUpSimple && hasPreviousInstances && needsRestore + + if currentStatus.Stopped() || isWakingUp { + if isWakingUp || newSpecLifecycle == nil || newSpecLifecycle.Phase != "stopped" { + return LifecycleActionWakeUp + } + } + + if newSpecLifecycle != nil && + newSpecLifecycle.Phase == "stopped" && + !currentStatus.Stopping() && + !currentStatus.Stopped() { + return LifecycleActionHibernate + } + + return LifecycleActionNone +} + +// handleHibernateAndWakeUp detects a hibernate/wake-up transition and asks +// syncStateLocked to mutate + persist the spec+status and run the full sync +// body. +// For hibernate it then waits inline for pods to terminate and persist Stopped. +// For wake-up the sync's existing defer transitions Updating → Running +// once resources exist. +// +// Returns (handled bool, err error): +// - (true, nil) lifecycle transition completed; Update() should return early +// - (false, nil) no lifecycle transition; Update() proceeds normally +// - (true, err) transition attempted but failed; caller returns the error +func (c *Cluster) handleHibernateAndWakeUp(newSpec *acidv1.Postgresql) (bool, error) { + action := detectLifecycleTransition( + &c.Status, + newSpec.Spec.Lifecycle, + newSpec.Spec.NumberOfInstances, + newSpec.Status.PreviousNumberOfInstances, + ) + + if action == LifecycleActionNone { + return false, nil + } + + // syncStateLocked → prepareLifecycleTransition: detects the transition, + // mutates the spec via initiateHibernate/initiateWakeUp, and immediately + // persists spec+status. + if err := c.syncStateLocked(newSpec); err != nil { + return true, fmt.Errorf("could not sync after lifecycle transition: %w", err) + } + + // Hibernate: wait inline for pods to actually terminate and persist Stopped. + // Doing this here (outside syncStateLocked) means a timeout returns an error + // without triggering the defer's SyncFailed status — the operator controller + // requeues and the next Sync (or this same Update on retry) resumes the wait. + if action == LifecycleActionHibernate { + if err := c.completeStoppingTransition(newSpec); err != nil { + return true, fmt.Errorf("hibernate failed: %w", err) + } + } + + return true, nil +} + +// prepareLifecycleTransition inspects newSpec for a lifecycle transition +// On detection it mutates the in-memory spec via initiateHibernate/initiateWakeUp +// and immediately persists BOTH spec and status to the K8s API so the new status +// (Stopping/Updating) is visible in same reconciliation pass — without waiting for +// syncStateLocked's defer. +// Uses oldSpec for detection so it works correctly when handleHibernateAndWakeUp +// has already mutated newSpec before calling syncStateLocked. +// +// Returns: +// - (true, nil): proceed with the rest of syncStateLocked (no transition, +// or transition was persisted) +// - (false, nil): skip sync — cluster is Stopped with no transition requested +// - (false, err): error persisting the spec or status; caller propagates to +// controller (next Update will retry the whole transition) +func (c *Cluster) prepareLifecycleTransition(newSpec **acidv1.Postgresql, oldSpec acidv1.Postgresql) (bool, error) { + spec := *newSpec + action := detectLifecycleTransition( + &oldSpec.Status, + spec.Spec.Lifecycle, + spec.Spec.NumberOfInstances, + spec.Status.PreviousNumberOfInstances, + ) + + // Stopped cluster with lifecycle=stopped: no reconciliation work. + if action == LifecycleActionNone && c.Status.Stopped() { + return false, nil + } + + switch action { + case LifecycleActionHibernate: + c.initiateHibernate(spec) + case LifecycleActionWakeUp: + c.initiateWakeUp(spec) + } + + if action == LifecycleActionHibernate || action == LifecycleActionWakeUp { + // Persist spec first (writes numInst/lifecycle + advances rv to N+1), + // then write the status subresource (advances rv to N+2). + pgUpdated, err := c.KubeClient.UpdatePostgresCR(c.clusterName(), spec) + if err != nil { + return false, fmt.Errorf("could not update spec for lifecycle action: %w", err) + } + // UpdatePostgresCR returns the CR with a fresh resourceVersion but the + // status subresource may be empty/stale. Re-apply the lifecycle status + // fields before writing the status. + pgUpdated.Status.PreviousNumberOfInstances = spec.Status.PreviousNumberOfInstances + pgUpdated.Status.PreviousPoolerInstances = spec.Status.PreviousPoolerInstances + pgUpdated.Status.PostgresClusterStatus = spec.Status.PostgresClusterStatus + + pgUpdated, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), pgUpdated) + if err != nil { + return false, fmt.Errorf("could not set status for lifecycle action: %w", err) + } + c.setSpec(pgUpdated) + } + + return true, nil +} + +// initiateHibernate prepares the cluster for hibernation by: +// - Storing current numberOfInstances in PreviousNumberOfInstances +// - Setting numberOfInstances to 0 +// - Setting status to Stopping +// - Scaling down connection pooler deployments +// - Suspending logical backup CronJob +// Errors during pooler/backup operations are logged but do not fail the transition. +func (c *Cluster) initiateHibernate(newSpec *acidv1.Postgresql) { + newSpec.Status.PreviousNumberOfInstances = newSpec.Spec.NumberOfInstances + newSpec.Spec.NumberOfInstances = 0 + newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusStopping + + c.logger.Infof("[lifecycle] initiating hibernate: stored previousNumberOfInstances=%d", + newSpec.Status.PreviousNumberOfInstances) + + if err := c.scalePoolerDown(newSpec); err != nil { + c.logger.Warningf("[lifecycle] failed to scale pooler during hibernate: %v", err) + } + + if err := c.suspendLogicalBackupJob(); err != nil { + c.logger.Warningf("[lifecycle] failed to suspend logical backup job: %v", err) + } +} + +// initiateWakeUp prepares the cluster for wake-up by: +// - Restoring numberOfInstances from PreviousNumberOfInstances (if > 0) +// - Scaling up connection pooler deployments (consumes PreviousPoolerInstances) +// - Resuming logical backup CronJob +// - Setting status to Updating +// - Clearing PreviousNumberOfInstances / PreviousPoolerInstances so they don't +// linger in the status subresource after the wake-up completes. The next +// hibernate overwrites them with fresh values. +// +// If PreviousNumberOfInstances is 0, logs a warning but still sets status to +// Updating (operator-restart catch-up may have already cleared it). +// Errors during pooler/backup operations are logged but do not fail the transition. +func (c *Cluster) initiateWakeUp(newSpec *acidv1.Postgresql) { + if newSpec.Status.PreviousNumberOfInstances > 0 { + newSpec.Spec.NumberOfInstances = newSpec.Status.PreviousNumberOfInstances + c.logger.Infof("[lifecycle] initiating wake-up: restoring numberOfInstances=%d", + newSpec.Status.PreviousNumberOfInstances) + } else { + c.logger.Warningf("[lifecycle] cluster is waking up but previousNumberOfInstances is 0, cannot restore") + } + + newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusUpdating + + if err := c.scalePoolerUp(newSpec); err != nil { + c.logger.Warningf("[lifecycle] failed to scale pooler during wake-up: %v", err) + } + + if err := c.unsuspendLogicalBackupJob(); err != nil { + c.logger.Warningf("[lifecycle] failed to resume logical backup job: %v", err) + } + + newSpec.Status.PreviousNumberOfInstances = 0 + newSpec.Status.PreviousPoolerInstances = nil +} + +// persistStoppingCompletedTransition persists the Stopping->Stopped transition to Kubernetes +// by updating only the status (status=Stopped). This is called when StatefulSet replicas +// have actually reached 0. +// Returns (handled=true, nil) on success, (handled=false, error) on failure. +func (c *Cluster) persistStoppingCompletedTransition(newSpec *acidv1.Postgresql) (bool, error) { + pgUpdated, err := c.KubeClient.SetPostgresCRDStatus(c.clusterName(), newSpec) + if err != nil { + return false, fmt.Errorf("could not update status during stopping completed: %w", err) + } + + c.setSpec(pgUpdated) + c.logger.Info("[lifecycle] stopping completed: cluster is stopped") + return true, nil +} + +// completeStoppingTransition waits for the StatefulSet pods to actually terminate +// and transitions the cluster status from Stopping to Stopped. On timeout, returns +// an error so the caller can propagate it to the controller which will retry. +func (c *Cluster) completeStoppingTransition(newSpec *acidv1.Postgresql) error { + if err := c.waitStatefulsetPodsGone(); err != nil { + return fmt.Errorf("could not wait for pods to terminate: %w", err) + } + + // Re-fetch the latest resourceVersion before writing Stopped to stop 409 and drift. + latest, err := c.KubeClient.Postgresqls(c.clusterNamespace()).Get( + context.TODO(), c.Name, metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("could not refresh postgresql before persisting Stopped: %w", err) + } + newSpec.ResourceVersion = latest.ResourceVersion + + newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusStopped + c.setSpec(newSpec) + if _, err := c.persistStoppingCompletedTransition(newSpec); err != nil { + return err + } + c.logger.Info("[lifecycle] cluster has stopped") + return nil +} + +// suspendLogicalBackupJob suspends the logical backup CronJob by setting spec.suspend=true. +// If the job was previously loaded but has been deleted externally, clears the cached reference. +// Returns nil if job is not loaded (no-op) or if job was not found (clears cache). +// Returns error only for actual failures (network errors, etc). +func (c *Cluster) suspendLogicalBackupJob() error { + if c.LogicalBackupJob == nil { + c.logger.Debug("logical backup job is not loaded, skipping suspend") + return nil + } + + // Check if job still exists (handles externally deleted jobs) + _, err := c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Get( + context.TODO(), c.getLogicalBackupJobName(), metav1.GetOptions{}) + if k8serrors.IsNotFound(err) { + c.logger.Info("logical backup job not found during suspend, clearing cached reference") + c.LogicalBackupJob = nil + return nil + } + if err != nil { + return fmt.Errorf("could not get logical backup job: %w", err) + } + + patchData := `{"spec":{"suspend":true}}` + cronJob, err := c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Patch( + context.TODO(), + c.getLogicalBackupJobName(), + types.MergePatchType, + []byte(patchData), + metav1.PatchOptions{}, + "", + ) + if err != nil { + return fmt.Errorf("could not suspend logical backup job: %w", err) + } + c.LogicalBackupJob = cronJob + c.logger.Info("logical backup job suspended") + + return nil +} + +// unsuspendLogicalBackupJob resumes the logical backup CronJob by setting spec.suspend=false. +// If the job was previously loaded but has been deleted externally, clears the cached reference. +// Returns nil if job is not loaded (no-op) or if job was not found (clears cache). +// Returns error only for actual failures (network errors, etc). +func (c *Cluster) unsuspendLogicalBackupJob() error { + if c.LogicalBackupJob == nil { + c.logger.Debug("logical backup job is not loaded, skipping unsuspend") + return nil + } + + // Check if job still exists (handles externally deleted jobs) + _, err := c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Get( + context.TODO(), c.getLogicalBackupJobName(), metav1.GetOptions{}) + if k8serrors.IsNotFound(err) { + c.logger.Info("logical backup job not found during unsuspend, clearing cached reference") + c.LogicalBackupJob = nil + return nil + } + if err != nil { + return fmt.Errorf("could not get logical backup job: %w", err) + } + + patchData := `{"spec":{"suspend":false}}` + cronJob, err := c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Patch( + context.TODO(), + c.getLogicalBackupJobName(), + types.MergePatchType, + []byte(patchData), + metav1.PatchOptions{}, + "", + ) + if err != nil { + return fmt.Errorf("could not resume logical backup job: %w", err) + } + c.LogicalBackupJob = cronJob + c.logger.Info("logical backup job resumed") + + return nil +} + +// getPoolerReplicas returns the current replica count for a pooler deployment. +// Returns 0 if pooler doesn't exist, hasn't been synced yet, or has nil Replicas. +func (c *Cluster) getPoolerReplicas(role PostgresRole) int32 { + if c.ConnectionPooler == nil || c.ConnectionPooler[role] == nil || + c.ConnectionPooler[role].Deployment == nil || + c.ConnectionPooler[role].Deployment.Spec.Replicas == nil { + return 0 + } + return *c.ConnectionPooler[role].Deployment.Spec.Replicas +} + +// scalePoolerDown scales all connection pooler deployments to 0 and stores their current +// replica counts in newSpec.Status.PreviousPoolerInstances. +// Should be called during hibernate initiation. +// Errors are returned immediately if a patch fails (partial state possible). +func (c *Cluster) scalePoolerDown(newSpec *acidv1.Postgresql) error { + if c.ConnectionPooler == nil { + return nil + } + + for role := range c.ConnectionPooler { + replicas := c.getPoolerReplicas(role) + + if newSpec.Status.PreviousPoolerInstances == nil { + newSpec.Status.PreviousPoolerInstances = make(map[string]int32) + } + newSpec.Status.PreviousPoolerInstances[string(role)] = replicas + + if replicas > 0 { + if err := c.patchPoolerReplicas(role, 0); err != nil { + return err + } + c.logger.Infof("[lifecycle] pooler %s scaled to 0 (was %d)", role, replicas) + } + } + return nil +} + +// scalePoolerUp restores connection pooler deployments to their previous replica counts +// from newSpec.Status.PreviousPoolerInstances. +// Should be called during wake-up. +// Errors are returned immediately if a patch fails (partial state possible). +func (c *Cluster) scalePoolerUp(newSpec *acidv1.Postgresql) error { + if newSpec.Status.PreviousPoolerInstances == nil { + return nil + } + + for roleStr, replicas := range newSpec.Status.PreviousPoolerInstances { + role := PostgresRole(roleStr) + + if err := c.patchPoolerReplicas(role, replicas); err != nil { + return err + } + c.logger.Infof("[lifecycle] pooler %s scaled to %d", role, replicas) + } + return nil +} + +// patchPoolerReplicas patches a pooler deployment's replica count. +// If the deployment doesn't exist (not found), returns nil (no-op). +// Returns error for other failures (network errors, etc). +func (c *Cluster) patchPoolerReplicas(role PostgresRole, replicas int32) error { + _, err := c.KubeClient.Deployments(c.Namespace).Get( + context.TODO(), c.connectionPoolerName(role), metav1.GetOptions{}) + if err != nil { + if k8serrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("could not get pooler deployment for %s: %w", role, err) + } + + patchData := fmt.Sprintf(`{"spec":{"replicas":%d}}`, replicas) + _, err = c.KubeClient.Deployments(c.Namespace).Patch( + context.TODO(), c.connectionPoolerName(role), types.MergePatchType, []byte(patchData), metav1.PatchOptions{}) + if err != nil { + return fmt.Errorf("could not patch pooler deployment %s replicas: %w", role, err) + } + return nil +} diff --git a/pkg/cluster/lifecycle_test.go b/pkg/cluster/lifecycle_test.go new file mode 100644 index 000000000..9e074f804 --- /dev/null +++ b/pkg/cluster/lifecycle_test.go @@ -0,0 +1,1008 @@ +package cluster + +import ( + "context" + "fmt" + "testing" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1" + fakeacidv1 "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/fake" + "github.com/zalando/postgres-operator/pkg/util/config" + "github.com/zalando/postgres-operator/pkg/util/k8sutil" + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" + "k8s.io/client-go/tools/record" +) + +var lifecycleLogger = logrus.New().WithField("test", "lifecycle") +var lifecycleEventRecorder = record.NewFakeRecorder(10) + +func int32Ptr(i int32) *int32 { return &i } + +func newTestPoolerObjects(role PostgresRole, replicas int32) *ConnectionPoolerObjects { + return &ConnectionPoolerObjects{ + Deployment: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("test-%s-pooler", role), + Namespace: "default", + }, + Spec: appsv1.DeploymentSpec{ + Replicas: int32Ptr(replicas), + }, + }, + Name: fmt.Sprintf("test-%s-pooler", role), + ClusterName: "test-cluster", + Namespace: "default", + Role: role, + } +} + +func newFakeK8sClientForLifecycle() (*k8sutil.KubernetesClient, *fake.Clientset, *fakeacidv1.Clientset) { + clientSet := fake.NewSimpleClientset() + acidClientSet := fakeacidv1.NewSimpleClientset() + + client := &k8sutil.KubernetesClient{ + DeploymentsGetter: clientSet.AppsV1(), + PostgresqlsGetter: acidClientSet.AcidV1(), + StatefulSetsGetter: clientSet.AppsV1(), + ServicesGetter: clientSet.CoreV1(), + SecretsGetter: clientSet.CoreV1(), + ConfigMapsGetter: clientSet.CoreV1(), + PodsGetter: clientSet.CoreV1(), + EndpointsGetter: clientSet.CoreV1(), + CronJobsGetter: clientSet.BatchV1(), + } + + return client, clientSet, acidClientSet +} + +// newLifecycleCluster builds a Cluster with the given current status and spec +// snapshot. The Postgres CR is pre-created in the fake clientset so K8s API +// calls work without a separate Create step. +func newLifecycleCluster( + client *k8sutil.KubernetesClient, + status string, + numberOfInstances int32, + lifecyclePhase string, + previousNumberOfInstances int32, + previousPoolerInstances map[string]int32, +) *Cluster { + pg := &acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster", + Namespace: "default", + }, + Spec: acidv1.PostgresSpec{ + TeamID: "test-team", + NumberOfInstances: numberOfInstances, + Volume: acidv1.Volume{Size: "1Gi"}, + }, + Status: acidv1.PostgresStatus{ + PostgresClusterStatus: status, + }, + } + if lifecyclePhase != "" { + pg.Spec.Lifecycle = &acidv1.LifecycleSpec{Phase: lifecyclePhase} + } + if previousNumberOfInstances > 0 { + pg.Status.PreviousNumberOfInstances = previousNumberOfInstances + } + if previousPoolerInstances != nil { + pg.Status.PreviousPoolerInstances = previousPoolerInstances + } + + created, err := client.Postgresqls("default").Create(context.TODO(), pg, metav1.CreateOptions{}) + if err != nil { + panic(fmt.Sprintf("failed to pre-create Postgresql: %v", err)) + } + + return &Cluster{ + Config: Config{ + OpConfig: config.Config{ + PodManagementPolicy: "ordered_ready", + LogicalBackup: config.LogicalBackup{ + LogicalBackupJobPrefix: "logical-backup-", + }, + }, + }, + Postgresql: *created, + KubeClient: *client, + logger: lifecycleLogger, + eventRecorder: lifecycleEventRecorder, + } +} + +func TestDetectLifecycleTransition(t *testing.T) { + tests := []struct { + name string + currentStatus string + newLifecyclePhase string // "" means nil lifecycle + newNumberOfInstances int32 + newPreviousNumberOfInstances int32 + want LifecycleAction + }{ + { + name: "Running + lifecycle.phase=stopped -> Hibernate", + currentStatus: acidv1.ClusterStatusRunning, + newLifecyclePhase: "stopped", + newNumberOfInstances: 3, + newPreviousNumberOfInstances: 0, + want: LifecycleActionHibernate, + }, + { + name: "Running + no lifecycle -> None", + currentStatus: acidv1.ClusterStatusRunning, + newLifecyclePhase: "", + newNumberOfInstances: 3, + newPreviousNumberOfInstances: 0, + want: LifecycleActionNone, + }, + { + name: "Stopping + lifecycle.phase=stopped -> None (already stopping)", + currentStatus: acidv1.ClusterStatusStopping, + newLifecyclePhase: "stopped", + newNumberOfInstances: 0, + newPreviousNumberOfInstances: 3, + want: LifecycleActionNone, + }, + { + name: "Stopped + lifecycle.phase=stopped -> None (still hibernated)", + currentStatus: acidv1.ClusterStatusStopped, + newLifecyclePhase: "stopped", + newNumberOfInstances: 0, + newPreviousNumberOfInstances: 3, + want: LifecycleActionNone, + }, + { + name: "Stopped + lifecycle cleared + has previous instances + numInst=0 -> WakeUp", + currentStatus: acidv1.ClusterStatusStopped, + newLifecyclePhase: "", + newNumberOfInstances: 0, + newPreviousNumberOfInstances: 3, + want: LifecycleActionWakeUp, + }, + { + name: "Running + lifecycle cleared + previous instances + numInst=0 -> WakeUp", + currentStatus: acidv1.ClusterStatusRunning, + newLifecyclePhase: "", + newNumberOfInstances: 0, + newPreviousNumberOfInstances: 3, + want: LifecycleActionWakeUp, + }, + { + name: "Running + lifecycle cleared + previous instances + numInst>0 -> None", + currentStatus: acidv1.ClusterStatusRunning, + newLifecyclePhase: "", + newNumberOfInstances: 3, + newPreviousNumberOfInstances: 3, + want: LifecycleActionNone, + }, + { + name: "Running + lifecycle cleared + no previous instances -> None", + currentStatus: acidv1.ClusterStatusRunning, + newLifecyclePhase: "", + newNumberOfInstances: 3, + newPreviousNumberOfInstances: 0, + want: LifecycleActionNone, + }, + { + name: "Stopped + lifecycle cleared + no previous instances -> WakeUp (operator restart catch-up)", + currentStatus: acidv1.ClusterStatusStopped, + newLifecyclePhase: "", + newNumberOfInstances: 0, + newPreviousNumberOfInstances: 0, + want: LifecycleActionWakeUp, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var lifecycle *acidv1.LifecycleSpec + if tt.newLifecyclePhase != "" { + lifecycle = &acidv1.LifecycleSpec{Phase: tt.newLifecyclePhase} + } + + status := acidv1.PostgresStatus{PostgresClusterStatus: tt.currentStatus} + + got := detectLifecycleTransition( + &status, + lifecycle, + tt.newNumberOfInstances, + tt.newPreviousNumberOfInstances, + ) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestInitiateHibernate(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusRunning, 3, "", 0, nil) + + newSpec := &acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}, + Spec: acidv1.PostgresSpec{ + NumberOfInstances: 3, + }, + Status: acidv1.PostgresStatus{ + PostgresClusterStatus: acidv1.ClusterStatusRunning, + }, + } + + c.initiateHibernate(newSpec) + + assert.Equal(t, int32(0), newSpec.Spec.NumberOfInstances, "numberOfInstances should be set to 0") + assert.Equal(t, int32(3), newSpec.Status.PreviousNumberOfInstances, "previousNumberOfInstances should be stored") + assert.Equal(t, acidv1.ClusterStatusStopping, newSpec.Status.PostgresClusterStatus, "status should be Stopping") +} + +func TestInitiateWakeUp(t *testing.T) { + t.Run("restores numberOfInstances and clears Previous fields", func(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusStopped, 0, "", 0, nil) + + newSpec := &acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}, + Spec: acidv1.PostgresSpec{ + NumberOfInstances: 0, + }, + Status: acidv1.PostgresStatus{ + PostgresClusterStatus: acidv1.ClusterStatusStopped, + PreviousNumberOfInstances: 3, + PreviousPoolerInstances: map[string]int32{"master": 2, "replica": 0}, + }, + } + + c.initiateWakeUp(newSpec) + + assert.Equal(t, int32(3), newSpec.Spec.NumberOfInstances, "numberOfInstances should be restored") + assert.Equal(t, acidv1.ClusterStatusUpdating, newSpec.Status.PostgresClusterStatus, "status should be Updating") + assert.Equal(t, int32(0), newSpec.Status.PreviousNumberOfInstances, "previousNumberOfInstances should be cleared") + assert.Nil(t, newSpec.Status.PreviousPoolerInstances, "previousPoolerInstances should be cleared") + }) + + t.Run("previousNumberOfInstances=0 still transitions to Updating", func(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusStopped, 0, "", 0, nil) + + newSpec := &acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}, + Spec: acidv1.PostgresSpec{ + NumberOfInstances: 0, + }, + Status: acidv1.PostgresStatus{ + PostgresClusterStatus: acidv1.ClusterStatusStopped, + PreviousPoolerInstances: map[string]int32{"master": 1}, + }, + } + + c.initiateWakeUp(newSpec) + + assert.Equal(t, int32(0), newSpec.Spec.NumberOfInstances, "numberOfInstances stays 0 when previous is 0") + assert.Equal(t, acidv1.ClusterStatusUpdating, newSpec.Status.PostgresClusterStatus, "status should still be Updating") + assert.Equal(t, int32(0), newSpec.Status.PreviousNumberOfInstances, "previousNumberOfInstances stays 0") + assert.Nil(t, newSpec.Status.PreviousPoolerInstances, "previousPoolerInstances should be cleared") + }) +} + +func TestPrepareLifecycleTransition_Hibernate(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusRunning, 3, "", 0, nil) + + newSpec := c.Postgresql.DeepCopy() + newSpec.Spec.Lifecycle = &acidv1.LifecycleSpec{Phase: "stopped"} + + oldSpec := acidv1.Postgresql{ + Status: acidv1.PostgresStatus{PostgresClusterStatus: acidv1.ClusterStatusRunning}, + } + + proceed, err := c.prepareLifecycleTransition(&newSpec, oldSpec) + + assert.NoError(t, err) + assert.True(t, proceed, "hibernate should proceed with sync") + assert.Equal(t, int32(0), newSpec.Spec.NumberOfInstances, "numberOfInstances should be set to 0") + assert.Equal(t, int32(3), newSpec.Status.PreviousNumberOfInstances, "previousNumberOfInstances should be stored") + assert.Equal(t, acidv1.ClusterStatusStopping, newSpec.Status.PostgresClusterStatus, "status should be Stopping") + + persisted, err := client.Postgresqls("default").Get(context.TODO(), "test-cluster", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, acidv1.ClusterStatusStopping, persisted.Status.PostgresClusterStatus, "persisted status should be Stopping") + assert.Equal(t, int32(3), persisted.Status.PreviousNumberOfInstances, "persisted previousNumberOfInstances should be 3") + assert.Equal(t, int32(0), persisted.Spec.NumberOfInstances, "persisted numberOfInstances should be 0") +} + +func TestPrepareLifecycleTransition_WakeUp(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster( + client, + acidv1.ClusterStatusStopped, + 0, + "", + 3, + map[string]int32{"master": 2, "replica": 0}, + ) + + newSpec := c.Postgresql.DeepCopy() + + oldSpec := acidv1.Postgresql{ + Status: acidv1.PostgresStatus{PostgresClusterStatus: acidv1.ClusterStatusStopped}, + } + + proceed, err := c.prepareLifecycleTransition(&newSpec, oldSpec) + + assert.NoError(t, err) + assert.True(t, proceed, "wake-up should proceed with sync") + assert.Equal(t, int32(3), newSpec.Spec.NumberOfInstances, "numberOfInstances should be restored") + assert.Equal(t, acidv1.ClusterStatusUpdating, newSpec.Status.PostgresClusterStatus, "status should be Updating") + assert.Equal(t, int32(0), newSpec.Status.PreviousNumberOfInstances, "previousNumberOfInstances should be cleared after persistence") + assert.Nil(t, newSpec.Status.PreviousPoolerInstances, "previousPoolerInstances should be cleared after persistence") + + persisted, err := client.Postgresqls("default").Get(context.TODO(), "test-cluster", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, acidv1.ClusterStatusUpdating, persisted.Status.PostgresClusterStatus, "persisted status should be Updating") + assert.Equal(t, int32(3), persisted.Spec.NumberOfInstances, "persisted numberOfInstances should be 3") + assert.Equal(t, int32(0), persisted.Status.PreviousNumberOfInstances, "persisted previousNumberOfInstances should be 0") + assert.Nil(t, persisted.Status.PreviousPoolerInstances, "persisted previousPoolerInstances should be nil") +} + +func TestPrepareLifecycleTransition_StoppedNoTransition(t *testing.T) { + client, _, acidClientSet := newFakeK8sClientForLifecycle() + c := newLifecycleCluster( + client, + acidv1.ClusterStatusStopped, + 0, + "stopped", + 3, + nil, + ) + + updateCalled := false + acidClientSet.PrependReactor("update", "postgresqls", func(action k8stesting.Action) (bool, runtime.Object, error) { + updateCalled = true + return false, nil, nil + }) + + newSpec := c.Postgresql.DeepCopy() + + oldSpec := acidv1.Postgresql{ + Status: acidv1.PostgresStatus{PostgresClusterStatus: acidv1.ClusterStatusStopped}, + } + + proceed, err := c.prepareLifecycleTransition(&newSpec, oldSpec) + + assert.NoError(t, err) + assert.False(t, proceed, "stopped cluster with lifecycle=stopped should skip sync") + assert.False(t, updateCalled, "no K8s API writes should happen for stopped-no-transition") + assert.Equal(t, int32(0), newSpec.Spec.NumberOfInstances, "numberOfInstances should be unchanged") + assert.Equal(t, acidv1.ClusterStatusStopped, newSpec.Status.PostgresClusterStatus, "status should remain Stopped") +} + +func TestPrepareLifecycleTransition_NoTransitionRunning(t *testing.T) { + client, _, acidClientSet := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusRunning, 3, "", 0, nil) + + updateCalled := false + acidClientSet.PrependReactor("update", "postgresqls", func(action k8stesting.Action) (bool, runtime.Object, error) { + updateCalled = true + return false, nil, nil + }) + + newSpec := c.Postgresql.DeepCopy() + + oldSpec := acidv1.Postgresql{ + Status: acidv1.PostgresStatus{PostgresClusterStatus: acidv1.ClusterStatusRunning}, + } + + proceed, err := c.prepareLifecycleTransition(&newSpec, oldSpec) + + assert.NoError(t, err) + assert.True(t, proceed, "no-transition running cluster should proceed with sync") + assert.False(t, updateCalled, "no K8s API writes should happen for no-transition") + assert.Equal(t, int32(3), newSpec.Spec.NumberOfInstances, "numberOfInstances should be unchanged") + assert.Equal(t, acidv1.ClusterStatusRunning, newSpec.Status.PostgresClusterStatus, "status should remain Running") +} + +func TestPrepareLifecycleTransition_UpdateSpecFails(t *testing.T) { + client, _, acidClientSet := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusRunning, 3, "", 0, nil) + + acidClientSet.PrependReactor("update", "postgresqls", func(action k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, fmt.Errorf("api server unavailable") + }) + + newSpec := c.Postgresql.DeepCopy() + newSpec.Spec.Lifecycle = &acidv1.LifecycleSpec{Phase: "stopped"} + + oldSpec := acidv1.Postgresql{ + Status: acidv1.PostgresStatus{PostgresClusterStatus: acidv1.ClusterStatusRunning}, + } + + proceed, err := c.prepareLifecycleTransition(&newSpec, oldSpec) + + assert.Error(t, err) + assert.Contains(t, err.Error(), "could not update spec for lifecycle action") + assert.False(t, proceed, "should not proceed when spec write fails") +} + +func TestPersistStoppingCompletedTransition(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusStopping, 0, "stopped", 3, nil) + + newSpec := c.Postgresql.DeepCopy() + newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusStopped + + handled, err := c.persistStoppingCompletedTransition(newSpec) + + assert.NoError(t, err) + assert.True(t, handled, "should report handled=true on success") + + persisted, err := client.Postgresqls("default").Get(context.TODO(), "test-cluster", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, acidv1.ClusterStatusStopped, persisted.Status.PostgresClusterStatus, "persisted status should be Stopped") + assert.Equal(t, acidv1.ClusterStatusStopped, c.Postgresql.Status.PostgresClusterStatus, "cache should reflect Stopped") +} + +func TestGetPoolerReplicas(t *testing.T) { + tests := []struct { + name string + poolerObjs map[PostgresRole]*ConnectionPoolerObjects + role PostgresRole + want int32 + }{ + { + name: "nil ConnectionPooler map", + poolerObjs: nil, + role: Master, + want: 0, + }, + { + name: "ConnectionPooler for role is nil", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{Master: nil}, + role: Master, + want: 0, + }, + { + name: "Deployment is nil", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{Master: {Deployment: nil}}, + role: Master, + want: 0, + }, + { + name: "Replicas is nil", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{ + Master: {Deployment: &appsv1.Deployment{Spec: appsv1.DeploymentSpec{Replicas: nil}}}, + }, + role: Master, + want: 0, + }, + { + name: "Master with 2 replicas", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{ + Master: newTestPoolerObjects(Master, 2), + }, + role: Master, + want: 2, + }, + { + name: "Master with 0 replicas", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{ + Master: newTestPoolerObjects(Master, 0), + }, + role: Master, + want: 0, + }, + { + name: "Replica with 3 replicas", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{ + Replica: newTestPoolerObjects(Replica, 3), + }, + role: Replica, + want: 3, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := &Cluster{ + ConnectionPooler: tt.poolerObjs, + } + got := c.getPoolerReplicas(tt.role) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestPatchPoolerReplicas(t *testing.T) { + tests := []struct { + name string + replicas int32 + setupClient func(clientSet *fake.Clientset) + wantErr bool + errContains string + }{ + { + name: "deployment exists, patch succeeds", + replicas: 0, + setupClient: func(clientSet *fake.Clientset) { + _, _ = clientSet.AppsV1().Deployments("default").Create(context.TODO(), &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster-pooler"}, + Spec: appsv1.DeploymentSpec{Replicas: int32Ptr(2)}, + }, metav1.CreateOptions{}) + }, + wantErr: false, + }, + { + name: "deployment not found - returns nil", + replicas: 2, + setupClient: func(clientSet *fake.Clientset) {}, + wantErr: false, + }, + { + name: "patch returns error", + replicas: 2, + setupClient: func(clientSet *fake.Clientset) { + _, _ = clientSet.AppsV1().Deployments("default").Create(context.TODO(), &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster-pooler"}, + Spec: appsv1.DeploymentSpec{Replicas: int32Ptr(2)}, + }, metav1.CreateOptions{}) + clientSet.PrependReactor("patch", "deployments", func(action k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, fmt.Errorf("network error") + }) + }, + wantErr: true, + errContains: "could not patch pooler deployment", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + clientSet := fake.NewSimpleClientset() + if tt.setupClient != nil { + tt.setupClient(clientSet) + } + + kubeClient := &k8sutil.KubernetesClient{ + DeploymentsGetter: clientSet.AppsV1(), + } + + c := &Cluster{ + KubeClient: *kubeClient, + Postgresql: acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster", + Namespace: "default", + }, + }, + } + + err := c.patchPoolerReplicas(Master, tt.replicas) + if tt.wantErr { + assert.Error(t, err) + if tt.errContains != "" { + assert.Contains(t, err.Error(), tt.errContains) + } + } else { + assert.NoError(t, err) + } + }) + } +} + +func TestScalePoolerDown(t *testing.T) { + tests := []struct { + name string + poolerObjs map[PostgresRole]*ConnectionPoolerObjects + wantStored map[string]int32 + }{ + { + name: "nil ConnectionPooler - no-op", + poolerObjs: nil, + wantStored: nil, + }, + { + name: "Master at 2 replicas", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{Master: newTestPoolerObjects(Master, 2)}, + wantStored: map[string]int32{"master": 2}, + }, + { + name: "Master already at 0 replicas", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{Master: newTestPoolerObjects(Master, 0)}, + wantStored: map[string]int32{"master": 0}, + }, + { + name: "Both Master and Replica", + poolerObjs: map[PostgresRole]*ConnectionPoolerObjects{ + Master: newTestPoolerObjects(Master, 2), + Replica: newTestPoolerObjects(Replica, 1), + }, + wantStored: map[string]int32{"master": 2, "replica": 1}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + clientSet := fake.NewSimpleClientset() + kubeClient := &k8sutil.KubernetesClient{ + DeploymentsGetter: clientSet.AppsV1(), + } + + c := &Cluster{ + KubeClient: *kubeClient, + ConnectionPooler: tt.poolerObjs, + logger: lifecycleLogger, + Postgresql: acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster", + Namespace: "default", + }, + }, + } + + newSpec := &acidv1.Postgresql{} + err := c.scalePoolerDown(newSpec) + + assert.NoError(t, err) + assert.Equal(t, tt.wantStored, newSpec.Status.PreviousPoolerInstances) + }) + } +} + +func TestScalePoolerUp(t *testing.T) { + tests := []struct { + name string + prevInst map[string]int32 + wantErr bool + }{ + { + name: "nil PreviousPoolerInstances - no-op", + prevInst: nil, + wantErr: false, + }, + { + name: "Restore master to 2", + prevInst: map[string]int32{"master": 2}, + wantErr: false, + }, + { + name: "Restore both roles", + prevInst: map[string]int32{"master": 2, "replica": 1}, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + clientSet := fake.NewSimpleClientset() + kubeClient := &k8sutil.KubernetesClient{ + DeploymentsGetter: clientSet.AppsV1(), + } + + c := &Cluster{ + KubeClient: *kubeClient, + logger: lifecycleLogger, + Postgresql: acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster", + Namespace: "default", + }, + }, + } + + newSpec := &acidv1.Postgresql{ + Status: acidv1.PostgresStatus{ + PreviousPoolerInstances: tt.prevInst, + }, + } + err := c.scalePoolerUp(newSpec) + + if tt.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} + +func TestSuspendLogicalBackupJob(t *testing.T) { + tests := []struct { + name string + jobExists bool + patchFails bool + wantErr bool + }{ + { + name: "job exists, suspend succeeds", + jobExists: true, + wantErr: false, + }, + { + name: "job does not exist - no-op", + jobExists: false, + wantErr: false, + }, + { + name: "job exists but patch fails", + jobExists: true, + patchFails: true, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + clientSet := fake.NewSimpleClientset() + jobName := "logical-backup-test-cluster" + + if tt.jobExists { + _, _ = clientSet.BatchV1().CronJobs("default").Create(context.TODO(), &batchv1.CronJob{ + ObjectMeta: metav1.ObjectMeta{ + Name: jobName, + Namespace: "default", + }, + Spec: batchv1.CronJobSpec{ + Schedule: "30 00 * * *", + }, + }, metav1.CreateOptions{}) + } + + if tt.patchFails { + clientSet.PrependReactor("patch", "cronjobs", func(action k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, fmt.Errorf("network error") + }) + } + + kubeClient := &k8sutil.KubernetesClient{ + CronJobsGetter: clientSet.BatchV1(), + } + + var job *batchv1.CronJob + if tt.jobExists { + job, _ = kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{}) + } + + c := New( + Config{ + OpConfig: config.Config{ + LogicalBackup: config.LogicalBackup{ + LogicalBackupJobPrefix: "logical-backup-", + }, + }, + }, + *kubeClient, + acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster", + Namespace: "default", + }, + }, + lifecycleLogger, + lifecycleEventRecorder, + ) + c.LogicalBackupJob = job + + err := c.suspendLogicalBackupJob() + + if tt.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + if tt.jobExists && !tt.patchFails { + updatedJob, _ := kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{}) + if updatedJob != nil { + assert.True(t, *updatedJob.Spec.Suspend, "job should be suspended") + } + } + } + }) + } +} + +func TestUnsuspendLogicalBackupJob(t *testing.T) { + tests := []struct { + name string + jobExists bool + patchFails bool + wantErr bool + }{ + { + name: "job exists, unsuspend succeeds", + jobExists: true, + wantErr: false, + }, + { + name: "job does not exist - no-op", + jobExists: false, + wantErr: false, + }, + { + name: "job exists but patch fails", + jobExists: true, + patchFails: true, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + clientSet := fake.NewSimpleClientset() + jobName := "logical-backup-test-cluster" + + if tt.jobExists { + suspendTrue := true + _, _ = clientSet.BatchV1().CronJobs("default").Create(context.TODO(), &batchv1.CronJob{ + ObjectMeta: metav1.ObjectMeta{ + Name: jobName, + Namespace: "default", + }, + Spec: batchv1.CronJobSpec{ + Schedule: "30 00 * * *", + Suspend: &suspendTrue, + }, + }, metav1.CreateOptions{}) + } + + if tt.patchFails { + clientSet.PrependReactor("patch", "cronjobs", func(action k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, fmt.Errorf("network error") + }) + } + + kubeClient := &k8sutil.KubernetesClient{ + CronJobsGetter: clientSet.BatchV1(), + } + + var job *batchv1.CronJob + if tt.jobExists { + job, _ = kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{}) + } + + c := New( + Config{ + OpConfig: config.Config{ + LogicalBackup: config.LogicalBackup{ + LogicalBackupJobPrefix: "logical-backup-", + }, + }, + }, + *kubeClient, + acidv1.Postgresql{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-cluster", + Namespace: "default", + }, + }, + lifecycleLogger, + lifecycleEventRecorder, + ) + c.LogicalBackupJob = job + + err := c.unsuspendLogicalBackupJob() + + if tt.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + if tt.jobExists && !tt.patchFails { + updatedJob, _ := kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{}) + if updatedJob != nil { + assert.False(t, *updatedJob.Spec.Suspend, "job should be unsuspended") + } + } + } + }) + } +} + +// blockLifecycleUpdate lives in cluster.go but is exercised here because the +// lifecycle subsystem owns its semantics (Stopped/Stopping states). +func TestBlockLifecycleUpdate(t *testing.T) { + tests := []struct { + name string + currentStatus string + lifecyclePhase string + wantBlocked bool + wantErr bool + errContains string + }{ + { + name: "Running cluster, allows update", + currentStatus: acidv1.ClusterStatusRunning, + wantBlocked: false, + wantErr: false, + }, + { + name: "Stopping state, blocks update", + currentStatus: acidv1.ClusterStatusStopping, + wantBlocked: true, + wantErr: true, + errContains: "cannot update cluster while it is stopping", + }, + { + name: "Stopped with lifecycle.phase=stopped, blocks update", + currentStatus: acidv1.ClusterStatusStopped, + lifecyclePhase: "stopped", + wantBlocked: true, + wantErr: true, + errContains: "cannot update cluster while stopped", + }, + { + name: "Stopped without lifecycle.phase, allows update (wake-up)", + currentStatus: acidv1.ClusterStatusStopped, + lifecyclePhase: "", + wantBlocked: false, + wantErr: false, + }, + { + name: "UpdateFailed state, allows update", + currentStatus: acidv1.ClusterStatusUpdateFailed, + wantBlocked: false, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, tt.currentStatus, 3, tt.lifecyclePhase, 0, nil) + + newSpec := c.Postgresql.DeepCopy() + if tt.lifecyclePhase != "" && tt.currentStatus != acidv1.ClusterStatusStopped { + newSpec.Spec.Lifecycle = &acidv1.LifecycleSpec{Phase: tt.lifecyclePhase} + } + + blocked, err := c.blockLifecycleUpdate(newSpec) + + if tt.wantErr { + assert.Error(t, err) + if tt.errContains != "" { + assert.Contains(t, err.Error(), tt.errContains) + } + } else { + assert.NoError(t, err) + } + assert.Equal(t, tt.wantBlocked, blocked) + }) + } +} + +func TestLifecycleUpdateBlocksDuringStopping(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusStopping, 0, "stopped", 3, nil) + + newSpec := c.Postgresql.DeepCopy() + blocked, err := c.blockLifecycleUpdate(newSpec) + + assert.True(t, blocked) + assert.Error(t, err) + assert.Contains(t, err.Error(), "cannot update cluster while it is stopping") +} + +func TestLifecycleUpdateBlocksWhenStoppedWithPhase(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusStopped, 0, "stopped", 3, nil) + + newSpec := c.Postgresql.DeepCopy() + blocked, err := c.blockLifecycleUpdate(newSpec) + + assert.True(t, blocked) + assert.Error(t, err) + assert.Contains(t, err.Error(), "cannot update cluster while stopped") +} + +func TestLifecycleUpdateAllowsWakeUp(t *testing.T) { + client, _, _ := newFakeK8sClientForLifecycle() + c := newLifecycleCluster(client, acidv1.ClusterStatusStopped, 0, "", 3, nil) + + newSpec := c.Postgresql.DeepCopy() + blocked, err := c.blockLifecycleUpdate(newSpec) + + assert.False(t, blocked) + assert.NoError(t, err) +} diff --git a/pkg/cluster/resources.go b/pkg/cluster/resources.go index 1925733de..b75f6626d 100644 --- a/pkg/cluster/resources.go +++ b/pkg/cluster/resources.go @@ -825,6 +825,8 @@ func (c *Cluster) deleteLogicalBackupJob() error { return nil } + + // GetServiceMaster returns cluster's kubernetes master Service func (c *Cluster) GetServiceMaster() *v1.Service { return c.Services[Master] diff --git a/pkg/cluster/sync.go b/pkg/cluster/sync.go index ffebd306c..946906a00 100644 --- a/pkg/cluster/sync.go +++ b/pkg/cluster/sync.go @@ -36,10 +36,26 @@ var requirePrimaryRestartWhenDecreased = []string{ // Sync syncs the cluster, making sure the actual Kubernetes objects correspond to what is defined in the manifest. // Unlike the update, sync does not error out if some objects do not exist and takes care of creating them. func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error { - var err error c.mu.Lock() defer c.mu.Unlock() + // operator-restart recovery for an in-flight hibernate. If the persisted + // status is Stopping but pods have not yet been confirmed gone, complete the + // transition inline. Lives at the Sync level so the existing defer's SyncFailed + // branch doesn't fire on the lifecycle wait timeout. + if c.Status.Stopping() { + return c.completeStoppingTransition(newSpec) + } + + return c.syncStateLocked(newSpec) +} + +// syncStateLocked performs reconciliation. The caller must hold c.mu. Detects +// lifecycle transitions in-memory and persists spec mutations before running +// the sync body, so a single call covers both lifecycle transitions (called +// from handleHibernateAndWakeUp) and periodic reconciliation. +func (c *Cluster) syncStateLocked(newSpec *acidv1.Postgresql) error { + var err error oldSpec := c.Postgresql c.setSpec(newSpec) @@ -47,7 +63,13 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error { if err != nil { c.logger.Warningf("error while syncing cluster state: %v", err) newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusSyncFailed - } else if !c.Status.Running() { + } else if !c.Status.Running() && !c.Status.Stopping() && !c.Status.Stopped() && + oldSpec.Spec.NumberOfInstances > 0 { + // Wake-up Updates carry oldSpec.Spec.NumberOfInstances == 0 because + // initiateWakeUp hasn't mutated newSpec yet when oldSpec was captured + // Skip the Updating->Running transition here: the cluster + // isn't ready (pods are still starting), the write would 409 on stale + // rv, and the next periodic Sync writes Running with fresh rv. newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusRunning } @@ -65,6 +87,22 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error { c.logger.Debugf("could not sync finalizers: %v", err) } + // Lifecycle handling lives in its own helper; syncStateLocked only does + // reconciliation work from here on. + proceed, perr := c.prepareLifecycleTransition(&newSpec, oldSpec) + if perr != nil { + return perr + } + if !proceed { + return nil + } + + // If prepareLifecycleTransition already persisted Stopping/Updating, resync + // oldSpec so the defer below doesn't redundantly re-write the same status (409). + if newSpec.Status.PostgresClusterStatus != oldSpec.Status.PostgresClusterStatus { + oldSpec.Status = *newSpec.Status.DeepCopy() + } + if err = c.initUsers(); err != nil { err = fmt.Errorf("could not init users: %v", err) return err diff --git a/pkg/cluster/util.go b/pkg/cluster/util.go index 81e927d94..22fac2759 100644 --- a/pkg/cluster/util.go +++ b/pkg/cluster/util.go @@ -469,6 +469,29 @@ func (c *Cluster) waitStatefulsetPodsReady() error { return nil } +// waitStatefulsetPodsGone blocks until the StatefulSet reports zero replicas in +// its Status (i.e. all pods have been terminated) or the wait times out. Each +// iteration refreshes c.Statefulset so the in-memory cache stays current. +// NotFound is treated as success (the cluster may have been deleted entirely). +// On timeout, returns an error so callers can surface it to the controller. +func (c *Cluster) waitStatefulsetPodsGone() error { + c.setProcessName("waiting for statefulset pods to terminate") + return retryutil.Retry(c.OpConfig.ResourceCheckInterval, c.OpConfig.ResourceCheckTimeout, + func() (bool, error) { + sts, err := c.KubeClient.StatefulSets(c.Namespace).Get( + context.TODO(), c.statefulSetName(), metav1.GetOptions{}) + if err != nil { + if k8sutil.ResourceNotFound(err) { + c.Statefulset = nil + return true, nil + } + return false, fmt.Errorf("could not get statefulset: %w", err) + } + c.Statefulset = sts + return sts.Status.Replicas == 0, nil + }) +} + // Returns labels used to create or list k8s objects such as pods // For backward compatibility, shouldAddExtraLabels must be false // when listing k8s objects. See operator PR #252 diff --git a/pkg/util/k8sutil/k8sutil.go b/pkg/util/k8sutil/k8sutil.go index c34faddd4..0e31112ad 100644 --- a/pkg/util/k8sutil/k8sutil.go +++ b/pkg/util/k8sutil/k8sutil.go @@ -200,6 +200,16 @@ func (client *KubernetesClient) SetPostgresCRDStatus(clusterName spec.Namespaced return pg, nil } +// UpdatePostgresCR of Postgres cluster (updates full resource including spec) +func (client *KubernetesClient) UpdatePostgresCR(clusterName spec.NamespacedName, pg *apiacidv1.Postgresql) (*apiacidv1.Postgresql, error) { + pg, err := client.PostgresqlsGetter.Postgresqls(clusterName.Namespace).Update(context.TODO(), pg, metav1.UpdateOptions{}) + if err != nil { + return pg, fmt.Errorf("could not update PostgresCR: %v", err) + } + + return pg, nil +} + // SetFinalizer of Postgres cluster func (client *KubernetesClient) SetFinalizer(clusterName spec.NamespacedName, pg *apiacidv1.Postgresql, finalizers []string) (*apiacidv1.Postgresql, error) { var (