ccserver/exceptions.py → cloudcontrol/server/exceptions.py

-#!/usr/bin/env python
-#coding=utf8
-
-
 class AlreadyRegistered(Exception):
     pass
 

ccserver/handlers.py → cloudcontrol/server/handlers.py

-#!/usr/bin/env python
-#coding=utf8
-
-from __future__ import absolute_import
-
 import inspect
-import logging
 
 from sjrpc.utils import RpcHandler
-from sjrpc.core import RpcError
-
-from ccserver.orderedset import OrderedSet
-from ccserver.conf import CCConf
-from ccserver.exceptions import (AlreadyRegistered, AuthenticationError,
-                                 RightError, ReservedTagError, BadObjectError,
-                                 BadRoleError, NotConnectedAccountError,
-                                 CloneError)
 
-from ccserver import __version__
+from cloudcontrol.server import __version__
 
 
 def listed(func):
@@ -25,9 +11,8 @@ def listed(func):
 
 
 class Reporter(object):
-    '''
-    Simple class used to report error, warning and success of command execution.
-    '''
+    """ Simple class used to report error, warning and success of command execution.
+    """
 
     def __init__(self):
         self._reports = []
@@ -69,16 +54,19 @@ class CCHandler(RpcHandler):
             # Filter the private members access:
             raise KeyError('Remote name %s is private.' % repr(name))
         else:
-            logging.debug('Called %s.%s', self.__class__.__name__, name)
+            self.logger.debug('Called %s.%s', self.__class__.__name__, name)
             return super(CCHandler, self).__getitem__(name)
 
+    @property
+    def logger(self):
+        return self.client.logger
+
     @listed
     def functions(self):
-        '''
-        Show the list of functions available to the peer.
+        """ Show the list of functions available to the peer.
 
         :return: list of dict with keys name and description.
-        '''
+        """
 
         cmd_list = []
 
@@ -96,9 +84,8 @@ class CCHandler(RpcHandler):
 
     @listed
     def version(self):
-        '''
-        Return the current server version.
+        """ Return the current server version.
 
         :return: the version
