Skip to content

Commit 0f4eabe

Browse files
adshin21adshin21
committed
SCF-830: Added cronjob pause/resume while cluster hibernation
1 parent 9424f7c commit 0f4eabe

5 files changed

Lines changed: 275 additions & 4 deletions

File tree

‎docs/user.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -964,6 +964,7 @@ When you apply this manifest, the operator will:
964964
* Set `spec.numberOfInstances` to 0
965965
* Scale down the StatefulSet to 0 replicas
966966
* Scale down the connection pooler deployments to 0 replicas
967+
* Suspend the logical backup CronJob (if enabled)
967968
* Set `status.PostgresClusterStatus` to "Stopping", then "Stopped"
968969

969970
### Waking up a Cluster
@@ -986,6 +987,7 @@ When you apply this manifest, the operator will:
986987

987988
* Restore `numberOfInstances` from `status.previousNumberOfInstances`
988989
* Restore the connection pooler replica counts from `status.previousPoolerInstances`
990+
* Resume the logical backup CronJob (if enabled)
989991
* Scale up the StatefulSet to the previous replica count
990992
* Scale up the connection pooler deployments to the previous replica counts
991993
* Set `status.PostgresClusterStatus` to "Updating", then "Running"
@@ -1019,6 +1021,16 @@ The connection pooler is automatically scaled alongside the cluster:
10191021
Note: If the connection pooler was already at 0 replicas before hibernate, it
10201022
will remain at 0 after wake-up.
10211023

1024+
### Logical Backup Behavior
1025+
1026+
The logical backup CronJob is automatically suspended during hibernate:
1027+
1028+
* When the cluster hibernates, the backup CronJob is suspended (`.spec.suspend: true`)
1029+
* When the cluster wakes up, the backup CronJob is automatically resumed
1030+
* The backup schedule is preserved and resumes from its normal schedule
1031+
1032+
This prevents failed backup jobs from running when the database is unavailable.
1033+
10221034
## Setting up a standby cluster
10231035

