WIP: Make admin node a web-app

The goal is to get rid off the neoctl command-line tool, and to manage the
cluster via a web browser, or tools like 'wget'. Then, it will be possible to
provide an web user interface to connect to the underlying DB of any storage
node, usually a SQL client.

The design of admin app is finished:
- it's threaded like clients
- it's a WSGI app

I also hacked a HTTP API as quickly as possible to make all tests pass.

TODO:
- define a better HTTP API
- there's no UI at all yet
- remove all unused packets from the protocol (those that were only used
  between neoctl and admin node)

There's currently no UI implemented.

There are a few dead files, not deleted yet, in case that they contain a few
pieces of useful code:
 neo/neoctl/app.py
 neo/neoctl/handler.py
 neo/scripts/neoctl.py

neo/admin/app.py

@@ -14,123 +14,202 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import json
+import thread
+import weakref
+from bottle import Bottle, HTTPError, request, response
+from copy import deepcopy
+from logging import ERROR, INFO
+from wsgiref.simple_server import WSGIRequestHandler
+from . import handler
 from neo.lib import logging
-from neo.lib.app import BaseApplication
-from neo.lib.connection import ListeningConnection
 from neo.lib.exception import PrimaryFailure
-from .handler import AdminEventHandler, MasterEventHandler, \
-    MasterRequestEventHandler
 from neo.lib.bootstrap import BootstrapManager
 from neo.lib.pt import PartitionTable
-from neo.lib.protocol import ClusterStates, Errors, \
-    NodeTypes, NodeStates, Packets
-from neo.lib.debug import register as registerLiveDebugger
+from neo.lib.protocol import ClusterStates, NodeTypes, NodeStates, Packets
+from neo.lib.threaded_app import ThreadedApplication
+from neo.lib.util import p64, tidFromTime
 
-class Application(BaseApplication):
-    """The storage node application."""
+def raiseNotReady():
+    raise HTTPError(503, 'Not connected to a primary master')
 
-    def __init__(self, config):
-        super(Application, self).__init__(
-            config.getSSL(), config.getDynamicMasterList())
-        for address in config.getMasters():
-            self.nm.createMaster(address=address)
 
-        self.name = config.getCluster()
-        self.server = config.getBind()
+class ObjectBottle(object):
 
-        logging.debug('IP address is %s, port is %d', *self.server)
+    def __init__(self, weakref=False, *args, **kw):
+        self._app = Bottle(*args, **kw)
+        self._weakref = weakref
 
-        # The partition table is initialized after getting the number of
-        # partitions.
-        self.pt = None
-        self.uuid = config.getUUID()
-        self.request_handler = MasterRequestEventHandler(self)
-        self.master_event_handler = MasterEventHandler(self)
-        self.cluster_state = None
-        self.reset()
-        registerLiveDebugger(on_log=self.log)
+    def __getattr__(self, name):
+        return getattr(self._app, name)
 
-    def close(self):
-        self.listening_conn = None
-        super(Application, self).close()
+    def __get__(self, obj, cls):
+        if obj is None: return self
+        app = obj.bottle = deepcopy(self._app)
+        if self._weakref:
+            obj = weakref.ref(obj)
+            app.install(lambda f: lambda *a, **k: f(obj(), *a, **k))
+        else:
+            app.install(lambda f: lambda *a, **k: f(obj, *a, **k))
+        return app
 
-    def reset(self):
-        self.bootstrapped = False
-        self.master_conn = None
-        self.master_node = None
 
-    def log(self):
-        self.em.log()
-        self.nm.log()
-        if self.pt is not None:
-            self.pt.log()
+class RequestHandler(WSGIRequestHandler):
 
-    def run(self):
-        try:
-            self._run()
-        except Exception:
-            logging.exception('Pre-mortem data:')
-            self.log()
-            logging.flush()
-            raise
+    def _log(self, level, format, *args):
+        logging.log(level, "%s %s", self.client_address[0], format % args)
 
-    def _run(self):
-        """Make sure that the status is sane and start a loop."""
-        if len(self.name) == 0:
-            raise RuntimeError, 'cluster name must be non-empty'
+    def log_error(self, *args):
+        self._log(ERROR, *args)
+
+    def log_message(self, *args):
+        self._log(INFO, *args)
+
+
+class Application(ThreadedApplication):
+    """The storage node application."""
 
-        # Make a listening port.
-        handler = AdminEventHandler(self)
-        self.listening_conn = ListeningConnection(self, handler, self.server)
+    bottle = ObjectBottle(weakref=True)
+    cluster_state = None
 
+    def __init__(self, config):
+        super(Application, self).__init__(
+            config.getMasters(), config.getCluster(),
+            ssl=config.getSSL(),
+            dynamic_master_list=config.getDynamicMasterList())
+        self.master_event_handler = handler.MasterEventHandler(self)
+        self.notifications_handler = handler.MasterNotificationsHandler(self)
+        self.primary_handler = handler.PrimaryAnswersHandler(self)
+
+    def _run(self):
+        """Make sure that the status is sane and start a loop."""
+        try:
+            poll = self.em.poll
             while self.cluster_state != ClusterStates.STOPPING:
                 self.connectToPrimary()
                 try:
                     while True:
-                    self.em.poll(1)
+                        poll(1)
                 except PrimaryFailure:
+                    self.nm.log()
                     logging.error('primary master is down')
