Merge "Implement graceful termination for the executor"

zuul/cmd/executor.py

@@ -45,8 +45,6 @@ class Executor(zuul.cmd.ZuulDaemonApp):
 
     def exit_handler(self, signum, frame):
         self.executor.stop()
-        self.executor.join()
-        sys.exit(0)
 
     def start_log_streamer(self):
         pipe_read, pipe_write = os.pipe()
@@ -107,14 +105,8 @@ class Executor(zuul.cmd.ZuulDaemonApp):
 
         if self.args.nodaemon:
             signal.signal(signal.SIGTERM, self.exit_handler)
-            while True:
-                try:
-                    signal.pause()
-                except KeyboardInterrupt:
-                    print("Ctrl + C: asking executor to exit nicely...\n")
-                    self.exit_handler(signal.SIGINT, None)
-        else:
-            self.executor.join()
+
+        self.executor.join()
 
 
 def main():

zuul/executor/server.py

@@ -2796,6 +2796,8 @@ class ExecutorServer(BaseMergeServer):
         for _ in self.update_threads:
             self.update_queue.put(None)
 
+        self.command_socket.stop()
+
         # All job results should have been sent by now, shutdown the
         # gearman workers.
         if self.process_merge_jobs:
@@ -2811,7 +2813,6 @@ class ExecutorServer(BaseMergeServer):
             self.statsd.gauge(base_key + '.pct_used_ram', 0)
             self.statsd.gauge(base_key + '.running_builds', 0)
 
-        self.command_socket.stop()
         self.stop_repl()
         self.log.debug("Stopped")
 
@@ -2837,8 +2838,17 @@ class ExecutorServer(BaseMergeServer):
             super().unpause()
 
     def graceful(self):
-        # TODOv3: implement
-        pass
+        # This pauses the executor end shuts it down when there is no running
+        # build left anymore
+        self.log.info('Stopping graceful')
+        self.pause()
+        while self.job_workers:
+            self.log.debug('Waiting for %s jobs to end', len(self.job_workers))
+            time.sleep(30)
+        try:
+            self.stop()
+        except Exception:
+            self.log.exception('Error while stopping')
 
     def verboseOn(self):
         self.verbose = True