Package org.apache.storm.scheduler

Class Cluster

java.lang.Object

org.apache.storm.scheduler.Cluster

All Implemented Interfaces:

ISchedulingState

Direct Known Subclasses:

SingleTopologyCluster

public class Cluster
extends Object
implements ISchedulingState

The current state of the storm cluster. Cluster is not currently thread safe.

Constructor Summary

Constructors

Constructor	Description
Cluster(Cluster src)	Copy constructor.
Cluster(Cluster src, Topologies topologies)	Testing Constructor that takes an existing cluster and replaces the topologies in it.
Cluster(INimbus nimbus, ResourceMetrics resourceMetrics, Map<String,SupervisorDetails> supervisors, Map<String,? extends SchedulerAssignment> assignments, Topologies topologies, Map<String,Object> conf)

Method Summary

Modifier and Type	Method	Description
protected void	assertValidTopologyForModification(String topologyId)	Check if the given topology is allowed for modification right now.
void	assign(SchedulerAssignment assignment, boolean ignoreSingleExceptions)	Assign everything for the given topology.
void	assign(WorkerSlot slot, String topologyId, Collection<ExecutorDetails> executors)	Assign the slot to the executors for this topology.
void	blacklistHost(String host)
void	freeSlot(WorkerSlot slot)	Free the specified slot.
void	freeSlots(Collection<WorkerSlot> slots)	free the slots.
NormalizedResourceRequest	getAllScheduledResourcesForNode(String nodeId)	Get all scheduled resources for node.
Set<Integer>	getAssignablePorts(SupervisorDetails supervisor)	Get the ports that are not blacklisted.
List<WorkerSlot>	getAssignableSlots()	Get all non-blacklisted slots in the cluster.
List<WorkerSlot>	getAssignableSlots(SupervisorDetails supervisor)	Return all non-blacklisted slots on this supervisor.
static double	getAssignedMemoryForSlot(Map<String,Object> topConf)	Get heap memory usage for a worker's main process and logwriter process.
int	getAssignedNumWorkers(TopologyDetails topology)	Get the number of workers assigned to a topology.
SchedulerAssignment	getAssignmentById(String topologyId)	get the current assignment for the topology.
Map<String,SchedulerAssignment>	getAssignments()	Get all the assignments.
Set<Integer>	getAvailablePorts(SupervisorDetails supervisor)	Return the available ports of this supervisor.
NormalizedResourceOffer	getAvailableResources(SupervisorDetails sd)	Get the resources on the supervisor that are available to be scheduled.
List<WorkerSlot>	getAvailableSlots()	Get all the available worker slots in the cluster.
List<WorkerSlot>	getAvailableSlots(SupervisorDetails supervisor)	Return all the available slots on this supervisor.
Set<String>	getBlacklistedHosts()	Get all of the hosts that are blacklisted.
double	getClusterTotalCpuResource()	Get the total amount of CPU resources in cluster.
Map<String,Double>	getClusterTotalGenericResources()	Get the total amount of generic resources (excluding CPU and memory) in cluster.
double	getClusterTotalMemoryResource()	Get the total amount of memory resources in cluster.
Map<String,Object>	getConf()	Get the nimbus configuration.
List<String>	getGreyListedSupervisors()
String	getHost(String supervisorId)	Map a supervisor to a given host.
INimbus	getINimbus()
double	getMinWorkerCpu()
Map<String,List<ExecutorDetails>>	getNeedsSchedulingComponentToExecutors(TopologyDetails topology)	Get the component name to executor list for executors that need to be scheduled.
Map<ExecutorDetails,String>	getNeedsSchedulingExecutorToComponents(TopologyDetails topology)	Get the executor to component name map for executors that need to be scheduled.
Map<String,List<String>>	getNetworkTopography()	Get the network topography (rackId -> nodes in the rack).
List<WorkerSlot>	getNonBlacklistedAvailableSlots(List<String> blacklistedSupervisorIds)	Get all the available worker slots in the cluster, that are not blacklisted.
NormalizedResourceOffer	getNonBlacklistedClusterAvailableResources(Collection<String> blacklistedSupervisorIds)	Get the resources in the cluster that are available for scheduling.
ResourceMetrics	getResourceMetrics()
double	getScheduledCpuForNode(String nodeId)	Get the total cpu currently scheduled on a node.
double	getScheduledMemoryForNode(String nodeId)	Get the total memory currently scheduled on a node.
String	getStatus(String topoId)
Map<String,String>	getStatusMap()	Get all topology scheduler statuses.
SupervisorDetails	getSupervisorById(String nodeId)	Get a specific supervisor with the nodeId.
Map<String,SupervisorDetails>	getSupervisors()	Get all the supervisors.
List<SupervisorDetails>	getSupervisorsByHost(String host)	Get all the supervisors on the specified host.
Map<String,SupervisorResources>	getSupervisorsResourcesMap()	Get the amount of used and free resources on a supervisor.
Topologies	getTopologies()	Get all of the topologies.
Map<String,TopologyResources>	getTopologyResourcesMap()	Get the amount of resources used by topologies.
Collection<ExecutorDetails>	getUnassignedExecutors(TopologyDetails topology)	get the unassigned executors of the topology.
Set<Integer>	getUsedPorts(SupervisorDetails supervisor)	Get all the used ports of this supervisor.
Collection<WorkerSlot>	getUsedSlots()	Get all currently occupied slots.
Collection<WorkerSlot>	getUsedSlotsByTopologyId(String topologyId)	get slots used by a topology.
WorkerResources	getWorkerResources(WorkerSlot ws)	Get the resources for a given slot.
Map<String,Map<WorkerSlot,WorkerResources>>	getWorkerResourcesMap()	Gets the reference to the full topology->worker resource map.
boolean	isBlackListed(String supervisorId)	Check is a given supervisor is on a blacklisted host.
boolean	isBlacklistedHost(String host)	Check if a given host is blacklisted.
boolean	isSlotOccupied(WorkerSlot slot)	Check if a slot is occupied or not.
boolean	needsScheduling(TopologyDetails topology)	Does the topology need scheduling.
boolean	needsSchedulingRas(TopologyDetails topology)	Like ISchedulingState.needsScheduling(TopologyDetails) but does not take into account the number of workers requested.
List<TopologyDetails>	needsSchedulingTopologies()	Get all of the topologies that need scheduling.
void	setAssignments(Map<String,? extends SchedulerAssignment> newAssignments, boolean ignoreSingleExceptions)	Set assignments for cluster.
void	setBlacklistedHosts(Set<String> hosts)	Set the list of hosts that are blacklisted.
void	setGreyListedSupervisors(Set<String> greyListedSupervisors)
void	setNetworkTopography(Map<String,List<String>> networkTopography)
void	setStatus(String topologyId, String statusMessage)	set scheduler status for a topology.
void	setStatus(TopologyDetails td, String statusMessage)	set scheduler status for a topology.
void	setStatusIfAbsent(String topologyId, String statusMessage)
void	setStatusMap(Map<String,String> statusMap)	set scheduler status map.
void	unassign(String topoId)	Unassign everything for the given topology id.
void	updateFrom(Cluster other)	Update the assignments and status from the other cluster.
boolean	wouldFit(WorkerSlot ws, ExecutorDetails exec, TopologyDetails td, NormalizedResourceOffer resourcesAvailable, double maxHeap)	Would scheduling exec on ws fit? With a heap <= maxHeap total memory added <= memoryAvailable and cpu added <= cpuAvailable.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.storm.scheduler.ISchedulingState

