Method processEvent

pkg/controller/postgresql.go:184–353

(event ClusterEvent)

}

func (c *Controller) processEvent(event ClusterEvent) {
	var clusterName spec.NamespacedName
	var clHistory ringlog.RingLogger
	var err error

	lg := c.logger.WithField("worker", event.WorkerID)

	if event.EventType == EventAdd || event.EventType == EventSync || event.EventType == EventRepair {
		clusterName = util.NameFromMeta(event.NewSpec.ObjectMeta)
	} else {
		clusterName = util.NameFromMeta(event.OldSpec.ObjectMeta)
	}
	lg = lg.WithField("cluster-name", clusterName)

	c.clustersMu.RLock()
	cl, clusterFound := c.clusters[clusterName]
	if clusterFound {
		clHistory = c.clusterHistory[clusterName]
	}
	c.clustersMu.RUnlock()

	defer c.curWorkerCluster.Store(event.WorkerID, nil)

	if event.EventType == EventRepair {
		runRepair, lastOperationStatus := cl.NeedsRepair()
		if !runRepair {
			lg.Debugf("observed cluster status %s, repair is not required", lastOperationStatus)
			return
		}
		lg.Debugf("observed cluster status %s, running sync scan to repair the cluster", lastOperationStatus)
		event.EventType = EventSync
	}

	if event.EventType == EventAdd || event.EventType == EventUpdate || event.EventType == EventSync {
		// handle deprecated parameters by possibly assigning their values to the new ones.
		if event.OldSpec != nil {
			c.mergeDeprecatedPostgreSQLSpecParameters(&event.OldSpec.Spec)
		}
		if event.NewSpec != nil {
			c.warnOnDeprecatedPostgreSQLSpecParameters(&event.NewSpec.Spec)
			c.mergeDeprecatedPostgreSQLSpecParameters(&event.NewSpec.Spec)
		}

		if err = c.submitRBACCredentials(event); err != nil {
			c.logger.Warnf("pods and/or Patroni may misfunction due to the lack of permissions: %v", err)
		}

	}

	switch event.EventType {
	case EventAdd:
		if clusterFound {
			lg.Infof("received add event for already existing Postgres cluster")
			return
		}

		lg.Infof("creating a new Postgres cluster")