10241036
Standby cluster is a [Patroni feature](https://github.com/zalando/patroni/blob/master/docs/replica_bootstrap.rst#standby-cluster)

‎pkg/cluster/cluster.go‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1286,14 +1286,22 @@ func (c *Cluster) handleHibernateAndWakeUp(newSpec *acidv1.Postgresql) (bool, er
12861286
return false, fmt.Errorf("could not scale pooler during hibernate: %w", err)
12871287
}
12881288

1289+
// Suspend logical backup CronJob if enabled
1290+
if c.Spec.EnableLogicalBackup {
1291+
if err := c.suspendLogicalBackupJob(); err != nil {
1292+
return false, fmt.Errorf("could not suspend logical backup job during hibernate: %w", err)
1293+
}
1294+
c.logger.Info("[lifecycle] logical backup job suspended")
1295+
}
1296+
12891297
c.logger.Infof("[lifecycle] hibernate initiated: setting numberOfInstances=0, previousNumberOfInstances=%d", newSpec.Status.PreviousNumberOfInstances)
12901298

12911299
// Update spec first (Update only updates spec when CR has status subresource)
12921300
pgUpdated, err := c.KubeClient.UpdatePostgresCR(c.clusterName(), newSpec)
12931301
if err != nil {
12941302
return false, fmt.Errorf("could not update spec during hibernate: %w", err)
12951303
}
1296-
c.logger.Infof("[lifecycle] hibernate: spec updated successfully")
1304+
c.logger.Info("[lifecycle] hibernate: spec updated successfully")
12971305

12981306
// Update status separately - preserve status values since UpdatePostgresCR returns object with status zeroed
12991307
pgUpdated.Status.PreviousNumberOfInstances = newSpec.Status.PreviousNumberOfInstances
@@ -1320,6 +1328,14 @@ func (c *Cluster) handleHibernateAndWakeUp(newSpec *acidv1.Postgresql) (bool, er
13201328
return false, fmt.Errorf("could not scale pooler during wake-up: %w", err)
13211329
}
13221330

1331+
// Resume logical backup CronJob if enabled
1332+
if c.Spec.EnableLogicalBackup {
1333+
if err := c.unsuspendLogicalBackupJob(); err != nil {
1334+
return false, fmt.Errorf("could not resume logical backup job during wake-up: %w", err)
1335+
}
1336+
c.logger.Info("[lifecycle] logical backup job resumed")
1337+
}
1338+
13231339
// Restore numberOfInstances from previousNumberOfInstances
13241340
newSpec.Spec.NumberOfInstances = newSpec.Status.PreviousNumberOfInstances
13251341
newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusUpdating
@@ -1329,7 +1345,7 @@ func (c *Cluster) handleHibernateAndWakeUp(newSpec *acidv1.Postgresql) (bool, er
13291345
if err != nil {
13301346
return false, fmt.Errorf("could not update spec during wake-up: %w", err)
13311347
}
1332-
c.logger.Infof("[lifecycle] wake-up: spec updated successfully")
1348+
c.logger.Info("[lifecycle] wake-up: spec updated successfully")
13331349

13341350
// Update status separately, and clear previousNumberOfInstances and previousPoolerInstances after restore
13351351
pgUpdated.Status.PreviousNumberOfInstances = 0
@@ -1340,7 +1356,7 @@ func (c *Cluster) handleHibernateAndWakeUp(newSpec *acidv1.Postgresql) (bool, er
13401356
if err != nil {
13411357
return false, fmt.Errorf("could not update status during wake-up: %w", err)
13421358
}
1343-
c.logger.Infof("[lifecycle] wake-up: status updated successfully, previousNumberOfInstances cleared")
1359+
c.logger.Info("[lifecycle] wake-up: status updated successfully, previousNumberOfInstances cleared")
13441360

13451361
c.setSpec(pgUpdated)
13461362
return true, nil

‎pkg/cluster/lifecycle.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ func (c *Cluster) manageHibernateState(oldSpec acidv1.Postgresql, newSpec *acidv
5555
if newSpec.Status.Stopping() {
5656
if c.Statefulset != nil && *c.Statefulset.Spec.Replicas == 0 {
5757
newSpec.Status.PostgresClusterStatus = acidv1.ClusterStatusStopped
58-
c.logger.Infof("[lifecycle] cluster has stopped, all pods are terminated")
58+
c.logger.Info("[lifecycle] cluster has stopped, all pods are terminated")
5959
}
6060
return true
6161
}

‎pkg/cluster/lifecycle_test.go‎

Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
"github.com/zalando/postgres-operator/pkg/util/config"
1313
"github.com/zalando/postgres-operator/pkg/util/k8sutil"
1414
appsv1 "k8s.io/api/apps/v1"
15+
batchv1 "k8s.io/api/batch/v1"
1516
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
1617
"k8s.io/apimachinery/pkg/runtime"
1718
"k8s.io/client-go/kubernetes/fake"
@@ -1232,4 +1233,194 @@ func TestManageHibernateState_StateTransitionSequence(t *testing.T) {
12321233
assert.True(t, continueSync)
12331234
assert.Equal(t, int32(3), newSpec.Spec.NumberOfInstances)
12341235
})
1236+
}
1237+
1238+
func TestSuspendLogicalBackupJob(t *testing.T) {
1239+
tests := []struct {
1240+
name string
1241+
jobExists bool
1242+
patchFails bool
1243+
wantErr bool
1244+
}{
1245+
{
1246+
name: "job exists, suspend succeeds",
1247+
jobExists: true,
1248+
wantErr: false,
1249+
},
1250+
{
1251+
name: "job does not exist - no-op",
1252+
jobExists: false,
1253+
wantErr: false,
1254+
},
1255+
{
1256+
name: "job exists but patch fails",
1257+
jobExists: true,
1258+
patchFails: true,
1259+
wantErr: true,
1260+
},
1261+
}
1262+
1263+
for _, tt := range tests {
1264+
t.Run(tt.name, func(t *testing.T) {
1265+
clientSet := fake.NewSimpleClientset()
1266+
jobName := "logical-backup-test-cluster"
1267+
1268+
if tt.jobExists {
1269+
clientSet.BatchV1().CronJobs("default").Create(context.TODO(), &batchv1.CronJob{
1270+
ObjectMeta: metav1.ObjectMeta{
1271+
Name: jobName,
1272+
Namespace: "default",
1273+
},
1274+
Spec: batchv1.CronJobSpec{
1275+
Schedule: "30 00 * * *",
1276+
},
1277+
}, metav1.CreateOptions{})
1278+
}
1279+
1280+
if tt.patchFails {
1281+
clientSet.PrependReactor("patch", "cronjobs", func(action k8stesting.Action) (bool, runtime.Object, error) {
1282+
return true, nil, fmt.Errorf("network error")
1283+
})
1284+
}
1285+
1286+
kubeClient := &k8sutil.KubernetesClient{
1287+
CronJobsGetter: clientSet.BatchV1(),
1288+
}
1289+
1290+
var job *batchv1.CronJob
1291+
if tt.jobExists {
1292+
job, _ = kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{})
1293+
}
1294+
1295+
c := New(
1296+
Config{
1297+
OpConfig: config.Config{
1298+
LogicalBackup: config.LogicalBackup{
1299+
LogicalBackupJobPrefix: "logical-backup-",
1300+
},
1301+
},
1302+
},
1303+
*kubeClient,
1304+
acidv1.Postgresql{
1305+
ObjectMeta: metav1.ObjectMeta{
1306+
Name: "test-cluster",
1307+
Namespace: "default",
1308+
},
1309+
},
1310+
lifecycleLogger,
1311+
lifecycleEventRecorder,
1312+
)
1313+
c.LogicalBackupJob = job
1314+
1315+
err := c.suspendLogicalBackupJob()
1316+
1317+
if tt.wantErr {
1318+
assert.Error(t, err)
1319+
} else {
1320+
assert.NoError(t, err)
1321+
if tt.jobExists && !tt.patchFails {
1322+
updatedJob, _ := kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{})
1323+
if updatedJob != nil {
1324+
assert.True(t, *updatedJob.Spec.Suspend, "job should be suspended")
1325+
}
1326+
}
1327+
}
1328+
})
1329+
}
1330+
}
1331+
1332+
func TestUnsuspendLogicalBackupJob(t *testing.T) {
1333+
tests := []struct {
1334+
name string
1335+
jobExists bool
1336+
patchFails bool
1337+
wantErr bool
1338+
}{
1339+
{
1340+
name: "job exists, unsuspend succeeds",
1341+
jobExists: true,
1342+
wantErr: false,
1343+
},
1344+
{
1345+
name: "job does not exist - no-op",
1346+
jobExists: false,
1347+
wantErr: false,
1348+
},
1349+
{
1350+
name: "job exists but patch fails",
1351+
jobExists: true,
1352+
patchFails: true,
1353+
wantErr: true,
1354+
},
1355+
}
1356+
1357+
for _, tt := range tests {
1358+
t.Run(tt.name, func(t *testing.T) {
1359+
clientSet := fake.NewSimpleClientset()
1360+
jobName := "logical-backup-test-cluster"
1361+
1362+
if tt.jobExists {
1363+
suspendTrue := true
1364+
clientSet.BatchV1().CronJobs("default").Create(context.TODO(), &batchv1.CronJob{
1365+
ObjectMeta: metav1.ObjectMeta{
1366+
Name: jobName,
1367+
Namespace: "default",
1368+
},
1369+
Spec: batchv1.CronJobSpec{
1370+
Schedule: "30 00 * * *",
1371+
Suspend: &suspendTrue,
1372+
},
1373+
}, metav1.CreateOptions{})
1374+
}
1375+
1376+
if tt.patchFails {
1377+
clientSet.PrependReactor("patch", "cronjobs", func(action k8stesting.Action) (bool, runtime.Object, error) {
1378+
return true, nil, fmt.Errorf("network error")
1379+
})
1380+
}
1381+
1382+
kubeClient := &k8sutil.KubernetesClient{
1383+
CronJobsGetter: clientSet.BatchV1(),
1384+
}
1385+
1386+
var job *batchv1.CronJob
1387+
if tt.jobExists {
1388+
job, _ = kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{})
1389+
}
1390+
1391+
c := New(
1392+
Config{
1393+
OpConfig: config.Config{
1394+
LogicalBackup: config.LogicalBackup{
1395+
LogicalBackupJobPrefix: "logical-backup-",
1396+
},
1397+
},
1398+
},
1399+
*kubeClient,
1400+
acidv1.Postgresql{
1401+
ObjectMeta: metav1.ObjectMeta{
1402+
Name: "test-cluster",
1403+
Namespace: "default",
1404+
},
1405+
},
1406+
lifecycleLogger,
1407+
lifecycleEventRecorder,
1408+
)
1409+
c.LogicalBackupJob = job
1410+
1411+
err := c.unsuspendLogicalBackupJob()
1412+
1413+
if tt.wantErr {
1414+
assert.Error(t, err)
1415+
} else {
1416+
assert.NoError(t, err)
1417+
if tt.jobExists && !tt.patchFails {
1418+
updatedJob, _ := kubeClient.CronJobs("default").Get(context.TODO(), jobName, metav1.GetOptions{})
1419+
if updatedJob != nil {
1420+
assert.False(t, *updatedJob.Spec.Suspend, "job should be unsuspended")
1421+
}
1422+
}
1423+
}
1424+
})
1425+
}
12351426
}

