[copr] master: [backend] check for aux process state and restart if needed (8456ca5)
by vgologuz@fedoraproject.org
Repository : http://git.fedorahosted.org/cgit/copr.git
On branch : master
>---------------------------------------------------------------
commit 8456ca5176ef17a2ea7c5380998c866bc27e17ea
Author: Valentin Gologuzov <vgologuz(a)redhat.com>
Date: Mon May 18 19:01:50 2015 +0200
[backend] check for aux process state and restart if needed
>---------------------------------------------------------------
backend/backend/daemons/backend.py | 55 ++++++++++++++++++++--------
backend/backend/daemons/dispatcher.py | 2 +
backend/backend/daemons/vm_master.py | 1 +
backend/backend/vm_manage/event_handle.py | 2 +-
backend/backend/vm_manage/terminate.py | 2 +-
backend/tests/deamons/test_backend.py | 6 ++-
6 files changed, 48 insertions(+), 20 deletions(-)
diff --git a/backend/backend/daemons/backend.py b/backend/backend/daemons/backend.py
index e07107f..2269733 100644
--- a/backend/backend/daemons/backend.py
+++ b/backend/backend/daemons/backend.py
@@ -67,7 +67,7 @@ class CoprBackend(object):
# event format is a dict {when:time, who:[worker|logger|job|main],
# what:str}
self.frontend_client = FrontendClient(self.opts)
- self.abort = False
+ self.is_running = False
if not os.path.exists(self.opts.worker_logdir):
os.makedirs(self.opts.worker_logdir, mode=0o750)
@@ -101,38 +101,59 @@ class CoprBackend(object):
self.clean_task_queues()
- def init_sub_process(self):
- """
- - Create backend logger
- - Create job grabber
- """
-
+ def _start_log_handler(self):
self.redis_log_handler = RedisLogHandler(self.opts)
self.redis_log_handler.start()
time.sleep(1)
+ def _start_job_grab(self):
self.log.info("Starting up Job Grabber")
self._jobgrab = CoprJobGrab(opts=self.opts,
frontend_client=self.frontend_client,
lock=self.lock)
self._jobgrab.start()
- self.log.info("Starting up VM Spawner")
+ def _start_vm_master(self):
+ self.log.info("Starting up VM Master")
+ self.vmm_daemon = VmMaster(self.vm_manager)
+ self.vmm_daemon.start()
+
+ def init_sub_process(self):
+ """
+ - Create backend logger
+ - Create job grabber
+ """
+
+ self._start_log_handler()
+ self._start_job_grab()
+
+ self.log.info("Creating up VM Spawner")
self.spawner = Spawner(self.opts)
- self.log.info("Starting up VM Health checker")
+ self.log.info("Creating up VM Health checker")
self.checker = HealthChecker(self.opts)
- self.log.info("Starting up VM Terminator")
+ self.log.info("Creating up VM Terminator")
self.terminator = Terminator(self.opts)
- self.log.info("Starting up VM Master")
self.vm_manager = VmManager(opts=self.opts,
logger=self.log,
checker=self.checker,
spawner=self.spawner,
terminator=self.terminator)
self.vm_manager.post_init()
- self.vmm_daemon = VmMaster(self.vm_manager)
- self.vmm_daemon.start()
+ self._start_vm_master()
+
+ def ensure_sub_processes_alive(self):
+ if self.is_running:
+ for proc, start_method in [
+ (self.redis_log_handler, self._start_log_handler),
+ (self._jobgrab, self._start_job_grab),
+
+ ]:
+ if not proc.is_alive():
+ self.log.error("Process `{}` died unexpectedly, restarting".format(proc))
+ proc.terminate()
+ proc.join()
+ start_method()
def event(self, what):
"""
@@ -209,7 +230,7 @@ class CoprBackend(object):
And also clean all task queues as they would survive copr restart
"""
- self.abort = True
+ self.is_running = False
for group in self.opts.build_groups:
group_id = group["id"]
for w in self.workers_by_group_id[group_id][:]:
@@ -229,11 +250,13 @@ class CoprBackend(object):
self.log.info("Initial config: {}".format(self.opts))
self.log.info("Sub processes was started")
- self.abort = False
- while not self.abort:
+ self.is_running = True
+ while self.is_running:
# re-read config into opts
self.update_conf()
+ self.ensure_sub_processes_alive()
+
for group in self.opts.build_groups:
group_id = group["id"]
diff --git a/backend/backend/daemons/dispatcher.py b/backend/backend/daemons/dispatcher.py
index 66d1af8..1b43ede 100644
--- a/backend/backend/daemons/dispatcher.py
+++ b/backend/backend/daemons/dispatcher.py
@@ -376,6 +376,8 @@ class Worker(multiprocessing.Process):
setproctitle(title)
def notify_job_grab_about_task_end(self, job, do_reschedule=False):
+ # TODO: Current notification method is unreliable,
+ # we should retask and use redis + lua for atomic acquire/release tasks
request = {
"action": "reschedule" if do_reschedule else "remove",
"build_id": job.build_id,
diff --git a/backend/backend/daemons/vm_master.py b/backend/backend/daemons/vm_master.py
index db32ef7..c8ccb0e 100644
--- a/backend/backend/daemons/vm_master.py
+++ b/backend/backend/daemons/vm_master.py
@@ -193,6 +193,7 @@ class VmMaster(Process):
self.event_handler = EventHandler(self.vmm)
self.event_handler.start()
+ self.log.info("VM master process started")
while not self.kill_received:
time.sleep(Thresholds.cycle_timeout)
try:
diff --git a/backend/backend/vm_manage/event_handle.py b/backend/backend/vm_manage/event_handle.py
index 8c0b328..460440c 100644
--- a/backend/backend/vm_manage/event_handle.py
+++ b/backend/backend/vm_manage/event_handle.py
@@ -118,7 +118,7 @@ class EventHandler(Process):
def on_vm_termination_result(self, msg):
if msg["result"] == "OK" and "vm_name" in msg:
self.log.debug("Vm terminated, removing from pool ip: {}, name: {}, msg: {}"
- .format(msg.get("vm_ip"), msg.get("vm_name"), msg.get("msg")))
+ .format(msg.get("vm_ip"), msg.get("vm_name"), msg.get("msg")))
self.vmm.remove_vm_from_pool(msg["vm_name"])
elif "vm_name" not in msg:
self.log.debug("Vm termination event missing vm name, msg: {}".format(msg))
diff --git a/backend/backend/vm_manage/terminate.py b/backend/backend/vm_manage/terminate.py
index 64a826b..ebf183a 100644
--- a/backend/backend/vm_manage/terminate.py
+++ b/backend/backend/vm_manage/terminate.py
@@ -37,9 +37,9 @@ def terminate_vm(opts, terminate_playbook, group, vm_name, vm_ip):
"topic": EventTopics.VM_TERMINATED,
"result": "OK"
}
+ start_time = time.time()
try:
log.info("starting terminate vm with args: {}".format(term_args))
- start_time = time.time()
run_ansible_playbook_cli(args, "terminate instance", log)
result["result"] = "OK"
except Exception as error:
diff --git a/backend/tests/deamons/test_backend.py b/backend/tests/deamons/test_backend.py
index c3bf2b3..6115d3a 100644
--- a/backend/tests/deamons/test_backend.py
+++ b/backend/tests/deamons/test_backend.py
@@ -302,7 +302,7 @@ class TestBackend(object):
self.be.terminate()
assert self.be.clean_task_queues.called
- assert self.be.abort
+ assert not self.be.is_running
assert worker_alive.terminate_instance.called
assert worker_dead.terminate_instance.called
@@ -314,6 +314,7 @@ class TestBackend(object):
self.be.clean_task_queues = MagicMock()
self.be.init_sub_process = MagicMock()
+ self.be.ensure_sub_processes_alive = MagicMock()
# self.be.init_task_queues = MagicMock()
self.be.update_conf = MagicMock()
self.be.spin_up_workers_by_group = MagicMock()
@@ -334,7 +335,8 @@ class TestBackend(object):
mock.call(self.opts.build_groups[1]),
]
assert self.be.update_conf.called
- assert self.be.abort
+ assert self.be.ensure_sub_processes_alive.called
+ assert not self.be.is_running
assert not self.be.workers_by_group_id[0]
assert not self.be.workers_by_group_id[1]