Add repl server for debug purposes

Sometimes it can be necessary to debug complex problems in a running
instance. A repl server can be useful in this case. However this can
be highly dangerous in case the port is not correctly firewalled
of. Thus add an undocumented command to enable and disable it.

To enable:
zuul-scheduler repl

To disable
zuul-scheduler norepl

Change-Id: I860db5d088c6b07a66b4390cdfd66637341e7a7a
Co-Authored-By: Tobias Henkel <tobias.henkel@bmw.de>

zuul/executor/server.py

@@ -40,6 +40,7 @@ from zuul.lib import filecomments
 
 import gear
 
+import zuul.lib.repl
 import zuul.merger.merger
 import zuul.ansible.logconfig
 from zuul.executor.sensors.cpu import CPUSensor
@@ -51,7 +52,7 @@ from zuul.lib import commandsocket
 
 BUFFER_LINES_FOR_SYNTAX = 200
 COMMANDS = ['stop', 'pause', 'unpause', 'graceful', 'verbose',
-            'unverbose', 'keep', 'nokeep']
+            'unverbose', 'keep', 'nokeep', 'repl', 'norepl']
 DEFAULT_FINGER_PORT = 7900
 DEFAULT_STREAM_PORT = 19885
 BLACKLISTED_ANSIBLE_CONNECTION_TYPES = [
@@ -2268,8 +2269,11 @@ class ExecutorServer(object):
             unverbose=self.verboseOff,
             keep=self.keep,
             nokeep=self.nokeep,
+            repl=self.start_repl,
+            norepl=self.stop_repl,
         )
         self.log_console_port = log_console_port
+        self.repl = None
 
         statsd_extra_keys = {'hostname': self.hostname}
         self.statsd = get_statsd(config, statsd_extra_keys)
@@ -2495,6 +2499,7 @@ class ExecutorServer(object):
             self.statsd.gauge(base_key + '.running_builds', 0)
 
         self.command_socket.stop()
+        self.stop_repl()
         self.log.debug("Stopped")
 
     def join(self):
@@ -2526,6 +2531,19 @@ class ExecutorServer(object):
     def nokeep(self):
         self.keep_jobdir = False
 
+    def start_repl(self):
+        if self.repl:
+            return
+        self.repl = zuul.lib.repl.REPLServer(self)
+        self.repl.start()
+
+    def stop_repl(self):
+        if not self.repl:
+            # not running
+            return
+        self.repl.stop()
+        self.repl = None
+
     def runCommand(self):
         while self._command_running:
             try:

zuul/lib/repl.py

