Repository: NetApp/harvest
Pull request #444 (2021-08-25 12:13:26)

fix: collector tasks time interval correction during recover


Patch:
diff --git a/cmd/poller/collector/collector.go b/cmd/poller/collector/collector.go
index 4ee6d850..1fdcc1d9 100644
--- a/cmd/poller/collector/collector.go
+++ b/cmd/poller/collector/collector.go
@@ -157,7 +157,7 @@ func Init(c Collector) error {
 
 		if m := reflect.ValueOf(c).MethodByName(methodName); m.IsValid() {
 			if foo, ok := m.Interface().(func() (*matrix.Matrix, error)); ok {
-				if err := s.NewTaskString(task.GetNameS(), task.GetContentS(), foo, true); err != nil {
+				if err := s.NewTaskString(task.GetNameS(), task.GetContentS(), foo, true, "Collector_"+c.GetName()+"_"+c.GetObject()); err != nil {
 					return errors.New(errors.INVALID_PARAM, "schedule ("+task.GetNameS()+"): "+err.Error())
 				}
 			} else {
@@ -307,12 +307,15 @@ func (me *AbstractCollector) Start(wg *sync.WaitGroup) {
 						retryDelay *= 4
 					}
 					if !me.Schedule.IsStandBy() {
-						//logger.Error(me.Prefix, err.Error())
-						me.Logger.Warn().Msgf("target unreachable, entering standby mode (retry to connect in %d s)", retryDelay)
+						me.Logger.Warn().
+							Str("task", task.Name).
+							Int("retryDelay", retryDelay).
+							Msgf("target unreachable, entering standby mode (retry in retryDelay s)")
 					}
 					me.Logger.Debug().
-						Int("retryDelay", retryDelay).
 						Err(err).
+						Int("retryDelay", retryDelay).
+						Str("task", task.Name).
 						Msg("Target unreachable, entering standby mode (retry in retryDelay s)")
 					me.Schedule.SetStandByMode(task, time.Duration(retryDelay)*time.Second)
 					me.SetStatus(1, errors.ERR_CONNECTION)
@@ -320,15 +323,21 @@ func (me *AbstractCollector) Start(wg *sync.WaitGroup) {
 				case errors.IsErr(err, errors.ERR_NO_INSTANCE):
 					me.Schedule.SetStandByMode(task, 5*time.Minute)
 					me.SetStatus(1, errors.ERR_NO_INSTANCE)
-					me.Logger.Info().Msgf("no [%s] instances on system, entering standby mode", me.Object)
+					me.Logger.Info().
+						Str("task", task.Name).
+						Str("object", me.Object).
+						Msg("no instances of object on system, entering standby mode")
 				// no metrics available
 				case errors.IsErr(err, errors.ERR_NO_METRIC):
 					me.SetStatus(1, errors.ERR_NO_METRIC)
 					me.Schedule.SetStandByMode(task, 1*time.Hour)
-					me.Logger.Warn().Msgf("no [%s] metrics on system, entering standby mode", me.Object)
+					me.Logger.Info().
+						Str("task", task.Name).
+						Str("object", me.Object).
+						Msg("no metrics of object on system, entering standby mode")
 				// not an error we are expecting, so enter failed state and terminate
 				default:
-					me.Logger.Error().Stack().Err(err).Msg("")
+					me.Logger.Error().Stack().Err(err).Str("task", task.Name).Msg("")
 					if errmsg := errors.GetClass(err); errmsg != "" {
 						me.SetStatus(2, errmsg)
 					} else {
@@ -343,7 +352,7 @@ func (me *AbstractCollector) Start(wg *sync.WaitGroup) {
 				me.Schedule.Recover()
 				retryDelay = 1
 				me.SetStatus(0, "running")
-				me.Logger.Info().Msg("recovered from standby mode, back to normal schedule")
+				me.Logger.Info().Str("task", task.Name).Msg("recovered from standby mode, back to normal schedule")
 			}
 
 			if data != nil {
diff --git a/cmd/poller/poller.go b/cmd/poller/poller.go
index 00e95cc5..c9d64eb7 100644
--- a/cmd/poller/poller.go
+++ b/cmd/poller/poller.go
@@ -278,7 +278,7 @@ func (p *Poller) Init() error {
 		pollerSchedule = *p.params.PollerSchedule
 	}
 	p.schedule = schedule.New()
-	if err = p.schedule.NewTaskString("poller", pollerSchedule, nil, true); err != nil {
+	if err = p.schedule.NewTaskString("poller", pollerSchedule, nil, true, "poller_"+p.name); err != nil {
 		logger.Error().Stack().Err(err).Msg("set schedule:")
 		return err
 	}
@@ -310,7 +310,7 @@ func (p *Poller) Init() error {
 					p.firstAutoSupport()
 				})
 			}
-			if err = p.schedule.NewTaskString("asup", asupSchedule, p.startAsup, p.options.Asup); err != nil {
+			if err = p.schedule.NewTaskString("asup", asupSchedule, p.startAsup, p.options.Asup, "asup_"+p.name); err != nil {
 				return err
 			}
 			logger.Debug().Msgf("Autosupport scheduled with %s frequency", asupSchedule)
diff --git a/cmd/poller/schedule/schedule.go b/cmd/poller/schedule/schedule.go
index 41cb386d..a65c6fc3 100644
--- a/cmd/poller/schedule/schedule.go
+++ b/cmd/poller/schedule/schedule.go
@@ -40,10 +40,11 @@ import (
 
 // Task represents a scheduled task
 type task struct {
-	Name     string                         // name of the task
-	interval time.Duration                  // the schedule interval
-	timer    time.Time                      // last time task was executed
-	foo      func() (*matrix.Matrix, error) // pointer to the function that executes the task
+	Name       string                         // name of the task
+	interval   time.Duration                  // the schedule interval
+	timer      time.Time                      // last time task was executed
+	foo        func() (*matrix.Matrix, error) // pointer to the function that executes the task
+	identifier string                         // optional additional information about schedule i.e. collector name
 }
 
 // Start marks the task as started by updating timer
@@ -83,11 +84,11 @@ func (t *task) IsDue() bool {
 
 // Schedule contains a collection of tasks and the current state of the schedule
 type Schedule struct {
-	tasks          []*task       // list of tasks that Schedule needs to run
-	standByMode    bool          // if true, Schedule waitsfor a stalled task
-	standByTask    *task         // stalled task in standByMode
-	standByTimer   time.Time     // timer for stalled task in standByMode
-	cachedInterval time.Duration // normal interval of the stalled task
+	tasks          []*task                  // list of tasks that Schedule needs to run
+	standByMode    bool                     // if true, Schedule waitsfor a stalled task
+	standByTask    *task                    // stalled task in standByMode
+	standByTimer   time.Time                // timer for stalled task in standByMode
+	cachedInterval map[string]time.Duration // normal interval of the stalled tasks
 }
 
 // New creates and initializes an empty Schedule.
@@ -95,6 +96,7 @@ func New() *Schedule {
 	s := Schedule{}
 	s.tasks = make([]*task, 0)
 	s.standByMode = false
+	s.cachedInterval = make(map[string]time.Duration)
 	return &s
 }
 
@@ -104,14 +106,13 @@ func (s *Schedule) IsStandBy() bool {
 	return s.standByMode
 }
 
-// SetStandByMode intiailzes StandbyMode: Schedule will suspend all tasks until
+// SetStandByMode initializes StandbyMode: Schedule will suspend all tasks until
 // the critical task t has succeeded. The temporary interval i will be used for
 // the task until Schedule recovers to normal mode.
 func (s *Schedule) SetStandByMode(t *task, i time.Duration) {
 	for _, x := range s.tasks {
 		if x.Name == t.Name {
 			s.standByTask = t
-			s.cachedInterval = t.interval // remember normal interval of task
 			t.interval = i
 			t.timer = time.Now()
 			s.standByMode = true
@@ -126,9 +127,11 @@ func (s *Schedule) Recover() {
 
 	if s.standByMode {
 		for _, t := range s.tasks {
+			if interval, ok := s.cachedInterval[t.Name]; ok {
+				t.interval = interval
+			}
 			// reset timer of the critical task, assume that it just completed
 			if t.Name == s.standByTask.Name {
-				t.interval = s.cachedInterval
 				t.timer = time.Now()
 				// all the other tasks that were suspended need to run asap
 			} else {
@@ -148,10 +151,11 @@ func (s *Schedule) Recover() {
 // should be positive.
 // The order in which tasks are added is maintained: GetTasks() will
 // return tasks in FIFO order.
-func (s *Schedule) NewTask(n string, i time.Duration, f func() (*matrix.Matrix, error), runNow bool) error {
+func (s *Schedule) NewTask(n string, i time.Duration, f func() (*matrix.Matrix, error), runNow bool, identifier string) error {
 	if s.GetTask(n) == nil {
 		if i > 0 {
-			t := &task{Name: n, interval: i, foo: f}
+			t := &task{Name: n, interval: i, foo: f, identifier: identifier}
+			s.cachedInterval[n] = t.interval // remember normal interval of task
 			if runNow {
 				t.timer = time.Now().Add(-i) // set to run immediately
 			} else {
@@ -166,29 +170,9 @@ func (s *Schedule) NewTask(n string, i time.Duration, f func() (*matrix.Matrix,
 }
 
 // NewTaskString creates a new task, the interval is parsed from string i
-func (s *Schedule) NewTaskString(n, i string, f func() (*matrix.Matrix, error), runNow bool) error {
-	if d, err := time.ParseDuration(i); err == nil {
-		return s.NewTask(n, d, f, runNow)
-	} else {
-		return err
-	}
-}
-
-// SetInterval changes the interval of an existing task. Changing interval is safe
-// at any time.
-func (s *Schedule) SetInterval(t *task, i time.Duration) error {
-	if i > 0 {
-		t.interval = i
-		return nil
-	}
-	return errors.New(errors.ERR_SCHEDULE, "invalid interval :"+i.String())
-}
-
-// SetIntervalString changes the interval of an existing task. Interval is parsed
-// from string i.
-func (s *Schedule) SetIntervalString(t *task, i string) error {
+func (s *Schedule) NewTaskString(n, i string, f func() (*matrix.Matrix, error), runNow bool, identifier string) error {
 	if d, err := time.ParseDuration(i); err == nil {
-		return s.SetInterval(t, d)
+		return s.NewTask(n, d, f, runNow, identifier)
 	} else {
 		return err
 	}


Test patch:
diff --git a/cmd/poller/schedule/schedule_test.go b/cmd/poller/schedule/schedule_test.go
new file mode 100644
index 00000000..04a955c8
--- /dev/null
+++ b/cmd/poller/schedule/schedule_test.go
@@ -0,0 +1,42 @@
+package schedule
+
+import (
+	"testing"
+	"time"
+)
+
+func setupSchedule() (s *Schedule) {
+	s = New()
+	s.NewTaskString("counter", "1200s", nil, false, "")
+	s.NewTaskString("data", "180s", nil, false, "")
+	s.NewTaskString("instance", "600s", nil, false, "")
+	return s
+}
+
+func setStandByTasks(s *Schedule) {
+	retryDelay := 1024
+	for _, task := range s.GetTasks() {
+		s.SetStandByMode(task, time.Duration(retryDelay)*time.Second)
+	}
+}
+
+func TestSchedule_Recover(t *testing.T) {
+	s := setupSchedule()
+	setStandByTasks(s)
+	s.Recover()
+	for _, task := range s.GetTasks() {
+		if "counter" == task.Name {
+			if task.interval != 1.2e+12 {
+				t.Errorf("expected = %v, got %d", 1.2e+12, task.interval)
+			}
+		} else if "data" == task.Name {
+			if task.interval != 1.8e+11 {
+				t.Errorf("expected = %b, got %b", 1.8e+11, task.interval)
+			}
+		} else if "instance" == task.Name {
+			if task.interval != 6e+11 {
+				t.Errorf("expected = %b, got %b", 6e+11, task.interval)
+			}
+		}
+	}
+}