-        '''
+        """
         return __version__

cloudcontrol/server/jobs/__init__.py

+from threading import Lock
+
+from cloudcontrol.server.jobs.hotmigration import HotMigrationJob
+from cloudcontrol.server.jobs.coldmigration import ColdMigrationJob
+from cloudcontrol.server.jobs.clone import CloneJob
+from cloudcontrol.server.jobs.killclient import KillClientJob
+from cloudcontrol.server.jobs.killoldcli import KillOldCliJob
+
+from cloudcontrol.server.exceptions import (BadJobTypeError, UnknownJobError)
+
+
+class JobsManager(object):
+
+    """ Manage the current job list.
+
+    :param server: The :class:`CCServer` instance.
+    """
+
+    JOBS_TYPES = {
+        'kill': KillClientJob,
+        'kill_oldcli': KillOldCliJob,
+        'cold_migrate': ColdMigrationJob,
+        'hot_migrate': HotMigrationJob,
+        'clone': CloneJob,
+    }
+
+    def __init__(self, logger, server):
+        self.logger = logger
+        # The main job dict, the keys are id of jobs.
+        self._jobs = {}
+
+        # The server:
+        self.server = server
+
+        # The id of the next job and it's lock:
+        self._current_id = 1
+        self._current_id_lock = Lock()
+
+    def create(self, jtype, **kwargs):
+        """ Create a new job.
+
+        :param jtype: the type of the new job
+        :param \*\*kwargs: arguments to pass to the job
+        :raise BadJobTypeError: when invalid jtype is passed
+        """
+
+        jobid = self.get_id()
+        jobtype = JobsManager.JOBS_TYPES.get(jtype)
+
+        if jobtype is None:
+            raise BadJobTypeError('Invalid job type %r' % jtype)
+
+        job = jobtype(self.logger.getChild(str(jobid)), self, id=jobid, **kwargs)
+        self._jobs[jobid] = job
+        job.daemon = True
+        job.start()
+        return job
+
+    def get_id(self):
+        """ Get the current id and increment the counter.
+        """
+
+        with self._current_id_lock:
+            jobid = self._current_id
+            self._current_id += 1
+
+        return jobid
+
+    def cancel(self, jobid):
+        """ Cancel the provided job.
+        """
+
+        job = self._jobs.get(jobid)
+
+        if job is None:
+            raise UnknownJobError('Invalid job id: %r' % jobid)
+        elif job.get('_hidden', False):
+            raise UnknownJobError('Invalid job id: %r (hidden)' % jobid)
+        else:
+            job.cancel()
+
+    def purge(self):
+        """ Purge all done jobs.
+        """
+
+        for job in self._jobs.values():
+            if job['done']:
+                del self._jobs[job['id']]
+
+    def iterjobs(self, show_done=True, show_running=True):
+        """ Iter over jobs.
+
+        :param done: If set, iterate over done or not done jobs.
+        """
+
+        for job in self._jobs.itervalues():
+            if (show_done and job['done'] or show_running and not job['done']
+                and not job.get('_hidden')):
+                yield job

cloudcontrol/server/jobs/base.py

+import time
+from threading import Thread
+from datetime import datetime
+
+from cloudcontrol.server.exceptions import JobError
+
+
+class JobCancelError(Exception):
+    """ Exception used by jobs to stop it when a cancel signal is sent.
+    """
+    pass
+
+
+class BaseJob(dict, Thread, object):
+
+    """ A base class to define a job.
+
+    The standards job items are:
+
+     * id: id of the job
+     * status: message explaining the current job status
+     * done: True if the job is done
+     * cancelled: True if job has been cancelled by user
+     * created: job date of creation
+     * ended: job date of end (or None if done = False)
+     * duration: duration in seconds of the job (processed on export)
+     * author: author login of the job
+
+    :param manager: the :class:`JobsManager` instance.
+    """
+
+    def __init__(self, logger, manager, *args, **kwargs):
+        # Initialize the inherited classes:
+        dict.__init__(self, *args, **kwargs)
+        Thread.__init__(self)
+
+        self.logger = logger
+        # The manager of this job:
+        self.manager = manager
+
+        # Define default job properties:
+        self['status'] = 'pending'
+        self['done'] = False
+        self['cancelled'] = False
+        self['created'] = datetime.now()
+        self['ended'] = None
+        self['duration'] = 0
+
+        #~ assert self.get('author') is not None, 'author is not defined'
+
+        # List of actions to do by the rollback method:
+        self._wayback = []
+
+        # Set the thread name:
+        self.name = 'job-%s' % self['id']
+
+    def __hash__(self):
+        return self['id'].__hash__()
+
+    def __setitem__(self, key, value):
+        if key == 'id':
+            raise KeyError('Key %r in read-only.' % key)
+        else:
+            super(BaseJob, self).__setitem__(key, value)
+
+    def __delitem__(self, key):
+        if key == 'id':
+            raise KeyError('Key %r in read-only.' % key)
+        else:
+            super(BaseJob, self).__delitem__(key)
+
+    def report(self, status, done=None):
+        """ Report the status of the job.
+
+        :param status: the status to set to the job
+        :param done: is the job done, None to keep current value
+        """
+
+        self['status'] = status
+        if done is not None:
+            self['ended'] = datetime.now()
+            self['done'] = done
+
+    def run(self):
+        """ Run the job itself.
+        """
+
+        try:
+            self.job()
+        except JobCancelError as err:
+            self._rollback('%s' % err)
+        except Exception as err:
+            self.logger.error('Error while executing job: %s, %r', err, err)
+            self._rollback('%s' % err)
+        else:
+            self.report('success', done=True)
+
+    def job(self):
+        """ Method to override to define the job's behavior.
+        """
+
+        pass
+
+    def checkpoint(self, func=None):
+        """ Check if job is not cancelled, else raise the CancellJobError. Also
+            add the provided function (optionnal) to the wayback list.
+
+        :param func: callable to add to the wayback list.
+        """
+
+        if self['cancelled']:
+            raise JobCancelError('Job has been cancelled by user')
+
+        if func is not None:
+            self._wayback.append(func)
+
+    def _rollback(self, error):
+        """ Rollback the job.
+        """
+
+        self.report('rollbacking')
+        try:
+            for func in reversed(self._wayback):
+                func()
+        except Exception as err:
+            self.report('rollback failed: %s' % err, done=True)
+        else:
+            self.report('cancelled: %s' % error, done=True)
+
+    def cancel(self):
+        """ Cancel the job.
+
+        .. note::
+            You can override this method to trigger an action when user cancel
+            the job.
+        """
+
+        if self['done']:
+            raise JobError('Job is done')
+        if self['cancelled']:
+            raise JobError('Job is already cancelled')
+
+        self['cancelled'] = True
+        self.report('cancelling')
+
+    def export(self, props=None):
+        """ Export the job in a simple dict format.
+        """
+
+        exported = {}
+        for key, val in self.iteritems():
+            if key.startswith('_') or (props is not None and key not in props):
+                continue
+            if isinstance(val, datetime):
+                val = int(time.mktime(val.timetuple()))
+
+            if key == 'duration':
+                if self['done']:
+                    dt = self['ended'] - self['created']
+                    val = dt.seconds + dt.days * 86400
+                else:
+                    now = datetime.now()
+                    dt = now - self['created']
+                    val = dt.seconds + dt.days * 86400
+
+            exported[key] = str(val)
+        return exported

cloudcontrol/server/jobs/clone.py

+import re
+
+from sjrpc.core import AsyncWatcher
+
+from cloudcontrol.server.jobs.base import BaseJob, JobCancelError
+from cloudcontrol.server.utils import AcquiresAllOrNone
+
+
+class CloneJob(BaseJob):
+
+    """ A clone job.
+
+    Mandatory items:
+     * vm_name: name of the vm to migrate
+     * new_vm_name: the name of the cloned vm
+     * hv_source: name of the hv which execute the VM
+     * hv_dest: the destination hypervisor
+     * author: login of the author cli
+    """
+
+    def job(self):
+        vm_id = '%s.%s' % (self['hv_source'], self['vm_name'])
+
+        self['title'] = 'Clone %s --> %s' % (vm_id, self['hv_dest'])
+        self.logger.info('Job-%s: Started clone for %s', self['id'], vm_id)
+
+        # Cancel the job if the user has not the right to clone the vm or to
+        # select an hypervisor as destination:
+        right_check = self.manager.server.check
+
+        tql = 'id=%s' % vm_id
+        if not right_check(self['author'], 'clone', tql):
+            raise JobCancelError('author have no rights to migrate this VM')
+
+        tql = 'id=%s' % self['hv_dest']
+        if not right_check(self['author'], 'clone_dest', tql):
+            raise JobCancelError('author have no right to migrate to this hv')
+
+        # Update the VM object:
+        vm = self.manager.server.db.get_by_id(vm_id)
+        if vm is None:
+            raise JobCancelError('Source VM not found')
+
+        # Get the source and destination hv clients:
+        try:
+            source = self.manager.server.get_client(self['hv_source'])
+        except KeyError:
+            raise JobCancelError('source hypervisor is not connected')
+
+        try:
+            dest = self.manager.server.get_client(self['hv_dest'])
+        except KeyError:
+            raise JobCancelError('destination hypervisor is not connected')
+
+        self.checkpoint()
+
+        self.report('waiting lock for source and dest hypervisors')
+        self.logger.info('Job-%s: Trying to acquire locks', self['id'])
+
+        with AcquiresAllOrNone(source.hvlock, dest.hvlock):
+            self.logger.info('Job-%s: Locks acquired', self['id'])
+            self.checkpoint()
+
+            # Create storages on destination:
+            old_new_disk_mapping = {} # Mapping between old and new disk names
+            names = {}
+            self.report('create volumes')
+            for disk in vm.get('disk', '').split():
+                # Getting informations about the disk:
+                pool = vm.get('disk%s_pool' % disk)
+                name = vm.get('disk%s_vol' % disk)
+                size = vm.get('disk%s_size' % disk)
+                assert pool is not None, 'pool tag doesn\'t exists'
+                assert name is not None, 'name tag doesn\'t exists'
+                assert size is not None, 'size tag doesn\'t exists'
+
+                # Change the name of the disk:
+                old_name = name
+                if name.startswith(self['vm_name']):
+                    suffix = name[len(self['vm_name']):]
+                    name = self['new_vm_name'] + suffix
+                else:
+                    name = '%s_%s' % (self['new_vm_name'], name)
+
+                names[disk] = name
+
+                fulloldname = '/dev/%s/%s' % (pool, old_name)
+                fullnewname = '/dev/%s/%s' % (pool, name)
+                old_new_disk_mapping[fulloldname] = fullnewname
+
+                # Create the volume on destination:
+                dest.proxy.vol_create(pool, name, int(size))
+                self.logger.info('Job-%s: Created volume %s/%s on destination '
+                             'hypervisor (was %s)', self['id'], pool, name,
+                             old_name)
+
+                # Rollback stuff for this action:
+                def rb_volcreate():
+                    dest.proxy.vol_delete(pool, name)
+                self.checkpoint(rb_volcreate)
+
+            # Define VM:
+            self.report('define vm')
+            self.logger.info('Job-%s: XML configuration transfert', self['id'])
+            vm_config = source.proxy.vm_export(self['vm_name'])
+
+            # Change vm configuration XML to update it with new name:
+            new_vm_config = self._update_xml(vm_config, self['vm_name'],
+                                             self['new_vm_name'],
+                                             old_new_disk_mapping)
+
+            dest.proxy.vm_define(new_vm_config)
+
+            # Rollback stuff for vm definition:
+            def rb_define():
+                dest.proxy.vm_undefine(name)
+            self.checkpoint(rb_define)
+
+            # Copy all source disk on destination disk:
+            for disk, name in names.iteritems():
+                self._copy_disk(source, dest, vm, disk, name)
+
+            self.logger.info('Job-%s: Clonage completed with success', self['id'])
+
+
+    def _update_xml(self, vm_config, old_name, name, old_new_name_mapping):
+        """ Update the XML definition of the VM with the new vm name, the new
+            vm disk, and remove the uuid tag.
+        """
+
+        vm_config = vm_config.replace('<name>%s</name>' % old_name,
+                                      '<name>%s</name>' % name)
+        vm_config = re.sub('<uuid>.*?</uuid>\n', '', vm_config)
+        # delete MAC address, then it will be regenerated by libvirt
+        vm_config = re.sub('<mac .*?/>\n', '', vm_config)
+        for old, new in old_new_name_mapping.iteritems():
+            vm_config = vm_config.replace("='%s'" % old,
+                                          "='%s'" % new)
+        return vm_config
+
+    def _copy_disk(self, source, dest, vm, disk, new_disk):
+        """ Copy the specified disk name of the vm from source to dest.
+        """
+
+        # Get informations about the disk:
+        pool = vm.get('disk%s_pool' % disk)
+        name = vm.get('disk%s_vol' % disk)
+        self.logger.info('Job-%s: Started copy for %s/%s to %s/%s',
+                         self['id'], pool, name, pool, new_disk)
+        self.report('copy %s/%s' % (pool, name))
+
+        # Make the copy and wait for it end:
+        xferprop = dest.proxy.vol_import(pool, new_disk)
+
+        # Register the cancel function:
+        def cancel_xfer():
+            dest.proxy.vol_import_cancel(xferprop['id'])
+        self['func_cancel_xfer'] = cancel_xfer
+
+        # Wait for the end of transfert:
+        watcher = AsyncWatcher()
+        watcher.register(source.conn, 'vol_export', pool, name, dest.ip, xferprop['port'])
+        watcher.register(dest.conn, 'vol_import_wait', xferprop['id'])
+
+        msgs = watcher.wait()
+        del self['func_cancel_xfer']
+
+        # Compare checksum of two answers:
+        checksums = []
+        assert len(msgs) == 2
+
+        for msg in msgs:
+            if msg.get('error') is not None:
+                msg = 'error while copy: %s' % msg['error']['message']
+                raise JobCancelError(msg)
+            else:
+                checksums.append(msg['return'].get('checksum'))
+                self.checkpoint()
+
+        if checksums[0] != checksums[1]:
+            raise JobCancelError('checksum mismatches')
+
+    def cancel(self):
+        if self.get('func_cancel_xfer') is not None:
+            self.get('func_cancel_xfer')()
+        super(CloneJob, self).cancel()

cloudcontrol/server/jobs/coldmigration.py

+from sjrpc.core import AsyncWatcher
+
+from cloudcontrol.server.jobs.base import BaseJob, JobCancelError
+from cloudcontrol.server.utils import AcquiresAllOrNone
+
+
+class ColdMigrationJob(BaseJob):
+
+    """ A cold vm migration job.
+
+    Mandatory items:
+     * vm_name: name of the vm to migrate
+     * hv_source: name of the hv which execute the VM
+     * hv_dest: the destination hypervisor
+     * author: login of the author cli
+    """
+
+    def job(self):
+        vm_id = '%s.%s' % (self['hv_source'], self['vm_name'])
+
+        self['title'] = 'Cold migration %s --> %s' % (vm_id, self['hv_dest'])
+        self.logger.info('Job-%s: Started migration for %s', self['id'], vm_id)
+
+        # Cancel the job if the user has not the right to migrate the vm or to
+        # select an hypervisor as destination:
+        right_check = self.manager.server.check
+
+        tql = 'id=%s' % vm_id
+        if not right_check(self['author'], 'coldmigrate', tql):
+            raise JobCancelError('author have no rights to migrate this VM')
+
+        tql = 'id=%s' % self['hv_dest']
+        if not right_check(self['author'], 'coldmigrate_dest', tql):
+            raise JobCancelError('author have no right to migrate to this hv')
+
+        # Update the VM object:
+        vm = self.manager.server.db.get_by_id(vm_id)
+        if vm is None:
+            raise JobCancelError('Source VM not found')
+
+        # Get the source and destination hv clients:
+        try:
+            source = self.manager.server.get_client(self['hv_source'])
+        except KeyError:
+            raise JobCancelError('source hypervisor is not connected')
+
+        try:
+            dest = self.manager.server.get_client(self['hv_dest'])
+        except KeyError:
+            raise JobCancelError('destination hypervisor is not connected')
+
+        self.checkpoint()
+
+        self.report('waiting lock for source and dest hypervisors')
+        self.logger.info('Job-%s: Trying to acquire locks', self['id'])
+
+        with AcquiresAllOrNone(source.hvlock, dest.hvlock):
+            self.logger.info('Job-%s: Locks acquired', self['id'])
+            self.checkpoint()
+
+            if not vm['status'] == 'stopped':
+                raise JobCancelError('vm is not stopped')
+
+            # Create storages on destination:
+            self.report('create volumes')
+            for disk in vm.get('disk', '').split():
+                # Getting informations about the disk:
+                pool = vm.get('disk%s_pool' % disk)
+                name = vm.get('disk%s_vol' % disk)
+                size = vm.get('disk%s_size' % disk)
+                assert pool is not None, 'pool tag doesn\'t exists'
+                assert name is not None, 'name tag doesn\'t exists'
+                assert size is not None, 'size tag doesn\'t exists'
+
+                # Create the volume on destination:
+                dest.proxy.vol_create(pool, name, int(size))
+                self.logger.info('Job-%s: Created volume %s/%s on destination '
+                             'hypervisor', self['id'], pool, name)
+
+                # Rollback stuff for this action:
+                def rb_volcreate():
+                    dest.proxy.vol_delete(pool, name)
+                self.checkpoint(rb_volcreate)
+
+            # Define VM:
+            self.report('define vm')
+            self.logger.info('Job-%s: XML configuration transfert', self['id'])
+            vm_config = source.proxy.vm_export(self['vm_name'])
+            dest.proxy.vm_define(vm_config)
+
+            # Rollback stuff for vm definition:
+            def rb_define():
+                dest.proxy.vm_undefine(self['vm_name'])
+            self.checkpoint(rb_define)
+
+            # Copy all source disk on destination disk:
+            for disk in vm.get('disk', '').split():
+                self._copy_disk(source, dest, vm, disk)
+
+            # At this point, if operation is a success, all we need is just to
+            # cleanup source hypervisor from disk and vm. This operation *CAN'T*
+            # be cancelled or rollbacked if anything fails (unlikely). The
+            # migration must be considered as a success, and the only way to
+            # undo this is to start a new migration in the other way.
+
+            # Delete the rollback list.
+            # This is mandatory to avoid data loss if the cleanup
+            # code below fail.
+            self._wayback = []
+
+            # Cleanup the disks:
+            for disk in vm.get('disk', '').split():
+                pool = vm.get('disk%s_pool' % disk)
+                name = vm.get('disk%s_vol' % disk)
+
+                source.proxy.vol_delete(pool, name)
+
+            # Cleanup the VM:
+            source.proxy.vm_undefine(self['vm_name'])
+
+            self.logger.info('Job-%s: Migration completed with success', self['id'])
+
+
+    def _copy_disk(self, source, dest, vm, disk):
+        """ Copy the specified disk name of the vm from source to dest.
+        """
+
+        # Get informations about the disk:
+        pool = vm.get('disk%s_pool' % disk)
+        name = vm.get('disk%s_vol' % disk)
+        self.logger.info('Job-%s: Started copy for %s/%s', self['id'], pool, name)
+        self.report('copy %s/%s' % (pool, name))
+
+        # Make the copy and wait for it end:
+        xferprop = dest.proxy.vol_import(pool, name)
+
+        # Register the cancel function:
+        def cancel_xfer():
+            dest.proxy.vol_import_cancel(xferprop['id'])
+        self['func_cancel_xfer'] = cancel_xfer
+
+        # Wait for the end of transfert:
+        watcher = AsyncWatcher()
+        watcher.register(source.conn, 'vol_export', pool, name, dest.ip, xferprop['port'])
+        watcher.register(dest.conn, 'vol_import_wait', xferprop['id'])
+
+        msgs = watcher.wait()
+        del self['func_cancel_xfer']
+
+        # Compare checksum of two answers:
+        checksums = []
+        assert len(msgs) == 2
+        for msg in msgs:
+            if msg.get('error') is not None:
+                msg = 'error while copy: %s' % msg['error']['message']
+                raise JobCancelError(msg)
+            else:
+                checksums.append(msg['return'].get('checksum'))
+                self.checkpoint()
+
+        if checksums[0] != checksums[1]:
+            raise JobCancelError('checksum mismatches')
+
+    def _check_status(self, vm_id, status):
+        """
+        Check the status of the VM.
+        """
+
+        answer = self.manager.server.list('id=%s&status=%s' % (vm_id, status))
+        return bool(answer)
+
+    def cancel(self):
+        if self.get('func_cancel_xfer') is not None:
+            self.get('func_cancel_xfer')()
+        super(ColdMigrationJob, self).cancel()

cloudcontrol/server/jobs/hotmigration.py

+import time
+
+from sjrpc.core import AsyncWatcher
+
+from cloudcontrol.server.jobs.base import BaseJob, JobCancelError
+from cloudcontrol.server.utils import AcquiresAllOrNone
+
+
+class HotMigrationJob(BaseJob):
+
+    """ A hot vm migration job.
+
+    Mandatory items:
+     * vm_name: name of the vm to migrate
+     * hv_source: name of the hv which execute the VM
+     * hv_dest: the destination hypervisor
+     * author: login of the author cli
+    """
+
+    def job(self):
+        vm_id = '%s.%s' % (self['hv_source'], self['vm_name'])
+
+        self['title'] = 'Hot migration %s --> %s' % (vm_id, self['hv_dest'])
+        self.logger.info('Job-%s: Started hot migration for %s', self['id'], vm_id)
+
+        # Cancel the job if the user has not the right to migrate the vm or to
+        # select an hypervisor as destination:
+        right_check = self.manager.server.check
+
+        tql = 'id=%s' % vm_id
+        if not right_check(self['author'], 'hotmigrate', tql):
+            raise JobCancelError('author have no rights to migrate this VM')
+
+        tql = 'id=%s' % self['hv_dest']
+        if not right_check(self['author'], 'hotmigrate_dest', tql):
+            raise JobCancelError('author have no right to migrate to this hv')
+
+        # Update the VM object:
+        vm = self.manager.server.db.get_by_id(vm_id)
+        if vm is None:
+            raise JobCancelError('Source VM not found')
+
+        # Get the source and destination hv clients:
+        try:
+            source = self.manager.server.get_client(self['hv_source'])
+        except KeyError:
+            raise JobCancelError('source hypervisor is not connected')
+
+        try:
+            dest = self.manager.server.get_client(self['hv_dest'])
+        except KeyError:
+            raise JobCancelError('destination hypervisor is not connected')
+
+        self.checkpoint()
+
+        self.report('waiting lock for source and dest hypervisors')
+        self.logger.info('Job-%s: Trying to acquire locks', self['id'])
+
+        with AcquiresAllOrNone(source.hvlock, dest.hvlock):
+            self.logger.info('Job-%s: Locks acquired', self['id'])
+            self.checkpoint()
+
+            if not self._check_status(vm_id, 'running'):
+                raise JobCancelError('vm is not started')
+
+            to_cleanup = []
+
+            # Create storages on destination and start synchronization:
+            disks = vm.get('disk', '').split()
+            for disk in disks:
+                to_cleanup += self._sync_disk(vm, disk, source, dest)
+
+            # Libvirt tunnel setup:
+            tunres_src = source.proxy.tun_setup()
+            def rb_tun_src():
+                source.proxy.tun_destroy(tunres_src)
+            self.checkpoint(rb_tun_src)
+
+            tunres_dst = dest.proxy.tun_setup(local=False)
+            def rb_tun_dst():
+                dest.proxy.tun_destroy(tunres_dst)
+            self.checkpoint(rb_tun_dst)
+
+            source.proxy.tun_connect(tunres_src, tunres_dst, dest.ip)
+            dest.proxy.tun_connect_hv(tunres_dst, migration=False)
+
+            # Migration tunnel setup:
+            migtunres_src = source.proxy.tun_setup()
+            def rb_migtun_src():
+                source.proxy.tun_destroy(migtunres_src)
+            self.checkpoint(rb_migtun_src)
+
+            migtunres_dst = dest.proxy.tun_setup(local=False)
+            def rb_migtun_dst():
+                dest.proxy.tun_destroy(migtunres_dst)
+            self.checkpoint(rb_migtun_dst)
+
+            source.proxy.tun_connect(migtunres_src, migtunres_dst, dest.ip)
+            dest.proxy.tun_connect_hv(migtunres_dst, migration=True)
+
+            # Initiate the live migration:
+            self.report('migration in progress')
+            source.proxy.vm_migrate_tunneled(self['vm_name'], tunres_src,
+                                             migtunres_src, _timeout=None)
+
+            # At this point, if operation is a success, all we need is just to
+            # cleanup source hypervisor from disk and vm. This operation *CAN'T*
+            # be cancelled or rollbacked if anything fails (unlikely). The
+            # migration must be considered as a success, and the only way to
+            # undo this is to start a new migration in the other way.
+
+            # Delete the rollback list.
+            # This is mandatory to avoid data loss if the cleanup
+            # code below fail.
+            self.report('cleanup')
+            self._wayback = []
+
+            source.proxy.tun_destroy(tunres_src)
+            dest.proxy.tun_destroy(tunres_dst)
+            source.proxy.tun_destroy(migtunres_src)
+            dest.proxy.tun_destroy(migtunres_dst)
+
+            for cb_cleanup in reversed(to_cleanup):
+                cb_cleanup()
+
+            # Cleanup the disks:
+            for disk in disks:
+                pool = vm.get('disk%s_pool' % disk)
+                name = vm.get('disk%s_vol' % disk)
+
+                source.proxy.vol_delete(pool, name)
+
+            self.logger.info('Job-%s: Migration completed with success', self['id'])
+
+
+    def _sync_disk(self, vm, disk, source, dest):
+        to_cleanup = []
+
+        # getting informations about the disk:
+        pool = vm.get('disk%s_pool' % disk)
+        name = vm.get('disk%s_vol' % disk)
+        size = vm.get('disk%s_size' % disk)
+        assert pool is not None, 'pool tag doesn\'t exists'
+        assert name is not None, 'name tag doesn\'t exists'
+        assert size is not None, 'size tag doesn\'t exists'
+
+        status_msg = 'sync volume %s/%s (%%s)' % (pool, name)
+        self.report(status_msg % 'creation')
+
+        # create the volume on destination:
+        dest.proxy.vol_create(pool, name, int(size))
+        self.logger.info('job-%s: created volume %s/%s on destination '
+                     'hypervisor', self['id'], pool, name)
+
+        # rollback stuff for this action:
+        def rb_volcreate():
+            dest.proxy.vol_delete(pool, name)
+        self.checkpoint(rb_volcreate)
+
+        # setup the drbd synchronization with each hypervisors:
+        self.report(status_msg % 'setup')
+
+        res_src = source.proxy.drbd_setup(pool, name)
+        def rb_setupsrc():
+            source.proxy.drbd_shutdown(res_src)
+        self.checkpoint(rb_setupsrc)
+        to_cleanup.append(rb_setupsrc)
+
+        res_dst = dest.proxy.drbd_setup(pool, name)
+        def rb_setupdst():
+            dest.proxy.drbd_shutdown(res_dst)
+        self.checkpoint(rb_setupdst)
+        to_cleanup.append(rb_setupdst)
+
+        # start connection of drbd:
+        self.report(status_msg % 'connect')
+
+        watcher = AsyncWatcher()
+        watcher.register(source.conn, 'drbd_connect', res_src, res_dst, dest.ip)
+        watcher.register(dest.conn, 'drbd_connect', res_dst, res_src, source.ip)
+        msgs = watcher.wait(timeout=30)
+        for msg in msgs:
+            if 'error' in msg:
+                msg = 'error while drbd_connect: %s' % msg['error']['message']
+                raise JobCancelError(msg)
+
+        # setup topology as primary/secondary:
+        source.proxy.drbd_role(res_src, True)
+        dest.proxy.drbd_role(res_dst, False)
+
+        # Wait the end of the synchronization:
+        sync_running = True
+        while sync_running:
+            status = dest.proxy.drbd_sync_status(res_dst)
+            if status['done']:
+                sync_running = False
+
+            self.report(status_msg % 'sync %s%%' % status['completion'])
+            time.sleep(2)
+
+        dest.proxy.drbd_role(res_dst, True)
+
+        source.proxy.drbd_takeover(res_src, True)
+        def rb_takeover_src():
+            source.proxy.drbd_takeover(res_src, False)
+        self.checkpoint(rb_takeover_src)
+        to_cleanup.append(rb_takeover_src)
+
+        dest.proxy.drbd_takeover(res_dst, True)
+        def rb_takeover_dst():
+            dest.proxy.drbd_takeover(res_dst, False)
+        self.checkpoint(rb_takeover_dst)
+        to_cleanup.append(rb_takeover_dst)
+
+        return to_cleanup
+
+    def _check_status(self, vm_id, status):
+        """
+        Check the status of the VM.
+        """
+
+        answer = self.manager.server.list('id=%s&status=%s' % (vm_id, status))
+        return bool(answer)

cloudcontrol/server/jobs/killclient.py

+import time
+
+from cloudcontrol.server.jobs.base import BaseJob
+
+class KillClientJob(BaseJob):
+
+    """ A job used to kill connected accounts.
+
+    Mandatory items:
+     * account: the account login to kill
+
+    Optional items:
+     * gracetime: time before to kill the user
+    """
+
+    def job(self):
+        gracetime = self.get('gracetime')
+        account = self.get('account')
+        assert account is not None, 'Account not specified'
+
+        if gracetime is not None:
+            time.sleep(int(gracetime))
+
+        self.checkpoint()
+
+        self.manager.server.kill(account)

cloudcontrol/server/jobs/killoldcli.py

+import time
+
+from cloudcontrol.server.jobs.base import BaseJob
+
+class KillOldCliJob(BaseJob):
+
+    """ Typically an hidden job used to kill clients who are connected/idle
+        since too much time.
+
+    Mandatory items:
+     * maxcon: maximum connection time in minutes
+     * maxidle: maximum idle time in minutes
+
+    Optional items:
+     * delay: delay in secondes between two checks (default 1m)
+    """
+
+    DEFAULT_DELAY = 60
+
+    def job(self):
+        maxcon = self.get('maxcon')
+        assert maxcon is not None, 'maxcon is None'
+        maxidle = self.get('maxidle')
+        assert maxidle is not None, 'maxidle is None'
+        delay = self.get('delay', self.DEFAULT_DELAY)
+
+        while True:
+            self.checkpoint()
+            for client in list(self.manager.server.iterclients('cli')):
+                if client.uptime > (maxcon * 60) or client.idle > (maxidle * 60):
+                    self.manager.server.kill(client.login)
+                    self.logger.info('Disconnected %s because of its long time'
+                                     ' or idle connection', client.login)
+            time.sleep(delay)

ccserver/ccserver.py → cloudcontrol/server/server.py

-'''
-Main class of cc-server.
-'''
+""" Main class of cc-server.
+"""
 
-from __future__ import absolute_import
-
-import logging
 from fnmatch import fnmatch as glob
 
 from sjrpc.server import SSLRpcServer
 from sjrpc.utils import RpcHandler, pass_connection
 
-from ccserver.conf import CCConf
-from ccserver.exceptions import AlreadyRegistered, NotConnectedAccountError, AuthenticationError, BadRoleError
-from ccserver.jobs import JobsManager
-from ccserver.clients import Client
+from cloudcontrol.server.conf import CCConf
+from cloudcontrol.server.exceptions import (AlreadyRegistered,
+                                            NotConnectedAccountError,
+                                            AuthenticationError,
+                                            BadRoleError)
+from cloudcontrol.server.jobs import JobsManager
+from cloudcontrol.server.clients import Client
 
-#from cloudcontrol.common.tql.db.object import TqlObject
-from ccserver.db import SObject, SRequestor
+from cloudcontrol.server.db import SObject, SRequestor
 from cloudcontrol.common.tql.db.tag import StaticTag
 from cloudcontrol.common.tql.db.db import TqlDatabase
 
-# Import all enabled roles
-import ccserver.clients.cli
-import ccserver.clients.host
-import ccserver.clients.hv
-import ccserver.clients.bootstrap
-import ccserver.clients.spv
+# Import all enabled roles:
+import cloudcontrol.server.clients.cli
+import cloudcontrol.server.clients.host
+import cloudcontrol.server.clients.hv
+import cloudcontrol.server.clients.bootstrap
+import cloudcontrol.server.clients.spv
+
 
 class WelcomeHandler(RpcHandler):
     """ Default handler used on client connections of the server.
@@ -40,52 +39,52 @@ class WelcomeHandler(RpcHandler):
 
 
 class CCServer(object):
-    '''
-    CloudControl server main class.
+    """ CloudControl server main class.
 
     :param conf_dir: the directory that store the client configuration
     :param certfile: the path to the ssl certificate
     :param keyfile: the path to the ssl key
     :param address: the interface to bind
     :param port: the port to bind
-    '''
+    """
 
     # These tags are reserved and cannot be setted by an user:
     RESERVED_TAGS = ('id', 'a', 'r', 'close', 'con', 'ip', 'p')
 
