Merge pull request #44 from openvstorage/develop

Develop in master

Author: JeffreyDevloo

helpers/api.py

@@ -58,10 +58,8 @@ def __init__(self, *args, **kwargs):
 
 class TimeOutError(RuntimeError):
     """
-    Custom tineout class
+    Custom timeout class
     """
-    def __init__(self, status_code, *args, **kwargs):
-        super(TimeOutError, self).__init__(*args, **kwargs)
 
 
 class OVSClient(object):
@@ -187,7 +185,7 @@ def _process(self, response, overrule_raw=False):
             else:
                 raise HttpException(status_code, message)
 
-    def _call(self, api, params, function, **kwargs):
+    def _call(self, api, params, func, **kwargs):
         if not api.endswith('/'):
             api = '{0}/'.format(api)
         if not api.startswith('/'):
@@ -197,15 +195,15 @@ def _call(self, api, params, function, **kwargs):
         first_connect = self._token is None
         headers, url = self._prepare(params=params)
         try:
-            return self._process(function(url=url.format(api), headers=headers, verify=self._verify, **kwargs))
+            return self._process(func(url=url.format(api), headers=headers, verify=self._verify, **kwargs))
         except ForbiddenException:
             if self._volatile_client is not None:
                 self._volatile_client.delete(self._key)
             if first_connect is True:  # First connect, so no token was present yet, so no need to try twice without token
                 raise
             self._token = None
             headers, url = self._prepare(params=params)
-            return self._process(function(url=url.format(api), headers=headers, verify=self._verify, **kwargs))
+            return self._process(func(url=url.format(api), headers=headers, verify=self._verify, **kwargs))
         except Exception:
             if self._volatile_client is not None:
                 self._volatile_client.delete(self._key)
@@ -216,15 +214,15 @@ def delete(self, api):
         Executes a DELETE call
         :param api: Specification for to fill out in the URL, eg: /alba/backends/<albabackend_guid>
         """
-        return self._call(api=api, params={}, function=requests.delete)
+        return self._call(api=api, params={}, func=requests.delete)
 
     def get(self, api, params=None):
         """
         Executes a GET call
         :param api: Specification for to fill out in the URL, eg: /vpools/<vpool_guid>/shrink_vpool
         :param params: Additional query parameters, eg: _dynamics
         """
-        return self._call(api=api, params=params, function=requests.get)
+        return self._call(api=api, params=params, func=requests.get)
 
     def post(self, api, data=None, params=None):
         """
@@ -233,7 +231,7 @@ def post(self, api, data=None, params=None):
         :param data: Data to post
         :param params: Additional query parameters, eg: _dynamics
         """
-        return self._call(api=api, params=params, function=requests.post, data=self._to_json(data))
+        return self._call(api=api, params=params, func=requests.post, data=self._to_json(data))
 
     def put(self, api, data=None, params=None):
         """
@@ -242,7 +240,7 @@ def put(self, api, data=None, params=None):
         :param data: Data to put
         :param params: Additional query parameters, eg: _dynamics
         """
-        return self._call(api=api, params=params, function=requests.put, data=self._to_json(data))
+        return self._call(api=api, params=params, func=requests.put, data=self._to_json(data))
 
     def patch(self, api, data=None, params=None):
         """
@@ -251,7 +249,7 @@ def patch(self, api, data=None, params=None):
         :param data: Data to patch
         :param params: Additional query parameters, eg: _dynamics
         """
-        return self._call(api=api, params=params, function=requests.patch, data=self._to_json(data))
+        return self._call(api=api, params=params, func=requests.patch, data=self._to_json(data))
 
     def wait_for_task(self, task_id, timeout=None):
         """

helpers/hypervisor/apis/kvm/sdk.py

@@ -49,12 +49,12 @@ def _recurse(treeitem):
     return result
 
 
-def authenticated(function):
+def authenticated(func):
     """
     Decorator that make sure all required calls are running onto a connected SDK
     """
     def wrapper(self, *args, **kwargs):
-        self.__doc__ = function.__doc__
+        self.__doc__ = func.__doc__
         # determine if connection isn't closed.
         try:
             self._conn = self.connect(self.login, self.host)
@@ -64,7 +64,7 @@ def wrapper(self, *args, **kwargs):
             except:
                 pass
             raise
-        return function(self, *args, **kwargs)
+        return func(self, *args, **kwargs)
     return wrapper
 
 
