Vertica Cluster Grow and Shrink

Implements grow and shrink for the vertica datastore. The
number of nodes in the cluster must be greater than the
number required to satisfy the min_ksafety configuration setting.

Change-Id: I65f924042833b199ac96ac446dcc5bc1c8c72612
Implements: bp/vertica-grow-shrink

trove/common/cfg.py

@@ -1129,6 +1129,8 @@ vertica_opts = [
                help='Root controller implementation for Vertica.'),
     cfg.StrOpt('guest_log_exposed_logs', default='',
                help='List of Guest Logs to expose for publishing.'),
+    cfg.IntOpt('min_ksafety', default=0,
+               help='Minimum k-safety setting permitted for vertica clusters'),
 ]
 
 # DB2

trove/common/exception.py

@@ -531,6 +531,12 @@ class ClusterNumInstancesNotLargeEnough(TroveError):
                 "be at least %(num_instances)s.")
 
 
+class ClusterNumInstancesBelowSafetyThreshold(TroveError):
+    message = _("The number of instances in your cluster cannot "
+                "safely be lowered below the current level based"
+                "on your current fault-tolerance settings.")
+
+
 class ClusterShrinkMustNotLeaveClusterEmpty(TroveError):
     message = _("Must leave at least one instance in the cluster when "
                 "shrinking.")

trove/common/strategies/cluster/experimental/vertica/api.py

@@ -21,6 +21,7 @@ from trove.common import cfg
 from trove.common import exception
 from trove.common import remote
 from trove.common.strategies.cluster import base
+from trove.common import utils
 from trove.extensions.mgmt.clusters.views import MgmtClusterView
 from trove.instance import models as inst_models
 from trove.quota.quota import check_quotas
@@ -37,6 +38,25 @@ class VerticaAPIStrategy(base.BaseAPIStrategy):
     def cluster_class(self):
         return VerticaCluster
 
+    def _action_grow(self, cluster, body):
+        nodes = body['grow']
+        instances = []
+        for node in nodes:
+            instance = {
+                'flavor_id': utils.get_id_from_href(node['flavorRef'])
+            }
+            if 'name' in node:
+                instance['name'] = node['name']
+            if 'volume' in node:
+                instance['volume_size'] = int(node['volume']['size'])
+            instances.append(instance)
+        return cluster.grow(instances)
+
+    def _action_shrink(self, cluster, body):
+        nodes = body['shrink']
+        instance_ids = [node['id'] for node in nodes]
+        return cluster.shrink(instance_ids)
+
     @property
     def cluster_view_class(self):
         return VerticaClusterView
@@ -48,15 +68,18 @@ class VerticaAPIStrategy(base.BaseAPIStrategy):
 
 class VerticaCluster(models.Cluster):
 
-    @classmethod
-    def create(cls, context, name, datastore, datastore_version,
-               instances, extended_properties):
-        LOG.debug("Initiating cluster creation.")
+    @staticmethod
+    def _create_instances(context, db_info, datastore, datastore_version,
+                          instances, new_cluster):
         vertica_conf = CONF.get(datastore_version.manager)
         num_instances = len(instances)
 
+        existing = inst_models.DBInstance.find_all(cluster_id=db_info.id).all()
+        num_existing = len(existing)
+
         # Matching number of instances with configured cluster_member_count
-        if num_instances != vertica_conf.cluster_member_count:
+        if new_cluster \
+                and num_instances != vertica_conf.cluster_member_count:
             raise exception.ClusterNumInstancesNotSupported(
                 num_instances=vertica_conf.cluster_member_count)
 
@@ -98,36 +121,119 @@ class VerticaCluster(models.Cluster):
         azs = [instance.get('availability_zone', None)
                for instance in instances]
 