-    def __init__(self, conf_dir, maxcon, maxidle, certfile=None, keyfile=None,
-                 address='0.0.0.0', port=1984):
+    def __init__(self, logger, conf_dir, maxcon, maxidle, certfile=None,
+                 keyfile=None, address='0.0.0.0', port=1984):
+
+        self.logger = logger
 
         self._clients = {}  # Clients connected to the server
 
         # The interface object to the configuration directory:
-        self.conf = CCConf(conf_dir)
+        self.conf = CCConf(self.logger.getChild('conf'), conf_dir)
         # Some settings:
         self._maxcon = maxcon
         self._maxidle = maxidle
         # SSL configuration stuff:
         if certfile:
-            logging.info('SSL Certificate: %s', certfile)
+            self.logger.info('SSL Certificate: %s', certfile)
         if keyfile:
-            logging.info('SSL Key: %s', certfile)
+            self.logger.info('SSL Key: %s', certfile)
 
         self.db = TqlDatabase(default_requestor=SRequestor())
 
         # Create the rpc server:
-        logging.info('Listening on %s:%s', address, port)
+        self.logger.info('Listening on %s:%s', address, port)
         self.rpc = SSLRpcServer.from_addr(address, port, certfile=certfile,
                                           keyfile=keyfile,
                                           conn_kw=dict(handler=WelcomeHandler(self),
                                                        on_disconnect='on_disconnect'))
 
         # The jobs manager:
