THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
Status
Current state: [ UNDER DISCUSSION ]ACCEPTED
Discussion thread: http://mail-archives.apache.org/mod_mbox/samza-dev/202001.mbox/browser
JIRA: SAMZA-2373
Released: TBDSamza 1.5
Problem
Samza operates in a multi-tenant environment with cluster managers like Yarn and Mesos where a single host can run multiple Samza containers. Often due to soft limits configured for cluster managers like Yarn and no notion of dynamic workload balancing in Samza a host lands in a situation where it is underperforming and it is desired to move one or more containers from that host to other hosts. Today this is not possible without affecting other jobs on the hot host or restarting the affected job manually. In other use cases like resetting checkpoints of a single container or supporting canary or rolling bounces the ability to restart a single or a subset of containers without restarting the whole job is highly desirable.
...
Pros | Cons |
|
|
Solution 2. Container Placement
...
API [Accepted]
API design
On the basis of types of Control actions, the commands are the following:
...
API | containerStatus |
Description | Gives the status & info of the container placement request, for ex is it running, stopped what control commands are issued on it |
Parameters | processor-id: Samza resource id of container e.g 0, 1, 2 deploymentId: unique identifier of the deployed app for which the action is taken |
Status code | BAD_REQUEST, SUCCEEDED |
Returns | Status of the Container placement action |
...
| Code Block | ||
|---|---|---|
| ||
place-container --deployment-id 1581635852024-5117e303 --app-name snjain-test-cp --app.id = 1 --processor-id 4 --request-expiry 10 --destination-host abc.prod.com |
...
| Code Block |
|---|
@CommandLine.Command(name = "place-container", description = "Request to move/restart container at destination-host")
public class ContainerPlacementTool {
...
_appName = // read from commandline
MetadataStore metadataStore_appId = buildMetadataStore();// read from commandline
_deploymentId = // read from commandline
_processorId = // read from commandline
_destinationHost = // read from commandline
_requestExpiry = // read from commandline
MetadataStore metadataStore = buildMetadataStore(_appName, _appId);
try {
ContainerPlacementMetadataStore containerPlacementMetadataStore =
new ContainerPlacementMetadataStore(metadataStore);
containerPlacementMetadataStore.start();
Duration requestExpiry = _requestExpiry != null ? Duration.ofSeconds(_requestExpiry) : null;
UUID uuid = containerPlacementMetadataStore.writeContainerPlacementRequestMessage(_deploymentId, _processorId,
_destinationHost, _requestExpiry, System.currentTimeMillis());
System.out.println("Request received query the status using: " + uuid);
} finally {
samzaJob.close();
metadataStore.close();
}
} |
Public Interfaces
| Code Block | ||||||
|---|---|---|---|---|---|---|
| ||||||
/**
* Encapsulates the request or response payload information between the ContainerPlacementHandler service and external
* controllers issuing placement actions
*/
@InterfaceStability.Evolving
public abstract class ContainerPlacementMessage {
public enum StatusCode {
/**
* Indicates that the container placement action is created
*/
CREATED,
/**
* Indicates that the container placement action was rejected because request was deemed invalid
*/
BAD_REQUEST,
/**
* Indicates that the container placement action is accepted and waiting to be processed
*/
ACCEPTED,
/**
* Indicates that the container placement action is in progress
*/
IN_PROGRESS,
/**
* Indicates that the container placement action is in progress
*/
SUCCEEDED,
/**
* Indicates that the container placement action is in failed
*/
FAILED;
}
/**
* UUID attached to a message which helps in identifying duplicate request messages written to metastore and not
* retake actions even if metastore is eventually consistent
*/
protected final UUID uuid;
/**
* Unique identifier for a deployment so messages can be invalidated across a job restarts
* for ex yarn bases cluster manager should set this to app attempt id
*/
protected final String deploymentId;
// Logical container Id 0, 1, 2
protected final String processorId;
// Destination host where container is desired to be moved
protected final String destinationHost;
// Optional request expiry which acts as a timeout for any resource request to cluster resource manager
protected final Duration requestExpiry;
// Status of the current request
protected final StatusCode statusCode;
// Timestamp of the request or response message
protected final long timestamp;
protected ContainerPlacementMessage(UUID uuid, String deploymentId, String processorId, String destinationHost,
Duration requestExpiry, StatusCode statusCode, long timestamp) {…}
} |
...