Merge pull request #1407 from mickem/scheduler-run-on-startup

Run checks at NSClient++ startup

Author: mickem

docs/samples/Scheduler_samples.md

@@ -68,3 +68,41 @@ The syntax of the schedule is similar to a cron expression in that you have:
 | Day of month | 1-31           | , *                        |
 | Month        | 0-11           | , *                        |
 | Day of week  | 1-7            | , *                        |
+
+#### Running a check at startup
+
+Both interval and schedule only report for the first time once the interval (or
+the next matching time) has elapsed. With a long interval that leaves the
+monitoring server with the old result for a long time after a reboot - exactly
+when the status is most likely to have changed. Set `run on startup` to run the
+command once as soon as the agent has started:
+
+```
+[/settings/scheduler/schedules/uptime]
+interval = 1h
+channel = NSCA
+command = check_uptime
+run on startup = true
+```
+
+The schedule is otherwise unaffected: the next run follows an hour after the
+startup run. The startup run also happens after a configuration reload, so a
+schedule you just changed reports its new status right away.
+
+Setting it on the default section turns it on for every schedule which does not
+override it:
+
+```
+[/settings/scheduler/schedules/default]
+interval = 1h
+channel = NSCA
+run on startup = true
+```
+
+If you have a lot of schedules and do not want all of them to report at the very
+same instant, spread the startup runs out over a window:
+
+```
+[/settings/scheduler]
+startup window = 30s
+```

include/scheduler/simple_scheduler.cpp

@@ -82,26 +82,34 @@ void scheduler::stop() {
   log_trace(__FILE__, __LINE__, "Thread pool contains: " + str::xtos(threads_.count()));
 }
 