-        # Updating Cluster Task
+        # Creating member instances
+        minstances = []
+        for i in range(0, num_instances):
+            if i == 0 and new_cluster:
+                member_config = {"id": db_info.id, "instance_type": "master"}
+            else:
+                member_config = {"id": db_info.id, "instance_type": "member"}
+            instance_name = "%s-member-%s" % (db_info.name,
+                                              str(i + num_existing + 1))
+            minstances.append(
+                inst_models.Instance.create(context, instance_name,
+                                            flavor_id,
+                                            datastore_version.image_id,
+                                            [], [], datastore,
+                                            datastore_version,
+                                            volume_size, None,
+                                            nics=nics[i],
+                                            availability_zone=azs[i],
+                                            configuration_id=None,
+                                            cluster_config=member_config)
+            )
+        return minstances
+
+    @classmethod
+    def create(cls, context, name, datastore, datastore_version,
+               instances, extended_properties):
+        LOG.debug("Initiating cluster creation.")
+
+        vertica_conf = CONF.get(datastore_version.manager)
+        num_instances = len(instances)
+
+        # Matching number of instances with configured cluster_member_count
+        if num_instances != vertica_conf.cluster_member_count:
+            raise exception.ClusterNumInstancesNotSupported(
+                num_instances=vertica_conf.cluster_member_count)
+
         db_info = models.DBCluster.create(
             name=name, tenant_id=context.tenant,
             datastore_version_id=datastore_version.id,
             task_status=ClusterTasks.BUILDING_INITIAL)
 
-        # Creating member instances
-        for i in range(0, num_instances):
-            if i == 0:
-                member_config = {"id": db_info.id, "instance_type": "master"}
-            else:
-                member_config = {"id": db_info.id, "instance_type": "member"}
-            instance_name = "%s-member-%s" % (name, str(i + 1))
-            inst_models.Instance.create(context, instance_name,
-                                        flavor_id,
-                                        datastore_version.image_id,
-                                        [], [], datastore,
-                                        datastore_version,
-                                        volume_size, None,
-                                        nics=nics[i],
-                                        availability_zone=azs[i],
-                                        configuration_id=None,
-                                        cluster_config=member_config)
-
+        cls._create_instances(context, db_info, datastore, datastore_version,
+                              instances, new_cluster=True)
         # Calling taskmanager to further proceed for cluster-configuration
         task_api.load(context, datastore_version.manager).create_cluster(
             db_info.id)
 
         return VerticaCluster(context, db_info, datastore, datastore_version)
 
+    @staticmethod
+    def k_safety(n):
+        """
+        Vertica defines k-safety values of 0, 1 or 2:
+        https://my.vertica.com/docs/7.1.x/HTML/Content/Authoring/Glossary/
+        K-Safety.htm
+        """
+        if n < 3:
+            return 0
+        elif n < 5:
+            return 1
+        else:
+            return 2
+
+    def grow(self, instances):
+        LOG.debug("Growing cluster.")
+
+        self.validate_cluster_available()
+
+        context = self.context
+        db_info = self.db_info
+        datastore = self.ds
+        datastore_version = self.ds_version
+
+        db_info.update(task_status=ClusterTasks.GROWING_CLUSTER)
+
+        new_instances = self._create_instances(context, db_info, datastore,
+                                               datastore_version, instances,
+                                               new_cluster=False)
+
+        task_api.load(context, datastore_version.manager).grow_cluster(
+            db_info.id, [instance.id for instance in new_instances])
+
+        return VerticaCluster(context, db_info, datastore, datastore_version)
+
+    def shrink(self, instance_ids):
+        self.validate_cluster_available()
+
+        context = self.context
+        db_info = self.db_info
+        datastore_version = self.ds_version
+
+        db_instances = inst_models.DBInstance.find_all(cluster_id=db_info.id,
+                                                       deleted=False).all()
+
+        all_instance_ids = [db_instance.id for db_instance in db_instances]
+
+        left_instances = [instance_id for instance_id
+                          in all_instance_ids
+                          if instance_id not in instance_ids]
+
+        k = self.k_safety(len(left_instances))
+
+        vertica_conf = CONF.get(datastore_version.manager)
+        if k < vertica_conf.min_ksafety:
+            raise exception.ClusterNumInstancesBelowSafetyThreshold()
+
+        db_info.update(task_status=ClusterTasks.SHRINKING_CLUSTER)
+
+        task_api.load(context, datastore_version.manager).shrink_cluster(
+            self.db_info.id, instance_ids)
+        return VerticaCluster(self.context, db_info,
+                              self.ds, self.ds_version)
+
 
 class VerticaClusterView(ClusterView):
 

trove/common/strategies/cluster/experimental/vertica/guestagent.py

@@ -47,6 +47,21 @@ class VerticaGuestAgentAPI(guest_api.API):
         return self._call("install_cluster", CONF.cluster_usage_timeout,
                           self.version_cap, members=members)
 
