Backport of Queue pgbackup PR#915 to operator 3.5 (#939)

Author: flamingdumpster

apis/cr/v1/backup.go

@@ -23,6 +23,7 @@ import (
 const PgbackupResourcePlural = "pgbackups"
 
 // Backup job processing states - used by pgdump
+const PgBackupJobReSubmitted = "Backup Job Re-Submitted"
 const PgBackupJobSubmitted = "Backup Job Submitted"
 const PgBackupJobInProgress = "Backup Job In Progress"
 const PgBackupJobCompleted = "Backup Job Completed"

controller/backupcontroller.go

@@ -24,9 +24,13 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
+	"strings"
 
 	crv1 "github.com/crunchydata/postgres-operator/apis/cr/v1"
+	"github.com/crunchydata/postgres-operator/kubeapi"
+
 	backupoperator "github.com/crunchydata/postgres-operator/operator/backup"
+	"k8s.io/client-go/util/workqueue"
 )
 
 // PgbackupController holds connections required by the controller
@@ -35,19 +39,28 @@ type PgbackupController struct {
 	PgbackupScheme    *runtime.Scheme
 	PgbackupClientset *kubernetes.Clientset
 	Namespace         string
+	Queue	workqueue.RateLimitingInterface
+	UpdateQueue workqueue.RateLimitingInterface
 }
 
 // Run starts controller
 func (c *PgbackupController) Run(ctx context.Context) error {
 	log.Debugf("Watch Pgbackup objects")
 
+	//shut down the work queue to cause workers to end
+	defer c.Queue.ShutDown()
+	defer c.UpdateQueue.ShutDown()
+
 	_, err := c.watchPgbackups(ctx)
 	if err != nil {
 		log.Errorf("Failed to register watch for Pgbackup resource: %v", err)
 		return err
 	}
 
 	<-ctx.Done()
+
+	log.Debugf("Watch Pgbackup ending")
+
 	return ctx.Err()
 }
 
@@ -81,6 +94,148 @@ func (c *PgbackupController) watchPgbackups(ctx context.Context) (cache.Controll
 	return controller, nil
 }
 
