Use of temporarily available computing nodes for dynamic scaling of a cluster

• US 8,719,415 B1
• Filed: 06/28/2010
• Issued: 05/06/2014
• Est. Priority Date: 06/28/2010
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Sirota, Peter, Cole, Richard J., Aggarwal, Vaibhav, Hitchcock, Andrew J., Vasanth, Jai
• Primary Examiner(s): Moise, Emmanuel L
• Assistant Examiner(s): ABU ROUMI, MAHRAN Y
• Application Number:
  US12/825,097
• Time in Patent Office: 1,408 Days
• Field of Search: 709/221, 709/226
• US Class Current: 709/226
• CPC Class Codes: G06F 2209/5011 Pool
• CPC Class Codes: G06F 9/4881 Scheduling strategies for dispatcher, e.g. round robin, multi-level priority queues
• CPC Class Codes: G06F 9/5011 the resources being hardware resources other than CPUs, Servers and Terminals
• CPC Class Codes: G06F 9/5061 Partitioning or combining of resources
• CPC Class Codes: H04L 67/10 in which an application is distributed across nodes in the network software deployment G06F8/60; multiprogramming arrangements G06F9/46

Abstract

Techniques are described for managing distributed execution of programs, including by dynamically scaling a cluster of multiple computing nodes performing ongoing distributed execution of a program, such as to increase and/or decrease computing node quantity. An architecture may be used that has core nodes that each participate in a distributed storage system for the distributed program execution, and that has one or more other auxiliary nodes that do not participate in the distributed storage system. Furthermore, as part of performing the dynamic scaling of a cluster, computing nodes that are only temporarily available may be selected and used, such as computing nodes that might be removed from the cluster during the ongoing program execution to be put to other uses and that may also be available for a different fee (e.g., a lower fee) than other computing nodes that are available throughout the ongoing use of the cluster.

Claims

1. A computer-implemented method comprising:

receiving, by one or more configured computing systems of a distributed program execution service, configuration information regarding execution of an indicated program in a distributed manner that includes executing a plurality of jobs of the indicated program;

determining, by the one or more configured computing systems, multiple computing nodes for use in a cluster to perform the execution of the indicated program in accordance with the received configuration information, the multiple computing nodes for the cluster including a first group of one or more computing nodes to act as core computing nodes that each participate in a distributed storage system storing information used in the execution of the indicated program, the multiple computing nodes further including a second group of one or more computing nodes to act as auxiliary computing nodes that do not participate in the distributed storage system, wherein at least one auxiliary computing node of the second group has temporary availability while not otherwise being used and is selected for use in the second group based at least in part on the temporary availability;

initiating, by the one or more configured computing systems, the execution of the indicated program in the distributed manner on the multiple computing nodes of the cluster by executing one or more of the plurality of jobs on each of the multiple computing nodes; and

during the execution of the indicated program by the multiple computing nodes of the cluster, receiving an indication that the at least one auxiliary computing node of the second group is to be used for a distinct first use that is not related to the execution of the indicated program, and initiating, by the one or more configured computing systems, removal from the cluster of the at least one auxiliary computing node in the second group while the execution of the indicated program continues, to enable the removed at least one auxiliary computing node to be available for the distinct first use.

2. The method of claim 1 wherein the at least one auxiliary computing node is part of a pool of computing nodes that are temporarily available from a first source while they are not being used for the distinct first use, and wherein the determining of the multiple computing nodes includes selecting computing nodes for the first group that are available from a second source and are not allocated for any uses other than use in the cluster.

3. The method of claim 2 wherein the computing nodes temporarily available from the first source have an associated first price of use, wherein the computing nodes from the second source have an associated second price of use that is higher than the first price, and wherein selecting of the at least one auxiliary computing node for use in the second group is further based at least in part on the associated first price.

4. The method of claim 3 wherein the pool of computing nodes includes computing nodes at multiple distinct geographical locations, wherein the computing nodes in the pool have differing associated first prices of use based on the multiple distinct geographical locations, and wherein the determining of the multiple computing nodes further includes selecting the at least one auxiliary computing node based on a geographical location of the at least one auxiliary computing node and on the associated first price of use based on the geographical location of the at least one auxiliary computing node.

5. The method of claim 1 wherein the configuration information includes one or more instructions regarding selection of computing nodes for the cluster, and wherein the determining of the multiple computing nodes is further performed in accordance with the one or more instructions.

6. The method of claim 5 further comprising, at a time after the execution of the indicated program is initiated, selecting one or more additional computing nodes and adding the selected additional computing nodes to the cluster for use in additional execution of the indicated program, wherein the selected one or more additional computing nodes are of a type indicated in the one or more instructions.

7. The method of claim 1 wherein the distributed program execution service provides a plurality of configurable computing nodes for use by customers, wherein the determining of the multiple computing nodes includes selecting a subset of the plurality of configurable computing nodes for use in the cluster, and wherein the method further comprises, during the execution of the indicated program, modifying a quantity of the multiple computing nodes of the cluster by adding one or more additional computing nodes from the plurality of configurable computing nodes.