+    def grow_cluster(self, members):
+        LOG.debug("Growing Vertica cluster with members: %s." % members)
+        return self._call("grow_cluster", CONF.cluster_usage_timeout,
+                          self.version_cap, members=members)
+
+    def shrink_cluster(self, members):
+        LOG.debug("Shrinking Vertica cluster with members: %s." % members)
+        return self._call("shrink_cluster", CONF.cluster_usage_timeout,
+                          self.version_cap, members=members)
+
+    def mark_design_ksafe(self, k):
+        LOG.debug("Setting vertica k-safety level to : %s." % k)
+        return self._call("mark_design_ksafe", CONF.cluster_usage_timeout,
+                          self.version_cap, k=k)
+
     def cluster_complete(self):
         LOG.debug("Notifying cluster install completion.")
         return self._call("cluster_complete", guest_api.AGENT_HIGH_TIMEOUT,

trove/common/strategies/cluster/experimental/vertica/taskmanager.py

@@ -17,6 +17,8 @@ from oslo_log import log as logging
 from trove.common import cfg
 from trove.common.i18n import _
 from trove.common.strategies.cluster import base
+from trove.common.strategies.cluster.experimental.vertica.api import \
+    VerticaCluster
 from trove.instance.models import DBInstance
 from trove.instance.models import Instance
 from trove.taskmanager import api as task_api
@@ -47,7 +49,8 @@ class VerticaClusterTasks(task_models.ClusterTasks):
         def _create_cluster():
 
             # Fetch instances by cluster_id against instances table.
-            db_instances = DBInstance.find_all(cluster_id=cluster_id).all()
+            db_instances = DBInstance.find_all(cluster_id=cluster_id,
+                                               deleted=False).all()
             instance_ids = [db_instance.id for db_instance in db_instances]
 
             # Wait for cluster members to get to cluster-ready status.
@@ -106,6 +109,116 @@ class VerticaClusterTasks(task_models.ClusterTasks):
 
         LOG.debug("End create_cluster for id: %s." % cluster_id)
 
+    def grow_cluster(self, context, cluster_id, new_instance_ids):
+
+        def _grow_cluster():
+            LOG.debug("begin grow_cluster for Vertica cluster %s" % cluster_id)
+
+            db_instances = DBInstance.find_all(cluster_id=cluster_id,
+                                               deleted=False).all()
+
+            instance_ids = [db_instance.id for db_instance in db_instances]
+
+            # Wait for new cluster members to get to cluster-ready status.
+            if not self._all_instances_ready(new_instance_ids, cluster_id):
+                return
+
+            new_insts = [Instance.load(context, instance_id)
+                         for instance_id in new_instance_ids]
+
+            existing_instances = [Instance.load(context, instance_id)
+                                  for instance_id
+                                  in instance_ids
+                                  if instance_id not in new_instance_ids]
+
+            existing_guests = [self.get_guest(i) for i in existing_instances]
+            new_guests = [self.get_guest(i) for i in new_insts]
+            all_guests = new_guests + existing_guests
+
+            authorized_users_without_password = ['root', 'dbadmin']
+            new_ips = [self.get_ip(instance) for instance in new_insts]
+
+            for user in authorized_users_without_password:
+                pub_key = [guest.get_public_keys(user) for guest in all_guests]
+                for guest in all_guests:
+                    guest.authorize_public_keys(user, pub_key)
+
+            for db_instance in db_instances:
+                if db_instance['type'] == 'master':
+                    LOG.debug("Found 'master' instance, calling grow on guest")
+                    master_instance = Instance.load(context,
+                                                    db_instance.id)
+                    self.get_guest(master_instance).grow_cluster(new_ips)
+                    break
+
+            for guest in new_guests:
+                guest.cluster_complete()
+
+        timeout = Timeout(CONF.cluster_usage_timeout)
+
+        try:
+            _grow_cluster()
+            self.reset_task()
+        except Timeout as t:
+            if t is not timeout:
+                raise  # not my timeout
+            LOG.exception(_("Timeout for growing cluster."))
+            self.update_statuses_on_failure(cluster_id)
+        except Exception:
+            LOG.exception(_("Error growing cluster %s.") % cluster_id)
+            self.update_statuses_on_failure(cluster_id)
+        finally:
+            timeout.cancel()
+
+    def shrink_cluster(self, context, cluster_id, instance_ids):
+        def _shrink_cluster():
+            db_instances = DBInstance.find_all(cluster_id=cluster_id,
+                                               deleted=False).all()
+
+            all_instance_ids = [db_instance.id for db_instance in db_instances]
+
+            remove_instances = [Instance.load(context, instance_id)
+                                for instance_id in instance_ids]
+
+            left_instances = [Instance.load(context, instance_id)
+                              for instance_id
+                              in all_instance_ids
+                              if instance_id not in instance_ids]
+
+            remove_member_ips = [self.get_ip(instance)
+                                 for instance in remove_instances]
+
+            k = VerticaCluster.k_safety(len(left_instances))
+
+            for db_instance in db_instances:
+                if db_instance['type'] == 'master':
+                    master_instance = Instance.load(context,
+                                                    db_instance.id)
+                    if self.get_ip(master_instance) in remove_member_ips:
+                        raise RuntimeError(_("Cannot remove master instance!"))
+                    LOG.debug(_("Marking cluster k-safety: %s") % k)
+                    self.get_guest(master_instance).mark_design_ksafe(k)
+                    self.get_guest(master_instance).shrink_cluster(
+                        remove_member_ips)
+                    break
+
+            for r in remove_instances:
+                Instance.delete(r)
+
+        timeout = Timeout(CONF.cluster_usage_timeout)
+        try:
+            _shrink_cluster()
+            self.reset_task()
+        except Timeout as t:
+            if t is not timeout:
+                raise
+            LOG.exception(_("Timeout for shrinking cluster."))
+            self.update_statuses_on_failure(cluster_id)
+        finally:
+            timeout.cancel()
+
+        LOG.debug("end shrink_cluster for Vertica cluster id %s" % self.id)
+
 
 class VerticaTaskManagerAPI(task_api.API):
 

trove/guestagent/datastore/experimental/vertica/manager.py

@@ -109,3 +109,32 @@ class Manager(manager.Manager):
             LOG.exception(_('Cluster installation failed.'))
             self.appStatus.set_status(rd_ins.ServiceStatuses.FAILED)
             raise
+
+    def grow_cluster(self, context, members):
+        try:
+            LOG.debug("Growing cluster to members: %s." % members)
+            self.app.grow_cluster(members)
+            LOG.debug("grow_cluster call has finished.")
+        except Exception:
+            LOG.exception(_('Cluster grow failed.'))
+            self.appStatus.set_status(rd_ins.ServiceStatuses.FAILED)
+            raise
+
+    def shrink_cluster(self, context, members):
+        try:
+            LOG.debug("Shrinking cluster members: %s." % members)
+            self.app.shrink_cluster(members)
+            LOG.debug("shrink_cluster call has finished.")
+        except Exception:
+            LOG.exception(_('Cluster shrink failed.'))
+            self.appStatus.set_status(rd_ins.ServiceStatuses.FAILED)
+            raise
+
+    def mark_design_ksafe(self, context, k):
+        try:
+            LOG.debug("Setting vertica k-safety to %s." % k)
+            self.app.mark_design_ksafe(k)
+        except Exception:
+            LOG.exception(_('K-safety setting failed.'))
+            self.appStatus.set_status(rd_ins.ServiceStatuses.FAILED)
+            raise

trove/guestagent/datastore/experimental/vertica/service.py

@@ -153,6 +153,43 @@ class VerticaApp(object):
         finally:
             self.status.end_restart()
 
+    def add_db_to_node(self, members=netutils.get_my_ipv4()):
+        """Add db to host with admintools"""
+        LOG.info(_("Calling admintools to add DB to host"))
+        try:
+            # Create db after install
+            db_password = self._get_database_password()
+            create_db_command = (system.ADD_DB_TO_NODE % (members,
+                                                          DB_NAME,
+                                                          db_password))
+            system.shell_execute(create_db_command, "dbadmin")
+        except exception.ProcessExecutionError:
+            # Give vertica some time to get the the node up, won't be available
+            # by the time adminTools -t db_add_node completes
+            LOG.info(_("adminTools failed as expected - wait for node"))
+        self.wait_for_node_status()
+        LOG.info(_("Vertica add db to host completed."))
+
+    def remove_db_from_node(self, members=netutils.get_my_ipv4()):
+        """Remove db from node with admintools"""
+        LOG.info(_("Removing db from node"))
+        try:
+            # Create db after install
+            db_password = self._get_database_password()
+            create_db_command = (system.REMOVE_DB_FROM_NODE % (members,
+                                                               DB_NAME,
+                                                               db_password))
+            system.shell_execute(create_db_command, "dbadmin")
+        except exception.ProcessExecutionError:
+            # Give vertica some time to get the the node up, won't be available
+            # by the time adminTools -t db_add_node completes
+            LOG.info(_("adminTools failed as expected - wait for node"))
+
+        # Give vertica some time to take the node down - it won't be available
+        # by the time adminTools -t db_add_node completes
+        self.wait_for_node_status()
+        LOG.info(_("Vertica remove host from db completed."))
+
     def create_db(self, members=netutils.get_my_ipv4()):
         """Prepare the guest machine with a Vertica db creation."""
         LOG.info(_("Creating database on Vertica host."))
@@ -182,6 +219,18 @@ class VerticaApp(object):
         self._generate_database_password()
         LOG.info(_("install_vertica completed."))
 
+    def update_vertica(self, command, members=netutils.get_my_ipv4()):
+        LOG.info(_("Calling update_vertica with command %s") % command)
+        try:
+            update_vertica_cmd = (system.UPDATE_VERTICA % (command, members,
+                                                           MOUNT_POINT))
+            system.shell_execute(update_vertica_cmd)
+        except exception.ProcessExecutionError:
+            LOG.exception(_("update_vertica failed."))
+            raise RuntimeError(_("update_vertica failed."))
+        # self._generate_database_password()
+        LOG.info(_("update_vertica completed."))
+
     def add_udls(self):
         """Load the user defined load libraries into the database."""
         LOG.info(_("Adding configured user defined load libraries."))
@@ -287,6 +336,17 @@ class VerticaApp(object):
             LOG.exception(_("Failed to prepare for install_vertica."))
             raise
 
+    def mark_design_ksafe(self, k):
+        """Wrapper for mark_design_ksafe function for setting k-safety """
+        LOG.info(_("Setting Vertica k-safety to %s") % str(k))
+        out, err = system.exec_vsql_command(self._get_database_password(),
+                                            system.MARK_DESIGN_KSAFE % k)
+        # Only fail if we get an ERROR as opposed to a warning complaining
+        # about setting k = 0
+        if "ERROR" in err:
+            LOG.error(err)
+            raise RuntimeError(_("Failed to set k-safety level %s.") % k)
+
     def _create_user(self, username, password, role):
         """Creates a user, granting and enabling the given role for it."""
         LOG.info(_("Creating user in Vertica database."))
@@ -418,3 +478,41 @@ class VerticaApp(object):
         LOG.debug("Creating database with members: %s." % cluster_members)
         self.create_db(cluster_members)
         LOG.debug("Cluster configured on members: %s." % cluster_members)
+
+    def grow_cluster(self, members):
+        """Adds nodes to cluster."""
+        cluster_members = ','.join(members)
+        LOG.debug("Growing cluster with members: %s." % cluster_members)
+        self.update_vertica("--add-hosts", cluster_members)
+        self._export_conf_to_members(members)
+        LOG.debug("Creating database with members: %s." % cluster_members)
+        self.add_db_to_node(cluster_members)
+        LOG.debug("Cluster configured on members: %s." % cluster_members)
+
+    def shrink_cluster(self, members):
+        """Removes nodes from cluster."""
+        cluster_members = ','.join(members)
+        LOG.debug("Shrinking cluster with members: %s." % cluster_members)
+        self.remove_db_from_node(cluster_members)
+        self.update_vertica("--remove-hosts", cluster_members)
+
+    def wait_for_node_status(self, status='UP'):
+        """Wait until all nodes are the same status"""
+        # select node_state from nodes where node_state <> 'UP'
+        def _wait_for_node_status():
+            out, err = system.exec_vsql_command(self._get_database_password(),
+                                                system.NODE_STATUS % status)
+            LOG.debug("Polled vertica node states: %s" % out)
+
+            if err:
+                LOG.error(err)
+                raise RuntimeError(_("Failed to query for root user."))
+
+            return "0 rows" in out
+
+        try:
+            utils.poll_until(_wait_for_node_status, time_out=600,
+                             sleep_time=15)
+        except exception.PollTimeOut:
+            raise RuntimeError(_("Timed out waiting for cluster to"
+                                 "change to status %s") % status)

trove/guestagent/datastore/experimental/vertica/system.py

@@ -14,6 +14,10 @@
 from trove.common import utils
 
 ALTER_USER_PASSWORD = "ALTER USER %s IDENTIFIED BY '%s'"
+ADD_DB_TO_NODE = ("/opt/vertica/bin/adminTools -t db_add_node -a"
+                  " %s -d %s -p '%s'")
+REMOVE_DB_FROM_NODE = ("/opt/vertica/bin/adminTools -t db_remove_node -s"
+                       " %s -d %s -i -p '%s'")
 CREATE_DB = ("/opt/vertica/bin/adminTools -t create_db -s"
              " %s -d %s -c %s -D %s -p '%s'")
 CREATE_USER = "CREATE USER %s IDENTIFIED BY '%s'"
@@ -21,7 +25,10 @@ ENABLE_FOR_USER = "ALTER USER %s DEFAULT ROLE %s"
 GRANT_TO_USER = "GRANT %s to %s"
 INSTALL_VERTICA = ("/opt/vertica/sbin/install_vertica -s %s"
                    " -d %s -X -N -S default -r"
-                   " /vertica.deb -L CE -Y --no-system-checks")
+                   " /vertica.deb -L CE -Y --no-system-checks"
+                   " --ignore-aws-instance-type")
+MARK_DESIGN_KSAFE = "SELECT MARK_DESIGN_KSAFE(%s)"
+NODE_STATUS = "SELECT node_state FROM nodes where node_state <> '%s'"
 STOP_DB = "/opt/vertica/bin/adminTools -t stop_db -F -d %s -p '%s'"
 START_DB = "/opt/vertica/bin/adminTools -t start_db -d %s -p '%s'"
 STATUS_ACTIVE_DB = "/opt/vertica/bin/adminTools -t show_active_db"