-        self.jobs = JobsManager(self)
-        logging.info('Server started to running')
+        self.jobs = JobsManager(self.logger.getChild('jobs'), self)
+        self.logger.info('Server started to running')
 
     def _update_accounts(self):
-        '''
-        Update the database with accounts.
-        '''
+        """ Update the database with accounts.
+        """
 
         db_accounts = set((obj['a'].value for obj in self.db.objects if 'a' in obj))
         accounts = set(self.conf.list_accounts())
@@ -131,72 +130,38 @@ class CCServer(object):
                 raise AuthenticationError('Unknown login')
             else:
                 if 'close' in self.conf.show(login)['tags']:
-                    logging.warning(logmsg + 'account closed (%s)', conn.getpeername(), login)
+                    self.logger.warning(logmsg + 'account closed (%s)', conn.getpeername(), login)
                     raise AuthenticationError('Account is closed')
 
         if role is None:
-            logging.warning(logmsg + 'bad login/password (%s)', conn.getpeername(), login)
+            self.logger.warning(logmsg + 'bad login/password (%s)', conn.getpeername(), login)
             raise AuthenticationError('Bad login/password')
         else:
             if role not in Client.roles:
-                logging.warning(logmsg + 'bad role in account config (%s)', conn.getpeername(), login)
+                self.logger.warning(logmsg + 'bad role in account config (%s)', conn.getpeername(), login)
                 raise BadRoleError('%r is not a legal role' % role)
 