getAssignedRacks, getHostToRack

Constructor Detail

Cluster

public Cluster(INimbus nimbus,
               ResourceMetrics resourceMetrics,
               Map<String,SupervisorDetails> supervisors,
               Map<String,? extends SchedulerAssignment> assignments,
               Topologies topologies,
               Map<String,Object> conf)

Cluster

public Cluster(Cluster src)

Copy constructor.

Cluster

public Cluster(Cluster src,
               Topologies topologies)

Testing Constructor that takes an existing cluster and replaces the topologies in it.

Parameters:

src - the original cluster

topologies - the new topolgoies to use

Method Detail

getAssignedMemoryForSlot

public static double getAssignedMemoryForSlot(Map<String,Object> topConf)

Get heap memory usage for a worker's main process and logwriter process.

Parameters:

topConf - - the topology config

Returns:

the assigned memory (in MB)

assertValidTopologyForModification

protected void assertValidTopologyForModification(String topologyId)

Check if the given topology is allowed for modification right now. If not throw an IllegalArgumentException else go on.

Parameters:

topologyId - the id of the topology to check

getTopologies

public Topologies getTopologies()

Description copied from interface: ISchedulingState

Get all of the topologies.

Specified by:

getTopologies in interface ISchedulingState

Returns:

all of the topologies that are a part of the cluster.

