Skip to content
Snippets Groups Projects
Commit 8f5a8e21 authored by Murat Kabilov's avatar Murat Kabilov
Browse files

split processEvent function into several functions

parent d39cfaab
No related branches found
No related tags found
No related merge requests found
......@@ -74,36 +74,13 @@ func (c *Controller) clusterWatchFunc(options api.ListOptions) (watch.Interface,
return req.Watch()
}
func (c *Controller) processEvent(obj interface{}) error {
var clusterName spec.NamespacedName
event, ok := obj.(spec.ClusterEvent)
if !ok {
return fmt.Errorf("could not 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]
c.clustersMu.RUnlock()
switch event.EventType {
case spec.EventAdd:
if clusterFound {
logger.Debugf("Cluster '%s' already exists", clusterName)
return nil
}
func (c *Controller) processAddEvent(workerID uint32, clusterName spec.NamespacedName, spec *spec.Postgresql) {
log := c.logger.WithField("worker", workerID)
logger.Infof("Creation of the '%s' cluster started", clusterName)
log.Infof("Creation of the '%s' cluster started", clusterName)
stopCh := make(chan struct{})
cl = cluster.New(c.makeClusterConfig(), *event.NewSpec, logger)
cl := cluster.New(c.makeClusterConfig(), *spec, log)
cl.Run(stopCh)
c.clustersMu.Lock()
......@@ -113,37 +90,39 @@ func (c *Controller) processEvent(obj interface{}) error {
if err := cl.Create(); err != nil {
cl.Error = fmt.Errorf("could not create cluster: %v", err)
logger.Errorf("%v", cl.Error)
log.Errorf("%v", cl.Error)
return nil
return
}
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
log.Infof("Cluster '%s' has been created", clusterName)
}
if err := cl.Update(event.NewSpec); err != nil {
func (c *Controller) processUpdateEvent(workerID uint32, clusterName spec.NamespacedName, cl *cluster.Cluster, newSpec *spec.Postgresql) {
log := c.logger.WithField("worker", workerID)
log.Infof("Update of the '%s' cluster started", clusterName)
if err := cl.Update(newSpec); err != nil {
cl.Error = fmt.Errorf("could not update cluster: %s", err)
logger.Errorf("%v", cl.Error)
log.Errorf("%v", cl.Error)
return nil
return
}
cl.Error = 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
log.Infof("Cluster '%s' has been updated", clusterName)
}
func (c *Controller) processDeleteEvent(workerID uint32, clusterName spec.NamespacedName, cl *cluster.Cluster) {
log := c.logger.WithField("worker", workerID)
log.Infof("Deletion of the '%s' cluster started", clusterName)
if err := cl.Delete(); err != nil {
logger.Errorf("could not delete cluster '%s': %s", clusterName, err)
return nil
log.Errorf("could not delete cluster '%s': %s", clusterName, err)
return
}
close(c.stopChs[clusterName])
......@@ -152,14 +131,18 @@ func (c *Controller) processEvent(obj interface{}) error {
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)
log.Infof("Cluster '%s' has been deleted", clusterName)
}
func (c *Controller) processSyncEvent(workerID uint32, cl *cluster.Cluster, clusterFound bool, clusterName spec.NamespacedName, newSpec *spec.Postgresql) {
log := c.logger.WithField("worker", workerID)
log.Infof("Syncing of the '%s' cluster started", clusterName)
// no race condition because a cluster is always processed by single worker
if !clusterFound {
stopCh := make(chan struct{})
cl = cluster.New(c.makeClusterConfig(), *event.NewSpec, logger)
cl := cluster.New(c.makeClusterConfig(), *newSpec, log)
cl.Run(stopCh)
c.clustersMu.Lock()
......@@ -170,12 +153,58 @@ func (c *Controller) processEvent(obj interface{}) error {
if err := cl.Sync(); err != nil {
cl.Error = fmt.Errorf("could not sync cluster '%s': %s", clusterName, err)
logger.Errorf("%v", cl)
return nil
log.Errorf("%v", cl)
return
}
cl.Error = nil
logger.Infof("Cluster '%s' has been synced", clusterName)
log.Infof("Cluster '%s' has been synced", clusterName)
}
func (c *Controller) processEvent(obj interface{}) error {
var clusterName spec.NamespacedName
event, ok := obj.(spec.ClusterEvent)
if !ok {
return fmt.Errorf("could not cast to ClusterEvent")
}
log := 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]
c.clustersMu.RUnlock()
switch event.EventType {
case spec.EventAdd:
if clusterFound {
log.Debugf("Cluster '%s' already exists", clusterName)
return nil
}
c.processAddEvent(event.WorkerID, clusterName, event.NewSpec)
case spec.EventUpdate:
if !clusterFound {
log.Warnf("Cluster '%s' does not exist", clusterName)
return nil
}
c.processUpdateEvent(event.WorkerID, clusterName, cl, event.NewSpec)
case spec.EventDelete:
if !clusterFound {
log.Errorf("Unknown cluster: %s", clusterName)
return nil
}
c.processDeleteEvent(event.WorkerID, clusterName, cl)
case spec.EventSync:
c.processSyncEvent(event.WorkerID, cl, clusterFound, clusterName, event.NewSpec)
}
return nil
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment