Merge branch 'feature/remote-config' into 'master'

Feature/remote config

See merge request grafolean/grafolean-worker-ping!1

Author: grafolean

Pipfile

@@ -4,10 +4,14 @@ url = "https://pypi.org/simple"
 verify_ssl = true
 
 [dev-packages]
+pylint = "*"
 
 [packages]
 requests = "*"
 multiping = "*"
+ansicolors = "*"
+grafoleancollector = "*"
+python-dotenv = "*"
 
 [requires]
 python_version = "3.6"

Pipfile.lock

@@ -1,87 +1,97 @@
-from collections import defaultdict
-from multiping import MultiPing
 import os
+import time
+from collections import defaultdict
+from colors import color
+import logging
+from multiping import multi_ping
+from pytz import utc
 import requests
 import socket
-import time
+import dotenv
+
+
+from grafoleancollector import Collector
+
+
+logging.basicConfig(format='%(asctime)s | %(levelname)s | %(message)s',
+                    datefmt='%Y-%m-%d %H:%M:%S', level=logging.DEBUG)
+logging.addLevelName(logging.DEBUG, color("DBG", 7))
+logging.addLevelName(logging.INFO, "INF")
+logging.addLevelName(logging.WARNING, color('WRN', fg='red'))
+logging.addLevelName(logging.ERROR, color('ERR', bg='red'))
+log = logging.getLogger("{}.{}".format(__name__, "base"))
+
+
+def send_results_to_grafolean(backend_url, bot_token, account_id, values):
+    url = '{}/accounts/{}/values/?b={}'.format(backend_url, account_id, bot_token)
+
+    if len(values) == 0:
+        log.warning("No results available to be sent to Grafolean, skipping.")
+        return
+
+    log.info("Sending results to Grafolean")
+    try:
+        r = requests.post(url, json=values)
+        r.raise_for_status()
+        log.info("Results sent: {}".format(values))
+    except:
+        log.exception("Error sending data to Grafolean")
+
+
+class PingCollector(Collector):
+
+    def jobs(self):
+        for entity_info in self.fetch_job_configs('ping'):
+            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 }
+            job_id = str(entity_info["entity_id"])
+            yield job_id, intervals, PingCollector.do_ping, job_info
+
+    @staticmethod
+    def do_ping(*args, **job_info):
+        # filter out only those sensors that are supposed to run at this interval:
+        affecting_intervals, = args
+        hostname = job_info["details"]["ipv4"]
+        cred = job_info["credential_details"]
+
+        activated_sensors = [s for s in job_info["sensors"] if s["interval"] in affecting_intervals]
+        if not activated_sensors:
+            return
+
+        # perform ping:
+        values = []
+        addrs = [hostname]
+        timeout = float(cred.get("timeout", 2.0))
+        retry = int(cred.get("retry", 0))
+        n_packets = int(cred.get("n_packets", 3))
+        sleep_packets = float(cred.get("sleep_packets", 1.0))
+        output_path_prefix = f'entity.{job_info["entity_id"]}.ping'
+        n_ok = 0
+        for i in range(n_packets):
+            responses, _ = multi_ping(addrs, timeout=timeout, retry=retry)
+
+            # save results:
+            if len(responses) > 0:
+                values.append({'p': f"{output_path_prefix}.p{i}.ok", 'v': 1.0})
+                values.append({'p': f"{output_path_prefix}.p{i}.rtt", 'v': list(responses.values())[0]})
+                n_ok += 1
+            else:
+                values.append({'p': f"{output_path_prefix}.p{i}.ok", 'v': 0.0})
+            time.sleep(sleep_packets)
+
+        values.append({'p': f"{output_path_prefix}.success", 'v': float(n_ok) / n_packets})
+
+        send_results_to_grafolean(job_info['backend_url'], job_info['bot_token'], job_info['account_id'], values)
 
-N_PINGS = 3
-
-# This is copy-pasted from multiping package; the reason is that we need to get MultiPing
-# instance, because it holds the IP addresses which correspond to the addresses we wanted
-# pinged - and the key in ping results is the IP.
-def multi_ping(dest_addrs, timeout, retry=0, ignore_lookup_errors=False):
-    retry_timeout = float(timeout) / (retry + 1)
-
-    mp = MultiPing(dest_addrs, ignore_lookup_errors=ignore_lookup_errors)
-
-    results = {}
-    retry_count = 0
-    while retry_count <= retry:
-        # Send a batch of pings
-        mp.send()
-        single_results, no_results = mp.receive(retry_timeout)
-        # Add the results from the last sending of pings to the overall results
-        results.update(single_results)
-        if not no_results:
-            # No addresses left? We are done.
-            break
-        retry_count += 1
-
-    return results, no_results, mp
-
-def get_addr_for_ip_dict(addrs, mp):
-    # This is complicated, and a hack. Still... mp (MultiPing instance) holds two lists,
-    # self._dest_addrs and self._unprocessed_targets. List _unprocessed_targets has the addresses
-    # that couldn't be resolved. Others were resolved, and _dest_addrs has the IPs in the same
-    # order as original addresses.
-    resolved_addrs = [a for a in addrs if a not in mp._unprocessed_targets]
-    ip_to_addr = {k: v for k, v in zip(mp._dest_addrs, resolved_addrs)}
-    return ip_to_addr
-
-def do_ping(addrs):
-    results = defaultdict(list)
-    mp = None
-    ip_to_addr = {}
-    for i in range(N_PINGS):
-        print(".")
-        responses, no_responses, mp = multi_ping(addrs, timeout=2, retry=3, ignore_lookup_errors=True)
-
-        # Some addresses (like demo.grafolean.com) resolve to multiple IPs, so each call to multi_ping will
-        # resolve differently - we must find the new IP addresses every time:
-        ip_to_addr = get_addr_for_ip_dict(addrs, mp)
-
-        for no_resp in no_responses:
-            addr = ip_to_addr.get(no_resp, no_resp)
-            results[addr].append(None)
-        for resp, t in responses.items():
-            addr = ip_to_addr.get(resp, resp)
-            results[addr].append(t)
-
-        if i < N_PINGS - 1:
-            time.sleep(1)
-    return dict(results)
-
-def send_results_to_grafolean(base_url, account_id, bot_token, results):
-    url = '{}/api/accounts/{}/values/?b={}'.format(base_url, account_id, bot_token)
-    values = []
-    for ip in results:
-        for ping_index, ping_time in enumerate(results[ip]):
-            values.append({
-                'p': 'ping.{}.{}.success'.format(ip.replace('.', '_'), ping_index),
-                'v': 0 if ping_time is None else 1,
-            })
-            if ping_time is not None:
-                values.append({
-                    'p': 'ping.{}.{}.rtt'.format(ip.replace('.', '_'), ping_index),
-                    'v': ping_time,
-                })
-    print("Sending results to Grafolean")
-    r = requests.post(url, json=values)
-    print(r.text)
-    r.raise_for_status()
 
 if __name__ == "__main__":
-    addrs = ["8.8.8.8", "youtube.com", "127.0.0.1", "demo.grafolean.com", "grafolean.com", "whateverdoeesndfexist.com"]
-    results = do_ping(addrs)
-    send_results_to_grafolean(os.environ.get('BACKEND_URL'), 1, os.environ.get('BOT_TOKEN'), results)
+    dotenv.load_dotenv()
+
+    backend_url = os.environ.get('BACKEND_URL')
+    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 = PingCollector(backend_url, bot_token, jobs_refresh_interval)
+    c.execute()