-            create_object = False
-
             # If authentication is a success, try to register the client:
-            if role == 'bootstrap':
-                # Set a bootstrap id for the object:
-                login = '%s.%s' % (login, conn.get_fd())
-                # Real role of the node will be host:
-                role = 'host'
-                create_object = True
-
-            # Try to register the client:
-            for _ in xrange(5):
-                try:
-                    self.register(login, role, conn, create_object)
-                except AlreadyRegistered:
-                    if role == 'cli':
-                        try:
-                            self.kill(login)
-                        except NotConnectedAccountError:
-                            pass
-                else:
-                    break
-            else:
-                logging.warning(logmsg + 'already connected (%s)', conn.getpeername(), login)
-                raise AuthenticationError('Already connected')
-
-            logging.info('Authentication success from %s with login %s', conn.getpeername(), login)
-            return role
-
+            client = self.register(login, role, conn)
+            return client.role
 
-    def register(self, login, role, connection, create_object=False):
-        '''
-        Register a new connected account on the server.
+    def register(self, login, role, connection):
+        """ Register a new connected account on the server.
 
         :param login: login of the account
         :param connection: connection to register
         :param tags: tags to add for the client
-        :param create_object: create the object on objectdb
-        '''
-
-        # Create the object on objectdb if required:
-        if create_object:
-            tql_object = SObject(login)
-            tql_object.register(StaticTag('r', role))
-            self.db.register(tql_object)
-        else:
-            tql_object = self.db.get(login)
-            assert tql_object is not None
-
-        # Register the client:
-        if login in self._clients:
-            raise AlreadyRegistered('A client is already connected with this account.')
-        else:
-            client = Client.from_role(role, login, self, connection, tql_object)
-            self._clients[login] = client
+        """
+        client = Client.from_role(role, None, login, self, connection)
+        client.logger = self.logger.getChild('clients.%s' % client.login)
+        if client.login in self._clients:
+            if client.KILL_ALREADY_CONNECTED:
+                self.kill(client.login)
+            else:
+                raise AlreadyRegistered('A client is already connected with this account.')
+        client.attach()
+        self._clients[client.login] = client
+        return client
 
     def unregister(self, client):
         """ Unregister a client.
