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

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

_assignment, _conf, _ops, _port, _supervisorId, _symlinksDisabled, _topoConf, _topologyId, _type, _workerId

Constructor Summary

Constructors

Constructor and Description
BasicContainer(Container.ContainerType type, Map<String,Object> conf, String supervisorId, int port, LocalAssignment assignment, LocalState localState, String workerId) 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()
protected List<String>	frameworkClasspath()
static List<String>	getDependencyLocationsFor(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) Compute the classpath for the worker process
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 PREREQUISITE: setup has run and passed
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 PREREQUISITE: cleanUpForRestart has run and passed
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, getWorkerId, getWorkerUser, isProcessAlive, kill, kill, readHeartbeat, readTopoConf, saveWorkerUser, setup, toString, 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

_exitedEarly

protected volatile boolean _exitedEarly

Constructor Detail

BasicContainer

public BasicContainer(Container.ContainerType type,
                      Map<String,Object> conf,
                      String supervisorId,
                      int port,
                      LocalAssignment assignment,
                      LocalState localState,
                      String workerId)
               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.

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.

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.

Throws:

IOException

Method Detail

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()

Specified by:

didMainProcessExit in class Container

Returns:

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

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()

getWorkerClassPath

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

Compute the classpath for the worker process

Parameters:

stormJar - the topology jar

dependencyLocations - any dependencies from the topology

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

getDependencyLocationsFor

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

Throws:

IOException

javaCmd

protected String javaCmd(String cmd)

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