-int scheduler::add_task(const std::string &tag, const boost::posix_time::time_duration duration, const double jitter_factor) {
+int scheduler::add_task(const std::string &tag, const boost::posix_time::time_duration duration, const double jitter_factor, const bool schedule_first_run) {
   task item(tag, duration, jitter_factor);
   {
     boost::mutex::scoped_lock l(mutex_);
     item.id = ++schedule_id_;
     tasks_[item.id] = item;
   }
-  reschedule(item, now());
+  if (schedule_first_run) reschedule(item, now());
   return item.id;
 }
-int scheduler::add_task(const std::string &tag, const cron_parser::schedule &schedule) {
+int scheduler::add_task(const std::string &tag, const cron_parser::schedule &schedule, const bool schedule_first_run) {
   task item(tag, schedule);
   {
     boost::mutex::scoped_lock l(mutex_);
     item.id = ++schedule_id_;
     tasks_[item.id] = item;
   }
-  reschedule(item, now());
+  if (schedule_first_run) reschedule(item, now());
   return item.id;
 }
+void scheduler::run_now(const int id, const boost::posix_time::time_duration delay) {
+  const op_task_object item = get_task(id);
+  if (!item) {
+    log_error(__FILE__, __LINE__, "Cannot run unknown task: " + str::xtos(id));
+    return;
+  }
+  reschedule_at(item->tag, id, now() + delay, true);
+}
 void scheduler::remove_task(const int id) {
   boost::mutex::scoped_lock l(mutex_);
   const auto it = tasks_.find(id);
@@ -120,8 +128,14 @@ scheduler::op_task_object scheduler::get_task(const int id) {
 }
 
 void scheduler::clear_tasks() {
-  boost::mutex::scoped_lock l(mutex_);
-  tasks_.clear();
+  {
+    boost::mutex::scoped_lock l(mutex_);
+    tasks_.clear();
+  }
+  // Drop the pending instances as well: without a task behind them they would
+  // only produce "Task not found" errors as they come due, and on a reload
+  // they would race the freshly added tasks.
+  if (!queue_.clear()) log_error(__FILE__, __LINE__, "Failed to clear the schedule queue");
 }
 
 void scheduler::watch_dog(const int id) {
@@ -182,7 +196,7 @@ void scheduler::thread_proc(const int id) {
 
       try {
         boost::posix_time::time_duration off = now() - instance->time;
-        if (off.total_seconds() > error_threshold_) {
+        if (!instance->suppress_late_warning && off.total_seconds() > error_threshold_) {
           log_error(__FILE__, __LINE__,
                     "Ran scheduled item " + instance->tag + "(" + str::xtos(instance->schedule_id) + ") " + str::xtos(off.total_seconds()) +
                         " seconds to late from thread " + str::xtos(id));
@@ -256,11 +270,12 @@ void scheduler::reschedule(const task &item, boost::posix_time::ptime now_time)
     reschedule_at(item.tag, item.id, item.get_next(now_time));
   }
 }
-void scheduler::reschedule_at(const std::string &tag, const int id, boost::posix_time::ptime new_time) {
+void scheduler::reschedule_at(const std::string &tag, const int id, boost::posix_time::ptime new_time, const bool suppress_late_warning) {
   schedule_instance instance;
   instance.tag = tag;
   instance.schedule_id = id;
   instance.time = new_time;
+  instance.suppress_late_warning = suppress_late_warning;
   if (!queue_.push(instance)) {
     log_error(__FILE__, __LINE__, "Failed to reschedule item");
   }

include/scheduler/simple_scheduler.hpp

@@ -91,6 +91,12 @@ struct schedule_instance {
   boost::posix_time::ptime time;
   std::string tag;
   int schedule_id{};
+  // Startup runs (run_now) are all queued for the same instant, so with more
+  // tasks than worker threads the tail is inevitably "late" through no fault
+  // of the configuration. Such instances are exempt from the lateness error
+  // to keep boot from filling the log with false alarms; the watchdog still
+  // sees the lag and scales the pool.
+  bool suppress_late_warning{false};
   friend bool operator<(const schedule_instance& p1, const schedule_instance& p2) { return p1.time > p2.time; }
 };
 
@@ -131,6 +137,13 @@ class safe_schedule_queue {
     return ret;
   }
 
+  bool clear(const unsigned int timeout = 5) {
+    boost::unique_lock<boost::shared_mutex> lock(mutex_, boost::get_system_time() + boost::posix_time::seconds(timeout));
+    if (!lock) return false;
+    queue_ = schedule_queue_type();
+    return true;
+  }
+
   bool push(T instance, const unsigned int timeout = 5) {
     boost::unique_lock<boost::shared_mutex> lock(mutex_, boost::get_system_time() + boost::posix_time::seconds(timeout));
     if (!lock) {
@@ -192,8 +205,18 @@ class scheduler : public boost::noncopyable {
   std::size_t get_metric_ql();
   bool has_metrics() const;
 
-  int add_task(const std::string& tag, boost::posix_time::time_duration duration, double jitter_factor);
-  int add_task(const std::string& tag, const cron_parser::schedule& schedule);
+  // `schedule_first_run == false` registers the task without queueing
+  // anything, leaving it dormant until run_now() queues its first instance.
+  // Used by run-on-startup schedules, which must not be queued twice: every
+  // execution queues the following one, so an extra instance would make the
+  // task run twice per interval forever.
+  int add_task(const std::string& tag, boost::posix_time::time_duration duration, double jitter_factor, bool schedule_first_run = true);
+  int add_task(const std::string& tag, const cron_parser::schedule& schedule, bool schedule_first_run = true);
+  // Queue a run of an already added task `delay` from now. Only ever call this
+  // for a task added with schedule_first_run = false (or one that has since
+  // been abandoned) - calling it for a live task adds a second, parallel chain
+  // of instances for that task.
+  void run_now(int id, boost::posix_time::time_duration delay = boost::posix_time::seconds(0));
   void remove_task(int id);
   op_task_object get_task(int id);
   void clear_tasks();
@@ -220,7 +243,7 @@ class scheduler : public boost::noncopyable {
   void thread_proc(int id);
 
   void reschedule(const task& item, boost::posix_time::ptime now_time);
-  void reschedule_at(const std::string& tag, int id, boost::posix_time::ptime new_time);
+  void reschedule_at(const std::string& tag, int id, boost::posix_time::ptime new_time, bool suppress_late_warning = false);
   void start_threads();
 
   void log_error(const char* file, const int line, const std::string& err) const {

include/scheduler/simple_scheduler_test.cpp

@@ -359,3 +359,82 @@ TEST(simple_scheduler_running, no_handler_does_not_crash) {
   s.stop();
   SUCCEED();
 }
+
+// --- deferred first run (run on startup) ----------------------------------
+
+TEST(simple_scheduler_basic, add_task_without_first_run_queues_nothing) {
+  simple_scheduler::scheduler s;
+  const int id = s.add_task("deferred", boost::posix_time::seconds(3600), 0.0, false);
+  EXPECT_TRUE(s.get_task(id));
+  EXPECT_EQ(s.get_metric_ql(), 0u);
+}
+
+TEST(simple_scheduler_basic, add_cron_task_without_first_run_queues_nothing) {
+  simple_scheduler::scheduler s;
+  const int id = s.add_task("deferred", cron_parser::parse("0 * * * *"), false);
+  EXPECT_TRUE(s.get_task(id));
+  EXPECT_EQ(s.get_metric_ql(), 0u);
+}
+
+TEST(simple_scheduler_basic, run_now_queues_a_deferred_task) {
+  simple_scheduler::scheduler s;
+  const int id = s.add_task("deferred", boost::posix_time::seconds(3600), 0.0, false);
+  s.run_now(id);
+  EXPECT_EQ(s.get_metric_ql(), 1u);
+}
+
+TEST(simple_scheduler_basic, run_now_unknown_task_is_ignored) {
+  simple_scheduler::scheduler s;
+  s.run_now(4711);
+  EXPECT_EQ(s.get_metric_ql(), 0u);
+}
+
+TEST(simple_scheduler_basic, clear_tasks_also_drops_pending_instances) {
+  // Pending instances outlive their task otherwise, which on a reload means
+  // the schedules from before the reload keep firing alongside the new ones.
+  simple_scheduler::scheduler s;
+  s.add_task("one", boost::posix_time::seconds(3600), 0.0);
+  s.add_task("two", boost::posix_time::seconds(3600), 0.0);
+  ASSERT_EQ(s.get_metric_ql(), 2u);
+
+  s.clear_tasks();
+  EXPECT_EQ(s.get_metric_ql(), 0u);
+}
+
+TEST(simple_scheduler_running, run_now_fires_a_deferred_task) {
+  simple_scheduler::scheduler s;
+  counting_handler h;
+  h.reschedule_after = false;
+  s.set_handler(&h);
+  s.set_threads(2);
+  // An hour-long interval: without run_now the handler cannot possibly be
+  // called during the test, which is exactly the stale-until-the-first-tick
+  // behaviour run-on-startup schedules avoid.
+  const int id = s.add_task("startup", boost::posix_time::seconds(3600), 0.0, false);
+  s.start();
+  EXPECT_FALSE(wait_for([&] { return h.calls.load() >= 1; }, std::chrono::milliseconds(200)));
+
+  s.run_now(id);
+  EXPECT_TRUE(wait_for([&] { return h.calls.load() >= 1; }, std::chrono::seconds(5)));
+
+  s.stop();
+  s.unset_handler();
+}
+
+TEST(simple_scheduler_running, run_now_honours_the_delay) {
+  simple_scheduler::scheduler s;
+  counting_handler h;
+  h.reschedule_after = false;
+  s.set_handler(&h);
+  s.set_threads(2);
+  const int id = s.add_task("startup", boost::posix_time::seconds(3600), 0.0, false);
+  s.start();
+  s.run_now(id, boost::posix_time::seconds(2));
+
+  // The instance is queued, but it must not run before its delay has passed.
+  EXPECT_FALSE(wait_for([&] { return h.calls.load() >= 1; }, std::chrono::milliseconds(500)));
+  EXPECT_TRUE(wait_for([&] { return h.calls.load() >= 1; }, std::chrono::seconds(5)));
+
+  s.stop();
+  s.unset_handler();
+}

modules/Scheduler/Scheduler.cpp

@@ -12,6 +12,7 @@
 #include <nscapi/protobuf/functions_submit.hpp>
 #include <nscapi/settings/helper.hpp>
 #include <nscapi/settings/proxy.hpp>
+#include <str/format.hpp>
 #include <str/utf8.hpp>
 
 namespace sh = nscapi::settings_helper;
@@ -20,6 +21,11 @@ bool Scheduler::loadModuleEx(std::string alias, NSCAPI::moduleLoadMode mode) {
   if (mode == NSCAPI::reloadStart) {
     scheduler_.prepare_shutdown();
     scheduler_.stop();
+    // The worker threads are joined at this point, so the task list and the
+    // pending queue can be dropped safely. Without this the tasks (and their
+    // queued instances) from before the reload survive alongside the ones
+    // added below, so every schedule ends up running twice.
+    scheduler_.clear();
     schedules_.clear();
   }
 
@@ -36,6 +42,10 @@ bool Scheduler::loadModuleEx(std::string alias, NSCAPI::moduleLoadMode mode) {
   settings.alias().add_key_to_settings()
     .add_int("threads", sh::int_fun_key([this] (auto value) { scheduler_.set_threads(value); }, 5),
 	    "Threads", "Number of threads to use.")
+    .add_string("startup window", sh::string_fun_key([this] (const auto& value) { this->set_startup_window(value); }, "0s"),
+        "Startup window",
+        "Time over which schedules with 'run on startup' are spread out when the agent starts. The default (0s) runs them all immediately; raise it if you "
+        "have many startup schedules and do not want to hit the monitoring server with all of them at once.")
     .add_string("timezone", sh::string_fun_key([this] (const auto& value) { scheduler_.set_timezone(value); }, "local"),
         "Timezone",
         "Reference clock for cron expressions. Accepts 'local' (default — standard cron semantics), 'utc'/'gmt' (restores the pre-0.13 "
@@ -96,9 +106,32 @@ bool Scheduler::loadModuleEx(std::string alias, NSCAPI::moduleLoadMode mode) {
     scheduler_.set_handler(this);
     scheduler_.start();
   }
+  // On a normal boot the startup runs wait for startModule, which the core
+  // calls once every plugin is loaded - firing them here would query commands
+  // that later modules have not registered yet. A reload (or a module loaded
+  // into an already running agent) has no such problem and gets no second
+  // startModule, so fire them right away instead.
+  if (started_ && (mode == NSCAPI::normalStart || mode == NSCAPI::reloadStart)) {
+    scheduler_.run_startup_tasks(startup_window_);
+  }
   return true;
 }
 
+bool Scheduler::startModule() {
+  started_ = true;
+  scheduler_.run_startup_tasks(startup_window_);
+  return true;
+}
+
+void Scheduler::set_startup_window(const std::string &value) {
+  try {
+    startup_window_ = boost::posix_time::seconds(str::format::stox_as_time_sec<long>(value, "s"));
+  } catch (const std::exception &e) {
+    NSC_LOG_ERROR_EXR("Invalid 'startup window' value '" + value + "', falling back to 0s: ", e);
+    startup_window_ = boost::posix_time::seconds(0);
+  }
+}
+
 void Scheduler::add_schedule(const std::string &key, const std::string &arg) {
   try {
     schedules_.add(nscapi::settings_proxy::create(get_id(), get_core()), key, arg);

modules/Scheduler/Scheduler.h

@@ -12,19 +12,29 @@ class Scheduler : public schedules::task_handler, public nscapi::impl::simple_pl
  private:
   schedules::scheduler scheduler_;
   schedules::schedule_handler schedules_;
+  // Window over which "run on startup" schedules are spread, see the
+  // "startup window" setting.
+  boost::posix_time::time_duration startup_window_;
+  // True once startModule has run, i.e. once every other plugin is loaded and
+  // startup runs are safe to fire. Reloads happen after that point and are
+  // never followed by another startModule (the core calls it once per plugin
+  // lifetime), so loadModuleEx has to fire the startup runs itself when set.
+  bool started_;
 
  public:
-  Scheduler() { scheduler_.set_handler(this); }
+  Scheduler() : startup_window_(boost::posix_time::seconds(0)), started_(false) { scheduler_.set_handler(this); }
   virtual ~Scheduler() { scheduler_.set_handler(nullptr); }
   // Module calls
   bool loadModuleEx(std::string alias, NSCAPI::moduleLoadMode mode);
+  bool startModule();
   void prepareShutdown();
   bool unloadModule();
 
   // Metrics
   void fetchMetrics(PB::Metrics::MetricsMessage_Response* response);
 
   void add_schedule(const std::string& alias, const std::string& command);
+  void set_startup_window(const std::string& value);
   bool handle_schedule(schedules::target_object task);
 
   void on_error(const char* file, int line, std::string error);

modules/Scheduler/module.json

@@ -10,6 +10,8 @@
 
 	"metrics" : "produce",
 
+	"on_start" : true,
+
 	"prepare_shutdown" : true
 
 }