Idempotent scheduler and QueueItems
Make the scheduler idempotent. The idea is that after any event,
the scheduler should be able to run and examine the state of every
item in the queue and act accordingly. This is a change from the
current state where most events are dealt with in context. This
should ease maintenance as it should facilitate reasoning about
the different actions Zuul might take -- centralizing major
decisions into one function.
Also add a new class QueueItem, which represents a Change(ish)
in a queue. Currently, Change objects themselves are placed
in the queue, which is confusing information about a change (for
instance: it's number and patchset) as well as information about
the processing of that change in the queue (e.g., the build
history, current build set, merge status, etc.).
Change objects are now cached, which should reduce the number of
queries to Gerrit (except the current algorithm to update them is
very naive and queries Gerrit again on any event relating to a
change). Changes are expired from the cache when they are not
present or related to any change currently in a pipeline.
There are now two things that need to be asserted at the end of
each test, so use addCleanup in setUp to call a method that
performs those assertions after the test method completes. Also,
move the existing shutdown method to use addCleanup as well,
because testr experts say that's a best practice.
Change-Id: Id2bf4c484c9e681456c69d99787e7a5b3a247690
Reviewed-on: https://review.openstack.org/34653
Reviewed-by: Jeremy Stanley <fungi@yuggoth.org>
Reviewed-by: Clark Boylan <clark.boylan@gmail.com>
Approved: James E. Blair <corvus@inaugust.com>
Tested-by: Jenkins
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 07d5c7d..4de5d05 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -770,7 +770,16 @@
self.builds = self.worker.running_builds
self.history = self.worker.build_history
- def tearDown(self):
+ self.addCleanup(self.assertFinalState)
+ self.addCleanup(self.shutdown)
+
+ def assertFinalState(self):
+ # Make sure that the change cache is cleared
+ assert len(self.sched.trigger._change_cache.keys()) == 0
+ self.assertEmptyQueues()
+
+ def shutdown(self):
+ self.log.debug("Shutting down after tests")
self.launcher.stop()
self.worker.shutdown()
self.gearman_server.shutdown()
@@ -1031,7 +1040,6 @@
assert self.getJobFromHistory('project-test2').result == 'SUCCESS'
assert A.data['status'] == 'MERGED'
assert A.reported == 2
- self.assertEmptyQueues()
self.assertReportedStat('gerrit.event.comment-added', '1|c')
self.assertReportedStat('zuul.pipeline.gate.current_changes', '1|g')
@@ -1050,11 +1058,6 @@
self.fake_gerrit.addEvent(A.getChangeRestoredEvent())
self.waitUntilSettled()
- print self.builds
- print A.messages
-
- self.assertEmptyQueues()
-
assert len(self.history) == 2
self.history[0].name == 'project-test1'
self.history[1].name == 'project-test1'
@@ -1147,7 +1150,6 @@
assert A.reported == 2
assert B.reported == 2
assert C.reported == 2
- self.assertEmptyQueues()
def test_failed_changes(self):
"Test that a change behind a failed change is retested"
@@ -1178,7 +1180,6 @@
assert B.data['status'] == 'MERGED'
assert A.reported == 2
assert B.reported == 2
- self.assertEmptyQueues()
def test_independent_queues(self):
"Test that changes end up in the right queues"
@@ -1226,7 +1227,6 @@
assert A.reported == 2
assert B.reported == 2
assert C.reported == 2
- self.assertEmptyQueues()
def test_failed_change_at_head(self):
"Test that if a change at the head fails, jobs behind it are canceled"
@@ -1284,7 +1284,6 @@
assert A.reported == 2
assert B.reported == 2
assert C.reported == 2
- self.assertEmptyQueues()
def test_failed_change_at_head_with_queue(self):
"Test that if a change at the head fails, queued jobs are canceled"
@@ -1347,7 +1346,6 @@
assert A.reported == 2
assert B.reported == 2
assert C.reported == 2
- self.assertEmptyQueues()
def test_patch_order(self):
"Test that dependent patches are tested in the right order"
@@ -1392,7 +1390,6 @@
assert A.reported == 2
assert B.reported == 2
assert C.reported == 2
- self.assertEmptyQueues()
def test_can_merge(self):
"Test whether a change is ready to merge"
@@ -1403,13 +1400,13 @@
assert not self.sched.trigger.canMerge(a, mgr.getSubmitAllowNeeds())
A.addApproval('CRVW', 2)
- a = self.sched.trigger.getChange(1, 2)
+ a = self.sched.trigger.getChange(1, 2, refresh=True)
assert not self.sched.trigger.canMerge(a, mgr.getSubmitAllowNeeds())
A.addApproval('APRV', 1)
- a = self.sched.trigger.getChange(1, 2)
+ a = self.sched.trigger.getChange(1, 2, refresh=True)
assert self.sched.trigger.canMerge(a, mgr.getSubmitAllowNeeds())
- self.assertEmptyQueues()
+ self.sched.trigger.maintainCache([])
def test_build_configuration(self):
"Test that zuul merges the right commits for testing"
@@ -1444,7 +1441,6 @@
repo_messages.reverse()
correct_messages = ['initial commit', 'A-1', 'B-1', 'C-1']
assert repo_messages == correct_messages
- self.assertEmptyQueues()
def test_build_configuration_conflict(self):
"Test that merge conflicts are handled"
@@ -1481,7 +1477,6 @@
assert A.reported == 2
assert B.reported == 2
assert C.reported == 2
- self.assertEmptyQueues()
def test_post(self):
"Test that post jobs run"
@@ -1504,7 +1499,6 @@
job_names = [x.name for x in self.history]
assert len(self.history) == 1
assert 'project-post' in job_names
- self.assertEmptyQueues()
def test_build_configuration_branch(self):
"Test that the right commits are on alternate branches"
@@ -1539,7 +1533,6 @@
repo_messages.reverse()
correct_messages = ['initial commit', 'mp commit', 'A-1', 'B-1', 'C-1']
assert repo_messages == correct_messages
- self.assertEmptyQueues()
def test_build_configuration_branch_interaction(self):
"Test that switching between branches works"
@@ -1550,7 +1543,6 @@
repo = git.Repo(path)
repo.heads.master.commit = repo.commit('init')
self.test_build_configuration()
- self.assertEmptyQueues()
def test_build_configuration_multi_branch(self):
"Test that dependent changes on multiple branches are merged"
@@ -1595,7 +1587,6 @@
repo_messages.reverse()
correct_messages = ['initial commit', 'mp commit', 'B-1']
assert repo_messages == correct_messages
- self.assertEmptyQueues()
def test_one_job_project(self):
"Test that queueing works with one job"
@@ -1613,7 +1604,6 @@
assert A.reported == 2
assert B.data['status'] == 'MERGED'
assert B.reported == 2
- self.assertEmptyQueues()
def test_job_from_templates_launched(self):
"Test whether a job generated via a template can be launched"
@@ -1660,7 +1650,6 @@
assert C.data['status'] == 'NEW'
assert C.reported == 2
assert len(self.history) == 1
- self.assertEmptyQueues()
def test_head_is_dequeued_once(self):
"Test that if a change at the head fails it is dequeued only once"
@@ -1726,7 +1715,6 @@
assert A.reported == 2
assert B.reported == 2
assert C.reported == 2
- self.assertEmptyQueues()
def test_nonvoting_job(self):
"Test that non-voting jobs don't vote."
@@ -1747,7 +1735,6 @@
'SUCCESS')
assert (self.getJobFromHistory('nonvoting-project-test2').result ==
'FAILURE')
- self.assertEmptyQueues()
def test_check_queue_success(self):
"Test successful check queue jobs."
@@ -1762,7 +1749,6 @@
assert self.getJobFromHistory('project-merge').result == 'SUCCESS'
assert self.getJobFromHistory('project-test1').result == 'SUCCESS'
assert self.getJobFromHistory('project-test2').result == 'SUCCESS'
- self.assertEmptyQueues()
def test_check_queue_failure(self):
"Test failed check queue jobs."
@@ -1778,7 +1764,6 @@
assert self.getJobFromHistory('project-merge').result == 'SUCCESS'
assert self.getJobFromHistory('project-test1').result == 'SUCCESS'
assert self.getJobFromHistory('project-test2').result == 'FAILURE'
- self.assertEmptyQueues()
def test_dependent_behind_dequeue(self):
"test that dependent changes behind dequeued changes work"
@@ -1865,7 +1850,6 @@
assert self.countJobResults(self.history, 'ABORTED') == 15
assert len(self.history) == 44
- self.assertEmptyQueues()
def test_merger_repack(self):
"Test that the merger works after a repack"
@@ -1894,7 +1878,6 @@
assert self.getJobFromHistory('project-test2').result == 'SUCCESS'
assert A.data['status'] == 'MERGED'
assert A.reported == 2
- self.assertEmptyQueues()
def test_merger_repack_large_change(self):
"Test that the merger works with large changes after a repack"
@@ -1914,7 +1897,6 @@
assert self.getJobFromHistory('project1-test2').result == 'SUCCESS'
assert A.data['status'] == 'MERGED'
assert A.reported == 2
- self.assertEmptyQueues()
def test_nonexistent_job(self):
"Test launching a job that doesn't exist"
@@ -1945,7 +1927,6 @@
assert self.getJobFromHistory('project-test2').result == 'SUCCESS'
assert A.data['status'] == 'MERGED'
assert A.reported == 2
- self.assertEmptyQueues()
def test_single_nonexistent_post_job(self):
"Test launching a single post job that doesn't exist"
@@ -1969,7 +1950,6 @@
self.waitUntilSettled()
assert len(self.history) == 0
- self.assertEmptyQueues()
def test_new_patchset_dequeues_old(self):
"Test that a new patchset causes the old to be dequeued"
@@ -2014,7 +1994,6 @@
assert D.data['status'] == 'MERGED'
assert D.reported == 2
assert len(self.history) == 9 # 3 each for A, B, D.
- self.assertEmptyQueues()
def test_new_patchset_dequeues_old_on_head(self):
"Test that a new patchset causes the old to be dequeued (at head)"
@@ -2058,7 +2037,6 @@
assert D.data['status'] == 'MERGED'
assert D.reported == 2
assert len(self.history) == 7
- self.assertEmptyQueues()
def test_new_patchset_dequeues_old_without_dependents(self):
"Test that a new patchset causes only the old to be dequeued"
@@ -2090,7 +2068,6 @@
assert C.data['status'] == 'MERGED'
assert C.reported == 2
assert len(self.history) == 9
- self.assertEmptyQueues()
def test_new_patchset_dequeues_old_independent_queue(self):
"Test that a new patchset causes the old to be dequeued (independent)"
@@ -2119,7 +2096,6 @@
assert C.reported == 1
assert len(self.history) == 10
assert self.countJobResults(self.history, 'ABORTED') == 1
- self.assertEmptyQueues()
def test_zuul_refs(self):
"Test that zuul refs exist and have the right changes"
@@ -2207,7 +2183,6 @@
assert C.reported == 2
assert D.data['status'] == 'MERGED'
assert D.reported == 2
- self.assertEmptyQueues()
def test_statsd(self):
"Test each of the statsd methods used in the scheduler"
@@ -2240,7 +2215,6 @@
assert A.reported == 2
assert B.data['status'] == 'MERGED'
assert B.reported == 2
- self.assertEmptyQueues()
def test_test_config(self):
"Test that we can test the config"