@@ -204,9 +169,8 @@ class CCServer(object):
         del self._clients[client.login]
 
     def run(self):
-        '''
-        Run the server mainloop.
-        '''
+        """ Run the server mainloop.
+        """
 
         # Register accounts on the database:
         self._update_accounts()
@@ -215,7 +179,7 @@ class CCServer(object):
         self.jobs.create('kill_oldcli', maxcon=self._maxcon,
                          maxidle=self._maxidle, _hidden=True)
 
-        logging.debug('Running rpc mainloop')
+        self.logger.debug('Running rpc mainloop')
         self.rpc.run()
 
     def get_client(self, login):
@@ -227,13 +191,12 @@ class CCServer(object):
         return self._clients[login]
 
     def kill(self, login):
-        '''
-        Disconnect from the server the client identified by provided login.
+        """ Disconnect from the server the client identified by provided login.
 
         :param login: the login of the user to disconnect
         :throws NotConnectedAccount: when provided account is not connected (or
             if account doesn't exists).
-        '''
+        """
 
         client = self._clients.get(login)
         if client is None:
@@ -242,14 +205,13 @@ class CCServer(object):
         client.shutdown()
 
     def check(self, login, method, tql=None):
-        '''
-        Check if the user can call the specified method with specified TQL.
+        """ Check if the user can call the specified method with specified TQL.
 
         :param login: the login of the user
         :param method: the method to call
         :param tql: the tql passed in argument of the method
         :return: True if user have rights, else False
