Loading ccnode/node.py +121 −112 Original line number Diff line number Diff line import time import signal import logging from threading import Thread, Lock import logging.config import pyev from sjrpc.core import RpcConnection from sjrpc.utils import ConnectionProxy, RpcHandler from ccnode import __version__ from ccnode.config import NodeConfigParser from ccnode.tags import Tag, get_tags Loading Loading @@ -42,124 +45,130 @@ class DefaultHandler(RpcHandler): self.tags.pop(tag_name, None) class Node(Thread): """Main class for ccnode.""" def __init__(self, server_host, server_port, user_name, user_passwd): """ :param string server_host: hostname for cc-server :param int server_port: port for cc-server :param string user_name: account name for authentication to cc-server :param string user_passwd: password for cc-server authentication """ Thread.__init__(self) # run thread as daemon self.daemon = True # settings used as read only self.server_host = server_host self.server_port = int(server_port) self.user_name = user_name self.user_passwd = user_passwd #: ``sjRpc`` proxy self.proxy = None #: ``sjRpc`` connection self.connection = None #: role returned by cc-server (set to None unless the authentication #: has succeed) self.role = None self._connection_lock = Lock() def init_rpc(self): """Init a new connection to ``cc-server``, create a ``sjRpc`` proxy. """ self.connection = RpcConnection.from_addr_ssl( addr=self.server_host, port=self.server_port, handler=DefaultHandler(), class AuthHandler(object): """Handles rpc authentication to the remote cc-server.""" def __init__(self, loop): self.loop = loop self.watcher = loop.loop.timer(.5, 5, self.cb) self.auth_id = None def cb(self, watcher, revents): logger.debug('Callback auth') # check is fallback mode on sjrpc is set otherwise our call would block # the loop if not self.loop.rpc_con._event_fallback.is_set(): logger.debug('Will try authentication again latter') return if self.auth_id is not None: logger.error('Authentication is taking longer than expected') return # try to authenticate self.auth_id = self.loop.rpc_con.rpc.async_call_cb( self.cb_auth, 'authentify', self.loop.config.server_user, self.loop.config.server_passwd, ) self.proxy = ConnectionProxy(self.connection) def authentify(self): """Try to authenticate to the server. If successfull, import and then set :class:`Handler` corresponding to the role returned by the ``cc-server``. def cb_auth(self, call_id, response=None, error=None): assert call_id == self.auth_id :raise: exception raised by :func:`proxy.authentify` """ try: role = self.proxy.authentify(self.user_name, self.user_passwd) except Exception: logger.exception('Unknow exception while authentifying') raise if error is not None: # we got an error logger.error('Error while authenticating with cc-server: %s("%s")', error['exception'], error.get('message', '')) # try to recconnect in 5 seconds return # set handler according to which role was returned by the cc-server if role == u'host': if response == u'host': logger.debug('Role host affected') from ccnode.host import Handler as HostHandler self.connection.rpc.set_handler(HostHandler()) self.role = u'host' elif role == u'hv': self.loop.rpc_con.rpc.set_handler(HostHandler()) self.loop.role = u'host' elif response == u'hv': logger.debug('Role hypervisor affected') from ccnode.hypervisor import Handler as HypervisorHandler self.connection.rpc.set_handler(HypervisorHandler( proxy=self.proxy, hypervisor_name=self.user_name)) self.role = u'hv' self.loop.rpc_con.rpc.set_handler(HypervisorHandler( proxy=self.loop.proxy, hypervisor_name=self.loop.config.server_user)) self.loop.role = u'hv' else: logger.debug('Wrong role returned: %s', role) role = None time.sleep(2) self.role = role def rpc(self): """Runs ``sjRpc`` main loop. Catches exceptions. :func:`shutdown` the connection before returning.""" try: self.connection.run() except Exception: logger.exception('Unknown exception:') finally: self.shutdown() def run(self): """Node main loop.""" while True: # init rpc connection while True: try: self.init_rpc() except Exception as e: logger.exception('Error in init.') else: break time.sleep(2) # launch main rpc thread rpc_thread = Thread(target=self.rpc) rpc_thread.daemon = True rpc_thread.start() # launch auth thread, make sure rpc is still running while rpc_thread.is_alive() and self.role is None: auth_thread = Thread(target=self.authentify) auth_thread.daemon = True auth_thread.start() auth_thread.join() # FIXME for debug time.sleep(4) # wait for rpc thread to terminates (it means error) rpc_thread.join() logger.error('Reconnecting to server.') # reset settings self.role = None logger.info('Failed authentication, role returned: %s', response) self.loop.role = None self.auth_id = None return if self.loop.role is not None: self.watcher.stop() logger.info('Successfully authenticated with role %s', response) self.auth_id = None def shutdown(self): """Shutdown the ``sjRpc`` connection and reset object state.""" with self._connection_lock: if self.connection is not None: self.connection.shutdown() self.connection = None def start(self): self.watcher.start() class MainLoop(object): def __init__(self, config_path): self.loop = pyev.default_loop() self.config_path = config_path # set signal watchers self.signals = { signal.SIGINT: self.stop, signal.SIGTERM: self.stop, signal.SIGUSR1: self.reload, } # turn into real watchers self.signals = dict(( signal, self.loop.signal(signal, cb), ) for signal, cb in self.signals.iteritems()) # load config variables self.config = NodeConfigParser(self.config_path) # configure logging logging.config.fileConfig(self.config_path) # rpc connection self.rpc_con = None self.auth = AuthHandler(self) # role self.role = None def rpc_connect(self): # TODO async and error handling self.rpc_con = RpcConnection.from_addr_ssl( addr=self.config.server_host, port=self.config.server_port, handler=DefaultHandler(), loop=self.loop, ) self.proxy = ConnectionProxy(self.rpc_con) def start(self): for signal in self.signals.itervalues(): signal.start() logger.debug('About to connect') self.rpc_connect() self.auth.start() logger.debug('About to start ev_loop') self.loop.start() def stop(self, watcher=None, revents=None): logger.info('Exiting node...') if self.rpc_con is not None: self.rpc_con.shutdown() self.loop.stop() def reload(self, watcher=None, revents=None): logger.info(u'Reloading logging configuration...') logging.config.fileConfig(self.config_path) Loading
ccnode/node.py +121 −112 Original line number Diff line number Diff line import time import signal import logging from threading import Thread, Lock import logging.config import pyev from sjrpc.core import RpcConnection from sjrpc.utils import ConnectionProxy, RpcHandler from ccnode import __version__ from ccnode.config import NodeConfigParser from ccnode.tags import Tag, get_tags Loading Loading @@ -42,124 +45,130 @@ class DefaultHandler(RpcHandler): self.tags.pop(tag_name, None) class Node(Thread): """Main class for ccnode.""" def __init__(self, server_host, server_port, user_name, user_passwd): """ :param string server_host: hostname for cc-server :param int server_port: port for cc-server :param string user_name: account name for authentication to cc-server :param string user_passwd: password for cc-server authentication """ Thread.__init__(self) # run thread as daemon self.daemon = True # settings used as read only self.server_host = server_host self.server_port = int(server_port) self.user_name = user_name self.user_passwd = user_passwd #: ``sjRpc`` proxy self.proxy = None #: ``sjRpc`` connection self.connection = None #: role returned by cc-server (set to None unless the authentication #: has succeed) self.role = None self._connection_lock = Lock() def init_rpc(self): """Init a new connection to ``cc-server``, create a ``sjRpc`` proxy. """ self.connection = RpcConnection.from_addr_ssl( addr=self.server_host, port=self.server_port, handler=DefaultHandler(), class AuthHandler(object): """Handles rpc authentication to the remote cc-server.""" def __init__(self, loop): self.loop = loop self.watcher = loop.loop.timer(.5, 5, self.cb) self.auth_id = None def cb(self, watcher, revents): logger.debug('Callback auth') # check is fallback mode on sjrpc is set otherwise our call would block # the loop if not self.loop.rpc_con._event_fallback.is_set(): logger.debug('Will try authentication again latter') return if self.auth_id is not None: logger.error('Authentication is taking longer than expected') return # try to authenticate self.auth_id = self.loop.rpc_con.rpc.async_call_cb( self.cb_auth, 'authentify', self.loop.config.server_user, self.loop.config.server_passwd, ) self.proxy = ConnectionProxy(self.connection) def authentify(self): """Try to authenticate to the server. If successfull, import and then set :class:`Handler` corresponding to the role returned by the ``cc-server``. def cb_auth(self, call_id, response=None, error=None): assert call_id == self.auth_id :raise: exception raised by :func:`proxy.authentify` """ try: role = self.proxy.authentify(self.user_name, self.user_passwd) except Exception: logger.exception('Unknow exception while authentifying') raise if error is not None: # we got an error logger.error('Error while authenticating with cc-server: %s("%s")', error['exception'], error.get('message', '')) # try to recconnect in 5 seconds return # set handler according to which role was returned by the cc-server if role == u'host': if response == u'host': logger.debug('Role host affected') from ccnode.host import Handler as HostHandler self.connection.rpc.set_handler(HostHandler()) self.role = u'host' elif role == u'hv': self.loop.rpc_con.rpc.set_handler(HostHandler()) self.loop.role = u'host' elif response == u'hv': logger.debug('Role hypervisor affected') from ccnode.hypervisor import Handler as HypervisorHandler self.connection.rpc.set_handler(HypervisorHandler( proxy=self.proxy, hypervisor_name=self.user_name)) self.role = u'hv' self.loop.rpc_con.rpc.set_handler(HypervisorHandler( proxy=self.loop.proxy, hypervisor_name=self.loop.config.server_user)) self.loop.role = u'hv' else: logger.debug('Wrong role returned: %s', role) role = None time.sleep(2) self.role = role def rpc(self): """Runs ``sjRpc`` main loop. Catches exceptions. :func:`shutdown` the connection before returning.""" try: self.connection.run() except Exception: logger.exception('Unknown exception:') finally: self.shutdown() def run(self): """Node main loop.""" while True: # init rpc connection while True: try: self.init_rpc() except Exception as e: logger.exception('Error in init.') else: break time.sleep(2) # launch main rpc thread rpc_thread = Thread(target=self.rpc) rpc_thread.daemon = True rpc_thread.start() # launch auth thread, make sure rpc is still running while rpc_thread.is_alive() and self.role is None: auth_thread = Thread(target=self.authentify) auth_thread.daemon = True auth_thread.start() auth_thread.join() # FIXME for debug time.sleep(4) # wait for rpc thread to terminates (it means error) rpc_thread.join() logger.error('Reconnecting to server.') # reset settings self.role = None logger.info('Failed authentication, role returned: %s', response) self.loop.role = None self.auth_id = None return if self.loop.role is not None: self.watcher.stop() logger.info('Successfully authenticated with role %s', response) self.auth_id = None def shutdown(self): """Shutdown the ``sjRpc`` connection and reset object state.""" with self._connection_lock: if self.connection is not None: self.connection.shutdown() self.connection = None def start(self): self.watcher.start() class MainLoop(object): def __init__(self, config_path): self.loop = pyev.default_loop() self.config_path = config_path # set signal watchers self.signals = { signal.SIGINT: self.stop, signal.SIGTERM: self.stop, signal.SIGUSR1: self.reload, } # turn into real watchers self.signals = dict(( signal, self.loop.signal(signal, cb), ) for signal, cb in self.signals.iteritems()) # load config variables self.config = NodeConfigParser(self.config_path) # configure logging logging.config.fileConfig(self.config_path) # rpc connection self.rpc_con = None self.auth = AuthHandler(self) # role self.role = None def rpc_connect(self): # TODO async and error handling self.rpc_con = RpcConnection.from_addr_ssl( addr=self.config.server_host, port=self.config.server_port, handler=DefaultHandler(), loop=self.loop, ) self.proxy = ConnectionProxy(self.rpc_con) def start(self): for signal in self.signals.itervalues(): signal.start() logger.debug('About to connect') self.rpc_connect() self.auth.start() logger.debug('About to start ev_loop') self.loop.start() def stop(self, watcher=None, revents=None): logger.info('Exiting node...') if self.rpc_con is not None: self.rpc_con.shutdown() self.loop.stop() def reload(self, watcher=None, revents=None): logger.info(u'Reloading logging configuration...') logging.config.fileConfig(self.config_path)