getBlacklistedHosts

public Set<String> getBlacklistedHosts()

Description copied from interface: ISchedulingState

Get all of the hosts that are blacklisted.

Specified by:

getBlacklistedHosts in interface ISchedulingState

Returns:

all of the hosts that are blacklisted

setBlacklistedHosts

public void setBlacklistedHosts(Set<String> hosts)

Set the list of hosts that are blacklisted.

Parameters:

hosts - the new hosts that are blacklisted.

blacklistHost

public void blacklistHost(String host)

isBlackListed

public boolean isBlackListed(String supervisorId)

Description copied from interface: ISchedulingState

Check is a given supervisor is on a blacklisted host.

Specified by:

isBlackListed in interface ISchedulingState

Parameters:

supervisorId - the id of the supervisor

Returns:

true if it is else false

isBlacklistedHost

public boolean isBlacklistedHost(String host)

Description copied from interface: ISchedulingState

Check if a given host is blacklisted.

Specified by:

isBlacklistedHost in interface ISchedulingState

Parameters:

host - the name of the host

Returns:

true if it is else false.

getHost

public String getHost(String supervisorId)

Description copied from interface: ISchedulingState

Map a supervisor to a given host.

Specified by:

getHost in interface ISchedulingState

Parameters:

supervisorId - the id of the supervisor

Returns:

the actual host name the supervisor is on

needsSchedulingTopologies

public List<TopologyDetails> needsSchedulingTopologies()

Description copied from interface: ISchedulingState

Get all of the topologies that need scheduling.

Specified by:

needsSchedulingTopologies in interface ISchedulingState

Returns:

all of the topologies that are not fully scheduled.

needsScheduling

public boolean needsScheduling(TopologyDetails topology)

Description copied from interface: ISchedulingState

Does the topology need scheduling.

A topology needs scheduling if one of the following conditions holds:

• Although the topology is assigned slots, but is squeezed. i.e. the topology is assigned less slots than desired.
• There are unassigned executors in this topology

Specified by:

needsScheduling in interface ISchedulingState

needsSchedulingRas

public boolean needsSchedulingRas(TopologyDetails topology)

Description copied from interface: ISchedulingState

Like ISchedulingState.needsScheduling(TopologyDetails) but does not take into account the number of workers requested. This is because the number of workers is ignored in RAS

Specified by:

needsSchedulingRas in interface ISchedulingState

Parameters:

topology - the topology to check

Returns:

true if the topology needs scheduling else false.

getNeedsSchedulingExecutorToComponents

public Map<ExecutorDetails,String> getNeedsSchedulingExecutorToComponents(TopologyDetails topology)

Description copied from interface: ISchedulingState

Get the executor to component name map for executors that need to be scheduled.

Specified by:

getNeedsSchedulingExecutorToComponents in interface ISchedulingState

Parameters:

topology - the topology this is for

Returns:

a executor -> component-id map which needs scheduling in this topology.

getNeedsSchedulingComponentToExecutors

public Map<String,List<ExecutorDetails>> getNeedsSchedulingComponentToExecutors(TopologyDetails topology)

Description copied from interface: ISchedulingState

Get the component name to executor list for executors that need to be scheduled.

Specified by:

getNeedsSchedulingComponentToExecutors in interface ISchedulingState

Parameters:

topology - the topology this is for

Returns:

a component-id -> executors map which needs scheduling in this topology.