-        '''
+        """
 
         rights = self.conf.show(login)['rights']
         if tql is not None:
@@ -266,5 +228,7 @@ class CCServer(object):
         return False
 
     def list(self, query, show=None):
+        """ List objects from database.
+        """
         self._update_accounts()
         return self.db.query(query, show)

ccserver/utils.py → cloudcontrol/server/utils.py

-#!/usr/bin/env python
-#coding=utf8
-
-'''
-Some helpers used by cc-server.
-'''
+""" Some helpers used by cc-server.
+"""
 
 from threading import Lock
 
 class Acquires(object):
 
-    '''
-    Context manager used to acquire more than one lock at once. It works rely
-    on the fact that if locks are always acquired in the same order, we can't
-    enter in a deadlock situation.
+    """ Context manager used to acquire more than one lock at once. It works
+        rely on the fact that if locks are always acquired in the same order,
+        we can't enter in a deadlock situation.
 
     Usage is very simple:
 
@@ -24,7 +19,7 @@ class Acquires(object):
 
     .. seealso::
         http://dabeaz.blogspot.com/2009/11/python-thread-deadlock-avoidance_20.html
-    '''
+    """
 
     def __init__(self, *locks):
         self._locks = sorted(set(locks), key=lambda x: id(x))
@@ -40,10 +35,9 @@ class Acquires(object):
 
 class AcquiresAllOrNone(Acquires):
 