@@ -33,6 +40,18 @@ SEND_CONF_TO_SERVER = ("rsync -v -e 'ssh -o "
                        "StrictHostKeyChecking=no' --perms --owner --group "
                        "%s %s:%s")
 SSH_KEY_GEN = "ssh-keygen -f %s/.ssh/id_rsa -t rsa -N ''"
+UPDATE_VERTICA = ("/opt/vertica/sbin/update_vertica %s %s "
+                  " -d %s -X -N -S default -r"
+                  " /vertica.deb -L CE -Y --no-system-checks"
+                  " --ignore-aws-instance-type")
+UPDATE_REMOVE = ("/opt/vertica/sbin/update_vertica --remove-hosts %s "
+                 " -d %s -X -N -S default -r"
+                 " /vertica.deb -L CE -Y --no-system-checks"
+                 " --ignore-aws-instance-type")
+UPDATE_ADD = ("/opt/vertica/sbin/update_vertica --add-hosts %s "
+              " -d %s -X -N -S default -r"
+              " /vertica.deb -L CE -Y --no-system-checks"
+              " --ignore-aws-instance-type")
 USER_EXISTS = ("/opt/vertica/bin/vsql -w '%s' -c "
                "\"select 1 from users where user_name = '%s'\" "
                "| grep row | awk '{print $1}' | cut -c2-")