getUsedPorts

public Set<Integer> getUsedPorts(SupervisorDetails supervisor)

Description copied from interface: ISchedulingState

Get all the used ports of this supervisor.

Specified by:

getUsedPorts in interface ISchedulingState

getAvailablePorts

public Set<Integer> getAvailablePorts(SupervisorDetails supervisor)

Description copied from interface: ISchedulingState

Return the available ports of this supervisor.

Specified by:

getAvailablePorts in interface ISchedulingState

getAssignablePorts

public Set<Integer> getAssignablePorts(SupervisorDetails supervisor)

Description copied from interface: ISchedulingState

Get the ports that are not blacklisted.

Specified by:

getAssignablePorts in interface ISchedulingState

Parameters:

supervisor - the supervisor

Returns:

the ports that are not blacklisted

getNonBlacklistedAvailableSlots

public List<WorkerSlot> getNonBlacklistedAvailableSlots(List<String> blacklistedSupervisorIds)

Description copied from interface: ISchedulingState

Get all the available worker slots in the cluster, that are not blacklisted.

Specified by:

getNonBlacklistedAvailableSlots in interface ISchedulingState

Parameters:

blacklistedSupervisorIds - list of supervisor ids that should also be considered blacklisted.

getAvailableSlots

public List<WorkerSlot> getAvailableSlots()

Description copied from interface: ISchedulingState

Get all the available worker slots in the cluster.

Specified by:

getAvailableSlots in interface ISchedulingState

getAvailableSlots

public List<WorkerSlot> getAvailableSlots(SupervisorDetails supervisor)

Description copied from interface: ISchedulingState

Return all the available slots on this supervisor.

Specified by:

getAvailableSlots in interface ISchedulingState

getAssignableSlots

public List<WorkerSlot> getAssignableSlots(SupervisorDetails supervisor)

Description copied from interface: ISchedulingState

Return all non-blacklisted slots on this supervisor.

Specified by:

getAssignableSlots in interface ISchedulingState

Parameters:

supervisor - the supervisor

Returns:

the non-blacklisted slots

getAssignableSlots

public List<WorkerSlot> getAssignableSlots()

Description copied from interface: ISchedulingState

Get all non-blacklisted slots in the cluster.

Specified by:

getAssignableSlots in interface ISchedulingState

getUnassignedExecutors

public Collection<ExecutorDetails> getUnassignedExecutors(TopologyDetails topology)

Description copied from interface: ISchedulingState

get the unassigned executors of the topology.

Specified by:

getUnassignedExecutors in interface ISchedulingState

Parameters:

topology - the topology to check

Returns:

the unassigned executors of the topology.

getAssignedNumWorkers

public int getAssignedNumWorkers(TopologyDetails topology)

Description copied from interface: ISchedulingState

Get the number of workers assigned to a topology.

Specified by:

getAssignedNumWorkers in interface ISchedulingState

Parameters:

topology - the topology this is for

Returns:

the number of workers assigned to this topology.

getAvailableResources

public NormalizedResourceOffer getAvailableResources(SupervisorDetails sd)

Description copied from interface: ISchedulingState

Get the resources on the supervisor that are available to be scheduled.

Specified by:

getAvailableResources in interface ISchedulingState

Parameters:

sd - the supervisor.

Returns:

the resources available to be scheduled.

wouldFit

public boolean wouldFit(WorkerSlot ws,
                        ExecutorDetails exec,
                        TopologyDetails td,
                        NormalizedResourceOffer resourcesAvailable,
                        double maxHeap)

Description copied from interface: ISchedulingState

Would scheduling exec on ws fit? With a heap <= maxHeap total memory added <= memoryAvailable and cpu added <= cpuAvailable.

Specified by:

wouldFit in interface ISchedulingState

Parameters:

ws - the slot to put it in

exec - the executor to investigate

td - the topology detains for this executor

resourcesAvailable - all the available resources

maxHeap - the maximum heap size for ws

Returns:

true it fits else false

assign

public void assign(WorkerSlot slot,
                   String topologyId,
                   Collection<ExecutorDetails> executors)

Assign the slot to the executors for this topology.

