Merge pull request #2130 from ripienaar/2129

(#2129) Add skip_trigger_on_reenter to scheduler

aagent/watchers/metricwatcher/metric.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022, R.I. Pienaar and the Choria Project contributors
+// Copyright (c) 2020-2024, R.I. Pienaar and the Choria Project contributors
 //
 // SPDX-License-Identifier: Apache-2.0
 
@@ -235,8 +235,14 @@ func (w *Watcher) handleCheck(output []byte, err error) error {
 	if err == nil {
 		if bytes.HasPrefix(bytes.TrimSpace(output), []byte("{")) {
 			metric, err = w.parseJSONCheck(output)
+			if err != nil {
+				w.Errorf("Failed to parse metric output: %v", err)
+			}
 		} else {
 			metric, err = w.parseNagiosCheck(output)
+			if err != nil {
+				w.Errorf("Failed to parse perf data output: %v", err)
+			}
 		}
 	}
 

aagent/watchers/metricwatcher/prometheus.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022, R.I. Pienaar and the Choria Project contributors
+// Copyright (c) 2020-2024, R.I. Pienaar and the Choria Project contributors
 //
 // SPDX-License-Identifier: Apache-2.0
 
@@ -112,6 +112,7 @@ func savePromState(td string, log logger) error {
 	if err != nil {
 		return fmt.Errorf("failed to create prometheus metric in %q: %s", td, err)
 	}
+	defer tfile.Close()
 
 	for name, pm := range pmetrics {
 		if len(pm.values) == 0 {
@@ -125,7 +126,6 @@ func savePromState(td string, log logger) error {
 		}
 	}
 
-	tfile.Close()
 	os.Chmod(tfile.Name(), 0644)
 	return os.Rename(tfile.Name(), filepath.Join(td, "choria_machine_metrics_watcher_status.prom"))
 }

aagent/watchers/schedulewatcher/schedule.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2019-2022, R.I. Pienaar and the Choria Project contributors
+// Copyright (c) 2019-2024, R.I. Pienaar and the Choria Project contributors
 //
 // SPDX-License-Identifier: Apache-2.0
 
@@ -36,9 +36,10 @@ var stateNames = map[State]string{
 }
 
 type properties struct {
-	Duration   time.Duration
-	StartSplay time.Duration `mapstructure:"start_splay"`
-	Schedules  []string
+	Duration             time.Duration
+	StartSplay           time.Duration `mapstructure:"start_splay"`
+	SkipTriggerOnReenter bool          `mapstructure:"skip_trigger_on_reenter"`
+	Schedules            []string
 }
 
 type Watcher struct {
@@ -54,6 +55,8 @@ type Watcher struct {
 	ctrq chan int
 	ctr  int
 
+	triggered bool
+
 	state         State
 	previousState State
 
@@ -90,7 +93,7 @@ func (w *Watcher) watchSchedule(ctx context.Context, wg *sync.WaitGroup) {
 	for {
 		select {
 		case i := <-w.ctrq:
-			w.Infof("Handling state change counter %v while ctr=%v", i, w.ctr)
+			w.Debugf("Handling state change counter %v while ctr=%v", i, w.ctr)
 			w.mu.Lock()
 
 			w.ctr = w.ctr + i
@@ -101,10 +104,10 @@ func (w *Watcher) watchSchedule(ctx context.Context, wg *sync.WaitGroup) {
 			}
 
 			if w.ctr == 0 {
-				w.Infof("State going off due to ctr change to 0")
+				w.Debugf("State going off due to ctr change to 0")
 				w.state = Off
 			} else {
-				w.Infof("State going on due to ctr change of %v", i)
+				w.Debugf("State going on due to ctr change of %v", i)
 				w.state = On
 			}
 
@@ -139,10 +142,17 @@ func (w *Watcher) watch() (err error) {
 
 	switch w.state {
 	case Off, Unknown:
+		w.setTriggered(false)
 		w.NotifyWatcherState(w.CurrentState())
 		return w.FailureTransition()
 
 	case On:
+		if w.properties.SkipTriggerOnReenter && w.didTrigger() {
+			w.Debugf("Skipping success transition that's already fired in this schedule due to skip_trigger_on_reenter")
+			return nil
+		}
+
+		w.setTriggered(true)
 		w.setPreviousState(w.state)
 		w.NotifyWatcherState(w.CurrentState())
 		return w.SuccessTransition()
@@ -155,6 +165,19 @@ func (w *Watcher) watch() (err error) {
 	return nil
 }
 
+func (w *Watcher) setTriggered(s bool) {
+	w.mu.Lock()
+	w.triggered = s
+	w.mu.Unlock()
+}
+
+func (w *Watcher) didTrigger() bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	return w.triggered
+}
+
 func (w *Watcher) Run(ctx context.Context, wg *sync.WaitGroup) {
 	defer wg.Done()