-        self.listening_conn.close()
+                finally:
+                    self.master_conn = None
             while not self.em.isIdle():
-            self.em.poll(1)
+                poll(1)
+        finally:
+            self.interrupt_main()
+
+    interrupt_main = staticmethod(thread.interrupt_main)
 
     def connectToPrimary(self):
         """Find a primary master node, and connect to it.
 
         If a primary master node is not elected or ready, repeat
         the attempt of a connection periodically.
-
-        Note that I do not accept any connection from non-master nodes
-        at this stage.
         """
+        self.master_node = None
+        self.uuid = None
         self.cluster_state = None
         # search, find, connect and identify to the primary master
-        bootstrap = BootstrapManager(self, self.name, NodeTypes.ADMIN,
-                self.uuid, self.server)
-        data = bootstrap.getPrimaryConnection()
-        (node, conn, uuid, num_partitions, num_replicas) = data
-        self.master_node = node
-        self.master_conn = conn
-        self.uuid = uuid
-
-        if self.pt is None:
+        bootstrap = BootstrapManager(self, self.name, NodeTypes.ADMIN)
+        (self.master_node, self.master_conn, self.uuid,
+         num_partitions, num_replicas) = bootstrap.getPrimaryConnection()
         self.pt = PartitionTable(num_partitions, num_replicas)
-        elif self.pt.getPartitions() != num_partitions:
-            # XXX: shouldn't we recover instead of raising ?
-            raise RuntimeError('the number of partitions is inconsistent')
-        elif self.pt.getReplicas() != num_replicas:
-            # XXX: shouldn't we recover instead of raising ?
-            raise RuntimeError('the number of replicas is inconsistent')
-
-        # passive handler
+
         self.master_conn.setHandler(self.master_event_handler)
         self.master_conn.ask(Packets.AskClusterState())
         self.master_conn.ask(Packets.AskNodeInformation())
         self.master_conn.ask(Packets.AskPartitionTable())
-
-    def sendPartitionTable(self, conn, min_offset, max_offset, uuid):
-        # we have a pt
-        self.pt.log()
+        self.master_conn.setHandler(self.notifications_handler)
+        self.master_conn.convertToMT(self.dispatcher)
+
+    def _askPrimary(self, packet, **kw):
+        """ Send a request to the primary master and process its answer """
+        return self._ask(self._getMasterConnection(), packet,
+                         handler=self.primary_handler, **kw)
+
+    def _getMasterConnection(self):
+        conn = self.master_conn
+        if conn is None or conn.isClosed():
+            raiseNotReady()
+        return conn
+
+    def serve(self, **kw):
+        self.start()
+        self.bottle.run(server='wsgiref', handler_class=RequestHandler,
+                        quiet=1, **kw)
+
+    def asTID(self, value):
+        if '.' in value:
+            return tidFromTime(float(value))
+        return p64(int(value, 0))
+
+    @bottle.route('/getClusterState')
+    def getClusterState(self):
+        if self.cluster_state is not None:
+            return str(self.cluster_state)
+
+    def _setClusterState(self, state):
+        self._askPrimary(Packets.SetClusterState(state))
+
+    @bottle.route('/setClusterState')
+    def setClusterState(self):
+        self._setClusterState(getattr(ClusterStates, request.query.state))
+
+    @bottle.route('/startCluster')
+    def startCluster(self):
+        self._setClusterState(ClusterStates.VERIFYING)
+
+    @bottle.route('/enableStorageList')
+    def enableStorageList(self):
+        node_list = request.query.node_list
+        self._askPrimary(Packets.AddPendingNodes(map(int,
+            request.query.node_list.split(',')) if node_list else ()))
+
+    @bottle.route('/tweakPartitionTable')
+    def tweakPartitionTable(self):
+        node_list = request.query.node_list
+        self._askPrimary(Packets.TweakPartitionTable(map(int,
+            request.query.node_list.split(',')) if node_list else ()))
+
+    @bottle.route('/getNodeList')
+    def getNodeList(self):
+        node_type = request.query.node_type
+        if node_type:
+            node_type = getattr(NodeTypes, node_type)
+            node_filter = lambda node: node.getType() is node_type
+        else:
+            node_filter = None
+        node_list = []
+        self._getMasterConnection()
+        for node in self.nm.getList(node_filter):
+            node_type, address, uuid, state = node = node.asTuple()
+            node_list.append((str(node_type), address, uuid, str(state)))
+        response.content_type = 'application/json'
+        return json.dumps(node_list)
+
+    @bottle.route('/getPrimary')
+    def getPrimary(self):
+        return str(getattr(self.master_node, 'getUUID', raiseNotReady)())
+
+    def _setNodeState(self, node, state):
+        self._askPrimary(Packets.SetNodeState(node, state))
+
+    @bottle.route('/killNode')
+    def killNode(self):
+        self._setNodeState(int(request.query.node), NodeStates.UNKNOWN)
+
+    @bottle.route('/dropNode')
+    def killNode(self):
+        self._setNodeState(int(request.query.node), NodeStates.DOWN)
+
+    @bottle.route('/getPartitionRowList')
+    def getPartitionRowList(self):
+        min_offset = int(request.query.min_offset)
+        max_offset = int(request.query.max_offset)
+        uuid = request.query.node
+        uuid = int(uuid) if uuid else None
         row_list = []
         if max_offset == 0:
             max_offset = self.pt.getPartitions()
