Virtual Hadoop manager

• US 9,727,355 B2
• Filed: 06/23/2014
• Issued: 08/08/2017
• Est. Priority Date: 08/23/2013
• Status: Active Grant

• Current Assignee: Vmware LLC (Broadcom, Inc.)
• Original Assignee: VMware, Inc. (Broadcom, Inc.)
• Inventors: Hicken, George, Gummaraju, Jayanth, Govil, Kinshuk, Corrie, Benjamin J., Holler, Anne
• Primary Examiner(s): Chen, Qing
• Assistant Examiner(s): Lyons, Andrew M
• Application Number:
  US14/311,755
• Publication Number:
  US 20150058843A1
• Time in Patent Office: 1,142 Days
• Field of Search: 718 1
• CPC Class Codes: G06F 2009/4557 Distribution of virtual machine instances; Migration and load balancing
• CPC Class Codes: G06F 9/455 Emulation; Interpretation; Software simulation, e.g. virtualisation or emulation of application or operating system execution engines
• CPC Class Codes: G06F 9/45533 Hypervisors; Virtual machine monitors
• CPC Class Codes: G06F 9/45558 Hypervisor-specific management and integration aspects
• CPC Class Codes: G06F 9/48 Program initiating; Program switching, e.g. by interrupt
• CPC Class Codes: G06F 9/50 Allocation of resources, e.g. of the central processing unit [CPU]
• CPC Class Codes: G06F 9/5088 involving task migration

Abstract

A distributed computing application is described that provides a highly elastic and multi-tenant platform for Hadoop applications and other workloads running in a virtualized environment. Multiple instances of a distributed computing framework, such as Hadoop, may be executed concurrently. A centralized manager detects when contention for computing resources, such as memory and CPU, causes tasks to run slower on VMs executing on a given host, and scales up or scales down a cluster based on the detected resource contention.

Claims

1. A method for executing a multi-tenant distributed computing application within a virtualized computing environment, the method comprising:

receiving cluster metrics for each compute cluster in a plurality of compute clusters executing in a virtualized computing environment, wherein each compute cluster includes a workload scheduler and a plurality of worker nodes;

receiving resource-related metrics associated with performance of the virtualized computing environment, wherein the resource-related metrics comprise at least one of memory related metrics and CPU-related metrics;

making a determination of whether actual resource contention exists among the plurality of compute clusters for computing resources of a host based on the received cluster metrics and resource-related metrics;

responsive to making the determination that actual resource contention exists, shrinking at least one of the plurality of compute clusters executing, at least in part, on the host; and

responsive to making the determination that actual resource contention does not exist on the host, expanding at least one of the plurality of compute clusters executing, at least in part, on the host and having pending work.

2. The method of claim 1, wherein the determination of actual resource contention is made based on an amount of memory reclamation for a virtual machine executing on the host exceeding an amount of unused guest memory for the virtual machine.

3. The method of claim 1, wherein the determination of actual resource contention is made based on CPU ready metrics indicating cycles of time that a virtual machine in the virtualized computing environment was ready but could not get scheduled to run on a physical CPU and based on CPU Overlap metrics that indicate cycles of time a virtual machine in the virtualized computing environment was interrupted to perform system services on behalf of that virtual machine.

4. The method of claim 1, wherein a portion of the resource-related metrics associated with a VM that was powered-on within an initial time period are disregarded from making a determination of whether actual resource contention exists.

5. The method of claim 1, wherein shrinking at least one of the plurality of compute clusters executing, at least in part, on the host further comprises:

responsive to making the determination that both a memory-contended VM and a CPU-contended VM exist on the host,

selecting the memory-contended VM.

6. The method of claim 1,

wherein shrinking at least one of the plurality of compute clusters executing, at least in part, on the host further comprises:

decommissioning a task tracker node executing on a VM, and

powering off the VM; and

wherein expanding at least one of the plurality of compute clusters executing, at least in part, on the host and having pending work further comprises;

powering on a VM executing on the host, and

re-commissioning a task tracker to execute on the VM.

7. The method of claim 1, wherein the determination of actual resource contention is made according to a control-theory-based algorithm based on the received cluster metrics and resource-related metrics.

8. The method of claim 7, further comprising:

receiving a second set of cluster metrics and resource-related metrics; and

making a determination, during a subsequent action cycle, that actual resource contention exists based on the second set of cluster metrics and resource-related metrics and further based on the cluster metrics and resource-related metrics received during a prior action cycle.

9. The method of claim 1, further comprising:

generating one or more moving averages based on the received resource-related metrics associated with performance of the virtualized computing environment; and

discarding a second set of resource-related metrics received during a subsequent action cycle responsive to determining the second set of resource-related metrics are anomalous relative to the moving average.

10. A non-transitory computer readable storage medium having stored thereon computer software executable by a processor, the computer software embodying a method for executing a multi-tenant distributed computing application within a virtualized computing environment, the method comprising:

receiving cluster metrics for each compute cluster in a plurality of compute clusters executing in a virtualized computing environment, wherein each compute cluster includes a workload scheduler and a plurality of worker nodes;

receiving resource-related metrics associated with performance of the virtualized computing environment, wherein the resource-related metrics comprise at least one of memory related metrics and CPU-related metrics;

making a determination of whether actual resource contention exists among the plurality of compute clusters for computing resources of a host based on the received cluster metrics and resource-related metrics;

responsive to making the determination that actual resource contention exists, shrinking at least one of the plurality of compute clusters executing, at least in part, on the host; and

responsive to making the determination that actual resource contention does not exist on the host, expanding at least one of the plurality of compute clusters executing, at least in part, on the host and having pending work.

11. The non-transitory computer readable storage medium of claim 10, wherein the determination of actual resource contention is made based on an amount of memory reclamation for a virtual machine executing on the host exceeding an amount of unused guest memory for the virtual machine.

12. The non-transitory computer readable storage medium of claim 10, wherein the determination of actual resource contention is made based on CPU ready metrics indicating cycles of time that a virtual machine in the virtualized computing environment was ready but could not get scheduled to run on a physical CPU and based on CPU Overlap metrics that indicate cycles of time a virtual machine in the virtualized computing environment was interrupted to perform system services on behalf of that virtual machine.

13. The non-transitory computer readable storage medium of claim 10, wherein a portion of the resource-related metrics associated with a VM that was powered-on within an initial time period are disregarded from making a determination of whether actual resource contention exists.

14. The non-transitory computer readable storage medium of claim 10, wherein shrinking at least one of the plurality of compute clusters executing, at least in part, on the host further comprises:

responsive to making a determination that both a memory-contended VM and a CPU-contended VM exist on the host, selecting the memory-contended VM.

15. The non-transitory computer readable storage medium of claim 10,

wherein shrinking at least one of the plurality of compute clusters executing, at least in part, on the host further comprises:

decommissioning a task tracker node executing on a VM, and powering off the VM; and

wherein expanding at least one of the plurality of compute clusters executing, at least in part, on the host and having pending work further comprises;

powering on a VM executing on the host, and

re-commissioning a task tracker to execute on the VM.

16. The non-transitory computer readable storage medium of claim 10, wherein the determination of actual resource contention is made according to a control-theory-based algorithm based on the received cluster-metrics and resource-related metrics.

17. The non-transitory computer readable storage medium of claim 16, the method further comprising:

receiving a second set of cluster metrics and resource-related metrics; and

making a determination, during a subsequent action cycle, that actual resource contention exists based on the second set of cluster metrics and resource-related metrics and further based on the cluster metrics and resource-related metrics received during a prior action cycle.

18. The non-transitory computer readable storage medium of claim 10, the method further comprising:

generating one or more moving averages based on the received resource-related metrics associated with performance of the virtualized computing environment; and

discarding a second set of resource-related metrics received during a subsequent action cycle responsive to determining the second set of resource-related metrics are anomalous relative to the moving average.

19. A system for executing a distributed computing application within a virtualized computing environment, the host computer system comprising:

a memory device;

a processor coupled to the memory device that contains a program for the processor to carry out the steps of;

receiving cluster metrics for each compute cluster in a plurality of compute clusters executing in a virtualized computing environment, wherein each compute cluster includes a workload scheduler and a plurality of worker nodes;

receiving resource-related metrics associated with performance of the virtualized computing environment, wherein the resource-related metrics comprise at least one of memory-related metrics and CPU-related metrics;

making a determination of whether actual resource contention exists among the plurality of compute clusters for computing resources of a host based on the received cluster metrics and resource-related metrics;

responsive to making the determination that actual resource contention exists, shrinking at least one of the plurality of compute clusters executing, at least in part, on the host; and

responsive to making the determination that actual resource contention does not exist on the host, expanding at least one of the plurality of compute clusters executing, at least in part, on the host and having pending work.

20. The system of claim 19, wherein the determination of actual resource contention is made according to a control-theory-based algorithm based on the received cluster-metrics and resource-related metrics.