8. The method of claim 1 wherein the distributed storage system is a distributed file system, and wherein each of the core computing nodes stores a portion of the distributed file system on local storage of that core computing node.

9. The method of claim 1 wherein the indicated program is configured to perform one or more map functions on each of multiple input data subsets and to perform one or more reduce functions on results of the one or more map functions, and wherein the method further comprises generating at least some of the plurality of jobs to each implement at least one of the map functions or at least one of the reduce functions.

10. A non-transitory computer-readable medium having stored contents that configure a computing system to:

initiate, by the configured computing system, execution of an indicated program in a distributed manner on a cluster of multiple computing nodes at a first time, the initiating of the execution including executing one or more of a plurality of jobs of the indicated program on each of the multiple computing nodes;

at a second time subsequent to the first time and while the execution of the indicated program is ongoing, determine, by the configured computing system, to modify the cluster in a manner based at least in part on use of one or more computing nodes that have temporary availability until a distinct use of a higher priority occurs for the one or more computing nodes, the cluster including at the second time a first group of the multiple computing nodes that each are part of distributed storage for use during the execution of the indicated program; and

initiate, by the configured computing system, a change in the multiple computing nodes of the cluster while the execution of the indicated program is ongoing in response to the determining, the initiating of the change including selecting a second group of computing nodes of the cluster that are not part of the distributed storage and including performing a modification to the second group corresponding to the use of the one or more computing nodes having temporary availability, wherein the use of the one or more computing nodes having temporary availability is based at least in part on the temporary availability of the one or more computing nodes.

11. The non-transitory computer-readable medium of claim 10 wherein the computing nodes of the first group include multiple core computing nodes that are each part of a distributed storage system for use with the cluster, wherein the cluster further includes at one or more times one or more auxiliary computing nodes in the second group that are not part of the distributed storage system, and wherein the initiating of the change includes modifying a quantity of the auxiliary computing nodes in the second group of the cluster.

12. The non-transitory computer-readable medium of claim 11 wherein the one or more computing nodes having temporary availability are previously allocated for a first use distinct from use in the cluster, wherein the one or more auxiliary computing nodes in the second group at the second time include the one or more computing nodes having temporary availability, wherein the multiple core computing nodes of the first group are not allocated for any other uses distinct from use in the cluster, and wherein the performed modification to the second group includes removing at least one of the one or more computing nodes having temporary availability from the cluster based on an indication that the removed computing nodes are to be used for the distinct first use.

13. The non-transitory computer-readable medium of claim 11 wherein the one or more computing nodes having temporary availability are previously allocated for a first use distinct from use in the cluster, wherein the multiple core computing nodes of the first group are not allocated for any other uses distinct from use in the cluster, wherein the cluster does not include any computing nodes of the second group at the second time, and wherein the performed modification to the second group includes creating the second group at the second time by adding the one or more computing nodes having temporary availability to the cluster for use as at least part of the second group.

14. The non-transitory computer-readable medium of claim 11 wherein the configured computing system is part of a distributed program execution service that provides a plurality of computing nodes of multiple types that are configurable to perform distributed execution of programs for clients, wherein the initiating of the execution of the indicated program is performed for a first client, and wherein the multiple computing nodes of the cluster are a subset of the plurality of computing nodes that are provided for use on behalf of the first client.

15. The non-transitory computer-readable medium of claim 11 wherein the one or more computing nodes having temporary availability are previously allocated for a first use distinct from use in the cluster, wherein the multiple core computing nodes of the first group at the second time are not allocated for any other uses distinct from use in the cluster, and wherein the initiating of the change further includes adding at the second time at least one of the one or more computing nodes having temporary availability to the first group of the cluster.

16. The non-transitory computer-readable medium of claim 11 wherein the one or more computing nodes having temporary availability are previously allocated for a first use distinct from use in the cluster, wherein the multiple core computing nodes of the first group at the second time include the one or more computing nodes having temporary availability and include one or more other computing nodes that are not allocated for any other uses distinct from use in the cluster, and wherein the performed modification to the second group includes moving at least one of the one or more computing nodes having temporary availability from the first group of the cluster to the second group of the cluster in preparation for future use of the moved computing nodes for the distinct first use.

17. The non-transitory computer-readable medium of claim 10 wherein the initiating of the execution of the indicated program on the cluster of multiple computing nodes at the first time includes determining a subset of a first quantity of the plurality of jobs whose execution is to be initiated at the first time, and wherein the initiating of the change includes automatically increasing the quantity of the multiple computing nodes in the cluster at the second time to enable execution of a second quantity of the plurality of jobs on the multiple computing nodes in the cluster after the second time, the second quantity being larger than the first quantity.

18. The non-transitory computer-readable medium of claim 10 wherein the initiating of the execution of the indicated program on the cluster of multiple computing nodes at the first time includes determining a subset of a first quantity of the plurality of jobs whose execution is to be initiated at the first time, and wherein the initiating of the change includes automatically reducing the quantity of the multiple computing nodes in the cluster at the second time to reflect execution of a second quantity of the plurality of jobs on the multiple computing nodes in the cluster after the second time, the second quantity being smaller than the first quantity.