@@ -140,11 +219,30 @@ class Application(BaseApplication):
                 try:
                     for cell in self.pt.getCellList(offset):
                         if uuid is None or cell.getUUID() == uuid:
-                            row.append((cell.getUUID(), cell.getState()))
+                            row.append((cell.getUUID(), str(cell.getState())))
                 except TypeError:
                     pass
                 row_list.append((offset, row))
         except IndexError:
-            conn.notify(Errors.ProtocolError('invalid partition table offset'))
+            raise HTTPError(400, 'invalid partition table offset')
+        response.content_type = 'application/json'
+        return json.dumps((self.pt.getID(), row_list))
+
+    @bottle.route('/checkReplicas')
+    def checkReplicas(self):
+        partition_dict = {}
+        for partition in request.query.pt.split(','):
+            partition, source = partition.split(':')
+            source = int(source) if source else None
+            if partition:
+                partition_dict[int(partition)] = source
+            elif partition_dict:
+                raise HTTPError(400)
             else:
-            conn.answer(Packets.AnswerPartitionList(self.pt.getID(), row_list))
+                self._getMasterConnection() # just for correct error handling
+                partition_dict = dict.fromkeys(xrange(self.pt.getPartitions()),
+                                               source)
+        max_tid = request.query.max_tid
+        self._askPrimary(Packets.CheckReplicas(partition_dict,
+            self.asTID(request.query.min_tid),
+            self.asTID(max_tid) if max_tid else None))

neo/admin/handler.py

@@ -14,73 +14,18 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from neo.lib import logging, protocol
-from neo.lib.handler import EventHandler
-from neo.lib.protocol import uuid_str, Packets
+from bottle import HTTPError
+from neo.lib import logging
+from neo.lib.handler import AnswerBaseHandler, EventHandler, MTEventHandler
 from neo.lib.exception import PrimaryFailure
 
-def check_primary_master(func):
-    def wrapper(self, *args, **kw):
-        if self.app.bootstrapped:
-            return func(self, *args, **kw)
-        raise protocol.NotReadyError('Not connected to a primary master.')
-    return wrapper
-
-def forward_ask(klass):
-    return check_primary_master(lambda self, conn, *args, **kw:
-        self.app.master_conn.ask(klass(*args, **kw),
-                                 conn=conn, msg_id=conn.getPeerId()))
-
-class AdminEventHandler(EventHandler):
-    """This class deals with events for administrating cluster."""
-
-    @check_primary_master
-    def askPartitionList(self, conn, min_offset, max_offset, uuid):
-        logging.info("ask partition list from %s to %s for %s",
-                     min_offset, max_offset, uuid_str(uuid))
-        self.app.sendPartitionTable(conn, min_offset, max_offset, uuid)
-
-    @check_primary_master
-    def askNodeList(self, conn, node_type):
-        if node_type is None:
-            node_type = 'all'
-            node_filter = None
-        else:
-            node_filter = lambda n: n.getType() is node_type
-        logging.info("ask list of %s nodes", node_type)
-        node_list = self.app.nm.getList(node_filter)
-        node_information_list = [node.asTuple() for node in node_list ]
-        p = Packets.AnswerNodeList(node_information_list)
-        conn.answer(p)
-
-    @check_primary_master
-    def askClusterState(self, conn):
-        conn.answer(Packets.AnswerClusterState(self.app.cluster_state))
-
-    @check_primary_master
-    def askPrimary(self, conn):
-        master_node = self.app.master_node
-        conn.answer(Packets.AnswerPrimary(master_node.getUUID()))
-
-    askLastIDs = forward_ask(Packets.AskLastIDs)
-    askLastTransaction = forward_ask(Packets.AskLastTransaction)
-    addPendingNodes = forward_ask(Packets.AddPendingNodes)
-    tweakPartitionTable = forward_ask(Packets.TweakPartitionTable)
-    setClusterState = forward_ask(Packets.SetClusterState)
-    setNodeState = forward_ask(Packets.SetNodeState)
-    checkReplicas = forward_ask(Packets.CheckReplicas)
-
-
 class MasterEventHandler(EventHandler):
-    """ This class is just used to dispacth message to right handler"""
 
     def _connectionLost(self, conn):
-        app = self.app
-        if app.listening_conn: # if running
-            assert app.master_conn in (conn, None)
         conn.cancelRequests("connection to master lost")
-            app.reset()
-            app.uuid = None
+        self.app.nm.getByUUID(conn.getUUID()).setUnknown()
+        if self.app.master_conn is not None:
+            assert self.app.master_conn is conn
             raise PrimaryFailure
 
     def connectionFailed(self, conn):
@@ -89,18 +34,6 @@ class MasterEventHandler(EventHandler):
     def connectionClosed(self, conn):
         self._connectionLost(conn)
 
-    def dispatch(self, conn, packet, kw={}):
-        if 'conn' in kw:
-            # expected answer
-            if packet.isResponse():
-                packet.setId(kw['msg_id'])
-                kw['conn'].answer(packet)
-            else:
-                self.app.request_handler.dispatch(conn, packet, kw)
-        else:
-            # unexpected answers and notifications
-            super(MasterEventHandler, self).dispatch(conn, packet, kw)
-
     def answerClusterState(self, conn, state):
         self.app.cluster_state = state
 