-    '''
-    Class that extend Acquires to allow to release all lock if one of them
-    is not free.
-    '''
+    """ Class that extend Acquires to allow to release all lock if one of
+        them is not free.
+    """
 
     # Global acquire lock:
     acquirelock = Lock()

debian/changelog

+cc-server (24~rc1) unstable; urgency=low
+
+  * RC1.
+
+ -- Antoine Millet <antoine.millet@smartjog.com>  Fri, 06 Jul 2012 16:21:33 +0200
+
 cc-server (24~dev) unstable; urgency=low
 
   * Development release.

debian/control

@@ -2,13 +2,22 @@ Source: cc-server
 Section: python
 Priority: optional
 Maintainer: Antoine Millet <antoine.millet@smartjog.com>
-Build-Depends: debhelper (>= 7), python-central (>= 0.6), cdbs (>= 0.4.50), python-setuptools, python, python-docutils
+Build-Depends: debhelper (>= 7),
+               python-central (>= 0.6),
+               cdbs (>= 0.4.50),
+               python-setuptools,
+               python,
+               python-docutils
 XS-Python-Version: >= 2.6
 Standards-Version: 3.9.1
 
 Package: cc-server
 Architecture: all
-Depends: ${misc:Depends}, ${python:Depends}, python-sjrpc (>= 14), python-daemon
+Depends: ${misc:Depends},
+         ${python:Depends},
+         python-sjrpc (>= 17),
+         python-daemon,
+         cc-common
 XB-Python-Version: ${python:Versions}
 Description: CloudControl server
  This package provides the server of CloudControl.

debian/postinst

@@ -46,4 +46,14 @@ esac
 
 #DEBHELPER#
 
+# Workaround to restart cc-server after pycentral
+if [ -x "/etc/init.d/cc-server" ]; then
+    update-rc.d cc-server defaults >/dev/null
+    if [ -x "`which invoke-rc.d 2>/dev/null`" ]; then
+        invoke-rc.d cc-server start || exit $?
+    else
+        /etc/init.d/cc-server start || exit $?
+    fi
+fi
+
 exit 0

debian/rules

@@ -7,3 +7,8 @@ DEB_PYTHON_SYSTEM=pycentral
 # dh_python / dh_pycentral / dh_pysupport
 include /usr/share/cdbs/1/rules/debhelper.mk
 include /usr/share/cdbs/1/class/python-distutils.mk
+
+PYTHON_PACKAGES := cc-server
+
+$(patsubst %,binary-install/%,$(PYTHON_PACKAGES))::
+	dh_pycentral -p$(cdbs_curpkg)

doc/conf.py

@@ -22,7 +22,12 @@ sys.path.append(os.path.abspath('../'))
 
 # Add any Sphinx extension module names here, as strings. They can be extensions
 # coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
-extensions = ['sphinx.ext.autodoc', 'sphinx.ext.intersphinx', 'sphinx.ext.todo', 'sphinx.ext.pngmath', 'sphinx.ext.ifconfig']
+extensions = ['sphinx.ext.autodoc',
+              'sphinx.ext.autosummary',
+              'sphinx.ext.intersphinx',
+              'sphinx.ext.todo',
+              'sphinx.ext.pngmath',
+              'sphinx.ext.ifconfig',]
 
 # Add any paths that contain templates here, relative to this directory.
 templates_path = ['_templates']
@@ -45,7 +50,7 @@ copyright = u'2010, Smartjog'
 # built documents.
 #
 # The short X.Y version.
-from ccserver import __version__
+from cloudcontrol.server import __version__
 version = __version__
 # The full version, including alpha/beta/rc tags.
 release = version

etc/cc-server.conf

@@ -11,7 +11,7 @@ group = cc-server
 #pidfile =
 
 # Set the umask of the process:
-#umask = 0177
+#umask = 077
 
 # Certificates for the SSL machinery (mandatory):
 ssl_key = 

setup.py

-from setuptools import setup
+from setuptools import setup, find_packages
 from distutils.command.build import build
 import os
 import sys
 
 # Retrieval of version:
-from ccserver import __version__
+from cloudcontrol.server import __version__
 
 ldesc = open(os.path.join(os.path.dirname(__file__), 'README')).read()
 
@@ -19,15 +19,15 @@ class BuildMan(build):
     description = 'Build manual from RSt source'
 
     def run(self):
-	from docutils.core import publish_file
-	from docutils.writers import manpage
-	
+        from docutils.core import publish_file
+        from docutils.writers import manpage
+
         srcdir = os.path.split(os.path.abspath(__file__))[0]
 
-	for man in self.MANPAGES:
-	    publish_file(source_path=os.path.join(srcdir, 'doc/%s.rst' % man),
-			 destination_path=os.path.join(srcdir, '%s.1' % man),
-			 writer=manpage.Writer())
+        for man in self.MANPAGES:
+            publish_file(source_path=os.path.join(srcdir, 'doc/%s.rst' % man),
+                         destination_path=os.path.join(srcdir, '%s.1' % man),
+                         writer=manpage.Writer())
 
 build.sub_commands.insert(0, ('build_man', None))
 
@@ -40,9 +40,10 @@ setup(
     long_description=ldesc,
     author='Antoine Millet',
     author_email='antoine.millet@smartjog.com',
-    license='GPL2', 
-    packages=['ccserver', 'ccserver.clients'],
+    license='GPL2',
+    packages=find_packages(exclude=['ez_setup', 'examples', 'tests']),
     scripts=['bin/cc-server', 'bin/cc-addaccount'],
+    namespace_packages=['cloudcontrol'],
     data_files=(
         ('/etc/', ('etc/cc-server.conf',)),
     ),