def _scale_cluster_instances(self, cluster, node_group_id_map):
        ctx = context.ctx()

        aa_group = None
        old_aa_groups = None
        if cluster.anti_affinity:
            aa_group = self._find_aa_server_group(cluster)
            if not aa_group:
                old_aa_groups = self._generate_anti_affinity_groups(cluster)

        instances_to_delete = []
        node_groups_to_enlarge = set()
        node_groups_to_delete = set()

        for node_group in cluster.node_groups:
            new_count = node_group_id_map[node_group.id]

            if new_count < node_group.count:
                instances_to_delete += node_group.instances[new_count:
                                                            node_group.count]
                if new_count == 0:
                    node_groups_to_delete.add(node_group.id)
            elif new_count > node_group.count:
                node_groups_to_enlarge.add(node_group.id)
                if node_group.count == 0 and node_group.auto_security_group:
                    self._create_auto_security_group(node_group)

        if instances_to_delete:
            cluster = c_u.change_cluster_status(
                cluster, c_u.CLUSTER_STATUS_DELETING_INSTANCES)

            for instance in instances_to_delete:
                with context.set_current_instance_id(instance.instance_id):
                    self._shutdown_instance(instance)

        self._await_deleted(cluster, instances_to_delete)
        for ng in cluster.node_groups:
            if ng.id in node_groups_to_delete:
                self._delete_auto_security_group(ng)

        cluster = conductor.cluster_get(ctx, cluster)
        instances_to_add = []
        if node_groups_to_enlarge:

            cpo.add_provisioning_step(
                cluster.id, _("Add instances"),
                self._count_instances_to_scale(
                    node_groups_to_enlarge, node_group_id_map, cluster))

            cluster = c_u.change_cluster_status(
                cluster, c_u.CLUSTER_STATUS_ADDING_INSTANCES)
            for ng in cluster.node_groups:
                if ng.id in node_groups_to_enlarge:
                    count = node_group_id_map[ng.id]
                    for idx in six.moves.xrange(ng.count + 1, count + 1):
                        instance_id = self._start_instance(
                            cluster, ng, idx, aa_group, old_aa_groups)
                        instances_to_add.append(instance_id)

        return instances_to_add

    def start_cloudera_agents(self, instances):
        # instances non-empty
        cpo.add_provisioning_step(instances[0].cluster_id, _("Start Cloudera Agents"), len(instances))

        with context.ThreadGroup() as tg:
            for i in instances:
                tg.spawn("cdh-agent-start-%s" % i.instance_name, self._start_cloudera_agent, i)

    def _provision_cluster(self, name, cluster_spec, ambari_info,
                           servers, version):
        # TODO(jspeidel): encapsulate in another class

        if servers:
            cpo.add_provisioning_step(
                servers[0].cluster_id,
                _("Provision cluster via Ambari"), len(servers))

        with context.ThreadGroup() as tg:
            for server in servers:
                with context.set_current_instance_id(
                        server.instance['instance_id']):
                    tg.spawn(
                        "hdp-provision-instance-%s" %
                        server.instance.hostname(),
                        server.provision_ambari, ambari_info, cluster_spec)

        handler = self.version_factory.get_version_handler(version)
        ambari_client = handler.get_ambari_client()

        ambari_client.wait_for_host_registrations(len(servers), ambari_info)
        self._set_ambari_credentials(cluster_spec, ambari_info, version)

        ambari_client.provision_cluster(
            cluster_spec, servers, ambari_info, name)

        LOG.info(_LI('Cluster provisioned via Ambari Server: {server_ip}')
                 .format(server_ip=ambari_info.get_address()))

    def _await_networks(self, cluster, instances):
        if not instances:
            return

        cpo.add_provisioning_step(cluster.id, _("Assign IPs"), len(instances))

        ips_assigned = set()
        self._ips_assign(ips_assigned, cluster, instances)

        LOG.info(
            _LI("Cluster {cluster_id}: all instances have IPs assigned")
            .format(cluster_id=cluster.id))

        cluster = conductor.cluster_get(context.ctx(), cluster)
        instances = g.get_instances(cluster, ips_assigned)

        cpo.add_provisioning_step(
            cluster.id, _("Wait for instance accessibility"), len(instances))

        with context.ThreadGroup() as tg:
            for instance in instances:
                tg.spawn("wait-for-ssh-%s" % instance.instance_name,
                         self._wait_until_accessible, instance)

        LOG.info(_LI("Cluster {cluster_id}: all instances are accessible")
                 .format(cluster_id=cluster.id))

    def install_packages(self, instances, packages):
        # instances non-empty
        cpo.add_provisioning_step(instances[0].cluster_id, _("Install packages"), len(instances))

        with context.ThreadGroup() as tg:
            for i in instances:
                tg.spawn("cdh-inst-pkgs-%s" % i.instance_name, self._install_pkgs, i, packages)