@@ -109,23 +42,22 @@ class MasterEventHandler(EventHandler):
         # implemented for factorize code (as done for bootstrap)
         logging.debug("answerNodeInformation")
 
-    def notifyPartitionChanges(self, conn, ptid, cell_list):
-        self.app.pt.update(ptid, cell_list, self.app.nm)
+    def notifyNodeInformation(self, conn, node_list):
+        self.app.nm.update(node_list)
 
     def answerPartitionTable(self, conn, ptid, row_list):
         self.app.pt.load(ptid, row_list, self.app.nm)
-        self.app.bootstrapped = True
 
-    def sendPartitionTable(self, conn, ptid, row_list):
-        if self.app.bootstrapped:
-            self.app.pt.load(ptid, row_list, self.app.nm)
+class MasterNotificationsHandler(MasterEventHandler, MTEventHandler):
 
-    def notifyClusterInformation(self, conn, cluster_state):
-        self.app.cluster_state = cluster_state
+    notifyClusterInformation = MasterEventHandler.answerClusterState.im_func
+    sendPartitionTable = MasterEventHandler.answerPartitionTable.im_func
 
-    def notifyNodeInformation(self, conn, node_list):
-        self.app.nm.update(node_list)
+    def notifyPartitionChanges(self, conn, ptid, cell_list):
+        self.app.pt.update(ptid, cell_list, self.app.nm)
 
-class MasterRequestEventHandler(EventHandler):
+class PrimaryAnswersHandler(AnswerBaseHandler):
     """ This class handle all answer from primary master node"""
-    # XXX: to be deleted ?
+
+    def protocolError(self, conn, message):
+        raise HTTPError(400, message)

neo/lib/connection.py

@@ -592,6 +592,11 @@ class ClientConnection(Connection):
         handler.connectionStarted(self)
         self._connect()
 
+    def convertToMT(self, dispatcher):
+        assert self.__class__ is ClientConnection, self
+        self.__class__ = MTClientConnection
+        self._initMT(dispatcher)
+
     def _connect(self):
         try:
             connected = self.connector.makeClientConnection()
@@ -688,11 +693,14 @@ class MTClientConnection(ClientConnection):
         return wrapper
 
     def __init__(self, *args, **kwargs):
-        self.lock = lock = RLock()
-        self.dispatcher = kwargs.pop('dispatcher')
-        with lock:
+        self._initMT(kwargs.pop('dispatcher'))
+        with self.lock:
             super(MTClientConnection, self).__init__(*args, **kwargs)
 
+    def _initMT(self, dispatcher):
+        self.lock = RLock()
+        self.dispatcher = dispatcher
+
     def ask(self, packet, timeout=CRITICAL_TIMEOUT, on_timeout=None,
             queue=None, **kw):
         with self.lock:

neo/lib/threaded_app.py

@@ -135,10 +135,16 @@ class ThreadedApplication(BaseApplication):
             handler.dispatch(conn, packet, kw)
 
     def _ask(self, conn, packet, handler=None, **kw):
-        self.setHandlerData(None)
-        queue = self._thread_container.queue
-        msg_id = conn.ask(packet, queue=queue, **kw)
-        get = queue.get
+        # The following line is more than optimization. If an admin node sends
+        # a packet that causes the master to disconnect (e.g. stop a cluster),
+        # we want at least to return the answer for this request, even if the
+        # polling thread already exited and cleared self.__dict__: returning
+        # the result of getHandlerData() would raise an AttributeError.
+        # This is tested by testShutdown (neo.tests.threaded.test.Test).
+        thread_container = self._thread_container
+        thread_container.answer = None
+        msg_id = conn.ask(packet, queue=thread_container.queue, **kw)
+        get = thread_container.queue.get
         _handlePacket = self._handlePacket
         while True:
             qconn, qpacket, kw = get(True)
@@ -152,7 +158,6 @@ class ThreadedApplication(BaseApplication):
                         raise ValueError, 'ForgottenPacket for an ' \
                             'explicitely expected packet.'
                     _handlePacket(qconn, qpacket, kw, handler)
-                    break
+                    return thread_container.answer # see above comment
             if not is_forgotten and qpacket is not None:
                 _handlePacket(qconn, qpacket, kw)
-        return self.getHandlerData()

neo/neoctl/neoctl.py

@@ -14,157 +14,100 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from neo.lib.app import BaseApplication
-from neo.lib.connection import ClientConnection
-from neo.lib.protocol import ClusterStates, NodeStates, ErrorCodes, Packets
-from .handler import CommandEventHandler
+import json, socket
+from urllib import URLopener, urlencode
+from neo.lib.protocol import CellStates, ClusterStates, NodeTypes, NodeStates, \
+    ZERO_TID
+from neo.lib.util import u64
 
 class NotReadyException(Exception):
     pass
 
