org.apache.storm.daemon.supervisor

Class BasicContainer

java.lang.Object

org.apache.storm.daemon.supervisor.Container

org.apache.storm.daemon.supervisor.BasicContainer

All Implemented Interfaces:

Killable

Direct Known Subclasses:

RunAsUserContainer

public class BasicContainer
extends Container

A container that runs processes on the local box.

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.storm.daemon.supervisor.Container

Container.ContainerType

Field Summary

Fields

Modifier and Type	Field and Description
protected boolean	_exitedEarly
protected LocalState	_localState
protected String	_profileCmd
protected String	_stormHome
protected double	hardMemoryLimitMultiplier
protected long	hardMemoryLimitOver
protected long	lowMemoryThresholdMB
protected long	mediumMemoryGracePeriodMs
protected long	mediumMemoryThresholdMb
protected long	memoryLimitExceededStart
protected long	memoryLimitMB

Fields inherited from class org.apache.storm.daemon.supervisor.Container

_assignment, _conf, _ops, _port, _resourceIsolationManager, _supervisorId, _supervisorPort, _symlinksDisabled, _topoConf, _topologyId, _type, _workerId, containerMemoryTracker

Constructor Summary

Constructors

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

Method Summary

Modifier and Type	Method and Description
void	cleanUpForRestart() Clean up the container partly preparing for restart.
protected void	createNewWorkerId() Create a new worker ID for this process and store in in this object and in the local state.
boolean	didMainProcessExit() Return true if the main process exited, else false.
protected List<String>	frameworkClasspath(SimpleVersion topoVersion)
static List<String>	getDependencyLocationsFor(Map<String,Object> conf, String topologyId, AdvancedFSOps ops, String stormRoot)
long	getMemoryReservationMb() Get the current memory reservation of this container.
long	getMemoryUsageMb() Get the current memory usage of this container.
static String	getStormVersionFor(Map<String,Object> conf, String topologyId, AdvancedFSOps ops, String stormRoot)
protected String	getWildcardDir(File dir) Returns a path with a wildcard as the final element, so that the JVM will expand that to all JARs in the directory.
protected String	getWorkerClassPath(String stormJar, List<String> dependencyLocations, SimpleVersion topoVersion) Compute the classpath for the worker process.
protected String	getWorkerLogWriter(SimpleVersion topoVersion)
protected String	getWorkerMain(SimpleVersion topoVersion)
boolean	isMemoryLimitViolated(LocalAssignment withUpdatedLimits) Check if the container is over its memory limit AND needs to be killed.
protected String	javaCmd(String cmd)
protected String	javaLibraryPath(String stormRoot, Map<String,Object> conf) Compute the java.library.path that should be used for the worker.
void	launch() Launch the process for the first time.
protected void	launchWorkerProcess(List<String> command, Map<String,String> env, String logPrefix, ExitCodeCallback processExitCallback, File targetDir) Launch the worker process (non-blocking).
void	relaunch() Restart the processes in this container.
boolean	runProfiling(ProfileRequest request, boolean stop) Run a profiling request.
protected boolean	runProfilingCommand(List<String> command, Map<String,String> env, String logPrefix, File targetDir) Run the given command for profiling.
protected List<String>	substituteChildopts(Object value)
protected List<String>	substituteChildopts(Object value, int memOnheap)

Methods inherited from class org.apache.storm.daemon.supervisor.Container

areAllProcessesDead, cleanUp, createArtifactsLink, createBlobstoreLinks, deleteSavedWorkerUser, forceKill, forceKill, getAllPids, getRunWorkerAsUser, getTotalTopologyMemoryReserved, getTotalTopologyMemoryUsed, getTotalWorkersForThisTopology, getWorkerId, getWorkerUser, isProcessAlive, kill, kill, readHeartbeat, readTopoConf, saveWorkerUser, setup, toString, updateMemoryAccounting, writeLogMetadata

Methods inherited from class java.lang.Object

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

Field Detail

_localState

protected final LocalState _localState

_profileCmd

protected final String _profileCmd

_stormHome

protected final String _stormHome

hardMemoryLimitMultiplier

protected final double hardMemoryLimitMultiplier

hardMemoryLimitOver

protected final long hardMemoryLimitOver

lowMemoryThresholdMB

protected final long lowMemoryThresholdMB

mediumMemoryThresholdMb

protected final long mediumMemoryThresholdMb

mediumMemoryGracePeriodMs

protected final long mediumMemoryGracePeriodMs

_exitedEarly

protected volatile boolean _exitedEarly

memoryLimitMB

protected volatile long memoryLimitMB

memoryLimitExceededStart

protected volatile long memoryLimitExceededStart

Constructor Detail

BasicContainer

