Update jobs definitions periodically

Author: Anze

collector.py

@@ -10,7 +10,7 @@
 import concurrent.futures
 import traceback
 
-from apscheduler.schedulers.blocking import BlockingScheduler
+from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.base import BaseTrigger
 from apscheduler.executors.base import BaseExecutor
 from apscheduler.events import (
@@ -156,11 +156,12 @@ def run_job(job, jobstore_alias, run_times, logger_name):
 
 
 class Collector(object):
-    __slots__ = 'backend_url', 'bot_token'
+    __slots__ = 'backend_url', 'bot_token', 'scheduler', 'known_jobs'
 
     def __init__(self, backend_url, bot_token):
         self.backend_url = backend_url
         self.bot_token = bot_token
+        self.known_jobs = {}
 
     @abstractmethod
     def jobs(self):
@@ -246,6 +247,17 @@ def fetch_job_configs(self, protocol):
 
                 yield entity_info
 
+    def refresh_jobs(self):
+        for job_id, intervals, job_func, job_data in self.jobs():
+            # if the existing job's configuration is the same, leave it alone, otherwise the trigger will be reset:
+            if self.known_jobs.get(job_id) == job_data:
+                continue
+            self.known_jobs[job_id] = job_data
+
+            trigger = MultipleIntervalsTrigger(intervals)
+            logging.info(f"Adding job: {job_id}")
+            self.scheduler.add_job(job_func, id=job_id, trigger=trigger, executor='iaexecutor', kwargs=job_data, replace_existing=True)
+
     def execute(self):
         """
             Calls self.jobs() to get the list of the jobs, and executes them by using
@@ -256,15 +268,19 @@ def execute(self):
             'coalesce': True,  # if multiple jobs "misfire", re-run only one instance of a missed job
             'max_instances': 1,
         }
-        scheduler = BlockingScheduler(job_defaults=job_defaults, timezone=utc)
-        executor = IntervalsAwareProcessPoolExecutor(10)
-        scheduler.add_executor(executor, 'iaexecutor')
-
-        for job_id, intervals, job_func, job_data in self.jobs():
-            trigger = MultipleIntervalsTrigger(intervals)
-            scheduler.add_job(job_func, id=job_id, trigger=trigger, executor='iaexecutor', kwargs=job_data)
+        self.scheduler = BackgroundScheduler(job_defaults=job_defaults, timezone=utc)
+        self.scheduler.add_executor(IntervalsAwareProcessPoolExecutor(10), 'iaexecutor')
 
         try:
-            scheduler.start()
+            self.scheduler.start()
+            while True:
+                try:
+                    self.refresh_jobs()
+                except:
+                    logging.exception("Error refreshing jobs.")
+                time.sleep(120)
+
         except KeyboardInterrupt:
             logging.info("Got exit signal, exiting.")
+        finally:
+            self.scheduler.shutdown()

snmpcollector.py

@@ -77,7 +77,7 @@ def send_results_to_grafolean(backend_url, bot_token, account_id, values):
     try:
         r = requests.post(url, json=values)
         r.raise_for_status()
-        log.info("Results sent.")
+        log.info("Results sent: {}".format(values))
     except:
         log.exception("Error sending data to Grafolean")