Repository : http://git.fedorahosted.org/cgit/copr.git
On branch : master
commit 8456ca5176ef17a2ea7c5380998c866bc27e17ea Author: Valentin Gologuzov vgologuz@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]
copr-commits@lists.fedorahosted.org