public BasicContainer(Container.ContainerType type,
                      Map<String,Object> conf,
                      String supervisorId,
                      int supervisorPort,
                      int port,
                      LocalAssignment assignment,
                      ResourceIsolationInterface resourceIsolationManager,
                      LocalState localState,
                      String workerId,
                      StormMetricsRegistry metricsRegistry,
                      ContainerMemoryTracker containerMemoryTracker)
               throws IOException

Create a new BasicContainer.

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

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.

localState - the local state of the supervisor. May be null if partial recovery

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

metricsRegistry - The metrics registry.

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

Throws:

IOException

Method Detail

getDependencyLocationsFor

public static List<String> getDependencyLocationsFor(Map<String,Object> conf,
                                                     String topologyId,
                                                     AdvancedFSOps ops,
                                                     String stormRoot)
                                              throws IOException

Throws:

IOException

getStormVersionFor

public static String getStormVersionFor(Map<String,Object> conf,
                                        String topologyId,
                                        AdvancedFSOps ops,
                                        String stormRoot)
                                 throws IOException

Throws:

IOException

createNewWorkerId

protected void createNewWorkerId()

Create a new worker ID for this process and store in in this object and in the local state. Never call this if a worker is currently up and running. We will lose track of the process.

cleanUpForRestart

public void cleanUpForRestart()
                       throws IOException

Description copied from class: Container

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

Overrides:

cleanUpForRestart in class Container

Throws:

IOException - on any error

relaunch

public void relaunch()
              throws IOException

Description copied from class: Container

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

Specified by:

relaunch in class Container

Throws:

IOException - on any error

didMainProcessExit

public boolean didMainProcessExit()

Description copied from class: Container

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

Specified by:

didMainProcessExit in class Container

runProfilingCommand

protected boolean runProfilingCommand(List<String> command,
                                      Map<String,String> env,
                                      String logPrefix,
                                      File targetDir)
                               throws IOException,
                                      InterruptedException

Run the given command for profiling.

Parameters:

command - the command to run

env - the environment to run the command

logPrefix - the prefix to include in the logs

targetDir - the working directory to run the command in

Returns:

true if it ran successfully, else false

Throws:

IOException - on any error

InterruptedException - if interrupted wile waiting for the process to exit.

runProfiling

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

Description copied from class: Container

Run a profiling request.

Specified by:

runProfiling in class Container

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.

javaLibraryPath

protected String javaLibraryPath(String stormRoot,
                                 Map<String,Object> conf)

Compute the java.library.path that should be used for the worker. This helps it to load JNI libraries that are packaged in the uber jar.

Parameters:

stormRoot - the root directory of the worker process

conf - the config for the supervisor.

Returns:

the java.library.path/LD_LIBRARY_PATH to use so native libraries load correctly.

getWildcardDir

protected String getWildcardDir(File dir)

Returns a path with a wildcard as the final element, so that the JVM will expand that to all JARs in the directory.

Parameters:

dir - the directory to which a wildcard will be appended

Returns:

the path with wildcard (“*”) suffix

frameworkClasspath

protected List<String> frameworkClasspath(SimpleVersion topoVersion)

getWorkerMain

protected String getWorkerMain(SimpleVersion topoVersion)

getWorkerLogWriter

protected String getWorkerLogWriter(SimpleVersion topoVersion)

getWorkerClassPath

protected String getWorkerClassPath(String stormJar,
                                    List<String> dependencyLocations,
                                    SimpleVersion topoVersion)

Compute the classpath for the worker process.

Parameters:

stormJar - the topology jar

dependencyLocations - any dependencies from the topology

topoVersion - the version of the storm framework to use

Returns:

the full classpath

substituteChildopts

protected List<String> substituteChildopts(Object value)

substituteChildopts

protected List<String> substituteChildopts(Object value,
                                           int memOnheap)

launchWorkerProcess

protected void launchWorkerProcess(List<String> command,
                                   Map<String,String> env,
                                   String logPrefix,
                                   ExitCodeCallback processExitCallback,
                                   File targetDir)
                            throws IOException

Launch the worker process (non-blocking).

Parameters:

command - the command to run

env - the environment to run the command

processExitCallback - a callback for when the process exits

logPrefix - the prefix to include in the logs

targetDir - the working directory to run the command in

Throws:

IOException - on any error

javaCmd

protected String javaCmd(String cmd)

isMemoryLimitViolated

public boolean isMemoryLimitViolated(LocalAssignment withUpdatedLimits)
                              throws IOException

Description copied from class: Container

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.

Overrides:

isMemoryLimitViolated in class Container

Throws:

IOException - on any error

getMemoryUsageMb

public long getMemoryUsageMb()

Description copied from class: Container

Get the current memory usage of this container.

Overrides:

getMemoryUsageMb in class Container

getMemoryReservationMb

public long getMemoryReservationMb()

Description copied from class: Container

Get the current memory reservation of this container.

Overrides:

getMemoryReservationMb in class Container

launch

public void launch()
            throws IOException

Description copied from class: Container

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

Specified by:

launch in class Container

Throws:

IOException - on any error