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 AdvancedFSOps	_ops
protected int	_port
protected String	_supervisorId
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 port, LocalAssignment assignment, String workerId, Map<String,Object> topoConf, AdvancedFSOps ops) Create a new Container.

Method Summary

Modifier and Type	Method and Description
boolean	areAllProcessesDead()
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()
void	forceKill() Kill the processes in this container violently.
protected void	forceKill(long pid) Kill a given process
protected Set<Long>	getAllPids()
protected String	getRunWorkerAsUser() Returns the user that the worker process is running as.
String	getWorkerId()
protected String	getWorkerUser()
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 PREREQUISITE: setup has run and passed
LSWorkerHeartbeat	readHeartbeat() Read the Heartbeat for the current container.
protected Map<String,Object>	readTopoConf()
abstract void	relaunch() Restart the processes in this container PREREQUISITE: cleanUpForRestart has run and passed
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	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

_workerId

protected String _workerId

_topologyId

protected final String _topologyId

_supervisorId

protected final String _supervisorId

_port

protected final int _port

_assignment

protected final LocalAssignment _assignment

_ops

protected final AdvancedFSOps _ops

_type

protected Container.ContainerType _type

_symlinksDisabled

protected final boolean _symlinksDisabled

Constructor Detail

Container

protected Container(Container.ContainerType type,
                    Map<String,Object> conf,
                    String supervisorId,
                    int port,
                    LocalAssignment assignment,
                    String workerId,
                    Map<String,Object> topoConf,
                    AdvancedFSOps ops)
             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.

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.

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

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

forceKill

protected void forceKill(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

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

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

Returns:

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

Throws:

IOException

getWorkerUser

protected String getWorkerUser()
                        throws IOException

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

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

Returns:

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

Returns:

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