Fixing multithread issue

src/functions/__init__.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from multiprocessing import Process, Lock
 import subprocess
 import shlex
 
@@ -41,7 +42,7 @@ class Functions:
                 line = process.stdout.readline().rstrip()
                 output.append(line) if return_result else None
                 Functions.log(source, "info", line) if log_output else None
-                Functions.log(source, "error", process.stderr.readline())
+                Functions.log(source, "warning", process.stderr.readline())
                 return_code = process.poll()
                 if return_code is not None:
                     lines = []
@@ -49,9 +50,111 @@ class Functions:
                         output.append(line.rstrip()) if return_result else None
                         lines.append(line.rstrip())
                     Functions.log(source, "info", lines) if log_output else None
-                    Functions.log(source, "error", process.stderr.readlines())
+                    Functions.log(source, "warning", process.stderr.readlines())
                     break
 
             return output
         except Exception as e:
             Functions.log(source, 'error', "%s" % (e))
+
+
+
+    @staticmethod
+    def log_ts(source, level, message, lock):
+        if message is None or message == "":
+            return
+
+        if not isinstance(message, (list, tuple)):
+            message = [message]
+        
+        lock.acquire()
+        try:
+            for line in message:
+                print("[%s] %s [%s]: %s" % (source, datetime.now().strftime("%x %X"), level, line.rstrip()))
+        finally:
+            lock.release()
+
+    @staticmethod
+    def run_bash_ts(source, command, lock):
+        if not isinstance(command, (list, tuple)):
+            command = shlex.split(command)
+
+        try:
+            process = subprocess.Popen(command, 
+                            shell=False,
+                            stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE,
+                            bufsize=-1,
+                            universal_newlines=True)
+
+            with process.stdout:
+                for line in iter(process.stdout.readline, b''): 
+                    Functions.log(source, "info", line)
+
+            returncode = process.wait() 
+            Functions.log(source, "info", "Return code %s" % (returncode))
+
+        except Exception as e:
+            Functions.log_ts(source, 'error', "%s" % (e), lock)
+
+
+class DaemonizeHAProxy:
+    def __init__(self):
+        self.process = None
+        self.thread = None
+
+    def haproxy(self, action):
+        if action == "start":
+            self.__prepare("/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -S /var/run/haproxy.sock")
+        else:
+            pid = "".join(Functions().run_bash("HAPROXY", "cat /run/haproxy.pid", log_output=False))
+            self.__prepare("/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -x /var/run/haproxy.sock -sf %s" % (pid))
+
+        if self.process is None:
+            return
+
+        self.thread = Process(target=self.__start, args=())
+        self.thread.start()
+
+    def __prepare(self, command):
+        source = "HAPROXY"
+        if not isinstance(command, (list, tuple)):
+            command = shlex.split(command)
+
+        try:
+            self.process = subprocess.Popen(command, 
+                            shell=False,
+                            stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE,
+                            bufsize=-1,
+                            universal_newlines=True)
+
+        except Exception as e:
+            Functions.log_ts(source, 'error', "%s" % (e), lock)
+
+
+    def __start(self):
+        source = "HAPROXY"
+        try:
+            with self.process.stdout:
+                for line in iter(self.process.stdout.readline, b''): 
+                    Functions.log(source, "info", line)
+
+            returncode = self.process.wait() 
+            Functions.log(source, "info", "Return code %s" % (returncode))
+
+        except Exception as e:
+            Functions.log_ts(source, 'error', "%s" % (e), lock)
+
+    def is_alive(self):
+        return self.thread.is_alive()
+
+    def kill(self):
+        self.process.kill()
+        self.thread.kill()
+
+    def terminate(self):
+        self.process.terminate()
+        self.thread.terminate()
+
+    

src/main.py

@@ -1,8 +1,7 @@
-from functions import Functions
+from functions import Functions, DaemonizeHAProxy
 from processor import ProcessorInterface
 import os
 import time
-from threading import Thread
 from deepdiff import DeepDiff
 
 easyhaproxy_config = "/etc/haproxy/easyconfig.yml"
@@ -22,23 +21,27 @@ def start():
     processor_obj.save_certs(certs_haproxy)
     Functions.log('EASYHAPROXY', 'info', 'Found hosts: %s' % ", ".join(processor_obj.get_hosts())) # Needs to after save_config
 
-    #configs = Functions.run_bash('HAPROXY', 'ls /etc/haproxy/conf.d/*.cfg', log_output=False)
+    old_haproxy = None
-    x = Thread(target=Functions.run_bash, args=("HAPROXY", "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -S /var/run/haproxy.sock", True, False))
+    haproxy = DaemonizeHAProxy()
-    x.start()
+    haproxy.haproxy("start")
 
     while True:
         time.sleep(10)
+        if old_haproxy is not None:
+            old_haproxy.kill()
+            old_haproxy = None
         try:
             old_parsed = processor_obj.get_parsed_object()
             processor_obj.refresh()
-            if DeepDiff(old_parsed, processor_obj.get_parsed_object()) != {} or not x.is_alive():
+            if DeepDiff(old_parsed, processor_obj.get_parsed_object()) != {} or not haproxy.is_alive():
                 Functions.log('EASYHAPROXY', 'info', 'New configuration found. Reloading...')
                 processor_obj.save_config(haproxy_config)
                 processor_obj.save_certs(certs_haproxy)
                 Functions.log('EASYHAPROXY', 'info', 'Found hosts: %s' % ", ".join(processor_obj.get_hosts())) # Needs to after save_config
-                pid = "".join(Functions().run_bash("HAPROXY", "cat /run/haproxy.pid", log_output=False))
+                old_haproxy = haproxy
-                x = Thread(target=Functions.run_bash, args=("HAPROXY", "/usr/sbin/haproxy -W -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -x /var/run/haproxy.sock -sf %s" % (pid), True, False))
+                haproxy = DaemonizeHAProxy()
-                x.start()
+                haproxy.haproxy("reload")
+                old_haproxy.terminate()
         except Exception as e:
             Functions.log('EASYHAPROXY', 'error', "Err: %s" % (e))
         Functions.log('EASYHAPROXY', 'info', 'Heartbeat')