-class NeoCTL(BaseApplication):
-
-    connection = None
-    connected = False
-
-    def __init__(self, address, **kw):
-        super(NeoCTL, self).__init__(**kw)
-        self.server = self.nm.createAdmin(address=address)
-        self.handler = CommandEventHandler(self)
-        self.response_queue = []
-
-    def __getConnection(self):
-        if not self.connected:
-            self.connection = ClientConnection(self, self.handler, self.server)
-            # Never delay reconnection to master. This speeds up unit tests
-            # and it should not change anything for normal use.
-            self.connection.setReconnectionNoDelay()
-            while not self.connected:
-                self.em.poll(1)
-                if self.connection is None:
-                    raise NotReadyException('not connected')
-        return self.connection
-
-    def __ask(self, packet):
-        # TODO: make thread-safe
-        connection = self.__getConnection()
-        connection.ask(packet)
-        response_queue = self.response_queue
-        assert len(response_queue) == 0
-        while self.connected:
-            self.em.poll(1)
-            if response_queue:
-                break
-        else:
-            raise NotReadyException, 'Connection closed'
-        response = response_queue.pop()
-        if response[0] == Packets.Error and \
-           response[1] == ErrorCodes.NOT_READY:
-            raise NotReadyException(response[2])
-        return response
+class NeoCTL(object):
+
+    def __init__(self, address):
+        host, port = address
+        if ":" in host:
+            host = "[%s]" % host
+        self.base_url = "http://%s:%s/" % (host, port)
+        self._open = URLopener().open
+
+    def _ask(self, path, **kw):
+        if kw:
+            path += "?" + urlencode(sorted(x for x in kw.iteritems()
+                                             if '' is not x[1] is not None))
+        try:
+            return self._open(self.base_url + path).read()
+        except IOError, e:
+            e0 = e[0]
+            if e0 == 'socket error' or e0 == 'http error' and e[1] == 503:
+                raise NotReadyException
+            raise
 
     def enableStorageList(self, uuid_list):
         """
           Put all given storage nodes in "running" state.
         """
-        packet = Packets.AddPendingNodes(uuid_list)
-        response = self.__ask(packet)
-        if response[0] != Packets.Error or response[1] != ErrorCodes.ACK:
-            raise RuntimeError(response)
-        return response[2]
+        self._ask('enableStorageList', node_list=','.join(map(str, uuid_list)))
 
     def tweakPartitionTable(self, uuid_list=()):
-        response = self.__ask(Packets.TweakPartitionTable(uuid_list))
-        if response[0] != Packets.Error or response[1] != ErrorCodes.ACK:
-            raise RuntimeError(response)
-        return response[2]
+        self._ask('tweakPartitionTable', node_list=','.join(map(str, uuid_list)))
 
     def setClusterState(self, state):
         """
           Set cluster state.
         """
-        packet = Packets.SetClusterState(state)
-        response = self.__ask(packet)
-        if response[0] != Packets.Error or response[1] != ErrorCodes.ACK:
-            raise RuntimeError(response)
-        return response[2]
-
-    def _setNodeState(self, node, state):
-        """
-          Kill node, or remove it permanently
-        """
-        response = self.__ask(Packets.SetNodeState(node, state))
-        if response[0] != Packets.Error or response[1] != ErrorCodes.ACK:
-            raise RuntimeError(response)
-        return response[2]
+        self._ask('setClusterState', state=state)
 
     def getClusterState(self):
         """
           Get cluster state.
         """
-        packet = Packets.AskClusterState()
-        response = self.__ask(packet)
-        if response[0] != Packets.AnswerClusterState:
-            raise RuntimeError(response)
-        return response[1]
-
-    def getLastIds(self):
-        response = self.__ask(Packets.AskLastIDs())
-        if response[0] != Packets.AnswerLastIDs:
-            raise RuntimeError(response)
-        return response[1:]
-
-    def getLastTransaction(self):
-        response = self.__ask(Packets.AskLastTransaction())
-        if response[0] != Packets.AnswerLastTransaction:
-            raise RuntimeError(response)
-        return response[1]
+        state = self._ask('getClusterState')
+        if state:
+            return getattr(ClusterStates, state)
 
     def getNodeList(self, node_type=None):
         """
           Get a list of nodes, filtering with given type.
         """
-        packet = Packets.AskNodeList(node_type)
-        response = self.__ask(packet)
-        if response[0] != Packets.AnswerNodeList:
-            raise RuntimeError(response)
-        return response[1] # node_list
+        node_list = json.loads(self._ask('getNodeList', node_type=node_type))
+        return ((getattr(NodeTypes, node_type), address and tuple(address),
+                 uuid, getattr(NodeStates, state))
+            for node_type, address, uuid, state in node_list)
 
     def getPartitionRowList(self, min_offset=0, max_offset=0, node=None):
         """
           Get a list of partition rows, bounded by min & max and involving
           given node.
         """
-        packet = Packets.AskPartitionList(min_offset, max_offset, node)
-        response = self.__ask(packet)
-        if response[0] != Packets.AnswerPartitionList:
-            raise RuntimeError(response)
-        return response[1:3] # ptid, row_list
+        ptid, row_list = json.loads(self._ask('getPartitionRowList',
+            min_offset=min_offset, max_offset=max_offset, node=node))
+        return ptid, [(offset, [(node, getattr(CellStates, state))
+                                for node, state in row])
+                      for offset, row in row_list]
 
     def startCluster(self):
         """
           Set cluster into "verifying" state.
         """
-        return self.setClusterState(ClusterStates.VERIFYING)
+        self._ask('startCluster')
 
     def killNode(self, node):
-        return self._setNodeState(node, NodeStates.UNKNOWN)
+        self._ask('killNode', node=node)
 
     def dropNode(self, node):
