...
This doc addresses the problem of restarting/moving a single container of a job without affecting other containers of the same job or other job jobs running on the same host. X
...
- Build the ability to move container (non-am) of a running job from its existing host to a preferred host (if specified) or any host
- If an existing stateful container has a standby container, move to the active container to stand by standby container
- If an existing stateful container is without a standby container, spin up a standby container and then move the active to standBy once stand by has bootstrapped to reduce downtime [Stretch]
- Build the ability to restart containers with the same host preferences
- Build the ability to selectively spin up standby for one or a subset of containers [Stretch]
- Expose these APIs via a tool or dashboard for Samza team to easily use during oncall operability
Non-Goals
- This system does not intend to solve the problem of Dynamic Workload Balance i.e Cruise control. It may act as a building block for one later.
- Solving the canary problem for YARN based deployment mode model is out of the scope of this solution however system built should be easily extensible to support canary
- This system will not have a built-in intelligence to find a better match for host for a container it will take make simplistic decisions as per params passed by the user.
SLA / SCALE / LIMITS
- At a time AM for a single job will only serve one request per container, parallel requests across containers are still supported. If a control request is underway any other requests issued on the same container will be discardedqueued
- System The system should be capable of scaling to be used across different jobs at the same time
...
- Should be able to handle multiple move request for different containers at the time for a single job
- Should provide the good monitoring mechanism
- Should be general enough to be coupled with any system for example Control -planePlane
- Should allow queuing requests for multiple containers across multiple jobs at the same time
- Should be easily extensible to build canary support for yarn deployments
...
- Each control action will be associated with a deployment id of an app it is intended to be issued upon
- Across Job restarts, any requests issued for the previous deployment incarnation will not be respected. This includes AM restarts (today if the AM restarts that implies a job restart)
Glossary
Term | Description |
Cluster Manager | Resource manager used by samza eg. Yarn, Mesos, etc |
Job Coordinator (JC) { Also referred to as ApplicationMaster (AM) for Yarn } | Each Samza application has a JC that manages the application’s workload, asks RM for containers and handles notifications when one of its containers fails. |
Node Manager (NM) | A single node in a Yarn Cluster |
Resource Manager (RM) | Central Service that provides like scheduling, heartbeats, liveness monitoring to all applications running in the YARN cluster |
Host Affinity | The ability of Samza to allocate a container to the same machine across job restarts is referred to as host-affinity. |
Container processorId | Samza allocates containers resource Ids eg 0, 1, 2 which remains the same across restarts of a job |
Proposed
...
API design
On the basis of types of Control actions, the commands are the following:
1. Restart Container
...
API
...
restartContainer
...
Description
...
Active Container: Stop container process on source-host and starts it for
- Stateless Job on
- Destination-host (if specified, destination host can be source as well)
- Any host (otherwise except source-host)
- Stateful Job on
- Destination-host (if specified, destination host can be source as well)
- Standby Container (if present)
- Any host (otherwise except source host)
StandBy Container: Stop container process on source-host and starts it on:
- Destination-host (if specified & matches StandBy Constraints)
- Any host (otherwise which matches StandBy Constraints)
...
Parameters
...
processor-id: Samza resource id of container e.g 0, 1, 2
destination-host: container host url [optional]
...
Status code
...
202 : Accepted
401 : Unauthorized
409 : Forbidden (In case already a Control Action is active on a Container)
...
Returns
...
Since this is an ASYNC API, this command returns a request-id which user can then query for status
...
Failure Scenarios
...
There are two cases under which a request to restart might fail:
- When active container stop fails, in this case we mark the request failed
- When stopped active container fails to start on destination host in that case we can set a flexible policy to either attempt a start on source host or restart anywhere
Solution:
Solution 1. Write locality to Coordinator Stream (Metastore) and restart job [Rejected]
Pros | Cons |
|
|
Solution 2. Container Placement Service [Accepted]
API design
On the basis of types of Control actions, the commands are the following:
1. Container Placement Actions (Move / Restart)
API | registerContainerPlacementAction |
Description | Active Container: Stop container process on source-host and starts it for
StandBy Container: Stop container process on source-host and starts it on:
|
Parameters | uuid: unique identifier a request, populated by the client applicationId: unique identifier of the deployed app for which the action is taken |
Note: For supporting canary above parameter list can be easily extended to support following parameters
Parameters | user-version: user application version [optional] samza-version: samza framework version [optional] jvm-args: arbitrary string to be used as jvm arguments [optional] |
3. Container Status
API | containerStatus |
Description | Gives the status & info of the container, 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 | |
Status code | 202 : Accepted 401 : Unauthorized |
2. Enable & Disable StandBy [Stretch]
...
API
...
controlStandBy
...
Description
...
Starts or Stops a standBy container for the active container
...
Parameters
...
processor-id: Samza resource id of container e.g 0, 1, 2
...
Status code
...
202 : Accepted
401 : Unauthorized
Architecture
For implementing a scalable container placement control system, the proposed solution is divided into two parts:
Part 1. Container Placement Handler
- Control Plane as described plane above the job that allows taking control actions by multiple controllers like Samza Dashboard, Start points controller.
- ContainerPlacementHandler is handler registered to control plane that translates control actions to invoke Container Placement Service APIs
This control plane can be implemented in the following ways
Option 1: Samza metastore service [Preferred]
Samza Metastore will provide an API to write to the coordinator stream. One simple way to expose Container Placement API is, Container Placement handler can have a coordinator stream consumer polling control messages from coordinator stream & acting on them. CPH will take actions maintaining some in-memory state with Container Placement Service to not take an action twice. In addition since control actions are associated with a deployment id, they are automatically invalidated across restarts
...
Pros
...
Cons
...
- No need to build Authentication & Authorization, already handled by the Metadata auth service
- No need to enable Rate limiting since requests are queued so the flow of requests can be regulated at the consumer side
...
- If AM dies there can be still queued requests in Coordinator Stream, such requests have to be handled across AM restarts
- Coordinator stream is log compacted so control messages written to the coordinator stream need to be deleted to prevent it from growing to large sizes which can affect job start times
...
It's fairly simple to embed a REST endpoint in JC or extend the existing Rest endpoint in JC that exposes configs in JC to support apis listed above
...
Pros
...
Cons
...
- Simple to extend the existing REST endpoint
- No need to build Authentication since AM runs on hosts which are blacklisted for anyone except the Samza Team
- If the AM dies all the outstanding requests are discarded (no additional handling needed)
...
- Need to build Authorization layer around these rest endpoints
- Loading the already Heavy loaded Job coordinator with another service might cause an increase in memory used
- Need to build a service for discovery or rely on Yarn embedded Servlet
Open Sources Access:
To expose this API to open source users, a metadata store writer tool can be used. The same tool is going to give Open source users access to Start points APIs.
Part 2. Container Placement Service
Container Placement service is a set of API Built around AM to move/restart containers. The solution proposes to introduce a ContainerPlacementManager which is a single entity managing the container placements for stateful and stateless jobs & a metadata holder for container actions (ControlActionMetadata) forex request_id, current status, requested resources, etc.
2.1 Container Move
2.1.1 Stateless Container Move (Host Affinity Off)
Option 1: Reserve-Stop-Move: Request Resource first and Only make a move once resources are allocated by ClusterResource Manager (Preferred)
...
Pros
...
Cons
...
- Container only moves when a preferred resource is available for container to start
- Offers stronger move semantics more effective when used with a Workload balancer
...
- Complex implementation as compared to Option 2
Option 2: Stop-Reserve-Move: Stop the container and then make resource request for preferred resources and attempt a start (Rejected Alternative)
...
Pros
...
Cons
...
- Easier to implement
...
- Weaker Move semantics since if preferred resources are not available and since the container is already stopped then clearly this has a higher chance of moving container to anything other than a preferred host which is ineffective for a workload balancer
Orchestration of Stateless Container Move using Option 1: The image shows a sequence of steps involved in moving a container. Note that there is a specific number of retries configured for requesting preferred hosts to ClusterResourceManager.
- ContainerPlacementHandler dispatches the move request to ContainerProcessManager
- ContainerProcessManager registers move with ContainerPlacementManager and issues a resource request with ContainerAllocator (Async)
- Now there are two possible scenarios
3.1 Requested Resources are allocated by ClusterResourceManager
- In this case ContainerPlacementManager initiates a container shutdown and waits for the ContainerPlacementManager callback on successfully stopping the container
- ContainerProcessManager on receiving the callback checks if this was due to a move request & propagates it to ContainerPlacementManager which issues a start container request ClusterResourceManager to start container on allocated resources
- If the container start request succeeds then a success notification is sent (updating the Control Action Metadata) otherwise a new request to start this container on ANY_HOST is issued
3.2 Resource Request expires
- In cases where Resource Request expires and ClusterResourceManager is not able to allocate requested resources there a fixed number times a request is retried. If request does not get allocated even after retires, move is marked as failed and container stays on the host it is running on.
Failure Scenarios:
- If the ContainerPlacementManager is not able to stop the active container (3.1 #1 above fails) in that case we mark the control action request failed
- If ClusterResourceManager fails to start the stopped active container on accrued destination host, then we attempt to start the container back on source host, failing to do which will attempt to start the container on ANY_HOST
...
destination-host: valid hostname / “ANY_HOST” / “STANDBY” request-expiry-timeout: [optional]: timeout for any resource request to the cluster manager | |
Status code | CREATED, BAD_REQUEST, ACCEPTED, IN_PROGRESS, SUCCEEDED, FAILED |
Returns | Since this is an ASYNC API, the status can be queried by processorId using |
Failure Scenarios | There are following cases under which a request to place container might fail:
|
Note: For supporting canary above parameter list can be easily extended to support the following parameters
Parameters | user-version: user application version [optional] samza-version: samza framework version [optional] jvm-args: arbitrary string to be used as jvm arguments [optional] |
2. Container Status
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 applicationId: unique identifier of the deployed app for which the action is taken uuid: unique identifier a request |
Status code | ACCEPTED, UNAUTHORIZED |
Returns | Status of the Container placement action |
2. Enable & Disable StandBy [Stretch]
API | controlStandBy |
Description | Starts or Stops a standBy container for the active container |
Parameters | processor-id: Samza resource id of container e.g 0, 1, 2 |
Status code | ACCEPTED, UNAUTHORIZED |
Architecture
For implementing a scalable container placement control system, the proposed solution is divided into two parts:
Part 1. Container Placement Handler
- Control Plane as described plane above the job that allows taking control actions by multiple controllers like Samza Dashboard, Start points controller.
- ContainerPlacementHandler is handler registered to control plane that translates control actions to invoke Container Placement Service APIs
This control plane can be implemented in the following ways
Option 1: Samza metastore service [Preferred]
Samza Metastore will provide an API to write to the coordinator stream. One simple way to expose Container Placement API is, Container Placement handler can have a coordinator stream consumer polling control messages from coordinator stream & acting on them. CPH will take actions maintaining some in-memory state with Container Placement Service to not take an action twice. In addition, since control actions are associated with a deployment id, they are automatically invalidated across restarts
Pros | Cons |
|
|
Option 2: Stub a Rest Endpoint in JC [Rejected Alternative]
It's fairly simple to embed a REST endpoint in JC or extend the existing Rest endpoint in JC that exposes configs in JC to support apis listed above
Pros | Cons |
|
|
Implementation Details:
- ContainerPlacementHandler is a stateless handler dispatching ContainerPlacementRequestMessages from Metastore to Container Placement Service & ContainerPlacementResponseMessages from Container Placement Service to metastore for external controls to query the status of an action. (PR).
- Metastore used today by in Samza by default is Kafka (coordinator stream) which is used to store configs & container mappings & is log compacted
- ContainerPlacementRequestMessage & ContainerPlacementResponseMessage are maintained in individual namespaces using NamespaceAwareMetaStore
- Key for storing the ContainerPlacementRequestMessage & ContainerPlacementResponseMessage in Metastore is chosen to be processorId (logical container id 0,1,2, etc) because at the worst the messages in metastore will be 2n where n is the size of containers, this choice to ensure job boot up times are not affected due to such messages
- One way to delete stale ContainerPlacementMessages is to delete messages from the previous incarnation in the metastore on job restarts
Open Sources Access:
To expose this API to open source users, a metadata store writer tool can be used. The same tool is going to give Open source users access to Start points APIs.
Part 2. Container Placement Service
Container Placement service is a set of APIs built around AM to move/restart containers. The solution proposes to refactor & simplify the current AM code & introduce a ContainerManager which is a single entity managing container actions like start, stop for both active and standby containers. Enlisted are functions of ContainerManager & proposed refactoring around the AM code
- Remove the HostAwareContainerAllocator & ContainerAllocator, simplify Container Allocator as a simple lightweight entity allocating requests to available resources (PR1, PR2)
- Introduce ContainerManager which acts as a brain for validating and issuing any actions on containers in the Job Coordinator for both active & Standby containers. (PR)
- Transfer state & validation of container launch & expired request handling from ContainerAllocator to ContainerManager
- Transfer state & validation for callback handler (ContainerProcessManager) to ContainerManager
- Encapsulates logic and state related to container placement actions like move, restarts for active & standby container in ContainerManager (PR-1, TDB)
- It is ContainerManager’s duty to validate any ContainerPlacementRequestMessages & also invalidate messages from the previous deployment incarnation
- It is ContainerManager’s duty to write ContainerPlacementResponseMessages to Metastore for external control to query the status of the request
- ContainerPlacementMetadata is a metadata holder for container actions (ControlActionMetadata) for ex request_id, current status, requested resources etc
2.1 Container Move
2.1.1 Stateless Container Move & Stateful Container Move (without Standby)
Option 1: Reserve-Stop-Move: Request Resource first and Only make a move once resources are allocated by ClusterResource Manager (Preferred)
Pros | Cons |
|
|
Option 2: Stop-Reserve-Move: Stop the container and then make a resource request for preferred resources and attempt a start (Rejected Alternative)
Pros | Cons |
|
|
Orchestration of Stateless Container Move using Option 1: The image shows a sequence of steps involved in moving a container. Note that there is a specific number of retries configured for requesting preferred hosts to ClusterResourceManager.
The image shows a sequence of steps involved in moving a container. Note that there is a specific number of retries configured for requesting preferred hosts to ClusterResourceManager.
- ContainerPlacementHandler dispatches the move request to ContainerProcessManager
- ContainerProcessManager dispatches placement request to Containermanager which validates the action and issues a preferred resource request with ContainerAllocator (Async)
- Now there are two possible scenarios
- Requested Resources are allocated by ClusterResourceManager
- In this case, ContainerManager initiates a container shutdown and waits for the ContainerProcessManager callback on successfully stopping the container
- On getting successful stop callback, ContainerManager issues a start container start request to ClusterResourceManage on allocated resources
- If the container start request succeeds then a success notification is sent (updating the Control Action Metadata) otherwise a new resource request to start this container on source host is issued & and a failure notification is sent to the user
- Requested Resources are allocated by ClusterResourceManager
- Resource Request expires
- In cases where Resource Request expires since ClusterResourceManager is not able to allocate requested resources a failure notification is sent and the active container remains unaffected
Failure Scenarios:
- If the preferred resources are not able to be accrued the active container is never stopped and a failure notification is sent for the ContainerPacementRequest
- If the ContainerPlacementManager is not able to stop the active container (3.1 #1 above fails) in that case the request is marked failed & a failure notification is sent for the ContainerPacementRequest
- If ClusterResourceManager fails to start the stopped active container on the accrued destination host, then we attempt to start the container back on the source host and a failure notification is sent for the ContainerPacementRequest. If container fails to start on source host then an attempt is made to start on ANY_HOST
Option 3: Stateful without Standby (Spin Up StandBy container & then move) (Phase 2) [Strech]
If we expose an API to spin a StandBy Container for an active container, a client can then make two requests, one to spin up a StandBy then periodically monitor the lag for StandByCotntainer (Using Diagnostics). Once StandBy Container has a steady lag is ready to move, this case becomes Case 1 mentioned above. (More details TBD later)
2.1.2 Stateful Container Move with Standby Enabled
The stateful container is divided into two cases on the basis of whether the job has StandBy Container for it or not
Let's take a case when the job has a stateful container C on H1 and has a StandBy container enabled C’ on H2
Option 1: Stop-Reserve-Move: Initiate a Stop on C & issue a Standby failover (Preferred - Phase 1*)
In this option, C is stopped on H1 and failover is initiated from H1 to H2 with current semantics of Failover of an active Container in the Hot standby feature. This ensures C moves either to H2 (in the best case) or to some other hosts (when Cluster Manager cannot allocate capacity on H2, i.e H2 is under capacity)
Pros | Cons |
|
|
Option 2: Reserve-Stop-Move: Initiate a Standby move first then issue a move for Active Container (Preferred - Phase 2*)
- In this option request is first issued to stop C’
- Then request H2 from cluster manager to start C
- Issue a stop on C only H2 can be allocated (#2 succeeds).
- Then request C’ to start ANY_HOST except for H1, H2
Two ways to achieve this:
- A client can make two calls for this first move C’ to ANY_HOST apart from H2, H1 (similar to a stateless move), then move the request of C to C’ (similar to a stateless move) so the state maintenance of this lives on the client
- Maintain state in JC to accomplish this
Pros | Cons |
|
|
* Phase 1 & Phase 2 refer to the implementation phases, please see the section: Implementation Plan
Option 3: Reserve-Stop-Move: First Request H2 from Cluster Manager and initiate the failover once you get H2 as a preferred resource [Rejected Alternative]
- In this option C’ is stopped on H2 and failover is initiated from H1 to H2 with current semantics as developed by the feature Hot standby
- In case of the move failed only affects standby
Pros | Cons |
|
|
Orchestration of StandBy Enabled Container Move using Option 1 (Phase 1)
- ContainerPlacementHandler dispatches the move request to ContainerProcessManager
- ContainerProcessManager registers a move & initiates a failover with StandByContainerManager then the following failover sequence is followed
2.2 Container Restart
This is much simpler as compared to Move. We can either make restart equivalent to a move or issue a stop a container and ContainerProcessManager will try to start that container again. Let's discuss these options in detail:
Option 1: Reserve-Stop-Start: Restart is equivalent to move on the same host [Preferred]
- In this option, resources are requested on the same host first before issuing a stop on Container
- Once the resources are accrued a container is stopped & requested to start
- So this is equivalent to move semantics above
Pros | Cons |
|
|
Both stateless & stateful container restarts will be equivalent to Stateless Container Move on the same host. Since the container is restarted on the same host then
Option 2: Stop-Reserve-Start: Restart is equivalent to stopping container first then attempting to start it [Rejected Alternative]
- In this option, a container is issued a stop first
- Once the container stops, resources are requested for starting on the same host (last seen host)
- Once the resources are accrued, the container is issued a start on the same host
Pros | Cons |
|
|
Public Interfaces
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
/**
* Encapsulates the request or response payload information between the ContainerPlacementHandler service and external
* controllers issuing placement actions
*/
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 applicationId;
// 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;
protected ContainerPlacementMessage(UUID uuid, String applicationId, String processorId, String destinationHost,
Duration requestExpiry, StatusCode statusCode) {…}
} |
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
/**
* Encapsulates the request sent from the external controller to the JobCoordinator to take a container placement action
*/
public class ContainerPlacementRequestMessage extends ContainerPlacementMessage {
public ContainerPlacementRequestMessage(UUID uuid, String applicationId, String processorId, String destinationHost, Duration requestExpiry) {...}
public ContainerPlacementRequestMessage(UUID uuid, String applicationId, String processorId, String destinationHost) {...}
} |
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
/**
* Encapsulates the response sent from the JobCoordinator for a container placement action
*/
public class ContainerPlacementResponseMessage extends ContainerPlacementMessage {
// Returned status of the request
private String responseMessage;
public ContainerPlacementResponseMessage(UUID uuid, String applicationId, String processorId, String destinationHost,
Duration requestExpiry, StatusCode statusCode, String responseMessage) {.. }
public ContainerPlacementResponseMessage(UUID uuid, String applicationId, String processorId, String destinationHost,
StatusCode statusCode, String responseMessage) {
this(uuid, applicationId, processorId, destinationHost, null, statusCode, responseMessage);
}
|
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
public class ContainerPlacementHandler {
/**
* Writes a {@link ContainerPlacementRequestMessage} to the underlying metastore.
* This method should be used by external controllers to issue a request to JobCoordinator
*
* @param message container placement request
*/
public void writeContainerPlacementRequestMessage(ContainerPlacementRequestMessage message) {...}
/**
* Reads a {@link ContainerPlacementRequestMessage} from the underlying metastore
* @param processorId key of the message, logical processor id of a samza container 0,1,2
* @return ContainerPlacementRequestMessage is its present
*/
public Optional<ContainerPlacementRequestMessage> readContainerPlacementRequestMessage(String processorId) {...}
/**
* Reads a {@link ContainerPlacementResponseMessage} from the underlying metastore
* @param processorId key of the message, logical processor id of a samza container 0,1,2
* @return ContainerPlacementResponseMessage is its present
*/
public Optional<ContainerPlacementResponseMessage> readContainerPlacementResponseMessage(String processorId) {..}
/**
* Deletes a {@link ContainerPlacementRequestMessage} if present identified by the key {@code processorId}
* @param processorId logical processor id 0,1,2
*/
public void deleteContainerPlacementRequestMessage(String processorId) {..}
/**
* Deletes a {@link ContainerPlacementResponseMessage} if present identified by the key {@code processorId}
* @param processorId logical processor id 0,1,2
*/
public void deleteContainerPlacementResponseMessage(String processorId) {..}
/**
* Deletes all {@link ContainerPlacementRequestMessage} present in underlying metastore
*/
public void deleteAllContainerPlacementRequestMessages() {..}
/**
* Deletes all {@link ContainerPlacementResponseMessage} present in underlying metastore
*/
public void deleteAllContainerPlacementResponseMessages() {..}
/**
* Deletes all {@link ContainerPlacementMessage}
*/
public void deleteAllContainerPlacementMessages() {..}
/**
* Writes a {@link ContainerPlacementResponseMessage} to the underlying metastore.
* This method should be used by Job Coordinator only to write responses to Container Placement Action
* @param message
*/
void writeContainerPlacementResponseMessage {..}
} |
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
public class ContainerManager {
/**
* Registers a container placement action to move the running container to destination host
*
* @param requestMessage request containing details of placement request
* @param containerAllocator to request physical resources
*/
public void registerContainerPlacementAction(ContainerPlacementRequestMessage requestMessage, ContainerAllocator containerAllocator) {...}
/**
* Handles the container start action for both active & standby containers. This method is invoked by the allocator thread
*
* @param request pending request for the preferred host
* @param preferredHost preferred host to start the container
* @param allocatedResource resource allocated from {@link ClusterResourceManager}
* @param resourceRequestState state of request in {@link ContainerAllocator}
* @param allocator to request resources from @{@link ClusterResourceManager}
*
* @return true if the container launch is complete, false if the container launch is in progress.
*/
boolean handleContainerLaunch(SamzaResourceRequest request, String preferredHost, SamzaResource allocatedResource,
ResourceRequestState resourceRequestState, ContainerAllocator allocator) {..}
/**
* Handle the container launch failure for active containers and standby (if enabled).
*
* @param processorId logical id of the container eg 1,2,3
* @param containerId last known id of the container deployed
* @param preferredHost host on which container is requested to be deployed
* @param containerAllocator allocator for requesting resources
*/
void handleContainerLaunchFail(String processorId, String containerId, String preferredHost,
ContainerAllocator containerAllocator) {...}
/**
* Handles the state update on successful launch of a container
*
* @param processorId logical processor id of container 0,1,2
*/
void handleContainerLaunchSuccess(String processorId) {...}
/**
* Handles the action to be taken after the container has been stopped.
*
* @param processorId logical id of the container eg 1,2,3
* @param containerId last known id of the container deployed
* @param preferredHost host on which container was last deployed
* @param exitStatus exit code returned by the container
* @param preferredHostRetryDelay delay to be incurred before requesting resources
* @param containerAllocator allocator for requesting resources
*/
void handleContainerStop(String processorId, String containerId, String preferredHost, int exitStatus,
Duration preferredHostRetryDelay, ContainerAllocator containerAllocator) {..}
/**
* Handles an expired resource request for both active and standby containers.
*
* @param processorId logical id of the container
* @param preferredHost host on which container is requested to be deployed
* @param request pending request for the preferred host
* @param allocator allocator for requesting resources
* @param resourceRequestState state of request in {@link ContainerAllocator}
*/
void handleExpiredRequestForControlActionOrHostAffinityEnabled(String processorId, String preferredHost,
SamzaResourceRequest request, ContainerAllocator allocator, ResourceRequestState resourceRequestState) {..}
}
|
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
public class ContainerPlacementMetadata {
/**
* State to track container failover
*/
public enum ContainerStatus { RUNNING, STOP_IN_PROGRESS, STOPPED }
// Container Placement request message
private final ContainerPlacementRequestMessage requestMessage;
// Host where the container is actively running
private final String sourceHost;
// Resource requests issued during this failover
private final Set<SamzaResourceRequest> resourceRequests;
// State of the control action
private ContainerPlacementMessage.StatusCode actionStatus;
// State of the active container to track failover
private ContainerStatus containerStatus;
// Represents information on current status of action
private String responseMessage;
} |
...
2.1.2 Stateful Container Move (Host Affinity On)
The stateful container is divided into two cases on the basis of whether the job has StandBy Container for it or not
Case 1. Stateful Job has Standby Container Enabled
Let's take a case when the job has a stateful container C on H1 and has a StandBy container enabled C’ on H2
Option 1: Stop-Reserve-Move: Initiate a Stop on C & issue a Standby failover (Preferred - Phase 1*)
In this option, C is stopped on H1 and failover is initiated from H1 to H2 with current semantics of Failover of an active Container in the Hot standby feature . This ensures C moves either to H2 (in the best case) or to some other hosts (when Cluster Manager cannot allocate capacity on H2, i.e H2 is under capacity)
...
Pros
...
Cons
...
- Easy to implement, ability to do this already exists
...
- C always moves from H1 and there a slim chance that C won’t land on H2 as per the current semantics (stolen-host scenario: Cluster Manager fails to allocate H2 for C)
Option 2: Reserve-Stop-Move: Initiate a Standby move first then issue a move for Active Container (Preferred - Phase 2*)
- In this option request is first issued to stop C’
- Then request H2 from cluster manager to start C
- Issue a stop on C only H2 can be allocated (#2 succeeds).
- Then request C’ to start ANY_HOST except for H1, H2
Two ways to achieve this:
- A client can make two calls for this first move C’ to ANY_HOST apart from H2, H1 (similar to a stateless move), then move the request of C to C’ (similar to a stateless move) so the state maintenance of this lives on the client
- Maintain state in JC to accomplish this
...
Pros
...
Cons
...
- Stronger move semantics, since the active container won’t be stopped if it cannot be moved to the standby container
...
- More complex as compared to Option 1 needs to maintain more state
* Phase 1 & Phase 2 refer to the implementation phases, please see the section: Implementation Plan
Option 3: Reserve-Stop-Move: First Request H2 from Cluster Manager and initiate the failover once you get H2 as a preferred resource (Rejected Alternative)
- In this option C’ is stopped on H2 and failover is initiated from H1 to H2 with current semantics as developed by the feature Hot standby
- In case of the move failed only affects standby
...
Pros
...
Cons
...
- Stronger move semantics are guaranteed, the active container won’t be stopped unless it can be started with H2
...
- Since there is a standby already running on H2, there is a higher chance that cluster manager might fail to allocate for H2 and move requests are ineffective (Host is running full capacity)
- Need to design a new failover scheme hence more development time to implement
Orchestration of StandBy Enabled Container Move using Option 1 (Phase 1)
- ContainerPlacementHandler dispatches the move request to ContainerProcessManager
- ContainerProcessManager registers a move & initiates a failover with StandByContainerManager then the following failover sequence is followed
Case 2. Stateful Job has Standby Container Disabled [Stretch]
Option 1: Stateful containers move is equivalent to Stateless container Move (Phase 1)
In this option, Stateful container move like stateless container & does a bootstrap
Option 2: Optionally Spin Up StandBy container & then move (Phase 2)
If we expose an API to spin a StandBy Container for an active container, a client can then make two requests, one to spin up a StandBy then periodically monitor the lag for StandByCotntainer (Using Diagnostics). Once the StandBy Container has a steady lag is ready to move, this case becomes Case 1 mentioned above. (More details TBD later)
...
This is much simpler as compared to Move. We can either make restart equivalent to a move or issue a stop a container and ContainerProcessManager will try to start that container again. Let's discuss these options in detail:
Option 1: Reserve-Stop-Start: Restart is equivalent to move on the same host [Preferred]
- In this option, resources are requested on the same host first before issuing a stop on Container
- Once the resources are accrued a container is stopped & requested to start
- So this is equivalent to move semantics above
...
Pros
...
Cons
...
- Strong Restart semantics since the container is not stopped if it cannot be restarted on the same resource
...
- For a container to restart, it will be holding 2x resources on the same host for the time it has accrued resources to the time when the active container is issued a stop
Both stateless & stateful container restarts will be equivalent to Stateless Container Move on the same host. Since the container is restarted on the same host then
Option 2: Stop-Reserve-Start: Restart is equivalent to stopping container first then attempting to start it [Rejected Alternative]
- In this option, a container is issued a stop first
- Once the container stops, resources are requested for starting on the same host (last seen host)
- Once the resources are accrued, the container is issued a start on the same host
...
Pros
...
Cons
...
- The container at any point in time is only holding resources it needs to start on a host
...
- Weaker Restart semantics since there is a likely chance of restarting this container on any other host then the source host (since container is stopped & when Cluster Manager cannot return resources)
...
In this case, when User registers a restart request, ContainerPlacementManager issues a stop on the current container. On receiving a callback from ClusterResourceManager for container stop, ContainerProcessManager requests resources on last seen host. But Allocator thread assigns this request on ANY_HOST available since we have homogenous containers in Yarn & host affinity is off, the container can be started from any available resource.
Case 2.2: Host Affinity is ON
In this case, a Stop is issued on Container & the ContainerProcessManager makes a request for resources on last seen host to ClusterResourcesManager. Allocator Thread on having resources on the preferred host (here last seen host) issues a request to run a container on it. If this request expires then there are multiple possibilities:
- If StandBy Containers are enabled directly initiate a failover
- Else if there are resources on ANY_HOST then issues a container start request on that resource
- Else issue an ANY_HOST request to ClusterResourceManager
...
Implementation and Test Plan
Implementation Plan
Phase | Feature | Timeline |
Phase 1 |
| Q4 2019 |
Phase 2 |
| TBD |
Test Specifications
- Complete testing plan: https://docs.google.com/spreadsheets/d/1v-fw0pHxKRobGkALDCno4FuPCsBhdepQ86vIGLHWu54/edit?usp=drive_web&ouid=115242259601904072922
- Local deployment testing with Virtual Private Cluster (LXC with Samza)
- Create a test job using VPC or in the real cluster that automates testing for taking control actions on containers
...