| ...
| Pros |
Cons |
- Simple to extend the existing REST endpoint
- Need to build authentication
- 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
|
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 NamespaceAwareMetaStoreKey in namespace using NamespaceAwareMetaStore ("samza-place-container-v1")
Key-Value Format Key for storing the ContainerPlacementRequestMessage & ContainerPlacementResponseMessage in Metastore is chosen to be ... KV for ContainerPlacementRequestMessages & ContainerPlacementResponseMessage ... Key ... Value ... processorId uuid: unique identifier a request, populated by client applicationId: unique identifier of the deployed app for which the action is taken destination-host: valid hostname / “ANY_HOST” / “STANDBY” ... UUID + "." + messageType(ContainerPlacementResponseMessage or ContainerPlacementRequestMessage). Value will be payload container ContainerPlacementRequestMessage & ContainerPlacementResponseMessage. Messages are written and read to the Metastore through the MetadataStore abstraction. ContainerPlacementResponseMessage:
| Key |
Value |
Field Description |
Field Type |
| "UUID.subType" |
uuid |
Unique identifier of a response message |
Required |
|
processorId |
Logical processor id 0,1,2 of the container |
Required |
|
deploymentId |
Unique identifier for a deployment |
Required |
|
subType |
Type of message here: ContainerPlacementResponseMessage |
Required |
|
destinationHost |
Destination host where the container is desired to be moved |
Required |
|
statusCode |
Status of the current action |
Required |
|
responseMessage |
Response message in conjunction to status |
Required |
|
timestamp |
The timestamp of the response message |
Required |
|
requestExpiry |
Eequest expiry which acts as a timeout for any resource request to cluster resource manager |
Optional |
Sample KV
| Key |
Value |
| [1,"samza-place-container-v1","88b0d30c-d518-4307-9e8e-c8529eb30f04.ContainerPlacementResponseMessage"] |
{"processorId":"1","deploymentId":"app-atttempt-001","subType":"ContainerPlacementResponseMessage","responseMessage":"Request is accepted","uuid":"88b0d30c-d518-4307-9e8e-c8529eb30f04","destinationHost":"ANY_HOST","statusCode":"ACCEPTED","timestamp":1578694070875} |
GC policy for stale messages in metastore
- One way to delete stale ContainerPlacementMessages is to delete request / responses from the previous incarnation of the job in the metastore on job restarts
- Once the request is complete, ContainerPlacementService can issue an async delete to the metastore
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 ... |
