From 0687ba60a22ba4e2a451fc0828482212eee0bc55 Mon Sep 17 00:00:00 2001 From: Joao Gilberto Magalhaes Date: Wed, 24 Aug 2022 17:38:06 -0500 Subject: [PATCH] Fixing multithread issue --- src/functions/__init__.py | 107 +++++++++++++++++++++++++++++++++++++- src/main.py | 21 ++++---- 2 files changed, 117 insertions(+), 11 deletions(-) diff --git a/src/functions/__init__.py b/src/functions/__init__.py index b0614a8..154a035 100644 --- a/src/functions/__init__.py +++ b/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() + + \ No newline at end of file diff --git a/src/main.py b/src/main.py index c1e9b0c..c2749f9 100644 --- a/src/main.py +++ b/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) - 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)) - x.start() + old_haproxy = None + haproxy = DaemonizeHAProxy() + 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)) - 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)) - x.start() + old_haproxy = haproxy + haproxy = DaemonizeHAProxy() + haproxy.haproxy("reload") + old_haproxy.terminate() except Exception as e: Functions.log('EASYHAPROXY', 'error', "Err: %s" % (e)) Functions.log('EASYHAPROXY', 'info', 'Heartbeat')