org.apache.storm.daemon.supervisor

Class Container

java.lang.Object

org.apache.storm.daemon.supervisor.Container

All Implemented Interfaces:

Killable

Direct Known Subclasses:

BasicContainer, LocalContainer

public abstract class Container
extends Object
implements Killable

Represents a container that a worker will run in.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	Container.ContainerType

Field Summary

Fields

Modifier and Type	Field and Description
protected LocalAssignment	assignment
protected Map<String,Object>	conf
protected ContainerMemoryTracker	containerMemoryTracker
protected AdvancedFSOps	ops
protected int	port
protected ResourceIsolationInterface	resourceIsolationManager
protected String	supervisorId
protected int	supervisorPort
protected boolean	symlinksDisabled
protected Map<String,Object>	topoConf
protected String	topologyId
protected Container.ContainerType	type
protected String	workerId

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	Container(Container.ContainerType type, Map<String,Object> conf, String supervisorId, int supervisorPort, int port, LocalAssignment assignment, ResourceIsolationInterface resourceIsolationManager, String workerId, Map<String,Object> topoConf, AdvancedFSOps ops, StormMetricsRegistry metricsRegistry, ContainerMemoryTracker containerMemoryTracker) Create a new Container.

Method Summary

Modifier and Type	Method and Description
boolean	areAllProcessesDead() Check whether all processes are dead.
void	cleanUp() Clean up the container.
void	cleanUpForRestart() Clean up the container partly preparing for restart.
protected void	createArtifactsLink() Create symlink from the containers directory/artifacts to the artifacts directory.
protected void	createBlobstoreLinks() Create symlinks for each of the blobs from the container’s directory to corresponding links in the storm dist directory.
protected void	deleteSavedWorkerUser()
abstract boolean	didMainProcessExit() Return true if the main process exited, else false.
void	forceKill() Kill the processes in this container violently.
protected void	forceKill(long pid) Kill a given process.
protected Set<Long>	getAllPids() Get all PIDs.
long	getMemoryReservationMb() Get the current memory reservation of this container.
long	getMemoryUsageMb() Get the current memory usage of this container.
protected String	getRunWorkerAsUser() Returns the user that the worker process is running as.
long	getTotalTopologyMemoryReserved(LocalAssignment withUpdatedLimits) Get the total memory reserved.
long	getTotalTopologyMemoryUsed() Get the total memory used (on and off heap).
long	getTotalWorkersForThisTopology() Get the number of workers for this topology.
String	getWorkerId() Get the id of the container or null if there is no worker id right now.
protected String	getWorkerUser() Get worker user.
boolean	isMemoryLimitViolated(LocalAssignment withUpdatedLimits) Check if the container is over its memory limit AND needs to be killed.
protected boolean	isProcessAlive(long pid, String user) Is a process alive and running?.
void	kill() Kill the processes in this container nicely.
protected void	kill(long pid) Kill a given process.
abstract void	launch() Launch the process for the first time.
LSWorkerHeartbeat	readHeartbeat() Read the Heartbeat for the current container.
protected Map<String,Object>	readTopoConf()
abstract void	relaunch() Restart the processes in this container.
abstract boolean	runProfiling(ProfileRequest request, boolean stop) Run a profiling request.
protected void	saveWorkerUser(String user)
protected void	setup() Setup the container to run.
String	toString()
protected void	updateMemoryAccounting()
protected void	writeLogMetadata(String user) Write out the file used by the log viewer to allow/reject log access.

Methods inherited from class java.lang.Object

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

Field Detail

conf

protected final Map<String,Object> conf

topoConf

protected final Map<String,Object> topoConf

topologyId

protected final String topologyId

supervisorId

protected final String supervisorId

supervisorPort

protected final int supervisorPort

port

protected final int port

assignment

protected final LocalAssignment assignment

ops

protected final AdvancedFSOps ops

resourceIsolationManager

protected final ResourceIsolationInterface resourceIsolationManager

symlinksDisabled

protected final boolean symlinksDisabled

workerId

protected String workerId

type

protected Container.ContainerType type

containerMemoryTracker

protected ContainerMemoryTracker containerMemoryTracker

Constructor Detail

Container

protected Container(Container.ContainerType type,
                    Map<String,Object> conf,
                    String supervisorId,
                    int supervisorPort,
                    int port,
                    LocalAssignment assignment,
                    ResourceIsolationInterface resourceIsolationManager,
                    String workerId,
                    Map<String,Object> topoConf,
                    AdvancedFSOps ops,
                    StormMetricsRegistry metricsRegistry,
                    ContainerMemoryTracker containerMemoryTracker)
             throws IOException

Create a new Container.

Parameters:

type - the type of container being made.

conf - the supervisor config

supervisorId - the ID of the supervisor this is a part of.

supervisorPort - the thrift server port of the supervisor this is a part of.

port - the port the container is on. Should be <= 0 if only a partial recovery @param assignment the assignment for this container. Should be null if only a partial recovery.

resourceIsolationManager - used to isolate resources for a container can be null if no isolation is used.

workerId - the id of the worker to use. Must not be null if doing a partial recovery.

topoConf - the config of the topology (mostly for testing) if null and not a partial recovery the real conf is read.

ops - file system operations (mostly for testing) if null a new one is made

metricsRegistry - The metrics registry.

containerMemoryTracker - The shared memory tracker for the supervisor’s containers

Throws:

IOException - on any error.

Method Detail

toString

public String toString()

Overrides:

toString in class Object

readTopoConf

protected Map<String,Object> readTopoConf()
                                   throws IOException