trove/tests/unittests/cluster/test_vertica_cluster.py

@@ -70,11 +70,13 @@ class ClusterTest(trove_testtools.TestCase):
                            'instance_type': 'member'},
                           {'volume_size': 1, 'flavor_id': '1234',
                            'instance_type': 'member'}]
+        self.db_instances = [1, 2, 3]
 
     def tearDown(self):
         super(ClusterTest, self).tearDown()
 
-    def test_create_empty_instances(self):
+    @patch.object(inst_models.DBInstance, 'find_all')
+    def test_create_empty_instances(self, *args):
         self.assertRaises(exception.ClusterNumInstancesNotSupported,
                           Cluster.create,
                           Mock(),
@@ -83,7 +85,9 @@ class ClusterTest(trove_testtools.TestCase):
                           self.datastore_version,
                           [], None)
 
-    def test_create_flavor_not_specified(self):
+    @patch.object(DBCluster, 'create')
+    @patch.object(inst_models.DBInstance, 'find_all')
+    def test_create_flavor_not_specified(self, *args):
         instances = self.instances
         instances[0]['flavor_id'] = None
         self.assertRaises(exception.ClusterFlavorsNotEqual,
@@ -96,9 +100,11 @@ class ClusterTest(trove_testtools.TestCase):
                           None
                           )
 
