added parallel multiprocessing support

Author: konstantinpae

probe.py

@@ -39,6 +39,8 @@
     from miniprobe import MiniProbe
     import sensors
     import requests
+    import worker
+    import multiprocessing
 except Exception as e:
     print e
     #sys.exit()
@@ -111,16 +113,31 @@ def main():
             gc.collect()
             if str(json_response) != '[]':
                 json_response_chunks = [json_response[i:i + 10] for i in range(0, len(json_response), 10)]
+                procs = []
+                out_queue = multiprocessing.Queue()
                 for element in json_response_chunks:
                     for part in element:
                         if config['debug']:
                             logging.debug(part)
                         for sensor in sensor_list:
                             if part['kind'] == sensor.get_kind():
-                                json_payload_data.append(sensor.get_data(part))
+                                #json_payload_data.append(sensor.get_data(part))
+                                p = multiprocessing.Process(target=sensor.get_data, args=(part, out_queue),
+                                                            name=part['kind'])
+                                procs.append(p)
+                                p.start()
+                                print p
                             else:
                                 pass
                         gc.collect()
+                    for i in range(len(procs)):
+                        json_payload_data.append(out_queue.get())
+                    for p in procs:
+                        #json_payload_data.append(out_queue.get())
+                        print p
+                        p.join()
+                        procs = []
+
                     url_data = mini_probe.create_url(config, 'data')
                     try:
                         request_data = requests.post(url_data, data=json.dumps(json_payload_data), verify=False)
@@ -136,6 +153,7 @@ def main():
                         time.sleep((int(config['baseinterval']) * (9 / len(json_response))))
                     else:
                         time.sleep(int(config['baseinterval']) / 2)
+                del out_queue
             else:
                 logging.info("Nothing to do. Waiting for %s seconds." % (int(config['baseinterval']) / 3))
                 time.sleep(int(config['baseinterval']) / 3)

sensors/cpuload.py

@@ -52,7 +52,7 @@ def get_sensordef():
         return sensordefinition
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         cpuload = CPULoad()
         logging.info("Running sensor: %s" % cpuload.get_kind())
         try:
@@ -77,6 +77,7 @@ def get_data(data):
         }
         del cpuload
         gc.collect()
+        out_queue.put(data)
         return data
 
     @staticmethod

sensors/diskspace.py

@@ -53,7 +53,7 @@ def get_sensordef():
         return sensordefinition
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         diskspace = Diskspace()
         try:
             disk = diskspace.read_disk()
@@ -76,6 +76,7 @@ def get_data(data):
         }
         del diskspace
         gc.collect()
+        out_queue.put(data)
         return data
 
     def read_disk(self):

sensors/http.py

@@ -173,7 +173,7 @@ def request(self, url, request_method=None, auth_method=None, timeout=None, post
         return data
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         http = HTTP()
         try:
             http_data = http.request(data['url'], request_method=data["http_method"], auth_method=data["auth_method"],
@@ -204,4 +204,5 @@ def get_data(data):
         }
         del http
         gc.collect()
+        out_queue.put(data)
         return data

sensors/memory.py

@@ -52,7 +52,7 @@ def get_sensordef():
         return sensordefinition
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         memory = Memory()
         try:
             mem = memory.read_memory('/proc/meminfo')
@@ -77,6 +77,7 @@ def get_data(data):
         }
         del memory
         gc.collect()
+        out_queue.put(data)
         return data
 
     def read_memory(self, path):

sensors/ping.py

@@ -135,7 +135,7 @@ def ping(self, target, count, timeout, packetsize):
         return channel_list
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         ping = Ping()
         try:
             pingdata = ping.ping(data['host'], data['pingcount'], data['timeout'], data['packsize'])
@@ -157,4 +157,5 @@ def get_data(data):
         }
         del ping
         gc.collect()
+        out_queue.put(data)
         return data

sensors/port.py

@@ -112,26 +112,27 @@ def port(self, target, timeout, port):
         return channel_list
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         port = Port()
         try:
             port_data = port.port(data['host'], data['timeout'], data['targetport'])
             logging.info("Running sensor: %s" % port.get_kind())
         except Exception as e:
             logging.error("Ooops Something went wrong with '%s' sensor %s. Error: %s" % (port.get_kind(),
                                                                                          data['sensorid'], e))
-            sensor_data = {
+            data = {
                 "sensorid": int(data['sensorid']),
                 "error": "Exception",
                 "code": 1,
                 "message": "Port check failed. See log for details"
             }
-            return sensor_data
-        sensor_data = {
+            return data
+        data = {
             "sensorid": int(data['sensorid']),
             "message": "OK Port %s available" % data['targetport'],
             "channel": port_data
         }
         del port
         gc.collect()
-        return sensor_data
+        out_queue.put(data)
+        return data

sensors/probehealth.py

@@ -52,7 +52,7 @@ def get_sensordef():
         return sensordefinition
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         probehealth = Probehealth()
         try:
             mem = probehealth.read_memory('/proc/meminfo')
@@ -78,6 +78,7 @@ def get_data(data):
             "message": "OK",
             "channel": probedata
         }
+        out_queue.put(data)
         return data
 
     def read_memory(self, path):

sensors/snmpcustom.py

@@ -171,7 +171,7 @@ def snmp_get(self, oid, target, snmp_type, community, port, unit, multiplication
         return channellist
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         snmpcustom = SNMPCustom()
         try:
             snmp_data = snmpcustom.snmp_get(str(data['oid']), data['host'], data['value_type'],
@@ -197,4 +197,5 @@ def get_data(data):
         }
         del snmpcustom
         gc.collect()
+        out_queue.put(data)
         return data

sensors/snmptraffic.py

@@ -165,7 +165,7 @@ def snmp_get(self, target, countertype, community, port, ifindex):
         return channellist
 
     @staticmethod
-    def get_data(data):
+    def get_data(data, out_queue):
         snmptraffic = SNMPTraffic()
         try:
             snmp_data = snmptraffic.snmp_get(data['host'], data['snmp_counter'],
@@ -190,4 +190,5 @@ def get_data(data):
         }
         del snmptraffic
         gc.collect()
+        out_queue.put(data)
         return data