19. The non-transitory computer-readable medium of claim 10 wherein the multiple computing nodes of the first group provide the distributed storage by each storing a portion of the indicated input data on local storage of that computing node, and wherein the stored instructions further configure the computing system to, while the execution of the indicated program is ongoing and in response to determining that one of the computing nodes of the first group has become unavailable for use in the cluster, replace the one unavailable computing node in the cluster with a different replacement computing node that is added to the first group and is provided with the portion of the indicated input data that was stored on the local storage of the one unavailable computing node.

20. The non-transitory computer-readable medium of claim 19 wherein the stored instructions further configure the computing system to, in response to determining that a second of the multiple computing nodes of the cluster has become unavailable for use in the cluster, determine not to replace the second unavailable computing node in the cluster with a different replacement computing node based at least in part on the second computing node being part of the second group, and wherein the replacing of the one unavailable computing node with a different replacement computing node is based at least in part on the one unavailable computing node being part of the first group.

21. The non-transitory computer-readable medium of claim 10 wherein the one or more computing nodes having temporary availability are part of a pool of computing nodes that have an associated price of use that varies at differing times, and wherein the determining to modify the cluster at the second time includes initiating at the second time the use of the one or more computing nodes having temporary availability based on the associated price of use at the second time being below a threshold or includes terminating at the second time the use of the one or more computing nodes having temporary availability based on the associated price of use at the second time being above a threshold.

22. The non-transitory computer-readable medium of claim 21 wherein the stored contents further configure the computing system to monitor the associated price of use of the computing nodes of the pool during the execution of the indicated program, and wherein the determining to modify the cluster at the second time is initiated based on one or more defined criteria associated with the indicated program being determined to be satisfied at the second time, the one or more defined criteria being based at least in part on the monitored associated price of use.

23. The non-transitory computer-readable medium of claim 21 wherein the execution of the indicated program is initiated on behalf of a client, and wherein the stored contents further configure the computing system to monitor the associated price of use of the computing nodes of the pool during the execution of the indicated program, and to provide information to the client regarding the monitored associated price of use.

24. The non-transitory computer-readable medium of claim 10 wherein the performing of the modification to the second group includes initiating at the second time the use of the one or more computing nodes having temporary availability as part of the second group until the distinct higher priority use arises.

25. The non-transitory computer-readable medium of claim 10 wherein the stored contents include instructions that, when executed, program the configured computing system.

26. A computing system, comprising:

one or more processors; and

one or more modules that are configured to, when executed by at least one of the one or more processors, manage distributed execution of an indicated program for a client by;

receiving information from the client regarding execution of the indicated program;

initiating the execution of the indicated program in a distributed manner on a cluster of multiple computing nodes at a first time in accordance with the received information, the initiating of the execution including attempting to execute one or more of a plurality of jobs of the indicated program on each of one or more of the multiple computing nodes;

at a second time subsequent to the first time and while the execution of the indicated program is ongoing, determining to modify a quantity of first computing nodes in the cluster that each execute one or more of the plurality of jobs of the indicated program without being part of distributed storage for information used for the ongoing execution of the indicated program, the first computing nodes being distinct from at least some other computing nodes in the cluster at the second time that each are part of the distributed storage; and

initiating a change in the quantity of the first computing nodes of the cluster while the execution of the indicated program is ongoing and in response to the determining, wherein the initiated change in the quantity of the first computing nodes corresponds to use of one or more computing nodes that have temporarily availability for use in the cluster until a distinct use occurs for the one or more computing nodes, and wherein the use of the one or more computing nodes having temporary availability is based at least in part on the temporary availability of the one or more computing nodes.

27. The computing system of claim 26 wherein the at least some computing nodes in the cluster that are part of the distributed storage includes a first group of multiple core computing nodes that are each part of a distributed storage system for use with the cluster, wherein the cluster further includes at one or more times one or more of the first computing nodes in a second group of one or more other auxiliary computing nodes that are not part of the distributed storage system, and wherein the initiating of the change includes modifying a quantity of the auxiliary computing nodes in the second group of the cluster.

28. The computing system of claim 27 wherein the information received from the client is configuration information for the execution of the indicated program that includes an indication of a quantity of the multiple computing nodes in the cluster at the first time and that includes one or more criteria used in the determining at the second time to modify the quantity of the first computing nodes in the cluster, wherein the one or more computing nodes having temporary availability are previously allocated for a first use distinct from use in the cluster, and wherein the initiated change in the quantity of the first computing nodes includes removing from the cluster at least one of the one or more computing nodes having temporary availability based on an indication that the removed computing nodes are to be used for the distinct first use.

29. The computing system of claim 26 wherein the one or more modules are part of a distributed program execution service and each includes software instructions for execution by the computing system.

30. The computing system of claim 26 wherein the one or more modules consist of a means for performing the managing of the distributed execution of the program for the client.