implement ES backend

examples/config.test.json

@@ -5,7 +5,7 @@
     "monitors": [
         {
             "type":"uptime",
-            "freq":"5",
+            "freq":"30",
             "args":{}
         }
     ]

pymonitor/builtins/sysinfo.py

@@ -0,0 +1,13 @@
+from subprocess import Popen,PIPE
+
+def ipaddr():
+    """
+    Return first entry of hostname --all-ip-addresses
+    """
+    return Popen(["hostname", "--all-ip-addresses"], stdout=PIPE).communicate()[0].decode().split(" ")[0].strip()
+
+def hostname():
+    """
+    Return system hostname from hostname -f
+    """
+    return Popen(["hostname", "-f"], stdout=PIPE).communicate()[0].decode().strip()

pymonitor/daemon.py

@@ -1,7 +1,10 @@
 #!/usr/bin/env python3
 
 from threading import Thread
+from elasticsearch import Elasticsearch
 from time import time,sleep
+from pymonitor.builtins import sysinfo
+import datetime
 import logging
 import json
 import sys
@@ -12,6 +15,7 @@ class MonitorDaemon(Thread):
         Thread.__init__(self)
         self.config = config
         self.threads = []
+        self.backend = Backend(self.config["backend"]["url"])
     
     def run(self):
         """
@@ -26,7 +30,7 @@ class MonitorDaemon(Thread):
         # Create/start all monitoring threads
         logger.info("starting monitor threads")
         for instance in self.config["monitors"]:
-            monitor_thread = MonitorThread(instance)
+            monitor_thread = MonitorThread(instance, self.backend)
             monitor_thread.start()
             self.threads.append(monitor_thread)
         
@@ -45,13 +49,76 @@ class MonitorDaemon(Thread):
             monitor_thread.shutdown()
 
 
+class Backend:
+    def __init__(self, es_url):
+        """
+        Init elasticsearch client
+        """
+        self.logger = logging.getLogger("monitordaemon.backend")
+        
+        self.sysinfo = {}
+        self.update_sys_info()
+        logger.info("running on %(hostname)s (%(ipaddr)s)" % self.sysinfo)
+        
+        self.logger.info("connecting to backend at %s" % es_url)
+        self.es = Elasticsearch([es_url])
+        self.logger.info("connected to backend")
+        self.current_index = ""
+        self.check_before_entry()
+    
+    def update_sys_info(self):
+        """
+        Fetch generic system info that is sent with every piece of monitoring data
+        """
+        self.sysinfo["hostname"] = sysinfo.hostname()
+        self.sysinfo["ipaddr"] = sysinfo.ipaddr()
+    
+    def get_index_name(self):
+        """
+        Return name of current index such as 'monitor-2015.12.05'
+        """
+        return "monitor-%s" % datetime.datetime.now().strftime("%Y.%m.%d")
+    
+    def create_index(self, indexName):
+        """
+        Check if current index exists, and if not, create it
+        """
+        if not self.es.indices.exists(index=indexName):
+            self.logger.info("creating index %s" % indexName)
+            self.es.indices.create(index=indexName, ignore=400) # ignore already exists error
+        self.current_index = indexName
+    
+    def check_before_entry(self):
+        """
+        Called before adding any data to ES. Checks if a new index should be created due to date change
+        """
+        indexName = self.get_index_name()
+        if indexName != self.current_index:
+            self.create_index(indexName)
+    
+    def add_data(self, data_type, data):
+        """
+        Submit a piece of monitoring data
+        """
+        self.check_before_entry()
+        
+        doc = self.sysinfo.copy()
+        doc.update(data)
+        doc["@timestamp"] = datetime.datetime.utcnow().isoformat()
+        
+        self.logger.info("logging type %s: %s" % (data_type, doc))
+        res = self.es.index(index=self.current_index, doc_type=data_type, body=doc)
+        self.logger.info("%s created %s" % (data_type, res["_id"]))
+    
+
 class MonitorThread(Thread):
-    def __init__(self, config):
+    def __init__(self, config, backend):
         """
         Load checker function and prepare scheduler
         """
         Thread.__init__(self)
         self.config = config
+        self.backend = backend
         self.logger = logging.getLogger("monitordaemon.monitorthread.%s"%self.config["type"])
         self.logger.info("initing worker thread with config %s" % self.config)
         
@@ -80,6 +147,7 @@ class MonitorThread(Thread):
         """
         result = self.checker_func(**args)
         self.logger.info("result: %s" % (result,))
+        self.backend.add_data(self.config["type"], result)
     
     def shutdown(self):
         """