Backport: add queue to cluster controller for onAdd events (#942)

* Backport of Queue pgbackup PR#915 to operator 3.5

* add queue to cluster controller for onAdd events

* Remove commented code after testing.

Author: flamingdumpster

controller/clustercontroller.go

@@ -18,6 +18,8 @@ limitations under the License.
 import (
 	"context"
 	"fmt"
+	"strings"
+	"io/ioutil"
 	log "github.com/sirupsen/logrus"
 	crv1 "github.com/crunchydata/postgres-operator/apis/cr/v1"
 	"github.com/crunchydata/postgres-operator/kubeapi"
@@ -29,20 +31,25 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
 )
 
 // PgclusterController holds the connections for the controller
 type PgclusterController struct {
 	PgclusterClient    *rest.RESTClient
 	PgclusterScheme    *runtime.Scheme
 	PgclusterClientset *kubernetes.Clientset
+	Queue              workqueue.RateLimitingInterface
 	Namespace          string
 }
 
 // Run starts an pgcluster resource controller
 func (c *PgclusterController) Run(ctx context.Context) error {
 	log.Debug("Watch Pgcluster objects")
 
+	//shut down the work queue to cause workers to end
+	defer c.Queue.ShutDown()
+
 	_, err := c.watchPgclusters(ctx)
 	if err != nil {
 		log.Errorf("Failed to register watch for Pgcluster resource: %v", err)
@@ -95,34 +102,82 @@ func (c *PgclusterController) onAdd(obj interface{}) {
 		return
 	}
 
-	// NEVER modify objects from the store. It's a read-only, local cache.
-	// You can use clusterScheme.Copy() to make a deep copy of original object and modify this copy
-	// Or create a copy manually for better performance
-	copyObj := cluster.DeepCopyObject()
-	clusterCopy := copyObj.(*crv1.Pgcluster)
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err == nil {
+		log.Debugf("cluster putting key in queue %s", key)
+		c.Queue.Add(key)
+	}
+
+}
+
+
+func (c *PgclusterController) RunWorker() {
+
+	//process the 'add' work queue forever
+	for c.processNextItem() {
+	}
+}
 
-	clusterCopy.Status = crv1.PgclusterStatus{
-		State:   crv1.PgclusterStateProcessed,
-		Message: "Successfully processed Pgcluster by controller",
+func (c *PgclusterController) processNextItem() bool {
+	// Wait until there is a new item in the working queue
+	key, quit := c.Queue.Get()
+	if quit {
+		return false
 	}
 
-	err := c.PgclusterClient.Put().
-		Name(cluster.ObjectMeta.Name).
-		Namespace(cluster.ObjectMeta.Namespace).
-		Resource(crv1.PgclusterResourcePlural).
-		Body(clusterCopy).
-		Do().
-		Error()
+	log.Debugf("working on %s", key.(string))
+	keyParts := strings.Split(key.(string), "/")
+	keyNamespace := keyParts[0]
+	keyResourceName := keyParts[1]
+
+	log.Debugf("cluster add 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 cluster
+	// deployment exists , if so, then we don't create another
+	_, found, err := kubeapi.GetDeployment(c.PgclusterClientset, keyResourceName, keyNamespace)
+
+	if found {
+		log.Debugf("cluster add - dep already found, not creating again")
+		return true
+	}
+
+	//get the pgcluster
+	cluster := crv1.Pgcluster{}
+	found, err = kubeapi.Getpgcluster(c.PgclusterClient, &cluster, keyResourceName, keyNamespace)
+	if !found {
+		log.Debugf("cluster add - pgcluster not found, this is invalid")
+		return false
+	}
+
+
+	addIdentifier(&cluster)
+
+	state := crv1.PgclusterStateProcessed
+	message := "Successfully processed Pgcluster by controller"
+
+	err = kubeapi.PatchpgclusterStatus(c.PgclusterClient, state, message, &cluster, keyNamespace)
 	if err != nil {
 		log.Errorf("ERROR updating pgcluster status on add: %s", err.Error())
+		return false
 	}
 
 	log.Debugf("pgcluster added: %s", cluster.ObjectMeta.Name)
 
-	clusteroperator.AddClusterBase(c.PgclusterClientset, c.PgclusterClient, clusterCopy, cluster.ObjectMeta.Namespace)
+	clusteroperator.AddClusterBase(c.PgclusterClientset, c.PgclusterClient, &cluster, cluster.ObjectMeta.Namespace)
+
+	return true
 }
 
+
 // onUpdate is called when a pgcluster is updated
 func (c *PgclusterController) onUpdate(oldObj, newObj interface{}) {
 	oldcluster := oldObj.(*crv1.Pgcluster)
@@ -212,3 +267,12 @@ func getReadyStatus(pod *v1.Pod) (string, bool) {
 	return fmt.Sprintf("%d/%d", readyCount, containerCount), equal
 
 }
+
+func addIdentifier(clusterCopy *crv1.Pgcluster) {
+	u, err := ioutil.ReadFile("/proc/sys/kernel/random/uuid")
+	if err != nil {
+		log.Error(err)
+	}
+
+	clusterCopy.ObjectMeta.Labels[util.LABEL_PG_CLUSTER_IDENTIFIER] = string(u[:len(u)-1])
+}

kubeapi/pgcluster.go

@@ -16,10 +16,13 @@ package kubeapi
 */
 
 import (
+	"encoding/json"
 	log "github.com/sirupsen/logrus"
 	crv1 "github.com/crunchydata/postgres-operator/apis/cr/v1"
+	jsonpatch "github.com/evanphx/json-patch"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
 )
 
@@ -145,3 +148,44 @@ func Updatepgcluster(client *rest.RESTClient, cluster *crv1.Pgcluster, name, nam
 	log.Debugf("updated pgcluster %s", cluster.Name)
 	return err
 }
+
+
+// PatchpgclusterStatus - patches a pgcluster resource
+func PatchpgclusterStatus(restclient *rest.RESTClient, state crv1.PgclusterState, message string, oldCrd *crv1.Pgcluster, namespace string) error {
+
+	oldData, err := json.Marshal(oldCrd)
+	if err != nil {
+		return err
+	}
+
+	//change it
+	oldCrd.Status = crv1.PgclusterStatus{
+		State:   state,
+		Message: message,
+	}
+
+	//create the patch
+	var newData, patchBytes []byte
+	newData, err = json.Marshal(oldCrd)
+	if err != nil {
+		return err
+	}
+	patchBytes, err = jsonpatch.CreateMergePatch(oldData, newData)
+	if err != nil {
+		return err
+	}
+
+	log.Debug(string(patchBytes))
+
+	//apply patch
+	_, err6 := restclient.Patch(types.MergePatchType).
+		Namespace(namespace).
+		Resource(crv1.PgclusterResourcePlural).
+		Name(oldCrd.Spec.Name).
+		Body(patchBytes).
+		Do().
+		Get()
+
+	return err6
+
+}

postgres-operator.go

@@ -97,6 +97,7 @@ func main() {
 		PgclusterClient:    crdClient,
 		PgclusterScheme:    crdScheme,
 		PgclusterClientset: Clientset,
+		Queue:              workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 		Namespace:          Namespace,
 	}
 	pgReplicacontroller := controller.PgreplicaController{
@@ -141,6 +142,7 @@ func main() {
 	defer cancelFunc()
 	go pgTaskcontroller.Run(ctx)
 	go pgClustercontroller.Run(ctx)
+	go pgClustercontroller.RunWorker()
 	go pgReplicacontroller.Run(ctx)
 	go pgReplicacontroller.RunWorker()
 	go pgBackupcontroller.Run(ctx)

util/labels.go

@@ -20,6 +20,7 @@ const LABEL_NAME = "name"
 const LABEL_SELECTOR = "selector"
 const LABEL_OPERATOR = "postgres-operator"
 const LABEL_PG_CLUSTER = "pg-cluster"
+const LABEL_PG_CLUSTER_IDENTIFIER = "pg-cluster-id"
 const LABEL_PG_DATABASE = "pg-database"
 const LABEL_PGBACKUP = "pgbackup"
 const LABEL_PGTASK = "pg-task"