fix #2

Signed-off-by: Mikhail Scherba <[email protected]>

Author: miklezzzz

pkg/hook/controller/kubernetes_bindings_controller.go

@@ -85,12 +85,10 @@ func (c *kubernetesBindingsController) EnableKubernetesBindings() ([]BindingExec
 		if err != nil {
 			return nil, fmt.Errorf("run monitor: %s", err)
 		}
-		c.l.Lock()
-		c.BindingMonitorLinks[config.Monitor.Metadata.MonitorId] = &KubernetesBindingToMonitorLink{
+		c.setBindingMonitorLinks(config.Monitor.Metadata.MonitorId, &KubernetesBindingToMonitorLink{
 			MonitorId:     config.Monitor.Metadata.MonitorId,
 			BindingConfig: config,
-		}
-		c.l.Unlock()
+		})
 		// Start monitor's informers to fill the cache.
 		c.kubeEventsManager.StartMonitor(config.Monitor.Metadata.MonitorId)
 
@@ -106,9 +104,7 @@ func (c *kubernetesBindingsController) EnableKubernetesBindings() ([]BindingExec
 
 func (c *kubernetesBindingsController) UpdateMonitor(monitorId string, kind, apiVersion string) error {
 	// Find binding for monitorId
-	c.l.RLock()
-	link, ok := c.BindingMonitorLinks[monitorId]
-	c.l.RUnlock()
+	link, ok := c.getBindingMonitorLinksById(monitorId)
 	if !ok {
 		return nil
 	}
@@ -158,12 +154,11 @@ func (c *kubernetesBindingsController) UpdateMonitor(monitorId string, kind, api
 
 // UnlockEvents turns on eventCb for all monitors to emit events after Synchronization.
 func (c *kubernetesBindingsController) UnlockEvents() {
-	c.l.RLock()
-	for monitorID := range c.BindingMonitorLinks {
+	c.iterateBindingMonitorLinks(func(monitorID string) bool {
 		m := c.kubeEventsManager.GetMonitor(monitorID)
 		m.EnableKubeEventCb()
-	}
-	c.l.RUnlock()
+		return true
+	})
 }
 
 // UnlockEventsFor turns on eventCb for matched monitor to emit events after Synchronization.
@@ -179,30 +174,53 @@ func (c *kubernetesBindingsController) UnlockEventsFor(monitorID string) {
 // StopMonitors stops all monitors for the hook.
 // TODO handle error!
 func (c *kubernetesBindingsController) StopMonitors() {
-	c.l.RLock()
-	for monitorID := range c.BindingMonitorLinks {
+	c.iterateBindingMonitorLinks(func(monitorID string) bool {
 		_ = c.kubeEventsManager.StopMonitor(monitorID)
-	}
-	c.l.RUnlock()
+		return true
+	})
 }
 
 func (c *kubernetesBindingsController) CanHandleEvent(kubeEvent kemtypes.KubeEvent) bool {
-	c.l.RLock()
-	defer c.l.RUnlock()
-	for key := range c.BindingMonitorLinks {
-		if key == kubeEvent.MonitorId {
+	var canHandleEvent bool
+
+	c.iterateBindingMonitorLinks(func(monitorID string) bool {
+		if monitorID == kubeEvent.MonitorId {
+			canHandleEvent = true
 			return true
 		}
+		return false
+	})
+
+	return canHandleEvent
+}
+
+func (c *kubernetesBindingsController) iterateBindingMonitorLinks(doFn func(monitorID string) bool) {
+	c.l.RLock()
+	for monitorID := range c.BindingMonitorLinks {
+		if exit := doFn(monitorID); exit {
+			break
+		}
 	}
-	return false
+	c.l.RUnlock()
+}
+
+func (c *kubernetesBindingsController) getBindingMonitorLinksById(monitorId string) (*KubernetesBindingToMonitorLink, bool) {
+	c.l.RLock()
+	link, found := c.BindingMonitorLinks[monitorId]
+	c.l.RUnlock()
+	return link, found
+}
+
+func (c *kubernetesBindingsController) setBindingMonitorLinks(monitorId string, link *KubernetesBindingToMonitorLink) {
+	c.l.Lock()
+	c.BindingMonitorLinks[monitorId] = link
+	c.l.Unlock()
 }
 
 // HandleEvent receives event from KubeEventManager and returns a BindingExecutionInfo
 // to help create a new task to run a hook.
 func (c *kubernetesBindingsController) HandleEvent(kubeEvent kemtypes.KubeEvent) BindingExecutionInfo {
-	c.l.RLock()
-	link, hasKey := c.BindingMonitorLinks[kubeEvent.MonitorId]
-	c.l.RUnlock()
+	link, hasKey := c.getBindingMonitorLinksById(kubeEvent.MonitorId)
 	if !hasKey {
 		log.Error("Possible bug!!! Unknown kube event: no such monitor id registered", slog.String("monitorID", kubeEvent.MonitorId))
 		return BindingExecutionInfo{

pkg/task/queue/task_queue.go

@@ -413,18 +413,18 @@ func (q *TaskQueue) Start() {
 
 	if q.Handler == nil {
 		log.Error("should set handler before start in queue", slog.String("name", q.Name))
-		q.Status = "no handler set"
+		q.SetStatus("no handler set")
 		return
 	}
 
 	go func() {
-		q.Status = ""
+		q.SetStatus("")
 		var sleepDelay time.Duration
 		for {
 			q.debugf("queue %s: wait for task, delay %d", q.Name, sleepDelay)
 			t := q.waitForTask(sleepDelay)
 			if t == nil {
-				q.Status = "stop"
+				q.SetStatus("stop")
 				log.Info("queue stopped", slog.String("name", q.Name))
 				return
 			}
@@ -435,14 +435,14 @@ func (q *TaskQueue) Start() {
 
 			// Now the task can be handled!
 			var nextSleepDelay time.Duration
-			q.Status = "run first task"
+			q.SetStatus("run first task")
 			taskRes := q.Handler(t)
 
 			// Check Done channel after long-running operation.
 			select {
 			case <-q.ctx.Done():
 				log.Info("queue stopped after task handling", slog.String("name", q.Name))
-				q.Status = "stop"
+				q.SetStatus("stop")
 				return
 			default:
 			}
@@ -452,7 +452,7 @@ func (q *TaskQueue) Start() {
 				// Exponential backoff delay before retry.
 				nextSleepDelay = q.ExponentialBackoffFn(t.GetFailureCount())
 				t.IncrementFailureCount()
-				q.Status = fmt.Sprintf("sleep after fail for %s", nextSleepDelay.String())
+				q.SetStatus(fmt.Sprintf("sleep after fail for %s", nextSleepDelay.String()))
 			case Success, Keep:
 				// Insert new tasks right after the current task in reverse order.
 				q.withLock(func() {
@@ -474,16 +474,16 @@ func (q *TaskQueue) Start() {
 						q.addLast(newTask)
 					}
 				})
-				q.Status = ""
+				q.SetStatus("")
 			case Repeat:
 				// repeat a current task after a small delay
 				nextSleepDelay = q.DelayOnRepeat
-				q.Status = "repeat head task"
+				q.SetStatus("repeat head task")
 			}
 
 			if taskRes.DelayBeforeNextTask != 0 {
 				nextSleepDelay = taskRes.DelayBeforeNextTask
-				q.Status = fmt.Sprintf("sleep for %s", nextSleepDelay.String())
+				q.SetStatus(fmt.Sprintf("sleep for %s", nextSleepDelay.String()))
 			}
 
 			sleepDelay = nextSleepDelay
@@ -527,7 +527,7 @@ func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task {
 	q.cancelDelay = false
 	q.waitMu.Unlock()
 
-	origStatus := q.Status
+	origStatus := q.GetStatus()
 
 	defer func() {
 		checkTicker.Stop()
@@ -578,10 +578,10 @@ func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task {
 		// Wait loop still in progress: update queue status.
 		waitTime := time.Since(waitBegin).Truncate(time.Second)
 		if sleepDelay == 0 {
-			q.Status = fmt.Sprintf("waiting for task %s", waitTime.String())
+			q.SetStatus(fmt.Sprintf("waiting for task %s", waitTime.String()))
 		} else {
 			delay := sleepDelay.Truncate(time.Second)
-			q.Status = fmt.Sprintf("%s (%s left of %s delay)", origStatus, (delay - waitTime).String(), delay.String())
+			q.SetStatus(fmt.Sprintf("%s (%s left of %s delay)", origStatus, (delay - waitTime).String(), delay.String()))
 		}
 	}
 }