def mount_to_instances(instances):
    if len(instances) == 0:
        return

    use_xfs = _can_use_xfs(instances)

    for instance in instances:
        with context.set_current_instance_id(instance.instance_id):
            devices = _find_instance_devices(instance)

            if devices:
                cpo.add_provisioning_step(
                    instance.cluster_id,
                    _("Mount volumes to {inst_name} instance").format(
                        inst_name=instance.instance_name), len(devices))

                formatted_devices = []
                lock = threading.Lock()
                with context.ThreadGroup() as tg:
                    # Since formating can take several minutes (for large
                    # disks) and can be done in parallel, launch one thread
                    # per disk.
                    for device in devices:
                        tg.spawn('format-device-%s' % device, _format_device,
                                 instance, device, use_xfs, formatted_devices,
                                 lock)

                conductor.instance_update(
                    context.current(), instance,
                    {"storage_devices_number": len(formatted_devices)})
                for idx, dev in enumerate(formatted_devices):
                    _mount_volume_to_node(instance, idx+1, dev, use_xfs)

    def start_secondarynamenodes(self, cluster):
        snns = vu.get_secondarynamenodes(cluster)
        if len(snns) == 0:
            return
        cpo.add_provisioning_step(cluster.id, utils.start_process_event_message("SecondaryNameNodes"), len(snns))

        for snn in snns:
            self._start_secondarynamenode(snn)

    def install_swift_integration(self, servers):
        if servers:
            cpo.add_provisioning_step(
                servers[0].cluster_id, _("Install Swift integration"),
                len(servers))

        for server in servers:
            server.install_swift_integration()

 def configure_os(self, instances):
     # instances non-empty
     cpo.add_provisioning_step(
         instances[0].cluster_id, _("Configure OS"), len(instances))
     with context.ThreadGroup() as tg:
         for inst in instances:
             tg.spawn('cdh-repo-conf-%s' % inst.instance_name,
                      self._configure_repo_from_inst, inst)

 def update_configs(self, instances):
     # instances non-empty
     cpo.add_provisioning_step(
         instances[0].cluster_id, _("Update configs"), len(instances))
     with context.ThreadGroup() as tg:
         for instance in instances:
             tg.spawn("update-configs-%s" % instance.instance_name,
                      self._update_configs, instance)

    def _await_deleted(self, cluster, instances):
        """Await all instances are deleted."""
        if not instances:
            return
        cpo.add_provisioning_step(
            cluster.id, _("Wait for instances to be deleted"), len(instances))

        deleted_ids = set()
        self._check_deleted(deleted_ids, cluster, instances)

def configure_instances(pctx, instances):
    if len(instances) == 0:
        return

    cpo.add_provisioning_step(instances[0].cluster_id, _("Configure instances"), len(instances))

    for instance in instances:
        with context.set_current_instance_id(instance.instance_id):
            _configure_instance(pctx, instance)

def configure_instances(pctx, instances):
    if len(instances) == 0:
        return

    cpo.add_provisioning_step(
        instances[0].cluster_id, _("Configure instances"), len(instances))

    for instance in instances:
        _configure_instance(pctx, instance)

    def configure_swift(self, cluster, instances=None):
        if self.c_helper.is_swift_enabled(cluster):
            if not instances:
                instances = u.get_instances(cluster)
            cpo.add_provisioning_step(cluster.id, _("Configure Swift"), len(instances))

            with context.ThreadGroup() as tg:
                for i in instances:
                    tg.spawn("cdh-swift-conf-%s" % i.instance_name, self._configure_swift_to_inst, i)
            swift_helper.install_ssl_certs(instances)

    def install_swift_integration(self, servers):
        if servers:
            cpo.add_provisioning_step(
                servers[0].cluster_id, _("Install Swift integration"),
                len(servers))

        for server in servers:
            with context.set_current_instance_id(
                    server.instance['instance_id']):
                server.install_swift_integration()

    def _start_slave_processes(self, sl_instances):
        if len(sl_instances) == 0:
            return
        cpo.add_provisioning_step(
            sl_instances[0].cluster_id, utils.start_process_event_message("Slave"), len(sl_instances)
        )

        with context.ThreadGroup() as tg:
            for i in sl_instances:
                tg.spawn("storm-start-sl-%s" % i.instance_name, self._start_slaves, i)

    def _start_datanode_processes(self, dn_instances):
        if len(dn_instances) == 0:
            return

        cpo.add_provisioning_step(
            dn_instances[0].cluster_id, utils.start_process_event_message("DataNodes"), len(dn_instances)
        )

        with context.ThreadGroup() as tg:
            for i in dn_instances:
                tg.spawn("spark-start-dn-%s" % i.instance_name, self._start_datanode, i)

    def _start_zookeeper_processes(self, zk_instances):
        if len(zk_instances) == 0:
            return

        cpo.add_provisioning_step(
            zk_instances[0].cluster_id, utils.start_process_event_message("Zookeeper"), len(zk_instances)
        )

        with context.ThreadGroup() as tg:
            for i in zk_instances:
                tg.spawn("storm-start-zk-%s" % i.instance_name, self._start_zookeeper, i)

 def _configure_topology_for_cluster(self, cluster, servers):
     if CONF.enable_data_locality:
         cpo.add_provisioning_step(
             cluster.id, _("Enable data locality for cluster"),
             len(servers))
         topology_data = th.generate_topology_map(
             cluster, CONF.enable_hypervisor_awareness)
         topology_str = "\n".join(
             [k + " " + v for k, v in topology_data.items()]) + "\n"
         for server in servers:
             server.configure_topology(topology_str)

        def wrapped(*args, **kwargs):
            cluster_context = _find_argument(
                cluster_context_reference, *args, **kwargs)
            instances = _find_argument(instances_reference, *args, **kwargs)

            cluster_id = cluster_context.cluster.id
            instance_count = len(instances)

            cpo.add_provisioning_step(cluster_id, name, instance_count)

            return function(*args, **kwargs)