Parallel cluster processing

Run operations concerning multiple clusters in parallel. Each cluster gets its own worker in order to create, update, sync or delete clusters. Each worker acquires the lock on a cluster. Subsequent operations on the same cluster have to wait until the current one finishes. There is a pool of parallel workers, configurable with the `workers` parameter in the configmap and set by default to 4. The cluster-related tasks are assigned to the workers based on a cluster name: the tasks for the same cluster will be always assigned to the same worker. There is no blocking between workers, although there is a chance that a single worker will become a bottleneck if too many clusters are assigned to it; therefore, for large-scale deployments it might be necessary to bump up workers from the default value.
2026-09-30 12:54:25 +02:00 · 2017-05-12 11:41:35 +02:00
parent a9c6c4861c
commit 2370659c69
13 changed files with 304 additions and 143 deletions
@@ -1,6 +1,7 @@
 package controller

 import (
+	"fmt"
 	"sync"

 	"github.com/Sirupsen/logrus"
@@ -28,14 +29,17 @@ type Controller struct {
 	Config
 	opConfig    *config.Config
 	logger      *logrus.Entry
-	clusters    map[spec.NamespacedName]*cluster.Cluster
-	stopChMap   map[spec.NamespacedName]chan struct{}
 	waitCluster sync.WaitGroup

+	clustersMu sync.RWMutex
+	clusters   map[spec.NamespacedName]*cluster.Cluster
+	stopChs    map[spec.NamespacedName]chan struct{}
+
 	postgresqlInformer cache.SharedIndexInformer
 	podInformer        cache.SharedIndexInformer
+	podCh              chan spec.PodEvent

-	podCh chan spec.PodEvent
+	clusterEventQueues []*cache.FIFO
 }

 func New(controllerConfig *Config, operatorConfig *config.Config) *Controller {
@@ -47,12 +51,12 @@ func New(controllerConfig *Config, operatorConfig *config.Config) *Controller {

 	controllerConfig.TeamsAPIClient = teams.NewTeamsAPI(operatorConfig.TeamsAPIUrl, logger)
 	return &Controller{
-		Config:    *controllerConfig,
-		opConfig:  operatorConfig,
-		logger:    logger.WithField("pkg", "controller"),
-		clusters:  make(map[spec.NamespacedName]*cluster.Cluster),
-		stopChMap: make(map[spec.NamespacedName]chan struct{}),
-		podCh:     make(chan spec.PodEvent),
+		Config:   *controllerConfig,
+		opConfig: operatorConfig,
+		logger:   logger.WithField("pkg", "controller"),
+		clusters: make(map[spec.NamespacedName]*cluster.Cluster),
+		stopChs:  make(map[spec.NamespacedName]chan struct{}),
+		podCh:    make(chan spec.PodEvent),
 	}
 }

@@ -64,6 +68,10 @@ func (c *Controller) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {

 	go c.runInformers(stopCh)

+	for i := range c.clusterEventQueues {
+		go c.processClusterEventsQueue(i)
+	}
+
 	c.logger.Info("Started working in background")
 }

@@ -117,6 +125,18 @@ func (c *Controller) initController() {
 	if err := c.initEtcdClient(c.opConfig.EtcdHost); err != nil {
 		c.logger.Fatalf("Can't get etcd client: %s", err)
 	}
+
+	c.clusterEventQueues = make([]*cache.FIFO, c.opConfig.Workers)
+	for i := range c.clusterEventQueues {
+		c.clusterEventQueues[i] = cache.NewFIFO(func(obj interface{}) (string, error) {
+			e, ok := obj.(spec.ClusterEvent)
+			if !ok {
+				return "", fmt.Errorf("Can't cast to ClusterEvent")
+			}
+
+			return fmt.Sprintf("%s-%s", e.EventType, e.UID), nil
+		})
+	}
 }

 func (c *Controller) runInformers(stopCh <-chan struct{}) {
@@ -65,7 +65,7 @@ func (c *Controller) podAdd(obj interface{}) {
 		ClusterName: c.PodClusterName(pod),
 		PodName:     util.NameFromMeta(pod.ObjectMeta),
 		CurPod:      pod,
-		EventType:   spec.PodEventAdd,
+		EventType:   spec.EventAdd,
 	}

 	c.podCh <- podEvent
@@ -87,7 +87,7 @@ func (c *Controller) podUpdate(prev, cur interface{}) {
 		PodName:     util.NameFromMeta(curPod.ObjectMeta),
 		PrevPod:     prevPod,
 		CurPod:      curPod,
-		EventType:   spec.PodEventUpdate,
+		EventType:   spec.EventUpdate,
 	}

 	c.podCh <- podEvent
@@ -103,7 +103,7 @@ func (c *Controller) podDelete(obj interface{}) {
 		ClusterName: c.PodClusterName(pod),
 		PodName:     util.NameFromMeta(pod.ObjectMeta),
 		CurPod:      pod,
-		EventType:   spec.PodEventDelete,
+		EventType:   spec.EventDelete,
 	}

 	c.podCh <- podEvent
@@ -114,7 +114,11 @@ func (c *Controller) podEventsDispatcher(stopCh <-chan struct{}) {
 	for {
 		select {
 		case event := <-c.podCh:
-			if subscriber, ok := c.clusters[event.ClusterName]; ok {
+			c.clustersMu.RLock()
+			subscriber, ok := c.clusters[event.ClusterName]
+			c.clustersMu.RUnlock()
+
+			if ok {
 				c.logger.Debugf("Sending %s event of Pod '%s' to the '%s' cluster channel", event.EventType, event.PodName, event.ClusterName)
 				go subscriber.ReceivePodEvent(event)
 			}
@@ -8,7 +8,9 @@ import (
 	"k8s.io/client-go/pkg/api/meta"
 	"k8s.io/client-go/pkg/fields"
 	"k8s.io/client-go/pkg/runtime"
+	"k8s.io/client-go/pkg/types"
 	"k8s.io/client-go/pkg/watch"
+	"k8s.io/client-go/tools/cache"

 	"github.bus.zalan.do/acid/postgres-operator/pkg/cluster"
 	"github.bus.zalan.do/acid/postgres-operator/pkg/spec"
@@ -35,24 +37,17 @@ func (c *Controller) clusterListFunc(options api.ListOptions) (runtime.Object, e
 		return nil, fmt.Errorf("Can't extract list of postgresql objects: %s", err)
 	}

-	clusterConfig := c.makeClusterConfig()
 	for _, obj := range objList {
 		pg, ok := obj.(*spec.Postgresql)
 		if !ok {
 			return nil, fmt.Errorf("Can't cast object to postgresql")
 		}
-		clusterName := util.NameFromMeta(pg.Metadata)
-		cl := cluster.New(clusterConfig, *pg, c.logger.Logger)
+		c.queueClusterEvent(nil, pg, spec.EventSync)

-		stopCh := make(chan struct{})
-		c.stopChMap[clusterName] = stopCh
-		c.clusters[clusterName] = cl
-		cl.LoadResources()
-		cl.ListResources()
-		cl.SyncCluster(stopCh)
+		c.logger.Debugf("Sync of the '%s' cluster has been queued", util.NameFromMeta(pg.Metadata))
 	}
-	if len(c.clusters) > 0 {
-		c.logger.Infof("There are %d clusters currently running", len(c.clusters))
+	if len(objList) > 0 {
+		c.logger.Infof("There are %d clusters currently running", len(objList))
 	} else {
 		c.logger.Infof("No clusters running")
 	}
@@ -60,6 +55,135 @@ func (c *Controller) clusterListFunc(options api.ListOptions) (runtime.Object, e
 	return object, err
 }

+func (c *Controller) processEvent(obj interface{}) error {
+	var clusterName spec.NamespacedName
+
+	event, ok := obj.(spec.ClusterEvent)
+	if !ok {
+		return fmt.Errorf("Can't cast to ClusterEvent")
+	}
+	logger := c.logger.WithField("worker", event.WorkerID)
+
+	if event.EventType == spec.EventAdd || event.EventType == spec.EventSync {
+		clusterName = util.NameFromMeta(event.NewSpec.Metadata)
+	} else {
+		clusterName = util.NameFromMeta(event.OldSpec.Metadata)
+	}
+
+	c.clustersMu.RLock()
+	cl, clusterFound := c.clusters[clusterName]
+	stopCh := c.stopChs[clusterName]
+	c.clustersMu.RUnlock()
+
+	switch event.EventType {
+	case spec.EventAdd:
+		if clusterFound {
+			logger.Debugf("Cluster '%s' already exists", clusterName)
+			return nil
+		}
+
+		logger.Infof("Creation of the '%s' cluster started", clusterName)
+
+		stopCh := make(chan struct{})
+		cl = cluster.New(c.makeClusterConfig(), *event.NewSpec, logger)
+
+		c.clustersMu.Lock()
+		c.clusters[clusterName] = cl
+		c.stopChs[clusterName] = stopCh
+		c.clustersMu.Unlock()
+
+		if err := cl.Create(stopCh); err != nil {
+			logger.Errorf("Can't create cluster: %s", err)
+			return nil
+		}
+
+		logger.Infof("Cluster '%s' has been created", clusterName)
+	case spec.EventUpdate:
+		logger.Infof("Update of the '%s' cluster started", clusterName)
+
+		if !clusterFound {
+			logger.Warnf("Cluster '%s' does not exist", clusterName)
+			return nil
+		}
+		if err := cl.Update(event.NewSpec); err != nil {
+			logger.Errorf("Can't update cluster: %s", err)
+			return nil
+		}
+		logger.Infof("Cluster '%s' has been updated", clusterName)
+	case spec.EventDelete:
+		logger.Infof("Deletion of the '%s' cluster started", clusterName)
+		if !clusterFound {
+			logger.Errorf("Unknown cluster: %s", clusterName)
+			return nil
+		}
+
+		if err := cl.Delete(); err != nil {
+			logger.Errorf("Can't delete cluster '%s': %s", clusterName, err)
+			return nil
+		}
+		close(c.stopChs[clusterName])
+
+		c.clustersMu.Lock()
+		delete(c.clusters, clusterName)
+		delete(c.stopChs, clusterName)
+		c.clustersMu.Unlock()
+
+		logger.Infof("Cluster '%s' has been deleted", clusterName)
+	case spec.EventSync:
+		logger.Infof("Syncing of the '%s' cluster started", clusterName)
+
+		// no race condition because a cluster is always processed by single worker
+		if !clusterFound {
+			cl = cluster.New(c.makeClusterConfig(), *event.NewSpec, logger)
+			stopCh = make(chan struct{})
+
+			c.clustersMu.Lock()
+			c.clusters[clusterName] = cl
+			c.stopChs[clusterName] = stopCh
+			c.clustersMu.Unlock()
+		}
+
+		cl.SyncCluster(stopCh)
+
+		logger.Infof("Cluster '%s' has been synced", clusterName)
+	}
+
+	return nil
+}
+
+func (c *Controller) processClusterEventsQueue(idx int) {
+	for {
+		c.clusterEventQueues[idx].Pop(cache.PopProcessFunc(c.processEvent))
+	}
+}
+
+func (c *Controller) queueClusterEvent(old, new *spec.Postgresql, eventType spec.EventType) {
+	var (
+		uid types.UID
+		clusterName spec.NamespacedName
+	)
+
+	if old != nil {
+		uid = old.Metadata.GetUID()
+		clusterName = util.NameFromMeta(old.Metadata)
+	} else {
+		uid = new.Metadata.GetUID()
+		clusterName = util.NameFromMeta(new.Metadata)
+	}
+	workerId := c.clusterWorkerId(clusterName)
+	clusterEvent := spec.ClusterEvent{
+		EventType: eventType,
+		UID:       uid,
+		OldSpec:   old,
+		NewSpec:   new,
+		WorkerID:  workerId,
+	}
+	//TODO: if we delete cluster, discard all the previous events for the cluster
+
+	c.clusterEventQueues[workerId].Add(clusterEvent)
+	c.logger.WithField("worker", workerId).Infof("%s of the '%s' cluster has been queued for", eventType, clusterName)
+}
+
 func (c *Controller) clusterWatchFunc(options api.ListOptions) (watch.Interface, error) {
 	return c.RestClient.Get().
 		Prefix("watch").
@@ -77,34 +201,12 @@ func (c *Controller) postgresqlAdd(obj interface{}) {
 		return
 	}

-	clusterName := util.NameFromMeta(pg.Metadata)
-
-	_, ok = c.clusters[clusterName]
-	if ok {
-		c.logger.Infof("Cluster '%s' already exists", clusterName)
-		return
-	}
-
-	c.logger.Infof("Creation of a new Postgresql cluster '%s' started", clusterName)
-	cl := cluster.New(c.makeClusterConfig(), *pg, c.logger.Logger)
-
-	c.clusters[clusterName] = cl
-	stopCh := make(chan struct{})
-	c.stopChMap[clusterName] = stopCh
-
-	cl.SetStatus(spec.ClusterStatusCreating)
-	if err := cl.Create(stopCh); err != nil {
-		c.logger.Errorf("Can't create cluster: %s", err)
-		cl.SetStatus(spec.ClusterStatusAddFailed)
-		return
-	}
-	cl.SetStatus(spec.ClusterStatusRunning) //TODO: are you sure it's running?
-
-	c.logger.Infof("Postgresql cluster '%s' has been created", clusterName)
+	// We will not get multiple Add events for the same cluster
+	c.queueClusterEvent(nil, pg, spec.EventAdd)
 }

 func (c *Controller) postgresqlUpdate(prev, cur interface{}) {
-	pgPrev, ok := prev.(*spec.Postgresql)
+	pgOld, ok := prev.(*spec.Postgresql)
 	if !ok {
 		c.logger.Errorf("Can't cast to postgresql spec")
 	}
@@ -112,51 +214,22 @@ func (c *Controller) postgresqlUpdate(prev, cur interface{}) {
 	if !ok {
 		c.logger.Errorf("Can't cast to postgresql spec")
 	}
-
-	clusterName := util.NameFromMeta(pgNew.Metadata)
-
-	//TODO: Do not update cluster which is currently creating
-	if pgPrev.Metadata.ResourceVersion == pgNew.Metadata.ResourceVersion {
-		c.logger.Infof("Skipping update with no resource version change")
+	if pgOld.Metadata.ResourceVersion == pgNew.Metadata.ResourceVersion {
 		return
 	}
-	pgCluster := c.clusters[clusterName] // current
-	if reflect.DeepEqual(pgPrev.Spec, pgNew.Spec) {
-		c.logger.Infof("Skipping update with no spec change")
+	if reflect.DeepEqual(pgOld.Spec, pgNew.Spec) {
 		return
 	}

-	pgCluster.SetStatus(spec.ClusterStatusUpdating)
-	if err := pgCluster.Update(pgNew); err != nil {
-		pgCluster.SetStatus(spec.ClusterStatusUpdateFailed)
-		c.logger.Errorf("Can't update cluster: %s", err)
-	} else {
-		c.logger.Infof("Cluster has been updated")
-	}
-	pgCluster.SetStatus(spec.ClusterStatusRunning)
+	c.queueClusterEvent(pgOld, pgNew, spec.EventUpdate)
 }

 func (c *Controller) postgresqlDelete(obj interface{}) {
-	pgCur, ok := obj.(*spec.Postgresql)
+	pg, ok := obj.(*spec.Postgresql)
 	if !ok {
 		c.logger.Errorf("Can't cast to postgresql spec")
 		return
 	}
-	clusterName := util.NameFromMeta(pgCur.Metadata)
-	pgCluster, ok := c.clusters[clusterName]
-	if !ok {
-		c.logger.Errorf("Unknown cluster: %s", clusterName)
-		return
-	}

-	c.logger.Infof("Starting deletion of the '%s' cluster", util.NameFromMeta(pgCur.Metadata))
-	if err := pgCluster.Delete(); err != nil {
-		c.logger.Errorf("Can't delete cluster '%s': %s", clusterName, err)
-		return
-	}
-
-	close(c.stopChMap[clusterName])
-	delete(c.clusters, clusterName)
-
-	c.logger.Infof("Cluster '%s' has been successfully deleted", clusterName)
+	c.queueClusterEvent(pg, nil, spec.EventDelete)
 }
@@ -2,24 +2,31 @@ package controller

 import (
 	"fmt"
+	"hash/crc32"

 	"k8s.io/client-go/pkg/api/v1"
 	extv1beta "k8s.io/client-go/pkg/apis/extensions/v1beta1"

 	"github.bus.zalan.do/acid/postgres-operator/pkg/cluster"
 	"github.bus.zalan.do/acid/postgres-operator/pkg/spec"
+	"github.bus.zalan.do/acid/postgres-operator/pkg/util/config"
 	"github.bus.zalan.do/acid/postgres-operator/pkg/util/constants"
 	"github.bus.zalan.do/acid/postgres-operator/pkg/util/k8sutil"
 )

 func (c *Controller) makeClusterConfig() cluster.Config {
+	infrastructureRoles := make(map[string]spec.PgUser)
+	for k, v := range c.InfrastructureRoles {
+		infrastructureRoles[k] = v
+	}
+
 	return cluster.Config{
 		KubeClient:          c.KubeClient,
 		RestClient:          c.RestClient,
 		EtcdClient:          c.EtcdClient,
 		TeamsAPIClient:      c.TeamsAPIClient,
-		OpConfig:            c.opConfig,
-		InfrastructureRoles: c.InfrastructureRoles,
+		OpConfig:            config.Copy(c.opConfig),
+		InfrastructureRoles: infrastructureRoles,
 	}
 }

@@ -55,6 +62,10 @@ func thirdPartyResource(TPRName string) *extv1beta.ThirdPartyResource {
 	}
 }

+func (c *Controller) clusterWorkerId(clusterName spec.NamespacedName) uint32 {
+	return crc32.ChecksumIEEE([]byte(clusterName.String())) % c.opConfig.Workers
+}
+
 func (c *Controller) createTPR() error {
 	TPRName := fmt.Sprintf("%s.%s", constants.TPRName, constants.TPRVendor)
 	tpr := thirdPartyResource(TPRName)