-        return self._setNodeState(node, NodeStates.DOWN)
+        self._ask('dropNode', node=node)
 
     def getPrimary(self):
         """
           Return the primary master UUID.
         """
-        packet = Packets.AskPrimary()
-        response = self.__ask(packet)
-        if response[0] != Packets.AnswerPrimary:
-            raise RuntimeError(response)
-        return response[1]
-
-    def checkReplicas(self, *args):
-        response = self.__ask(Packets.CheckReplicas(*args))
-        if response[0] != Packets.Error or response[1] != ErrorCodes.ACK:
-            raise RuntimeError(response)
-        return response[2]
+        return int(self._ask('getPrimary'))
+
+    def checkReplicas(self, partition_dict, min_tid=ZERO_TID, max_tid=None):
+        kw = {'pt': ','.join('%s:%s' % (k, '' if v is None else v)
+                             for k, v in partition_dict.iteritems())}
+        if max_tid is not None:
+            kw['max_tid'] = u64(max_tid)
+        self._ask('checkReplicas', min_tid=u64(min_tid), **kw)

neo/scripts/neoadmin.py

@@ -29,6 +29,7 @@ defaults = dict(
     masters = '127.0.0.1:10000',
 )
 
+
 def main(args=None):
     # build configuration dict from command line options
     (options, args) = parser.parse_args(args=args)
@@ -39,6 +40,5 @@ def main(args=None):
 
     # and then, load and run the application
     from neo.admin.app import Application
-    app = Application(config)
-    app.run()
-
+    host, port = config.getBind()
+    Application(config).serve(host=host, port=port)

neo/tests/functional/__init__.py

@@ -363,7 +363,7 @@ class NEOCluster(object):
                         pending_count += 1
                     if pending_count == target[0]:
                         neoctl.startCluster()
-            except (NotReadyException, RuntimeError):
+            except (NotReadyException, IOError):
                 pass
         if not pdb.wait(test, MAX_START_TIME):
             raise AssertionError('Timeout when starting cluster')

neo/tests/functional/testMaster.py

@@ -47,7 +47,7 @@ class MasterTests(NEOFunctionalTest):
                 break
         neoctl.killNode(uuid)
         self.neo.expectDead(master)
-        self.assertRaises(RuntimeError, neoctl.killNode, primary_uuid)
+        self.assertRaises(IOError, neoctl.killNode, primary_uuid)
 
     def testStoppingPrimaryWithTwoSecondaries(self):
         # Wait for masters to stabilize

neo/tests/functional/testStorage.py

@@ -173,7 +173,7 @@ class StorageTests(NEOFunctionalTest):
         self.neo.expectOudatedCells(2)
         self.neo.expectClusterRunning()
 
