Consolidate scheduler pause/exit as hibernation

The two properties, `_pause` and `_exit`, in scheduler are
both used for "hibernation", i.e., saving the queue in a
pickled file. The `resume` method is used to wake (load
the queue from the picked file) up the scheduler.

This is a preparation for pause/resume of a scheduler
which is needed in a multi-scheduler operation. This
hibernation functionality will be removed in the near
future and replaced with keeping the queue in Zookeeper.

Change-Id: I93343272b04eedcf10e963b3ba47042a287b6e9e
Story: 2007192
This commit is contained in:
Jan Kubovy
2020-04-03 14:49:59 +02:00
parent 6e9c922dd5
commit 9b612c0a95
3 changed files with 26 additions and 21 deletions
+1 -1
View File
@@ -3451,7 +3451,7 @@ class SchedulerTestApp:
self.sched.start()
executor_client.gearman.waitForServer()
self.sched.reconfigure(self.config)
self.sched.resume()
self.sched.wakeUp()
def fullReconfigure(self):
try:
+1 -1
View File
@@ -165,7 +165,7 @@ class Scheduler(zuul.cmd.ZuulDaemonApp):
self.sched.start()
self.sched.registerConnections(self.connections)
self.sched.reconfigure(self.config)
self.sched.resume()
self.sched.wakeUp()
except Exception:
self.log.exception("Error starting Zuul:")
# TODO(jeblair): If we had all threads marked as daemon,
+24 -19
View File
@@ -301,8 +301,7 @@ class Scheduler(threading.Thread):
'repl': self.start_repl,
'norepl': self.stop_repl,
}
self._pause = False
self._exit = False
self._hibernate = False
self._stopped = False
self._zuul_app = None
self.executor = None
@@ -698,8 +697,7 @@ class Scheduler(threading.Thread):
def exit(self):
self.log.debug("Prepare to exit")
self._pause = True
self._exit = True
self._hibernate = True
self.wake_event.set()
self.log.debug("Waiting for exit")
@@ -729,7 +727,8 @@ class Scheduler(threading.Thread):
"current mode is %o" % (key_dir, mode))
return key_dir
def _save_queue(self):
def _save_queue(self) -> None:
# TODO JK: Remove when queues in ZK
pickle_file = self._get_queue_pickle_file()
events = []
while not self.trigger_event_queue.empty():
@@ -739,7 +738,8 @@ class Scheduler(threading.Thread):
self.log.debug("Saving queue")
pickle.dump(events, open(pickle_file, 'wb'))
def _load_queue(self):
def _load_queue(self) -> None:
# TODO JK: Remove when queues in ZK
pickle_file = self._get_queue_pickle_file()
if os.path.exists(pickle_file):
self.log.debug("Loading queue")
@@ -750,13 +750,19 @@ class Scheduler(threading.Thread):
else:
self.log.debug("No queue file found")
def _delete_queue(self):
def _delete_queue(self) -> None:
# TODO JK: Remove when queues in ZK
pickle_file = self._get_queue_pickle_file()
if os.path.exists(pickle_file):
self.log.debug("Deleting saved queue")
os.unlink(pickle_file)
def resume(self):
def wakeUp(self) -> None:
"""
Wakes up scheduler by loading pickled queue.
TODO JK: Remove when queues in ZK
"""
try:
self._load_queue()
except Exception:
@@ -768,8 +774,9 @@ class Scheduler(threading.Thread):
self.log.debug("Resuming queue processing")
self.wake_event.set()
def _doPauseEvent(self):
if self._exit:
def _doHibernate(self) -> None:
# TODO JK: Remove when queues in ZK
if self._hibernate:
self.log.debug("Exiting")
self._save_queue()
os._exit(0)
@@ -1215,13 +1222,13 @@ class Scheduler(threading.Thread):
not self._stopped):
self.process_result_queue()
if not self._pause:
if not self._hibernate:
while (not self.trigger_event_queue.empty() and
not self._stopped):
self.process_event_queue()
if self._pause and self._areAllBuildsComplete():
self._doPauseEvent()
if self._hibernate and self._areAllBuildsComplete():
self._doHibernate()
for tenant in self.abide.tenants.values():
for pipeline in tenant.layout.pipelines.values():
@@ -1625,12 +1632,10 @@ class Scheduler(threading.Thread):
data['zuul_version'] = self.zuul_version
websocket_url = get_default(self.config, 'web', 'websocket_url', None)
if self._pause:
ret = 'Queue only mode: preparing to '
if self._exit:
ret += 'exit'
ret += ', queue length: %s' % self.trigger_event_queue.qsize()
data['message'] = ret
if self._hibernate:
data['message'] = 'Queue only mode: preparing to hibernate,' \
' queue length: %s'\
% self.trigger_event_queue.qsize()
data['trigger_event_queue'] = {}
data['trigger_event_queue']['length'] = \