def create(cls, context, name, datastore, datastore_version,
               instances, extended_properties, locality):
        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)

        cls._create_instances(context, db_info, datastore, datastore_version,
                              instances, extended_properties, locality,
                              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)

    def test_load_api(self, get_client_mock):
        context = Mock()
        manager = 'mongodb'

        self.assertIsInstance(task_api.load(context), task_api.API)
        self.assertIsInstance(task_api.load(context, manager),
                              MongoDbTaskManagerAPI)

    def test_load_api(self):
        context = Mock()
        manager = 'mongodb'

        self.assertTrue(isinstance(task_api.load(context), task_api.API))
        self.assertTrue(isinstance(task_api.load(context, manager),
                                   MongoDbTaskManagerAPI))

    def shrink(self, removal_ids):
        LOG.debug("Processing a request for shrinking cluster: %s" % self.id)

        self.validate_cluster_available()

        context = self.context
        db_info = self.db_info
        datastore = self.ds
        datastore_version = self.ds_version

        # we have to make sure that there are nodes with all the required
        # services left after the shrink
        remaining_instance_services = self._get_remaining_instance_services(
            db_info.id, removal_ids)
        try:
            self.validate_instance_types(
                remaining_instance_services, datastore_version.manager)
        except exception.ClusterInstanceTypeMissing as ex:
            raise exception.TroveError(
                _("The remaining instances would not be valid: %s") % str(ex))

        db_info.update(task_status=ClusterTasks.SHRINKING_CLUSTER)

        task_api.load(context, datastore_version.manager).shrink_cluster(
            db_info.id, removal_ids)

        return CouchbaseCluster(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

        for db_instance in self.db_instances:
            if db_instance.type == 'master':
                if db_instance.id in instance_ids:
                    raise exception.ClusterShrinkInstanceInUse(
                        id=db_instance.id,
                        reason="Cannot remove master node."
                    )

        all_instance_ids = [db_instance.id for db_instance
                            in self.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)

    def grow(self, instances):
        LOG.debug("Growing cluster %s." % self.id)

        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)
        try:
            # Get the network of the existing cluster instances.
            interface_ids = self._get_cluster_network_interfaces()
            for instance in instances:
                instance["nics"] = interface_ids

            locality = srv_grp.ServerGroup.convert_to_hint(self.server_group)
            new_instances = self._create_instances(
                context, db_info, datastore, datastore_version, instances,
                None, locality)

            task_api.load(context, datastore_version.manager).grow_cluster(
                db_info.id, [instance.id for instance in new_instances])
        except Exception:
            db_info.update(task_status=ClusterTasks.NONE)

        return self.__class__(context, db_info,
                              datastore, datastore_version)

    def create(cls, context, name, datastore, datastore_version,
               instances, extended_properties, locality, configuration):
        LOG.debug("Processing a request for creating a new cluster.")

        if configuration:
            raise exception.ConfigurationNotSupported()

        cls.validate_instance_types(instances, datastore_version.manager,
                                    for_grow=False)

        # Updating Cluster Task.
        db_info = models.DBCluster.create(
            name=name, tenant_id=context.tenant,
            datastore_version_id=datastore_version.id,
            task_status=ClusterTasks.BUILDING_INITIAL)

        cls._create_cluster_instances(
            context, db_info.id, db_info.name,
            datastore, datastore_version, instances, extended_properties,
            locality)

        # Calling taskmanager to further proceed for cluster-configuration.
        task_api.load(context, datastore_version.manager).create_cluster(
            db_info.id)

        return CouchbaseCluster(context, db_info, datastore, datastore_version)

    def grow(self, instances):
        LOG.debug("Growing cluster %s." % self.id)

        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)
        try:
            locality = srv_grp.ServerGroup.convert_to_hint(self.server_group)
            configuration_id = self.db_info.configuration_id
            new_instances = self._create_instances(
                context, db_info, datastore, datastore_version, instances,
                None, locality, configuration_id)

            task_api.load(context, datastore_version.manager).grow_cluster(
                db_info.id, [instance.id for instance in new_instances])
        except Exception:
            db_info.update(task_status=ClusterTasks.NONE)
            raise

        return self.__class__(context, db_info,
                              datastore, datastore_version)

    def test_load_api(self, get_client_mock):
        context = trove_testtools.TroveTestContext(self)
        manager = 'mongodb'

        self.assertTrue(isinstance(task_api.load(context), task_api.API))
        self.assertTrue(isinstance(task_api.load(context, manager),
                                   MongoDbTaskManagerAPI))

    def rolling_restart(self):
        self.validate_cluster_available()
        self.db_info.update(task_status=ClusterTasks.RESTARTING_CLUSTER)
        try:
            cluster_id = self.db_info.id
            task_api.load(self.context, self.ds_version.manager
                          ).restart_cluster(cluster_id)
        except Exception:
            self.db_info.update(task_status=ClusterTasks.NONE)
            raise

        return self.__class__(self.context, self.db_info,
                              self.ds, self.ds_version)

    def add_shard(self):

        if self.db_info.task_status != ClusterTasks.NONE:
            current_task = self.db_info.task_status.name
            msg = _("This action cannot be performed on the cluster while "
                    "the current cluster task is '%s'.") % current_task
            LOG.error(msg)
            raise exception.UnprocessableEntity(msg)

        db_insts = inst_models.DBInstance.find_all(cluster_id=self.id,
                                                   type='member').all()
        num_unique_shards = len(set([db_inst.shard_id for db_inst
                                     in db_insts]))
        arbitrary_shard_id = db_insts[0].shard_id
        members_in_shard = [db_inst for db_inst in db_insts
                            if db_inst.shard_id == arbitrary_shard_id]
        num_members_per_shard = len(members_in_shard)
        a_member = inst_models.load_any_instance(self.context,
                                                 members_in_shard[0].id)
        deltas = {'instances': num_members_per_shard}
        volume_size = a_member.volume_size
        if volume_size:
            deltas['volumes'] = volume_size * num_members_per_shard
        check_quotas(self.context.tenant, deltas)
        new_replica_set_name = "rs" + str(num_unique_shards + 1)
        new_shard_id = utils.generate_uuid()
        member_config = {"id": self.id,
                         "shard_id": new_shard_id,
                         "instance_type": "member",
                         "replica_set_name": new_replica_set_name}
        for i in range(1, num_members_per_shard + 1):
            instance_name = "%s-%s-%s" % (self.name, new_replica_set_name,
                                          str(i))
            inst_models.Instance.create(self.context, instance_name,
                                        a_member.flavor_id,
                                        a_member.datastore_version.image_id,
                                        [], [], a_member.datastore,
                                        a_member.datastore_version,
                                        volume_size, None,
                                        availability_zone=None,
                                        nics=None,
                                        configuration_id=None,
                                        cluster_config=member_config)

        self.update_db(task_status=ClusterTasks.ADDING_SHARD)
        manager = (datastore_models.DatastoreVersion.
                   load_by_uuid(db_insts[0].datastore_version_id).manager)
        task_api.load(self.context, manager).mongodb_add_shard_cluster(
            self.id,
            new_shard_id,
            new_replica_set_name)

    def shrink(self, instances):
        """Removes instances from a cluster."""
        LOG.debug("Shrinking cluster %s." % self.id)

        self.validate_cluster_available()
        removal_instances = [Instance.load(self.context, inst_id) for inst_id in instances]
        db_instances = DBInstance.find_all(cluster_id=self.db_info.id).all()
        if len(db_instances) - len(removal_instances) < 1:
            raise exception.ClusterShrinkMustNotLeaveClusterEmpty()

        self.db_info.update(task_status=ClusterTasks.SHRINKING_CLUSTER)
        task_api.load(self.context, self.ds_version.manager).shrink_cluster(
            self.db_info.id, [instance.id for instance in removal_instances]
        )

        return PXCCluster(self.context, self.db_info, self.ds, self.ds_version)

    def shrink(self, removal_ids):
        LOG.debug("Processing a request for shrinking cluster: %s" % self.id)

        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.SHRINKING_CLUSTER)

        task_api.load(context, datastore_version.manager).shrink_cluster(
            db_info.id, removal_ids)

        return CouchbaseCluster(context, db_info, datastore, datastore_version)

    def create(cls, context, name, datastore, datastore_version, instances, extended_properties):
        LOG.debug("Initiating PXC cluster creation.")
        cls._validate_cluster_instances(context, instances, datastore, datastore_version)
        # Updating Cluster Task
        db_info = models.DBCluster.create(
            name=name,
            tenant_id=context.tenant,
            datastore_version_id=datastore_version.id,
            task_status=ClusterTasks.BUILDING_INITIAL,
        )

        cls._create_instances(context, db_info, datastore, datastore_version, instances)

        # Calling taskmanager to further proceed for cluster-configuration
        task_api.load(context, datastore_version.manager).create_cluster(db_info.id)

        return PXCCluster(context, db_info, datastore, datastore_version)

    def rolling_upgrade(self, datastore_version):
        """Upgrades a cluster to a new datastore version."""
        LOG.debug("Upgrading cluster %s." % self.id)

        self.validate_cluster_available()
        self.db_info.update(task_status=ClusterTasks.UPGRADING_CLUSTER)
        try:
            cluster_id = self.db_info.id
            ds_ver_id = datastore_version.id
            task_api.load(self.context, self.ds_version.manager
                          ).upgrade_cluster(cluster_id, ds_ver_id)
        except Exception:
            self.db_info.update(task_status=ClusterTasks.NONE)
            raise

        return self.__class__(self.context, self.db_info,
                              self.ds, self.ds_version)

    def create(cls, context, name, datastore, datastore_version,
               instances, extended_properties):
        LOG.debug("Processing a request for creating a new cluster.")

        # Updating Cluster Task.
        db_info = models.DBCluster.create(
            name=name, tenant_id=context.tenant,
            datastore_version_id=datastore_version.id,
            task_status=ClusterTasks.BUILDING_INITIAL)

        cls._create_cluster_instances(
            context, db_info.id, db_info.name,
            datastore, datastore_version, instances, extended_properties)

        # Calling taskmanager to further proceed for cluster-configuration.
        task_api.load(context, datastore_version.manager).create_cluster(
            db_info.id)

        return CouchbaseCluster(context, db_info, datastore, datastore_version)

    def grow(self, instances):
        LOG.debug("Processing a request for growing cluster: %s" % self.id)

        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_cluster_instances(
            context, db_info.id, db_info.name, datastore, datastore_version,
            instances)

        task_api.load(context, datastore_version.manager).grow_cluster(
            db_info.id, [instance.id for instance in new_instances])

        return CouchbaseCluster(context, db_info, datastore, datastore_version)

    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 grow(self, instances):
        LOG.debug("Growing cluster %s." % self.id)

        self.validate_cluster_available()

        context = self.context
        db_info = self.db_info
        datastore = self.ds
        datastore_version = self.ds_version

        # Get the network of the existing cluster instances.
        interface_ids = self._get_cluster_network_interfaces()
        for instance in instances:
            instance["nics"] = interface_ids

        db_info.update(task_status=ClusterTasks.GROWING_CLUSTER)

        new_instances = self._create_instances(context, db_info, datastore, datastore_version, instances)

        task_api.load(context, datastore_version.manager).grow_cluster(
            db_info.id, [instance.id for instance in new_instances]
        )

        return PXCCluster(context, db_info, datastore, datastore_version)

    def _prep_resize(self):
        """Get information about the cluster's current state."""
        if self.db_info.task_status != ClusterTasks.NONE:
            current_task = self.db_info.task_status.name
            msg = (
                _("This action cannot be performed on the cluster while " "the current cluster task is '%s'.")
                % current_task
            )
            LOG.error(msg)
            raise exception.UnprocessableEntity(msg)

        def _instances_of_type(instance_type):
            return [db_inst for db_inst in self.db_instances if db_inst.type == instance_type]

        self.config_svrs = _instances_of_type("config_server")
        self.query_routers = _instances_of_type("query_router")
        self.members = _instances_of_type("member")
        self.shard_ids = set([member.shard_id for member in self.members])
        self.arbitrary_query_router = inst_models.load_any_instance(self.context, self.query_routers[0].id)
        self.manager = task_api.load(self.context, self.datastore_version.manager)