-        self.assertRaises(RuntimeError, self.neo.neoctl.killNode,
+        self.assertRaises(IOError, self.neo.neoctl.killNode,
             started[1].getUUID())
         started[1].stop()
         # Cluster not operational anymore. Only cells of second storage that
@@ -324,7 +324,7 @@ class StorageTests(NEOFunctionalTest):
         self.neo.expectStorageNotKnown(started[0])
         self.neo.expectAssignedCells(started[0], 0)
         self.neo.expectAssignedCells(started[1], 10)
-        self.assertRaises(RuntimeError, self.neo.neoctl.dropNode,
+        self.assertRaises(IOError, self.neo.neoctl.dropNode,
                           started[1].getUUID())
         self.neo.expectClusterRunning()
 

neo/tests/threaded/__init__.py

@@ -21,11 +21,13 @@ import traceback
 from collections import deque
 from ConfigParser import SafeConfigParser
 from contextlib import contextmanager
+from cStringIO import StringIO
 from itertools import count
 from functools import wraps
+from urllib import splitquery
 from zlib import decompress
 from mock import Mock
-import transaction, ZODB
+import bottle, transaction, ZODB
 import neo.admin.app, neo.master.app, neo.storage.app
 import neo.client.app, neo.neoctl.app
 from neo.client import Storage
@@ -162,13 +164,12 @@ class Serialized(object):
                 next_lock.release()
                 cls._sched_lock.acquire()
 
-    def __init__(self, app, busy=True):
+    def __init__(self, app):
         self._epoll = app.em.epoll
         app.em.epoll = self
         # XXX: It may have been initialized before the SimpleQueue is patched.
         thread_container = getattr(app, '_thread_container', None)
         thread_container is None or thread_container.__init__()
-        if busy:
         self._busy.add(self) # block tic until app waits for polling
 
     def __getattr__(self, attr):
@@ -207,20 +208,6 @@ class Serialized(object):
             cls._epoll.unregister(fd)
             self._release_next()
 
-class TestSerialized(Serialized):
-
-    def __init__(*args):
-        Serialized.__init__(busy=False, *args)
-
-    def poll(self, timeout):
-        if timeout:
-            while 1:
-                r = self._epoll.poll(0)
-                if r:
-                    return r
-                Serialized.tic(step=1)
-        return self._epoll.poll(timeout)
-
 
 class Node(object):
 
@@ -236,6 +223,24 @@ class Node(object):
     def filterConnection(self, *peers):
         return ConnectionFilter(self.getConnectionList(*peers))
 
+    def _run(self):
+        try:
+            super(Node, self)._run()
+        finally:
+            self._afterRun()
+            self.em.epoll.exit()
+
+    def _afterRun(self):
+        logging.debug('stopping %r', self)
+        try:
+            self.listening_conn.close()
+        except AttributeError:
+            pass
+
+    def start(self):
+        isinstance(self.em.epoll, Serialized) or Serialized(self)
+        super(Node, self).start()
+
 class ServerNode(Node):
 
     _server_class_dict = {}
@@ -299,33 +304,44 @@ class ServerNode(Node):
         self.close()
         self.__init__(**kw)
 
-    def start(self):
-        Serialized(self)
-        threading.Thread.start(self)
-
-    def run(self):
-        try:
-            super(ServerNode, self).run()
-        finally:
-            self._afterRun()
-            logging.debug('stopping %r', self)
-            self.em.epoll.exit()
-
-    def _afterRun(self):
-        try:
-            self.listening_conn.close()
-        except AttributeError:
-            pass
-
     def getListeningAddress(self):
         try:
             return self.listening_conn.getAddress()
         except AttributeError:
             raise ConnectorException
 
-class AdminApplication(ServerNode, neo.admin.app.Application):
+class ThreadedNode(Node):
+
+    def __init__(self, *args, **kw):
+        self._on_thread_exit = kw.pop("on_thread_exit", None)
+        super(ThreadedNode, self).__init__(*args, **kw)
+        self.poll_thread.node_name = getattr(self, 'node_name', self.name)
+
+    def _afterRun(self):
+        x = self._on_thread_exit
+        if x is not None:
+            self._on_thread_exit = None
+            x()
+
+class AdminApplication(ThreadedNode, ServerNode, neo.admin.app.Application):
+
+    def interrupt_main(self):
         pass
 
+    def start(self):
+        super(AdminApplication, self).start()
+        return
+        host, port = BIND
+        l = threading.Lock()
+        l.acquire()
+        self.start = l.release
+        try:
+            self.run = lambda: self.serve(host=host, port=port)
+            threading.Thread.start(self)
+            l.acquire()
+        finally:
+            del self.start, self.run
+
 class MasterApplication(ServerNode, neo.master.app.Application):
     pass
 
@@ -366,21 +382,7 @@ class StorageApplication(ServerNode, neo.storage.app.Application):
         (r,), = self.dm.query("SELECT COUNT(*) FROM " + table)
         return r
 
-class ClientApplication(Node, neo.client.app.Application):
-
-    def __init__(self, master_nodes, name, **kw):
-        super(ClientApplication, self).__init__(master_nodes, name, **kw)
-        self.poll_thread.node_name = name
-
-    def _run(self):
-        try:
-            super(ClientApplication, self)._run()
-        finally:
-            self.em.epoll.exit()
-
-    def start(self):
-        isinstance(self.em.epoll, Serialized) or Serialized(self)
-        super(ClientApplication, self).start()
+class ClientApplication(ThreadedNode, neo.client.app.Application):
 
     def getConnectionList(self, *peers):
         for peer in peers:
@@ -392,10 +394,25 @@ class ClientApplication(Node, neo.client.app.Application):
             yield conn
 
 class NeoCTL(neo.neoctl.app.NeoCTL):
+    # Bypass HTTP layer
 
-    def __init__(self, *args, **kw):
-        super(NeoCTL, self).__init__(*args, **kw)
-        TestSerialized(self)
+    base_url = '/'
+
+    def __init__(self, cluster):
+        self._cluster = cluster
+
+    def _open(self, path):
+        environ = {'REQUEST_METHOD': 'GET'}
+        environ['PATH_INFO'], environ['QUERY_STRING'] = splitquery(path)
+        bottle.request.bind(environ)
+        route, args = self._cluster.admin.bottle.match(environ)
+        try:
+            result = route.call(*args)
+            if isinstance(result, Exception):
+                raise result
+        except bottle.HTTPError, e:
+            raise IOError('http error', e.status_code)
+        return StringIO(result or '')
 
 
 class LoggerThreadName(str):
@@ -502,6 +519,7 @@ class ConnectionFilter(object):
 
 class NEOCluster(object):
 
+    admin = None
     SSL = None
 
     def __init__(orig, self): # temporary definition for SimpleQueue patch
@@ -564,9 +582,9 @@ class NEOCluster(object):
                        for _ in xrange(master_count)]
         self.master_nodes = ' '.join('%s:%s' % x for x in master_list)
         weak_self = weakref.proxy(self)
-        kw = dict(cluster=weak_self, getReplicas=replicas, getAdapter=adapter,
-                  getPartitions=partitions, getReset=clear_databases,
-                  getSSL=self.SSL)
+        self._server_kw = kw = dict(cluster=weak_self, getReplicas=replicas,
+            getAdapter=adapter, getPartitions=partitions,
+            getReset=clear_databases, getSSL=self.SSL)
         if upstream is not None:
             self.upstream = weakref.proxy(upstream)
             kw.update(getUpstreamCluster=upstream.name,
@@ -602,14 +620,13 @@ class NEOCluster(object):
             kw["getAdapter"] = "Importer"
         self.storage_list = [StorageApplication(getDatabase=db % x, **kw)
                              for x in db_list]
-        self.admin_list = [AdminApplication(**kw)]
-        self.neoctl = NeoCTL(self.admin.getVirtualAddress(), ssl=self.SSL)
+        self.neoctl = NeoCTL(weak_self)
 
     def __repr__(self):
         return "<%s(%s) at 0x%x>" % (self.__class__.__name__,
                                      self.name, id(self))
 
-    # A few shortcuts that work when there's only 1 master/storage/admin
+    # A few shortcuts that work when there's only 1 master/storage
     @property
     def master(self):
         master, = self.master_list
@@ -618,10 +635,6 @@ class NEOCluster(object):
     def storage(self):
         storage, = self.storage_list
         return storage
-    @property
-    def admin(self):
-        admin, = self.admin_list
-        return admin
     ###
 
     @property
@@ -630,20 +643,19 @@ class NEOCluster(object):
         return master
 
     def reset(self, clear_database=False):
-        for node_type in 'master', 'storage', 'admin':
-            kw = {}
-            if node_type == 'storage':
-                kw['clear_database'] = clear_database
-            for node in getattr(self, node_type + '_list'):
-                node.resetNode(**kw)
-        self.neoctl.close()
-        self.neoctl = NeoCTL(self.admin.getVirtualAddress(), ssl=self.SSL)
+        for node in self.master_list:
+            node.resetNode()
+        for node in self.storage_list:
+            node.resetNode(clear_database=clear_database)
 
     def start(self, storage_list=None, fast_startup=False):
         self._patch()
-        for node_type in 'master', 'admin':
-            for node in getattr(self, node_type + '_list'):
+        for node in self.master_list:
             node.start()
+        self.admin = AdminApplication(
+            on_thread_exit=lambda: delattr(self, "admin"),
+            **self._server_kw)
+        self.admin.start()
         Serialized.tic()
         if fast_startup:
             self.startCluster()
@@ -659,20 +671,13 @@ class NEOCluster(object):
         assert state in (ClusterStates.RUNNING, ClusterStates.BACKINGUP), state
         self.enableStorageList(storage_list)
 
-    def newClient(self):
-        return ClientApplication(name=self.name, master_nodes=self.master_nodes,
-                                 compress=self.compress, ssl=self.SSL)
-
     @cached_property
     def client(self):
-        client = self.newClient()
-        # Make sure client won't be reused after it was closed.
-        def close():
-            client = self.client
-            del self.client, client.close
-            client.close()
-        client.close = close
-        return client
+        return self.newClient(on_thread_exit=lambda: delattr(self, "client"))
+
+    def newClient(self, **kw):
+        return ClientApplication(name=self.name, master_nodes=self.master_nodes,
+                                 compress=self.compress, ssl=self.SSL, **kw)
 
     @cached_property
     def db(self):
@@ -681,7 +686,7 @@ class NEOCluster(object):
     def startCluster(self):
         try:
             self.neoctl.startCluster()
-        except RuntimeError:
+        except IOError:
             Serialized.tic()
             if self.neoctl.getClusterState() not in (
                       ClusterStates.BACKINGUP,
@@ -696,8 +701,13 @@ class NEOCluster(object):
         for node in storage_list:
             assert self.getNodeState(node) == NodeStates.RUNNING
 
-    def join(self, thread_list, timeout=5):
+    def join(self, thread_list, threaded_node_list=(), timeout=5):
         timeout += time.time()
+        for node in threaded_node_list:
+            try:
+                thread_list.append(node.poll_thread)
+            except AttributeError:
+                pass # node is None or thread is already stopped
         while thread_list:
             assert time.time() < timeout
             Serialized.tic()
@@ -705,16 +715,14 @@ class NEOCluster(object):
 
     def stop(self):
         logging.debug("stopping %s", self)
+        admin = self.admin
+        admin is None or admin.close()
         client = self.__dict__.get("client")
         client is None or self.__dict__.pop("db", client).close()
-        node_list = self.admin_list + self.storage_list + self.master_list
+        node_list = self.storage_list + self.master_list
         for node in node_list:
             node.em.wakeup(True)
-        try:
-            node_list.append(client.poll_thread)
-        except AttributeError: # client is None or thread is already stopped
-            pass
-        self.join(node_list)
+        self.join(node_list, (admin, client))
         logging.debug("stopped %s", self)
         self._unpatch()
 
@@ -761,8 +769,7 @@ class NEOCluster(object):
 
     def __del__(self, __print_exc=traceback.print_exc):
         try:
-            self.neoctl.close()
-            for node_type in 'admin', 'storage', 'master':
+            for node_type in 'storage', 'master':
                 for node in getattr(self, node_type + '_list'):
                     node.close()
         except:

neo/tests/threaded/test.py

@@ -534,9 +534,8 @@ class Test(NEOThreadedTest):
             # tell admin to shutdown the cluster
             cluster.neoctl.setClusterState(ClusterStates.STOPPING)
             # all nodes except clients should exit
-            cluster.join(cluster.master_list
-                       + cluster.storage_list
-                       + cluster.admin_list)
+            cluster.join(cluster.master_list + cluster.storage_list,
+                         (cluster.admin,))
         finally:
             cluster.stop()
         cluster.reset() # reopen DB to check partition tables

setup.py

@@ -26,7 +26,7 @@ if not os.path.exists('mock.py'):
 zodb_require = ['ZODB3>=3.10', 'ZODB3<3.11dev']
 
 extras_require = {
-    'admin': [],
+    'admin': ['bottle'],
     'client': zodb_require,
     'ctl': [],
     'master': [],