Repository :
http://git.fedorahosted.org/cgit/copr.git
On branch : master
---------------------------------------------------------------
commit c6cd80a7e0b774c07bec797593119e0060ba3acc
Author: Valentin Gologuzov <vgologuz(a)redhat.com>
Date: Mon Apr 20 12:43:32 2015 +0200
[backend] notify job_grab to remove job from added when start_job failed
---------------------------------------------------------------
backend/backend/daemons/dispatcher.py | 91 +++++++++++++++++-------------
backend/tests/deamons/test_dispatcher.py | 57 ++++++++++---------
2 files changed, 81 insertions(+), 67 deletions(-)
diff --git a/backend/backend/daemons/dispatcher.py
b/backend/backend/daemons/dispatcher.py
index 08566ca..8897aa8 100644
--- a/backend/backend/daemons/dispatcher.py
+++ b/backend/backend/daemons/dispatcher.py
@@ -235,13 +235,30 @@ class Worker(multiprocessing.Process):
self.notify_job_grab_about_task_end(job)
self._announce_end(job)
+ def can_start_job(self, job):
+ """
+ Checks if we can and/or should start job
+ :type job: BuildJob
+ :rtype: bool
+ """
+ # Checking whether the build is not cancelled
+ if not self.starting_build(job):
+ self.log.info("Couldn't start job: {}".format(job))
+ return False
+
+ # Checking whether to build or skip
+ if self.pkg_built_before(job.pkg, job.chroot, job.destdir):
+ self.on_pkg_skip(job)
+ return False
+
+ return True
+
def obtain_job(self):
"""
Retrieves new build task from queue.
Checks if the new job can be started and not skipped.
"""
- self.update_process_title(suffix="No task")
-
+ # ToDo: remove retask, use redis lua fsm logic similiar to VMM
# this sometimes caused TypeError in random worker
# when another one picekd up a task to build
# why?
@@ -256,20 +273,6 @@ class Worker(multiprocessing.Process):
self.update_process_title(suffix="Task: {} chroot: {}, obtained at {}"
.format(job.build_id, job.chroot,
str(datetime.now())))
- # Checking whether the build is not cancelled
- if not self.starting_build(job):
- self.log.info("Couldn't start job: {}".format(job))
- return
-
- # Checking whether to build or skip
- if self.pkg_built_before(job.pkg, job.chroot, job.destdir):
- self.on_pkg_skip(job)
- return
-
- # FIXME
- # this is our best place to sanity check the job before starting
- # up any longer process
-
return job
def do_job(self, job):
@@ -382,15 +385,7 @@ class Worker(multiprocessing.Process):
self.rc.publish(JOB_GRAB_TASK_END_PUBSUB, json.dumps(request))
- def run_cycle(self):
- self.update_process_title(suffix="trying to acquire job")
-
- job = self.obtain_job()
- if not job:
- self.update_process_title(suffix="trying to acquire job")
- time.sleep(self.opts.sleeptime)
- return
-
+ def acquire_vm_for_job(self, job):
self.log.info("got job: {}, acquiring VM for build".format(str(job)))
start_vm_wait_time = time.time()
vmd = None
@@ -408,27 +403,43 @@ class Worker(multiprocessing.Process):
self.log.exception("Unhandled exception during VM acquire
:{}".format(error))
self.notify_job_grab_about_task_end(job, do_reschedule=True)
time.sleep(self.opts.sleeptime)
- return
+ break
+ return vmd
- try:
+ def run_cycle(self):
+ self.update_process_title(suffix="trying to acquire job")
+
+ time.sleep(self.opts.sleeptime)
+ job = self.obtain_job()
+ if not job:
+ return
+
+ if not self.can_start_job(job):
+ self.notify_job_grab_about_task_end(job)
+ return
+
+ vmd = self.acquire_vm_for_job(job)
+
+ if vmd is not None:
self.log.info("acquired VM: {} ip: {} for build
{}".format(vmd.vm_name, vmd.vm_ip, job.task_id))
# TODO: store self.vmd = vmd and use it
self.vm_name = vmd.vm_name
self.vm_ip = vmd.vm_ip
- self.do_job(job)
- self.notify_job_grab_about_task_end(job)
- except VmError as error:
- self.log.exception("Builder error, re-scheduling task:
{}".format(error))
- self.notify_job_grab_about_task_end(job, do_reschedule=True)
- except Exception as error:
- self.log.exception("Unhandled build error: {}".format(error))
- self.notify_job_grab_about_task_end(job, do_reschedule=True)
- finally:
- # clean up the instance
- self.vmm.release_vm(vmd.vm_name)
- self.vm_ip = None
- self.vm_name = None
+ try:
+ self.do_job(job)
+ self.notify_job_grab_about_task_end(job)
+ except VmError as error:
+ self.log.exception("Builder error, re-scheduling task:
{}".format(error))
+ self.notify_job_grab_about_task_end(job, do_reschedule=True)
+ except Exception as error:
+ self.log.exception("Unhandled build error: {}".format(error))
+ self.notify_job_grab_about_task_end(job, do_reschedule=True)
+ finally:
+ # clean up the instance
+ self.vmm.release_vm(vmd.vm_name)
+ self.vm_ip = None
+ self.vm_name = None
def run(self):
self.log.info("Starting worker")
diff --git a/backend/tests/deamons/test_dispatcher.py
b/backend/tests/deamons/test_dispatcher.py
index 70a2e91..143043e 100644
--- a/backend/tests/deamons/test_dispatcher.py
+++ b/backend/tests/deamons/test_dispatcher.py
@@ -338,64 +338,57 @@ class TestDispatcher(object):
mc_tq = MagicMock()
self.worker.task_queue = mc_tq
self.worker.starting_build = MagicMock()
- self.worker.pkg_built_before = MagicMock()
- self.worker.pkg_built_before.return_value = False
mc_tq.dequeue.return_value = MagicMock(data=self.task)
obtained_job = self.worker.obtain_job()
assert obtained_job.__dict__ == self.job.__dict__
- assert self.worker.pkg_built_before.called
- def test_obtain_job_skip_pkg(self, init_worker):
- mc_tq = MagicMock()
- self.worker.task_queue = mc_tq
+ def test_can_start_job_skip_pkg(self, init_worker):
self.worker.starting_build = MagicMock()
+ self.worker.starting_build.return_value = False
self.worker.pkg_built_before = MagicMock()
self.worker.pkg_built_before.return_value = True
- self.worker.mark_started = MagicMock()
- self.worker.return_results = MagicMock()
- self.worker.notify_job_grab_about_task_end = MagicMock()
+ assert self.worker.can_start_job(self.job) is False
- mc_tq.dequeue.return_value = MagicMock(data=self.task)
- assert self.worker.obtain_job() is None
- assert self.worker.pkg_built_before.called
- assert self.worker.notify_job_grab_about_task_end.called
+ def test_can_start_job_not_starting(self, init_worker):
+ self.worker.starting_build = MagicMock()
+ self.worker.starting_build.return_value = False
+ self.worker.pkg_built_before = MagicMock()
+ self.worker.pkg_built_before.return_value = False
- def test_obtain_job_dequeue_type_error(self, init_worker):
- mc_tq = MagicMock()
- self.worker.task_queue = mc_tq
+ assert self.worker.can_start_job(self.job) is False
+
+ def test_can_start_job(self, init_worker):
self.worker.starting_build = MagicMock()
+ self.worker.starting_build.return_value = True
self.worker.pkg_built_before = MagicMock()
self.worker.pkg_built_before.return_value = False
- mc_tq.dequeue.side_effect = TypeError()
- assert self.worker.obtain_job() is None
- assert not self.worker.starting_build.called
- assert not self.worker.pkg_built_before.called
+ assert self.worker.can_start_job(self.job) is True
- def test_obtain_job_dequeue_none_result(self, init_worker):
+ def test_obtain_job_dequeue_type_error(self, init_worker):
mc_tq = MagicMock()
self.worker.task_queue = mc_tq
self.worker.starting_build = MagicMock()
self.worker.pkg_built_before = MagicMock()
self.worker.pkg_built_before.return_value = False
- mc_tq.dequeue.return_value = None
+ mc_tq.dequeue.side_effect = TypeError()
assert self.worker.obtain_job() is None
assert not self.worker.starting_build.called
assert not self.worker.pkg_built_before.called
- def test_obtain_job_on_starting_build(self, init_worker):
+ def test_obtain_job_dequeue_none_result(self, init_worker):
mc_tq = MagicMock()
self.worker.task_queue = mc_tq
self.worker.starting_build = MagicMock()
- self.worker.starting_build.return_value = False
self.worker.pkg_built_before = MagicMock()
self.worker.pkg_built_before.return_value = False
- mc_tq.dequeue.return_value = MagicMock(data=self.task)
+ mc_tq.dequeue.return_value = None
assert self.worker.obtain_job() is None
+ assert not self.worker.starting_build.called
assert not self.worker.pkg_built_before.called
def test_dummy_run(self, init_worker, mc_time, mc_grc):
@@ -423,7 +416,6 @@ class TestDispatcher(object):
self.worker.update_process_title()
base_title = 'worker-{} {} '.format(self.group_id, self.worker_num)
assert mc_setproctitle.call_args[0][0] == base_title
-
#mc_setproctitle.reset_mock()
self.worker.vm_ip = self.vm_ip
self.worker.update_process_title()
@@ -513,7 +505,6 @@ class TestDispatcher(object):
assert
self.worker.notify_job_grab_about_task_end.call_args[1]["do_reschedule"]
assert self.worker.vmm.release_vm.called
-
### handle other errors
self.worker.notify_job_grab_about_task_end.reset_mock()
self.worker.vmm.release_vm.reset_mock()
@@ -522,3 +513,15 @@ class TestDispatcher(object):
assert
self.worker.notify_job_grab_about_task_end.call_args[1]["do_reschedule"]
assert self.worker.vmm.release_vm.called
+
+ def test_run_cycle_halt_on_can_start_job_false(self, init_worker):
+ self.worker.notify_job_grab_about_task_end = MagicMock()
+ self.worker.obtain_job = MagicMock()
+ self.worker.obtain_job.return_value = self.job
+ self.worker.can_start_job = MagicMock()
+ self.worker.can_start_job.return_value = False
+ self.worker.acquire_vm_for_job = MagicMock()
+
+ self.worker.run_cycle()
+ assert self.worker.can_start_job.called
+ assert not self.worker.acquire_vm_for_job.called