‎pkg/cluster/resources.go‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -825,6 +825,58 @@ func (c *Cluster) deleteLogicalBackupJob() error {
825825
return nil
826826
}
827827

828+
func (c *Cluster) suspendLogicalBackupJob() error {
829+
if c.LogicalBackupJob == nil {
830+
c.logger.Debug("logical backup job is not loaded, skipping suspend")
831+
return nil
832+
}
833+
834+
c.setProcessName("suspending logical backup job")
835+
836+
patchData := fmt.Sprintf(`{"spec":{"suspend":true}}`)
837+
cronJob, err := c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Patch(
838+
context.TODO(),
839+
c.getLogicalBackupJobName(),
840+
types.MergePatchType,
841+
[]byte(patchData),
842+
metav1.PatchOptions{},
843+
"",
844+
)
845+
if err != nil {
846+
return fmt.Errorf("could not suspend logical backup job: %w", err)
847+
}
848+
c.LogicalBackupJob = cronJob
849+
c.logger.Info("logical backup job suspended")
850+
851+
return nil
852+
}
853+
854+
func (c *Cluster) unsuspendLogicalBackupJob() error {
855+
if c.LogicalBackupJob == nil {
856+
c.logger.Debug("logical backup job is not loaded, skipping unsuspend")
857+
return nil
858+
}
859+
860+
c.setProcessName("resuming logical backup job")
861+
862+
patchData := fmt.Sprintf(`{"spec":{"suspend":false}}`)
863+
cronJob, err := c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Patch(
864+
context.TODO(),
865+
c.getLogicalBackupJobName(),
866+
types.MergePatchType,
867+
[]byte(patchData),
868+
metav1.PatchOptions{},
869+
"",
870+
)
871+
if err != nil {
872+
return fmt.Errorf("could not resume logical backup job: %w", err)
873+
}
874+
c.LogicalBackupJob = cronJob
875+
c.logger.Info("logical backup job resumed")
876+
877+
return nil
878+
}
879+
828880
// GetServiceMaster returns cluster's kubernetes master Service
829881
func (c *Cluster) GetServiceMaster() *v1.Service {
830882
return c.Services[Master]

0 commit comments

Comments
 (0)