Throws:

RuntimeException - if the specified slot is already occupied.

assign

public void assign(SchedulerAssignment assignment,
                   boolean ignoreSingleExceptions)

Assign everything for the given topology.

Parameters:

assignment - the new assignment to make

freeSlot

public void freeSlot(WorkerSlot slot)

Free the specified slot.

Parameters:

slot - the slot to free

freeSlots

public void freeSlots(Collection<WorkerSlot> slots)

free the slots.

Parameters:

slots - multiple slots to free

isSlotOccupied

public boolean isSlotOccupied(WorkerSlot slot)

Description copied from interface: ISchedulingState

Check if a slot is occupied or not.

Specified by:

isSlotOccupied in interface ISchedulingState

Parameters:

slot - the slot be to checked.

Returns:

true if the specified slot is occupied.

getAssignmentById

public SchedulerAssignment getAssignmentById(String topologyId)

Description copied from interface: ISchedulingState

get the current assignment for the topology.

Specified by:

getAssignmentById in interface ISchedulingState

getUsedSlotsByTopologyId

public Collection<WorkerSlot> getUsedSlotsByTopologyId(String topologyId)

Description copied from interface: ISchedulingState

get slots used by a topology.

Specified by:

getUsedSlotsByTopologyId in interface ISchedulingState

getSupervisorById

public SupervisorDetails getSupervisorById(String nodeId)

Description copied from interface: ISchedulingState

Get a specific supervisor with the nodeId.

Specified by:

getSupervisorById in interface ISchedulingState

getUsedSlots

public Collection<WorkerSlot> getUsedSlots()

Description copied from interface: ISchedulingState

Get all currently occupied slots.

Specified by:

getUsedSlots in interface ISchedulingState

getSupervisorsByHost

public List<SupervisorDetails> getSupervisorsByHost(String host)

Description copied from interface: ISchedulingState

Get all the supervisors on the specified host.

Specified by:

getSupervisorsByHost in interface ISchedulingState

Parameters:

host - hostname of the supervisor

Returns:

the SupervisorDetails object.

getAssignments

public Map<String,SchedulerAssignment> getAssignments()

Description copied from interface: ISchedulingState

Get all the assignments.

Specified by:

getAssignments in interface ISchedulingState

setAssignments

public void setAssignments(Map<String,? extends SchedulerAssignment> newAssignments,
                           boolean ignoreSingleExceptions)

Set assignments for cluster.

getSupervisors

public Map<String,SupervisorDetails> getSupervisors()

Description copied from interface: ISchedulingState

Get all the supervisors.

Specified by:

getSupervisors in interface ISchedulingState

getNonBlacklistedClusterAvailableResources

public NormalizedResourceOffer getNonBlacklistedClusterAvailableResources(Collection<String> blacklistedSupervisorIds)

Description copied from interface: ISchedulingState

Get the resources in the cluster that are available for scheduling.

Specified by:

getNonBlacklistedClusterAvailableResources in interface ISchedulingState

Parameters:

blacklistedSupervisorIds - other ids that are tentatively blacklisted.

getClusterTotalCpuResource

public double getClusterTotalCpuResource()

Description copied from interface: ISchedulingState

Get the total amount of CPU resources in cluster.

Specified by:

getClusterTotalCpuResource in interface ISchedulingState

getClusterTotalMemoryResource

public double getClusterTotalMemoryResource()

Description copied from interface: ISchedulingState

Get the total amount of memory resources in cluster.

Specified by:

getClusterTotalMemoryResource in interface ISchedulingState

getClusterTotalGenericResources

public Map<String,Double> getClusterTotalGenericResources()

Description copied from interface: ISchedulingState

Get the total amount of generic resources (excluding CPU and memory) in cluster.

Specified by:

getClusterTotalGenericResources in interface ISchedulingState

getNetworkTopography

public Map<String,List<String>> getNetworkTopography()

Description copied from interface: ISchedulingState

Get the network topography (rackId -> nodes in the rack).

Specified by:

getNetworkTopography in interface ISchedulingState

setNetworkTopography

public void setNetworkTopography(Map<String,List<String>> networkTopography)