+    @patch.object(DBCluster, 'create')
+    @patch.object(inst_models.DBInstance, 'find_all')
     @patch.object(remote, 'create_nova_client')
-    def test_create_invalid_flavor_specified(self,
-                                             mock_client):
+    def test_create_invalid_flavor_specified(self, mock_client,
+                                             mock_find_all, mock_create):
         instances = [{'flavor_id': '1234'},
                      {'flavor_id': '1234'},
                      {'flavor_id': '1234'}]
@@ -117,9 +123,11 @@ class ClusterTest(trove_testtools.TestCase):
                           None
                           )
 
+    @patch.object(DBCluster, 'create')
+    @patch.object(inst_models.DBInstance, 'find_all')
     @patch.object(remote, 'create_nova_client')
-    def test_create_volume_no_specified(self,
-                                        mock_client):
+    def test_create_volume_no_specified(self, mock_client, mock_find_all,
+                                        mock_create):
         instances = self.instances
         instances[0]['volume_size'] = None
         flavors = Mock()
@@ -134,11 +142,15 @@ class ClusterTest(trove_testtools.TestCase):
                           None
                           )
 
+    @patch.object(DBCluster, 'create')
+    @patch.object(inst_models.DBInstance, 'find_all')
     @patch.object(remote, 'create_nova_client')
     @patch.object(vertica_api, 'CONF')
     def test_create_storage_specified_with_no_volume_support(self,
                                                              mock_conf,
-                                                             mock_client):
+                                                             mock_client,
+                                                             mock_find_all,
+                                                             mock_create):
         mock_conf.get = Mock(
             return_value=FakeOptGroup(volume_support=False))
         instances = self.instances
