Merge "Add repl server for debug purposes"

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 = [
@@ -2272,8 +2273,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)
@@ -2499,6 +2503,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):
@@ -2530,6 +2535,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)