@@ -0,0 +1,81 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+# Based on ASL2 code from:
+# https://gist.github.com/tim-patterson/4471877
+
+import sys
+import io
+import threading
+import socketserver
+import code
+
+
+class ThreadLocalProxy(object):
+    def __init__(self, default):
+        self.files = {}
+        self.default = default
+
+    def __getattr__(self, name):
+        obj = self.files.get(threading.currentThread(), self.default)
+        return getattr(obj, name)
+
+    def register(self, obj):
+        self.files[threading.currentThread()] = obj
+
+    def unregister(self):
+        self.files.pop(threading.currentThread())
+
+
+class REPLHandler(socketserver.StreamRequestHandler):
+    def handle(self):
+        sys.stdout.register(io.TextIOWrapper(self.wfile, 'utf8'))
+        sys.stderr.register(io.TextIOWrapper(self.wfile, 'utf8'))
+        sys.stdin.register(io.TextIOWrapper(self.rfile, 'utf8'))
+        try:
+            console = code.InteractiveConsole(locals())
+            console.interact('Console:')
+        except Exception:
+            pass
+        finally:
+            sys.stdout.unregister()
+            sys.stderr.unregister()
+            sys.stdin.unregister()
+
+
+class REPLThreadedTCPServer(socketserver.ThreadingMixIn,
+                            socketserver.TCPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+    def __init__(self, scheduler, *args, **kw):
+        self.scheduler = scheduler
+        super(REPLThreadedTCPServer, self).__init__(*args, **kw)
+        sys.stdout = ThreadLocalProxy(sys.stdout)
+        sys.stderr = ThreadLocalProxy(sys.stderr)
+        sys.stdin = ThreadLocalProxy(sys.stdin)
+
+
+class REPLServer(object):
+    def __init__(self, scheduler):
+        self.server = REPLThreadedTCPServer(
+            scheduler, ('localhost', 3000), REPLHandler)
+
+    def start(self):
+        self.thread = threading.Thread(target=self.server.serve_forever)
+        self.thread.daemon = True
+        self.thread.start()
+
+    def stop(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(10)

zuul/scheduler.py

@@ -38,9 +38,10 @@ from zuul.lib.gear_utils import getGearmanFunctions
 from zuul.lib.logutil import get_annotated_logger
 from zuul.lib.statsd import get_statsd
 import zuul.lib.queue
+import zuul.lib.repl
 from zuul.model import Build
 
-COMMANDS = ['full-reconfigure', 'stop']
+COMMANDS = ['full-reconfigure', 'stop', 'repl', 'norepl']
 
 
 class ManagementEvent(object):
@@ -277,6 +278,8 @@ class Scheduler(threading.Thread):
         self.command_map = {
             'stop': self.stop,
             'full-reconfigure': self.fullReconfigureCommandHandler,
+            'repl': self.start_repl,
+            'norepl': self.stop_repl,
         }
         self._pause = False
         self._exit = False
@@ -287,6 +290,7 @@ class Scheduler(threading.Thread):
         self.connections = None
         self.statsd = get_statsd(config)
         self.rpc = rpclistener.RPCListener(config, self)
+        self.repl = None
         self.stats_thread = threading.Thread(target=self.runStats)
         self.stats_thread.daemon = True
         self.stats_stop = threading.Event()
@@ -353,6 +357,7 @@ class Scheduler(threading.Thread):
         self.stats_thread.join()
         self.rpc.stop()
         self.rpc.join()
+        self.stop_repl()
         self._command_running = False
         self.command_socket.stop()
         self.command_thread.join()
@@ -519,6 +524,18 @@ class Scheduler(threading.Thread):
     def fullReconfigureCommandHandler(self):
         self._zuul_app.fullReconfigure()
 
+    def start_repl(self):
+        if self.repl:
+            return
+        self.repl = zuul.lib.repl.REPLServer(self)
+        self.repl.start()
+
+    def stop_repl(self):
+        if not self.repl:
+            return
+        self.repl.stop()
+        self.repl = None
+
     def reconfigure(self, config):
         self.log.debug("Submitting reconfiguration event")
         event = ReconfigureEvent(config)

zuul/web/__init__.py

@@ -31,6 +31,7 @@ import threading
 
 import re2
 
+import zuul.lib.repl
 import zuul.model
 import zuul.rpcclient
 import zuul.zk
@@ -39,7 +40,7 @@ from zuul.lib import commandsocket
 STATIC_DIR = os.path.join(os.path.dirname(__file__), 'static')
 cherrypy.tools.websocket = WebSocketTool()
 
-COMMANDS = ['stop']
+COMMANDS = ['stop', 'repl', 'norepl']
 
 
 class SaveParamsTool(cherrypy.Tool):
@@ -750,8 +751,13 @@ class ZuulWeb(object):
         self.stream_manager = StreamManager()
 
         self.command_socket = commandsocket.CommandSocket(command_socket)
+
+        self.repl = None
+
         self.command_map = {
             'stop': self.stop,
+            'repl': self.start_repl,
+            'norepl': self.stop_repl,
         }
 
         route_map = cherrypy.dispatch.RoutesDispatcher()
@@ -866,6 +872,7 @@ class ZuulWeb(object):
         self.wsplugin.unsubscribe()
         self.stream_manager.stop()
         self.zk.disconnect()
+        self.stop_repl()
         self._command_running = False
         self.command_socket.stop()
         self.command_thread.join()
@@ -879,6 +886,18 @@ class ZuulWeb(object):
             except Exception:
                 self.log.exception("Exception while processing command")
 
+    def start_repl(self):
+        if self.repl:
+            return
+        self.repl = zuul.lib.repl.REPLServer(self)
+        self.repl.start()
+
+    def stop_repl(self):
+        if not self.repl:
+            return
+        self.repl.stop()
+        self.repl = None
+
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.DEBUG)