(
        self, new_target_capacity: float, group_targets: Optional[Mapping[str, float]],
    )
return marked_nodes
Choose nodes to kill in order to decrease the capacity on the cluster.

        The number of tasks killed is limited by self.max_tasks_to_kill, and the nodes are terminated in an order
        which (hopefully) reduces the impact on jobs running on the cluster.

        :param new_target_capacity: The total new target capacity for the pool. Most of the time, this is equal to
            self.target_capacity, but in some situations (such as when all resource groups are stale),
            modify_target_capacity cannot make self.target_capacity equal new_target_capacity. We'd rather this method
            aim for the actual target value.
        :param group_targets: a list of new resource group target_capacities; if None, use the existing
            target_capacities (this parameter is necessary in order for dry runs to work correctly)
        :returns: a dict of resource group ids -> list of nodes to terminate
Choose nodes to kill in order to decrease the capacity on the cluster.
Choose nodes to kill in order to decrease the capacity on the cluster .
def _choose_nodes_to_prune(
        self, new_target_capacity: float, group_targets: Optional[Mapping[str, float]],
    ) -> Mapping[str, List[ClusterNodeMetadata]]:
        """ Choose nodes to kill in order to decrease the capacity on the cluster.

        The number of tasks killed is limited by self.max_tasks_to_kill, and the nodes are terminated in an order
        which (hopefully) reduces the impact on jobs running on the cluster.

        :param new_target_capacity: The total new target capacity for the pool. Most of the time, this is equal to
            self.target_capacity, but in some situations (such as when all resource groups are stale),
            modify_target_capacity cannot make self.target_capacity equal new_target_capacity. We'd rather this method
            aim for the actual target value.
        :param group_targets: a list of new resource group target_capacities; if None, use the existing
            target_capacities (this parameter is necessary in order for dry runs to work correctly)
        :returns: a dict of resource group ids -> list of nodes to terminate
        """

        # If dry_run is True in modify_target_capacity, the resource group target_capacity values will not have changed,
        # so this function would not choose to terminate any nodes (see case #2 in the while loop below).  So
        # instead we take a list of new target capacities to use in this computation.
        #
        # We leave the option for group_targets to be None in the event that we want to call
        # prune_excess_fulfilled_capacity outside the context of a modify_target_capacity call
        if not group_targets:
            group_targets = {group_id: rg.target_capacity for group_id, rg in self.resource_groups.items()}

        curr_capacity = self.fulfilled_capacity
        # we use new_target_capacity instead of self.target_capacity here in case they are different (see docstring)
        if curr_capacity <= new_target_capacity:
            return {}

        prioritized_killable_nodes = self._get_prioritized_killable_nodes()
        logger.info(
            "Killable instance IDs in kill order:\n{instance_ids}".format(
                instance_ids=[node_metadata.instance.instance_id for node_metadata in prioritized_killable_nodes],
            )
        )

        if not prioritized_killable_nodes:
            return {}
        rem_group_capacities = {group_id: rg.fulfilled_capacity for group_id, rg in self.resource_groups.items()}

        # How much capacity is actually up and available in Mesos.
        remaining_non_orphan_capacity = self.non_orphan_fulfilled_capacity

        # Iterate through all of the idle agents and mark one at a time for removal until we reach our target capacity
        # or have reached our limit of tasks to kill.
        marked_nodes: Mapping[str, List[ClusterNodeMetadata]] = defaultdict(list)
        removed_weight, killed_task_count = 0.0, 0
        for node_metadata in prioritized_killable_nodes:
            # Try to mark the node for removal; this could fail in a few different ways:
            #  0) We've gone over our limit for max weight to remove
            #  1) The resource group the node belongs to can't be reduced further.
            #  2) Killing the node's tasks would take over the maximum number of tasks we are willing to kill.
            #  3) Killing the node would bring us under our target_capacity of non-orphaned nodes.
            # In each of the cases, the node has been removed from consideration and we jump to the next iteration.

            instance_id = node_metadata.instance.instance_id
            group_id = node_metadata.instance.group_id
            instance_weight = node_metadata.instance.weight

            new_group_capacity = rem_group_capacities[group_id] - instance_weight
            if instance_weight + removed_weight > self.max_weight_to_remove:  # case 0
                logger.info(
                    f"Killing instance {instance_id} with weight {instance_weight} would take us "
                    f"over our max_weight_to_remove of {self.max_weight_to_remove}. Skipping this instance."
                )
                continue

            if new_group_capacity < group_targets[group_id]:  # case 1
                logger.info(f"Resource group {group_id} is at target capacity; skipping {instance_id}")
                continue

            if killed_task_count + node_metadata.agent.task_count > self.max_tasks_to_kill:  # case 2
                logger.info(
                    f"Killing instance {instance_id} with {node_metadata.agent.task_count} tasks would take us "
                    f"over our max_tasks_to_kill of {self.max_tasks_to_kill}. Skipping this instance."
                )
                continue

            if node_metadata.agent.state != AgentState.ORPHANED:
                if remaining_non_orphan_capacity - instance_weight < new_target_capacity:  # case 3
                    logger.info(
                        f"Killing instance {instance_id} with weight {instance_weight} would take us under "
                        f"our target_capacity for non-orphan boxes. Skipping this instance."
                    )
                    continue

            logger.info(f"marking {instance_id} for termination")
            marked_nodes[group_id].append(node_metadata)
            rem_group_capacities[group_id] -= instance_weight
            curr_capacity -= instance_weight
            killed_task_count += node_metadata.agent.task_count
            removed_weight += instance_weight
            if node_metadata.agent.state != AgentState.ORPHANED:
                remaining_non_orphan_capacity -= instance_weight

            if curr_capacity <= new_target_capacity:
                logger.info("Seems like we've picked enough nodes to kill; finishing")
                break

        return marked_nodes