@@ -622,7 +622,7 @@ def create_vm(self, name, vcpus, ram, disks, cdrom_iso=None, os_type=None, os_va
 
         ovs_vm = False
         if edge_configuration is not None:
-            required_edge_params = {'port': (int, {'min': 1, 'max': 65565}),
+            required_edge_params = {'port': (int, {'min': 1, 'max': 65535}),
                                     'protocol': (str, ['tcp', 'udp', 'rdma']),
                                     'hostname': (str, None),
                                     'username': (str, None, False),

helpers/hypervisor/apis/vmware/sdk.py

@@ -38,23 +38,23 @@ def authenticated(force=False):
     Decorator to make that a login is executed in case the current session isn't valid anymore
     @param force: Force a (re)login, as some methods also work when not logged in
     """
-    def wrapper(function):
+    def wrapper(func):
         def new_function(self, *args, **kwargs):
-            self.__doc__ = function.__doc__
+            self.__doc__ = func.__doc__
             try:
                 if force:
                     self._login()
-                return function(self, *args, **kwargs)
+                return func(self, *args, **kwargs)
             except WebFault as fault:
                 if 'The session is not authenticated' in str(fault):
                     logger.debug('Received WebFault authentication failure, logging in...')
                     self._login()
-                    return function(self, *args, **kwargs)
+                    return func(self, *args, **kwargs)
                 raise
             except NotAuthenticatedException:
                 logger.debug('Received NotAuthenticatedException, logging in...')
                 self._login()
-                return function(self, *args, **kwargs)
+                return func(self, *args, **kwargs)
         return new_function
     return wrapper
 

helpers/system.py

@@ -13,12 +13,11 @@
 #
 # Open vStorage is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY of any kind.
-
+import time
 from ovs.log.log_handler import LogHandler
-from ovs.extensions.generic.system import System
+from ovs.extensions.services.service import ServiceManager
 from ovs.extensions.generic.sshclient import SSHClient
-from ovs.extensions.packages.package import PackageManager
-from ..helpers.init_manager import InitManager, InitManagerSupported
+from ovs.extensions.generic.system import System
 
 
 class SystemHelper(object):
@@ -31,40 +30,20 @@ def __init__(self):
         pass
 
     @staticmethod
-    def get_non_running_ovs_services(storagerouter_ip):
+    def get_non_running_ovs_services(client):
         """
         get all non-running ovs services
-        :param storagerouter_ip: ip address of a existing storagerouter
-        :type storagerouter_ip: str
+        :param client: sshclient instance
         :return: list of non running ovs services
         :rtype: list
         """
-        client = SSHClient(storagerouter_ip, username='root')
-        if InitManager.INIT_MANAGER == InitManagerSupported.INIT:
-            ovs_services = [service for service in client.dir_list(InitManager.UPSTART_BASEDIR) if 'ovs-' in service]
-            return [ovs_service.split('.')[0] for ovs_service in ovs_services
-                    if not InitManager.service_running(ovs_service.split('.')[0], storagerouter_ip)]
-        elif InitManager.INIT_MANAGER == InitManagerSupported.SYSTEMD:
-            ovs_services = [service for service in client.dir_list(InitManager.SYSTEMD_BASEDIR) if 'ovs-' in service]
-            return [ovs_service.split('.')[0] for ovs_service in ovs_services
-                    if not InitManager.service_running(ovs_service.split('.')[0], storagerouter_ip)]
-
-    @staticmethod
-    def get_missing_packages(ip, required_packages):
-        """
-        Get all missing packages based on required packages
-
-        :param ip: ip address of a server
-        :type ip: str
-        :param required_packages: a list of required packages (e.g. ['openvstorage', 'qemu', 'fio'])
-        :type required_packages: list
-        :return: list of missing packages
-        :rtype: list
-        """
-        client = SSHClient(ip, username='root')
-        return [package for package in required_packages
-                if client.run("dpkg -s {0} | grep Status | cut -d ' ' -f 3".format(package),
-                              allow_insecure=True) != "ok"]
+        non_running_ovs_services = []
+        for service in ServiceManager.list_services(client):
+            if not service.startswith('ovs-'):
+                continue
+            if ServiceManager.get_service_status(service, client) != 'active':
+                non_running_ovs_services.append(service)
+        return non_running_ovs_services
 
     @staticmethod
     def get_local_storagerouter():
@@ -75,31 +54,76 @@ def get_local_storagerouter():
         """
         return System.get_my_storagerouter()
 
-    @staticmethod
-    def get_ovs_version(client):
+    @classmethod
+    def get_ovs_version(cls, storagerouter=None):
         """
         Gets the installed ovs version
-        :param client: Sshclient instance
-        :type client: ovs.extensions.generic.sshclient.SSHClient
+        :param storagerouter: Storagerouter to fetch info from
         :return: ovs version identifier. Either ee or ose
         :rtype: str
         """
-        # @todo replace with storagerouter.features instead
-        # Version mapping with identifier
-        mapping = {'ee': 'volumedriver-ee-base',
-                   'ose': 'volumedriver-no-dedup-base'}
-        installed_versions = PackageManager.get_installed_versions(client=client)
-        for ovs_version, detection_key in mapping.iteritems():
-            if detection_key in installed_versions:
-                return ovs_version
+        if storagerouter is None:
+            storagerouter = cls.get_local_storagerouter()
+        if storagerouter.features['alba']['edition'] == 'community':
+            return 'ose'
+        else:
+            return 'ee'
 
     @staticmethod
-    def upper_case_first_letter(x):
+    def idle_till_ovs_is_up(ip, username, password=None, connection_timeout=300, service_timeout=60, logger=LOGGER):
         """
-        Upper case the first letter of a string
-        :param x: a normal string
-        :type x: str
-        :return: a normal string with the first letter uppercases
-        :rtype: str
+        wait until a node is back up and all ovs related are running (or potentially stuck)
+        :param ip: ip of the node
+        :param username: username to login with
+        :param password: password to login with
+        :param connection_timeout: raise when not online after these seconds
+        :param service_timeout: poll for x seconds when checking services
+        :param logger: logging instance
+        :raise RuntimeError: when the timeout has been reached
+        :return: dict with services mapped by their state
         """
-        return x[0].upper() + x[1:]
+        # neutral_states = ['inactive', 'deactivating']
+        failed_states = ['failed', 'error']
+        active_states = ['active', 'reloading']
+        activating_state = 'activating'
+        start_time = time.time()
+        client = None
+        while client is None:
+            delta = time.time() - start_time
+            if delta > connection_timeout:
+                raise RuntimeError('Idling has timed out after {0}s'.format(delta))
+            try:
+                client = SSHClient(ip, username=username, password=password)
+            except:
+                logger.debug('Could not establish a connection yet to {0} after {1}s'.format(ip, delta))
+            time.sleep(1)
+        ovs_services = [service for service in ServiceManager.list_services(client) if service.startswith('ovs-')]
+        active_services = []
+        failed_service = []
+        activating_services = []
+        # Initially class these services
+        for service in ovs_services:
+            logger.debug('Initially classifying {0}'.format(service))
+            service_state = ServiceManager.get_service_status(service, client)
+            logger.debug('Service {0} - State {1}'.format(service, service_state))
+            if service_state in failed_states:
+                failed_service.append(service)
+            elif service_state in active_states:
+                active_services.append(service)
+            elif service_state == activating_state:
+                activating_services.append(service)
+            else:
+                logger.error('Unable to process service state {0}'.format(service_state))
+        start_time = time.time()
+        while len(activating_services) > 0:
+            if time.time() - start_time > service_timeout:
+                break
+            service = activating_services.pop()
+            service_state = ServiceManager.get_service_status(service, client)
+            if service_state in failed_states:
+                failed_service.append(service)
+            elif service_state in active_states:
+                active_services.append(service)
+            elif service_state == activating_state:
+                activating_services.append(service)
+        return {'active': active_services, 'failed': failed_service, 'activating': activating_services}

helpers/thread.py

@@ -42,6 +42,7 @@ def start_thread_with_event(target, name, args=(), kwargs=None):
         args = args + (event,)
         thread = threading.Thread(target=target, args=tuple(args), kwargs=kwargs)
         thread.setName(str(name))
+        thread.setDaemon(True)
         thread.start()
         return thread, event
 
@@ -52,18 +53,22 @@ def start_thread(target, name, args=(), kwargs=None):
         ThreadHelper.LOGGER.info('Starting thread with target {0}'.format(target))
         thread = threading.Thread(target=target, args=tuple(args), kwargs=kwargs)
         thread.setName(str(name))
+        thread.setDaemon(True)
         thread.start()
         return thread
 
     @staticmethod
-    def stop_evented_threads(thread_pairs, r_semaphore=None, logger=LOGGER):
+    def stop_evented_threads(thread_pairs, r_semaphore=None, logger=LOGGER, timeout=300):
         for thread_pair in thread_pairs:
             if thread_pair[0].isAlive():
                 thread_pair[1].set()
             # Wait again to sync
             logger.info('Syncing threads')
         if r_semaphore is not None:
+            start = time.time()
             while r_semaphore.get_counter() < len(thread_pairs):  # Wait for the number of threads we currently have.
+                if time.time() - start > timeout:
+                    raise RuntimeError('Synching the thread with the r_semaphore has timed out.')
                 time.sleep(0.05)
             r_semaphore.wait()  # Unlock them to let them stop (the object is set -> wont loop)
         # Wait for threads to die

remove/vdisk.py

@@ -30,6 +30,24 @@ class VDiskRemover(object):
     def __init__(self):
         pass
 
+    @staticmethod
+    def remove_vdisks_with_structure(vdisks):
+        """
+        Remove many vdisks at once. Will keep the parent structure in mind
+        :param vdisks: list of vdisks 
+        :return: 
+        """
+        removed_guids = []
+        for vdisk in vdisks:
+            if vdisk.guid in removed_guids:
+                continue
+            if len(vdisk.child_vdisks_guids) > 0:
+                for vdisk_child_guid in vdisk.child_vdisks_guids:
+                    VDiskRemover.remove_vdisk(vdisk_child_guid)
+                    removed_guids.append(vdisk_child_guid)
+            VDiskRemover.remove_vdisk(vdisk.guid)
+            removed_guids.append(vdisk.guid)
+
     @staticmethod
     def remove_snapshot(snapshot_guid, vdisk_name, vpool_name, api, timeout=REMOVE_SNAPSHOT_TIMEOUT):
         """