@@ -155,11 +167,15 @@ class ClusterTest(trove_testtools.TestCase):
                           None
                           )
 
+    @patch.object(DBCluster, 'create')
+    @patch.object(inst_models.DBInstance, 'find_all')
     @patch.object(remote, 'create_nova_client')
     @patch.object(vertica_api, 'CONF')
     def test_create_storage_not_specified_and_no_ephemeral_flavor(self,
                                                                   mock_conf,
-                                                                  mock_client):
+                                                                  mock_client,
+                                                                  m_find_all,
+                                                                  mock_create):
         class FakeFlavor:
             def __init__(self, flavor_id):
                 self.flavor_id = flavor_id
@@ -188,8 +204,11 @@ class ClusterTest(trove_testtools.TestCase):
                           None
                           )
 
+    @patch.object(DBCluster, 'create')
+    @patch.object(inst_models.DBInstance, 'find_all')
     @patch.object(remote, 'create_nova_client')
-    def test_create_volume_not_equal(self, mock_client):
+    def test_create_volume_not_equal(self, mock_client, mock_find_all,
+                                     mock_create):
         instances = self.instances
         instances[0]['volume_size'] = 2
         flavors = Mock()
@@ -204,13 +223,14 @@ class ClusterTest(trove_testtools.TestCase):
                           None
                           )
 
+    @patch.object(inst_models.DBInstance, 'find_all')
     @patch.object(inst_models.Instance, 'create')
     @patch.object(DBCluster, 'create')
     @patch.object(task_api, 'load')
     @patch.object(QUOTAS, 'check_quotas')
     @patch.object(remote, 'create_nova_client')
     def test_create(self, mock_client, mock_check_quotas, mock_task_api,
-                    mock_db_create, mock_ins_create):
+                    mock_db_create, mock_ins_create, mock_find_all):
         instances = self.instances
         flavors = Mock()
         mock_client.return_value.flavors = flavors
@@ -224,6 +244,7 @@ class ClusterTest(trove_testtools.TestCase):
             mock_db_create.return_value.id)
         self.assertEqual(3, mock_ins_create.call_count)
 
+    @patch.object(inst_models.DBInstance, 'find_all')
     @patch.object(vertica_api, 'CONF')
     @patch.object(inst_models.Instance, 'create')
     @patch.object(DBCluster, 'create')
@@ -232,7 +253,8 @@ class ClusterTest(trove_testtools.TestCase):
     @patch.object(remote, 'create_nova_client')
     def test_create_with_ephemeral_flavor(self, mock_client, mock_check_quotas,
                                           mock_task_api, mock_db_create,
-                                          mock_ins_create, mock_conf):
+                                          mock_ins_create, mock_conf,
+                                          mock_find_all):
         class FakeFlavor:
             def __init__(self, flavor_id):
                 self.flavor_id = flavor_id