def _choose_nodes_to_prune ( self , new_target_capacity : float , group_targets : Optional [ Mapping [ str , float ] ] , ) -> Mapping [ str , List [ ClusterNodeMetadata ] ] : # If dry_run is True in modify_target_capacity, the resource group target_capacity values will not have changed, # so this function would not choose to terminate any nodes (see case #2 in the while loop below).  So # instead we take a list of new target capacities to use in this computation. # # We leave the option for group_targets to be None in the event that we want to call # prune_excess_fulfilled_capacity outside the context of a modify_target_capacity call if not group_targets : group_targets = { group_id : rg . target_capacity for group_id , rg in self . resource_groups . items ( ) } curr_capacity = self . fulfilled_capacity # we use new_target_capacity instead of self.target_capacity here in case they are different (see docstring) if curr_capacity <= new_target_capacity : return { } prioritized_killable_nodes = self . _get_prioritized_killable_nodes ( ) logger . info ( "Killable instance IDs in kill order:\n{instance_ids}" . format ( instance_ids = [ node_metadata . instance . instance_id for node_metadata in prioritized_killable_nodes ] , ) ) if not prioritized_killable_nodes : return { } rem_group_capacities = { group_id : rg . fulfilled_capacity for group_id , rg in self . resource_groups . items ( ) } # How much capacity is actually up and available in Mesos. remaining_non_orphan_capacity = self . non_orphan_fulfilled_capacity # Iterate through all of the idle agents and mark one at a time for removal until we reach our target capacity # or have reached our limit of tasks to kill. marked_nodes : Mapping [ str , List [ ClusterNodeMetadata ] ] = defaultdict ( list ) removed_weight , killed_task_count = 0.0 , 0 for node_metadata in prioritized_killable_nodes : # Try to mark the node for removal; this could fail in a few different ways: #  0) We've gone over our limit for max weight to remove #  1) The resource group the node belongs to can't be reduced further. #  2) Killing the node's tasks would take over the maximum number of tasks we are willing to kill. #  3) Killing the node would bring us under our target_capacity of non-orphaned nodes. # In each of the cases, the node has been removed from consideration and we jump to the next iteration. instance_id = node_metadata . instance . instance_id group_id = node_metadata . instance . group_id instance_weight = node_metadata . instance . weight new_group_capacity = rem_group_capacities [ group_id ] - instance_weight if instance_weight + removed_weight > self . max_weight_to_remove : # case 0 logger . info ( f"Killing instance {instance_id} with weight {instance_weight} would take us " f"over our max_weight_to_remove of {self.max_weight_to_remove}. Skipping this instance." ) continue if new_group_capacity < group_targets [ group_id ] : # case 1 logger . info ( f"Resource group {group_id} is at target capacity; skipping {instance_id}" ) continue if killed_task_count + node_metadata . agent . task_count > self . max_tasks_to_kill : # case 2 logger . info ( f"Killing instance {instance_id} with {node_metadata.agent.task_count} tasks would take us " f"over our max_tasks_to_kill of {self.max_tasks_to_kill}. Skipping this instance." ) continue if node_metadata . agent . state != AgentState . ORPHANED : if remaining_non_orphan_capacity - instance_weight < new_target_capacity : # case 3 logger . info ( f"Killing instance {instance_id} with weight {instance_weight} would take us under " f"our target_capacity for non-orphan boxes. Skipping this instance." ) continue logger . info ( f"marking {instance_id} for termination" ) marked_nodes [ group_id ] . append ( node_metadata ) rem_group_capacities [ group_id ] -= instance_weight curr_capacity -= instance_weight killed_task_count += node_metadata . agent . task_count removed_weight += instance_weight if node_metadata . agent . state != AgentState . ORPHANED : remaining_non_orphan_capacity -= instance_weight if curr_capacity <= new_target_capacity : logger . info ( "Seems like we've picked enough nodes to kill; finishing" ) break return marked_nodes