Throws:

IOException

kill

protected void kill(long pid)
             throws IOException

Kill a given process.

Parameters:

pid - the id of the process to kill

Throws:

IOException

kill

public void kill()
          throws IOException

Description copied from interface: Killable

Kill the processes in this container nicely. kill -15 equivalent

Specified by:

kill in interface Killable

Throws:

IOException - on any error

forceKill

protected void forceKill(long pid)
                  throws IOException

Kill a given process.

Parameters:

pid - the id of the process to kill

Throws:

IOException

forceKill

public void forceKill()
               throws IOException

Description copied from interface: Killable

Kill the processes in this container violently. kill -9 equivalent

Specified by:

forceKill in interface Killable

Throws:

IOException - on any error

readHeartbeat

public LSWorkerHeartbeat readHeartbeat()
                                throws IOException

Read the Heartbeat for the current container.

Returns:

the Heartbeat

Throws:

IOException - on any error

isProcessAlive

protected boolean isProcessAlive(long pid,
                                 String user)
                          throws IOException

Is a process alive and running?.

Parameters:

pid - the PID of the running process

user - the user that is expected to own that process

Returns:

true if it is, else false

Throws:

IOException - on any error

areAllProcessesDead

public boolean areAllProcessesDead()
                            throws IOException

Description copied from interface: Killable

Check whether all processes are dead.

Specified by:

areAllProcessesDead in interface Killable

Returns:

true if all of the processes are dead, else false

Throws:

IOException - on any error

cleanUp

public void cleanUp()
             throws IOException

Description copied from interface: Killable

Clean up the container. It is not coming back. by default do the same thing as when restarting.

Specified by:

cleanUp in interface Killable

Throws:

IOException - on any error

setup

protected void setup()
              throws IOException

Setup the container to run. By default this creates the needed directories/links in the local file system PREREQUISITE: All needed blobs and topology, jars/configs have been downloaded and placed in the appropriate locations

Throws:

IOException - on any error

writeLogMetadata

protected void writeLogMetadata(String user)
                         throws IOException

Write out the file used by the log viewer to allow/reject log access.

Parameters:

user - the user this is going to run as

Throws:

IOException - on any error

createArtifactsLink

protected void createArtifactsLink()
                            throws IOException

Create symlink from the containers directory/artifacts to the artifacts directory.

Throws:

IOException - on any error

createBlobstoreLinks

protected void createBlobstoreLinks()
                             throws IOException

Create symlinks for each of the blobs from the container’s directory to corresponding links in the storm dist directory.

Throws:

IOException - on any error.

getAllPids

protected Set<Long> getAllPids()
                        throws IOException

Get all PIDs.

Returns:

all of the pids that are a part of this container.

Throws:

IOException

getWorkerUser

protected String getWorkerUser()
                        throws IOException

Get worker user.

Returns:

the user that some operations should be done as.

Throws:

IOException - on any error

getRunWorkerAsUser

protected String getRunWorkerAsUser()

Returns the user that the worker process is running as.

The default behavior is to launch the worker as the user supervisor is running as (e.g. ‘storm’)

Returns:

the user that the worker process is running as.

saveWorkerUser

protected void saveWorkerUser(String user)
                       throws IOException

Throws:

IOException

deleteSavedWorkerUser

protected void deleteSavedWorkerUser()
                              throws IOException

Throws:

IOException

cleanUpForRestart

public void cleanUpForRestart()
                       throws IOException

Clean up the container partly preparing for restart. By default delete all of the temp directories we are going to get a new worker_id anyways. POST CONDITION: the workerId will be set to null

Throws:

IOException - on any error

isMemoryLimitViolated

public boolean isMemoryLimitViolated(LocalAssignment withUpdatedLimits)
                              throws IOException

Check if the container is over its memory limit AND needs to be killed. This does not necessarily mean that it just went over the limit.

Throws:

IOException - on any error

updateMemoryAccounting

protected void updateMemoryAccounting()

getTotalTopologyMemoryUsed

public long getTotalTopologyMemoryUsed()

Get the total memory used (on and off heap).

getTotalTopologyMemoryReserved

public long getTotalTopologyMemoryReserved(LocalAssignment withUpdatedLimits)

Get the total memory reserved.

Parameters:

withUpdatedLimits - the local assignment with shared memory

Returns:

the total memory reserved.

getTotalWorkersForThisTopology

public long getTotalWorkersForThisTopology()

Get the number of workers for this topology.

getMemoryUsageMb

public long getMemoryUsageMb()

Get the current memory usage of this container.

getMemoryReservationMb

public long getMemoryReservationMb()

Get the current memory reservation of this container.

launch

public abstract void launch()
                     throws IOException

Launch the process for the first time. PREREQUISITE: setup has run and passed

Throws:

IOException - on any error

relaunch

public abstract void relaunch()
                       throws IOException

Restart the processes in this container. PREREQUISITE: cleanUpForRestart has run and passed

Throws:

IOException - on any error

didMainProcessExit

public abstract boolean didMainProcessExit()

Return true if the main process exited, else false. This is just best effort return false if unknown.

runProfiling

public abstract boolean runProfiling(ProfileRequest request,
                                     boolean stop)
                              throws IOException,
                                     InterruptedException

Run a profiling request.

Parameters:

request - the request to run

stop - is this a stop request?

Returns:

true if it succeeded, else false

Throws:

IOException - on any error

InterruptedException - if running the command is interrupted.

getWorkerId

public String getWorkerId()

Get the id of the container or null if there is no worker id right now.