Merge branch 'feature/update-jobs' into 'master'

Feature/update jobs

See merge request grafolean/grafolean-worker-snmp!3

Author: grafolean

.env.example

@@ -1,2 +1,3 @@
 BACKEND_URL=https://grafolean.com/api
 BOT_TOKEN=
+JOBS_REFRESH_INTERVAL=120

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,16 +156,18 @@ 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', 'jobs_refresh_interval'
 
-    def __init__(self, backend_url, bot_token):
+    def __init__(self, backend_url, bot_token, jobs_refresh_interval):
         self.backend_url = backend_url
         self.bot_token = bot_token
+        self.jobs_refresh_interval = jobs_refresh_interval
+        self.known_jobs = {}
 
     @abstractmethod
     def jobs(self):
         """
-            Returns a list of (intervals, job_func, job_data) tuples. Usually calls
+            Returns a list of (job_id, intervals, job_func, job_data) tuples. Usually calls
             `fetch_job_configs` to get input data.
         """
 
@@ -246,6 +248,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 +269,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 intervals, job_func, job_data in self.jobs():
-            trigger = MultipleIntervalsTrigger(intervals)
-            scheduler.add_job(job_func, 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(self.jobs_refresh_interval)
+
         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")
 
@@ -198,7 +198,8 @@ def jobs(self):
         for entity_info in self.fetch_job_configs('snmp'):
             intervals = list(set([sensor_info["interval"] for sensor_info in entity_info["sensors"]]))
             job_info = { **entity_info, "backend_url": self.backend_url, "bot_token": self.bot_token }
-            yield intervals, SNMPCollector.do_snmp, job_info
+            job_id = str(entity_info["entity_id"])
+            yield job_id, intervals, SNMPCollector.do_snmp, job_info
 
 
 if __name__ == "__main__":
@@ -208,6 +209,7 @@ def jobs(self):
     bot_token = os.environ.get('BOT_TOKEN')
     if not backend_url or not bot_token:
         raise Exception("Please specify BACKEND_URL and BOT_TOKEN env vars.")
+    jobs_refresh_interval = int(os.environ.get('JOBS_REFRESH_INTERVAL', 120))
 
-    c = SNMPCollector(backend_url, bot_token)
+    c = SNMPCollector(backend_url, bot_token, jobs_refresh_interval)
     c.execute()