+func (c *PgbackupController) RunWorker() {
+
+	//process the 'add' work queue forever
+	for c.processNextItem() {
+	}
+}
+
+func (c *PgbackupController) RunUpdateWorker() {
+
+	//process the 'add' work queue forever
+	for c.processNextUpdateItem() {
+	}
+}
+
+func (c *PgbackupController) processNextUpdateItem() bool {
+	// Wait until there is a new item in the working queue
+	key, quit := c.UpdateQueue.Get()
+	if quit {
+		return false
+	}
+
+	log.Debugf("working on %s", key.(string))
+	keyParts := strings.Split(key.(string), "/")
+	keyNamespace := keyParts[0]
+	keyResourceName := keyParts[1]
+
+	log.Debugf("update queue got key ns=[%s] resource=[%s]", keyNamespace, keyResourceName)
+
+	// Tell the queue that we are done with processing this key. This unblocks the key for other workers
+	// This allows safe parallel processing because two pods with the same key are never processed in
+	// parallel.
+	defer c.UpdateQueue.Done(key)
+
+	// Invoke the method containing the business logic
+	// for pgbackups, the convention is the CRD name is always
+	// the same as the pg-cluster label value
+
+	// in this case, the de-dupe logic is to test whether a backup
+	// job is already running, if so, then we don't create another
+	// backup job
+	selector := "pg-cluster=" + keyResourceName + ",pgbackup=true"
+	jobs, err := kubeapi.GetJobs(c.PgbackupClientset, selector, keyNamespace)
+	if err != nil {
+		log.Errorf("update working...error found " + err.Error())
+		return true
+	}
+
+	jobRunning := false
+	for _, j := range jobs.Items {
+		if j.Status.Succeeded <= 0 {
+			jobRunning = true
+		}
+	}
+
+	if jobRunning {
+		log.Debugf("update working...found job already, would do nothing")
+	} else {
+		log.Debugf("update working...no job found, means we process")
+		b := crv1.Pgbackup{}
+		found, err := kubeapi.Getpgbackup(c.PgbackupClient, &b, keyResourceName, keyNamespace)
+		if found {
+			state := crv1.PgbackupStateProcessed
+			message := "Successfully processed Pgbackup by controller"
+			err = kubeapi.PatchpgbackupStatus(c.PgbackupClient, state, message, &b, b.ObjectMeta.Namespace)
+			if err != nil {
+				log.Errorf("ERROR updating pgbackup status: %s", err.Error())
+			}
+
+			backupoperator.AddBackupBase(c.PgbackupClientset, c.PgbackupClient, &b, b.ObjectMeta.Namespace)
+
+			//no error, tell the queue to stop tracking history
+			c.UpdateQueue.Forget(key)
+		}
+	}
+
+	return true
+}
+
+func (c *PgbackupController) processNextItem() bool {
+	// Wait until there is a new item in the working queue
+	key, quit := c.Queue.Get()
+	
+	if quit {
+		return false
+	}
+
+	log.Debugf("working on %s", key.(string))
+	keyParts := strings.Split(key.(string), "/")
+	keyNamespace := keyParts[0]
+	keyResourceName := keyParts[1]
+
+	log.Debugf("queue got key ns=[%s] resource=[%s]", keyNamespace, keyResourceName)
+
+	// Tell the queue that we are done with processing this key. This unblocks the key for other workers
+	// This allows safe parallel processing because two pods with the same key are never processed in
+	// parallel.
+	defer c.Queue.Done(key)
+
+	// Invoke the method containing the business logic
+	// for pgbackups, the convention is the CRD name is always
+	// the same as the pg-cluster label value
+
+	// in this case, the de-dupe logic is to test whether a backup
+	// job is already running, if so, then we don't create another
+	// backup job
+	selector := "pg-cluster=" + keyResourceName + ",pgbackup=true"
+	jobs, err := kubeapi.GetJobs(c.PgbackupClientset, selector, keyNamespace)
+	if err != nil {
+		log.Errorf("working...error found " + err.Error())
+		return true
+	}
+
+	jobRunning := false
+	for _, j := range jobs.Items {
+		if j.Status.Succeeded <= 0 {
+			jobRunning = true
+		}
+	}
+
+	if jobRunning {
+		log.Debugf("working...found job already, would do nothing")
+	} else {
+		log.Debugf("working...no job found, means we process")
+		b := crv1.Pgbackup{}
+		found, err := kubeapi.Getpgbackup(c.PgbackupClient, &b, keyResourceName, keyNamespace)
+		if found {
+			state := crv1.PgbackupStateProcessed
+			message := "Successfully processed Pgbackup by controller"
+			err = kubeapi.PatchpgbackupStatus(c.PgbackupClient, state, message, &b, b.ObjectMeta.Namespace)
+			if err != nil {
+				log.Errorf("ERROR updating pgbackup status: %s", err.Error())
+			}
+			backupoperator.AddBackupBase(c.PgbackupClientset, c.PgbackupClient, &b, b.ObjectMeta.Namespace)
+
+			//no error, tell the queue to stop tracking history
+			c.Queue.Forget(key)
+		}
+	}
+	return true
+}
+
+
 // onAdd is called when a pgbackup is added
 func (c *PgbackupController) onAdd(obj interface{}) {
 	backup := obj.(*crv1.Pgbackup)
@@ -93,37 +248,32 @@ func (c *PgbackupController) onAdd(obj interface{}) {
 		return
 	}
 
-	// NEVER modify objects from the store. It's a read-only, local cache.
-	// You can use backupScheme.Copy() to make a deep copy of original object and modify this copy
-	// Or create a copy manually for better performance
-	copyObj := backup.DeepCopyObject()
-
-	backupCopy := copyObj.(*crv1.Pgbackup)
-	backupCopy.Status = crv1.PgbackupStatus{
-		State:   crv1.PgbackupStateProcessed,
-		Message: "Successfully processed Pgbackup by controller",
-	}
-
-	err := c.PgbackupClient.Put().
-		Name(backup.ObjectMeta.Name).
-		Namespace(backup.ObjectMeta.Namespace).
-		Resource(crv1.PgbackupResourcePlural).
-		Body(backupCopy).
-		Do().
-		Error()
-
-	if err != nil {
-		log.Errorf("ERROR updating pgbackup status: %s", err.Error())
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err == nil {
+		log.Debugf("[PgbackupController] putting key in queue %s", key)
+		c.Queue.Add(key)
 	}
 
 	//handle new pgbackups
-	backupoperator.AddBackupBase(c.PgbackupClientset, c.PgbackupClient, backupCopy, backup.ObjectMeta.Namespace)
+	//	backupoperator.AddBackupBase(c.PgbackupClientset, c.PgbackupClient, backupCopy, backup.ObjectMeta.Namespace)
 }
 
 // onUpdate is called when a pgbackup is updated
 func (c *PgbackupController) onUpdate(oldObj, newObj interface{}) {
+	oldBackup := oldObj.(*crv1.Pgbackup)
 	backup := newObj.(*crv1.Pgbackup)
 	log.Debugf("[PgbackupController] ns=%s onUpdate %s", backup.ObjectMeta.Namespace, backup.ObjectMeta.SelfLink)
+
+	// check for re-submission of backup
+	if oldBackup.Spec.BackupStatus != crv1.PgBackupJobReSubmitted && backup.Spec.BackupStatus == crv1.PgBackupJobReSubmitted {
+		log.Debugf("[PgbackupController] ns=%s onUpdate %s re-submitted", backup.ObjectMeta.Namespace, backup.ObjectMeta.SelfLink)
+		key, err := cache.MetaNamespaceKeyFunc(oldObj)
+		if err == nil {
+			log.Debugf("[PgbackupController] putting key in update queue %s", key)
+			c.UpdateQueue.Add(key)
+	}
+}
+
 }
 
 // onDelete is called when a pgbackup is deleted

controller/jobcontroller.go

@@ -117,6 +117,7 @@ func (c *JobController) onUpdate(oldObj, newObj interface{}) {
 			status = crv1.JobErrorStatus
 		}
 
+		// change this to patchpgbackupbackupstatus ?? 
 		if labels[util.LABEL_BACKREST] != "true" {
 			err = util.Patch(c.JobClient, "/spec/backupstatus", status, "pgbackups", dbname, job.ObjectMeta.Namespace)
 			if err != nil {