setStatus

public void setStatus(TopologyDetails td,
                      String statusMessage)

set scheduler status for a topology.

setStatus

public void setStatus(String topologyId,
                      String statusMessage)

set scheduler status for a topology.

setStatusIfAbsent

public void setStatusIfAbsent(String topologyId,
                              String statusMessage)

getStatusMap

public Map<String,String> getStatusMap()

Description copied from interface: ISchedulingState

Get all topology scheduler statuses.

Specified by:

getStatusMap in interface ISchedulingState

setStatusMap

public void setStatusMap(Map<String,String> statusMap)

set scheduler status map.

getStatus

public String getStatus(String topoId)

getTopologyResourcesMap

public Map<String,TopologyResources> getTopologyResourcesMap()

Description copied from interface: ISchedulingState

Get the amount of resources used by topologies. Used for displaying resource information on the UI.

Specified by:

getTopologyResourcesMap in interface ISchedulingState

Returns:

a map that contains multiple topologies and the resources the topology requested and assigned. Key: topology id Value: an array that describes the resources the topology requested and assigned in the following format: {requestedMemOnHeap, requestedMemOffHeap, requestedCpu, assignedMemOnHeap, assignedMemOffHeap, assignedCpu}

getSupervisorsResourcesMap

public Map<String,SupervisorResources> getSupervisorsResourcesMap()

Description copied from interface: ISchedulingState

Get the amount of used and free resources on a supervisor. Used for displaying resource information on the UI

Specified by:

getSupervisorsResourcesMap in interface ISchedulingState

Returns:

a map where the key is the supervisor id and the value is a map that represents resource usage for a supervisor in the following format: {totalMem, totalCpu, usedMem, usedCpu}

getWorkerResourcesMap

public Map<String,Map<WorkerSlot,WorkerResources>> getWorkerResourcesMap()

Description copied from interface: ISchedulingState

Gets the reference to the full topology->worker resource map.

Specified by:

getWorkerResourcesMap in interface ISchedulingState

Returns:

map of topology -> map of worker slot ->resources for that worker

getWorkerResources

public WorkerResources getWorkerResources(WorkerSlot ws)

Description copied from interface: ISchedulingState

Get the resources for a given slot.

Specified by:

getWorkerResources in interface ISchedulingState

Parameters:

ws - the slot

Returns:

the resources currently assigned

getResourceMetrics

public ResourceMetrics getResourceMetrics()

getAllScheduledResourcesForNode

public NormalizedResourceRequest getAllScheduledResourcesForNode(String nodeId)

Description copied from interface: ISchedulingState

Get all scheduled resources for node.

Specified by:

getAllScheduledResourcesForNode in interface ISchedulingState

getScheduledMemoryForNode

public double getScheduledMemoryForNode(String nodeId)

Description copied from interface: ISchedulingState

Get the total memory currently scheduled on a node.

Specified by:

getScheduledMemoryForNode in interface ISchedulingState

Parameters:

nodeId - the id of the node

Returns:

the total memory currently scheduled on the node

getScheduledCpuForNode

public double getScheduledCpuForNode(String nodeId)

Description copied from interface: ISchedulingState

Get the total cpu currently scheduled on a node.

Specified by:

getScheduledCpuForNode in interface ISchedulingState

Parameters:

nodeId - the id of the node

Returns:

the total cpu currently scheduled on the node

getINimbus

public INimbus getINimbus()

getConf

public Map<String,Object> getConf()

Description copied from interface: ISchedulingState

Get the nimbus configuration.

Specified by:

getConf in interface ISchedulingState

unassign

public void unassign(String topoId)

Unassign everything for the given topology id.

Parameters:

topoId - the is of the topology to unassign

updateFrom

public void updateFrom(Cluster other)

Update the assignments and status from the other cluster.

Parameters:

other - the cluster to get the assignments and status from

getMinWorkerCpu

public double getMinWorkerCpu()

getGreyListedSupervisors

public List<String> getGreyListedSupervisors()

setGreyListedSupervisors

public void setGreyListedSupervisors(Set<String> greyListedSupervisors)