| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package org.apache.samza.clustermanager; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import java.time.Duration; |
| import java.util.Optional; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import org.apache.commons.lang3.tuple.ImmutablePair; |
| import org.apache.commons.lang3.tuple.Pair; |
| import org.apache.samza.clustermanager.container.placement.ContainerPlacementMetadataStore; |
| import org.apache.samza.clustermanager.container.placement.ContainerPlacementMetadata; |
| import org.apache.samza.container.LocalityManager; |
| import org.apache.samza.container.placement.ContainerPlacementMessage; |
| import org.apache.samza.container.placement.ContainerPlacementRequestMessage; |
| import org.apache.samza.container.placement.ContainerPlacementResponseMessage; |
| import org.apache.samza.job.model.ProcessorLocality; |
| import org.apache.samza.util.BoundedLinkedHashSet; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * ContainerManager is a centralized entity that manages container actions like start, stop for both active and standby containers |
| * ContainerManager acts as a brain for validating and issuing any actions on containers in the Job Coordinator. |
| * |
| * The requests to allocate resources resources made by {@link ContainerAllocator} can either expire or succeed. |
| * When the requests succeeds the ContainerManager validates those requests before starting the container |
| * When the requests expires the ContainerManager decides the next set of actions for the pending request. |
| * |
| * Callbacks issued from {@link ClusterResourceManager} aka {@link ContainerProcessManager} are intercepted |
| * by ContainerManager to handle container failure and completions for both active and standby containers |
| * |
| * ContainerManager encapsulates logic and state related to container placement actions like move, restarts for active container |
| * if issued externally. |
| * |
| */ |
| public class ContainerManager { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(ContainerManager.class); |
| private static final String ANY_HOST = ResourceRequestState.ANY_HOST; |
| private static final String LAST_SEEN = "LAST_SEEN"; |
| private static final String FORCE_RESTART_LAST_SEEN = "FORCE_RESTART_LAST_SEEN"; |
| private static final int UUID_CACHE_SIZE = 20000; |
| |
| /** |
| * Container placement metadata store to write responses to control actions |
| */ |
| private final ContainerPlacementMetadataStore containerPlacementMetadataStore; |
| /** |
| * Resource-manager, used to stop containers |
| */ |
| private final ClusterResourceManager clusterResourceManager; |
| private final SamzaApplicationState samzaApplicationState; |
| private final boolean hostAffinityEnabled; |
| |
| /** |
| * Map maintaining active container placement action meta data indexed by container's processorId eg 0, 1, 2 |
| * Key is chosen to be processorId since at a time only one placement action can be in progress on a container. |
| */ |
| private final ConcurrentHashMap<String, ContainerPlacementMetadata> actions; |
| /** |
| * In-memory cache of placement requests UUIDs de-queued from the metadata store. Used to de-dup requests with the same |
| * request UUID. Sized using max tolerable memory footprint and max likely duplicate-spacing. |
| */ |
| private final BoundedLinkedHashSet<UUID> placementRequestsCache; |
| |
| private final Optional<StandbyContainerManager> standbyContainerManager; |
| |
| private final LocalityManager localityManager; |
| |
| public ContainerManager(ContainerPlacementMetadataStore containerPlacementMetadataStore, |
| SamzaApplicationState samzaApplicationState, ClusterResourceManager clusterResourceManager, |
| boolean hostAffinityEnabled, boolean standByEnabled, LocalityManager localityManager) { |
| Preconditions.checkNotNull(localityManager, "Locality manager cannot be null"); |
| this.samzaApplicationState = samzaApplicationState; |
| this.clusterResourceManager = clusterResourceManager; |
| this.actions = new ConcurrentHashMap<>(); |
| this.placementRequestsCache = new BoundedLinkedHashSet<UUID>(UUID_CACHE_SIZE); |
| this.hostAffinityEnabled = hostAffinityEnabled; |
| this.containerPlacementMetadataStore = containerPlacementMetadataStore; |
| this.localityManager = localityManager; |
| // Enable standby container manager if required |
| if (standByEnabled) { |
| this.standbyContainerManager = |
| Optional.of(new StandbyContainerManager(samzaApplicationState, clusterResourceManager, localityManager)); |
| } else { |
| this.standbyContainerManager = Optional.empty(); |
| } |
| } |
| |
| /** |
| * Handles the container start action for both active & standby containers. This method is invoked by the allocator thread |
| * |
| * Case 1. If the container launch request is due to an existing container placement action, issue a stop on active |
| * container & wait for the active container to be stopped before issuing a start. |
| * Case 2. If StandbyContainer is present refer to {@code StandbyContainerManager#checkStandbyConstraintsAndRunStreamProcessor} |
| * Case 3. Otherwise just invoke a container start on the allocated resource for the pending request |
| * |
| * TODO: SAMZA-2399: Investigate & configure a timeout for container stop if needed |
| * |
| * @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. A container launch |
| * might be in progress when it is waiting for the previous container incarnation to stop in case of container |
| * placement actions |
| */ |
| boolean handleContainerLaunch(SamzaResourceRequest request, String preferredHost, SamzaResource allocatedResource, |
| ResourceRequestState resourceRequestState, ContainerAllocator allocator) { |
| if (hasActiveContainerPlacementAction(request.getProcessorId())) { |
| String processorId = request.getProcessorId(); |
| ContainerPlacementMetadata actionMetaData = getPlacementActionMetadata(processorId).get(); |
| ContainerPlacementMetadata.ContainerStatus actionStatus = actionMetaData.getContainerStatus(); |
| if (samzaApplicationState.runningProcessors.containsKey(processorId) && actionStatus == ContainerPlacementMetadata.ContainerStatus.RUNNING) { |
| LOG.debug("Requesting running container to shutdown due to existing ContainerPlacement action {}", actionMetaData); |
| actionMetaData.setContainerStatus(ContainerPlacementMetadata.ContainerStatus.STOP_IN_PROGRESS); |
| updateContainerPlacementActionStatus(actionMetaData, ContainerPlacementMessage.StatusCode.IN_PROGRESS, |
| "Active container stop in progress"); |
| clusterResourceManager.stopStreamProcessor(samzaApplicationState.runningProcessors.get(processorId)); |
| return false; |
| } else if (actionStatus == ContainerPlacementMetadata.ContainerStatus.STOP_IN_PROGRESS) { |
| LOG.info("Waiting for running container to shutdown due to existing ContainerPlacement action {}", actionMetaData); |
| return false; |
| } else if (actionStatus == ContainerPlacementMetadata.ContainerStatus.STOP_FAILED) { |
| LOG.info("Shutdown on running container failed for action {}", actionMetaData); |
| markContainerPlacementActionFailed(actionMetaData, |
| String.format("failed to stop container on current host %s", actionMetaData.getSourceHost())); |
| resourceRequestState.cancelResourceRequest(request); |
| return true; |
| } else if (actionStatus == ContainerPlacementMetadata.ContainerStatus.STOPPED) { |
| // If the job has standby containers enabled, always check standby constraints before issuing a start on container |
| // Note: Always check constraints against allocated resource, since preferred host can be ANY_HOST as well |
| if (standbyContainerManager.isPresent() && !standbyContainerManager.get().checkStandbyConstraints(request.getProcessorId(), allocatedResource.getHost())) { |
| LOG.info( |
| "Starting container {} on host {} does not meet standby constraints, falling back to source host placement metadata: {}", |
| request.getProcessorId(), preferredHost, actionMetaData); |
| // Release unstartable container |
| standbyContainerManager.get().releaseUnstartableContainer(request, allocatedResource, preferredHost, resourceRequestState); |
| // Fallback to source host since the new allocated resource does not meet standby constraints |
| allocator.requestResource(processorId, actionMetaData.getSourceHost()); |
| markContainerPlacementActionFailed(actionMetaData, |
| String.format("allocated resource %s does not meet standby constraints now, falling back to source host", allocatedResource)); |
| } else { |
| LOG.info("Status updated for ContainerPlacement action: ", actionMetaData); |
| allocator.runStreamProcessor(request, preferredHost); |
| } |
| return true; |
| } |
| } |
| |
| if (this.standbyContainerManager.isPresent()) { |
| standbyContainerManager.get() |
| .checkStandbyConstraintsAndRunStreamProcessor(request, preferredHost, allocatedResource, allocator, |
| resourceRequestState); |
| } else { |
| allocator.runStreamProcessor(request, preferredHost); |
| } |
| return true; |
| } |
| |
| /** |
| * Handles the action to be taken after the container has been stopped. If stop was issued due to existing control |
| * action, mark the container as stopped, otherwise |
| * |
| * Case 1. When standby is enabled, refer to {@link StandbyContainerManager#handleContainerStop} to check constraints. |
| * Case 2. When standby is disabled there are two cases according to host-affinity being enabled |
| * Case 2.1. When host-affinity is enabled resources are requested on host where container was last seen |
| * Case 2.2. When host-affinity is disabled resources are requested for ANY_HOST |
| * |
| * @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) { |
| if (hasActiveContainerPlacementAction(processorId)) { |
| ContainerPlacementMetadata metadata = getPlacementActionMetadata(processorId).get(); |
| LOG.info("Setting the container state with Processor ID: {} to be stopped because of existing ContainerPlacement action: {}", |
| processorId, metadata); |
| metadata.setContainerStatus(ContainerPlacementMetadata.ContainerStatus.STOPPED); |
| } else if (standbyContainerManager.isPresent()) { |
| standbyContainerManager.get() |
| .handleContainerStop(processorId, containerId, preferredHost, exitStatus, containerAllocator, |
| preferredHostRetryDelay); |
| } else { |
| // If StandbyTasks are not enabled, we simply make a request for the preferredHost |
| containerAllocator.requestResourceWithDelay(processorId, preferredHost, preferredHostRetryDelay); |
| } |
| } |
| |
| /** |
| * Handle the container launch failure for active containers and standby (if enabled). |
| * |
| * Case 1. If this launch was issued due to an existing container placement action update the metadata to report failure and issue |
| * a request for source host where the container was last seen and mark the container placement failed |
| * Case 2. When standby is enabled, refer to {@link StandbyContainerManager#handleContainerLaunchFail} to check behavior |
| * Case 3. When standby is disabled the allocator issues a request for ANY_HOST resources |
| * |
| * @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) { |
| if (processorId != null && hasActiveContainerPlacementAction(processorId)) { |
| ContainerPlacementMetadata metaData = getPlacementActionMetadata(processorId).get(); |
| // Issue a request to start the container on source host |
| String sourceHost = hostAffinityEnabled ? metaData.getSourceHost() : ResourceRequestState.ANY_HOST; |
| markContainerPlacementActionFailed(metaData, |
| String.format("failed to start container on destination host %s, attempting to start on source host %s", |
| preferredHost, sourceHost)); |
| containerAllocator.requestResource(processorId, sourceHost); |
| } else if (processorId != null && standbyContainerManager.isPresent()) { |
| standbyContainerManager.get().handleContainerLaunchFail(processorId, containerId, containerAllocator); |
| } else if (processorId != null) { |
| LOG.info("Falling back to ANY_HOST for Processor ID: {} since launch failed for Container ID: {} on host: {}", |
| processorId, containerId, preferredHost); |
| containerAllocator.requestResource(processorId, ResourceRequestState.ANY_HOST); |
| } else { |
| LOG.warn("Did not find a pending Processor ID for Container ID: {} on host: {}. " |
| + "Ignoring invalid/redundant notification.", containerId, preferredHost); |
| } |
| } |
| |
| /** |
| * Handle the container stop 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 containerHost host on which container is requested to be deployed |
| * @param containerAllocator allocator for requesting resources |
| * TODO: SAMZA-2512 Add integ test for handleContainerStopFail |
| */ |
| void handleContainerStopFail(String processorId, String containerId, String containerHost, |
| ContainerAllocator containerAllocator) { |
| if (processorId != null && hasActiveContainerPlacementAction(processorId)) { |
| // Assuming resource acquired on destination host will be relinquished by the containerAllocator, |
| // We mark the placement action as failed, and return. |
| ContainerPlacementMetadata metaData = getPlacementActionMetadata(processorId).get(); |
| metaData.setContainerStatus(ContainerPlacementMetadata.ContainerStatus.STOP_FAILED); |
| } else if (processorId != null && standbyContainerManager.isPresent()) { |
| standbyContainerManager.get().handleContainerStopFail(processorId, containerId, containerAllocator); |
| } else { |
| LOG.warn("Did not find a running Processor ID for Container ID: {} on host: {}. " |
| + "Ignoring invalid/redundant notification.", containerId, containerHost); |
| } |
| } |
| |
| /** |
| * Handles the state update on successful launch of a container, if this launch is due to a container placement action updates the |
| * related metadata to report success |
| * |
| * @param processorId logical processor id of container 0,1,2 |
| */ |
| void handleContainerLaunchSuccess(String processorId, String containerHost) { |
| if (hasActiveContainerPlacementAction(processorId)) { |
| ContainerPlacementMetadata metadata = getPlacementActionMetadata(processorId).get(); |
| // Mark the active container running again and dispatch a response |
| metadata.setContainerStatus(ContainerPlacementMetadata.ContainerStatus.RUNNING); |
| updateContainerPlacementActionStatus(metadata, ContainerPlacementMessage.StatusCode.SUCCEEDED, |
| "Successfully completed the container placement action started container on host " + containerHost); |
| } |
| } |
| |
| /** |
| * Handles an expired resource request for both active and standby containers. |
| * |
| * Case 1. If this expired request is due to a container placement action mark the request as failed and return |
| * Case 2: Otherwise for a normal resource request following cases are possible |
| * Case 2.1 If StandbyContainer is present refer to {@code StandbyContainerManager#handleExpiredResourceRequest} |
| * Case 2.2: host-affinity is enabled, allocator thread looks for allocated resources on ANY_HOST and issues a |
| * container start if available, otherwise issue an ANY_HOST request |
| * Case 2.2: host-affinity is disabled, allocator thread does not handle expired requests, it waits for cluster |
| * manager to return resources on ANY_HOST |
| * |
| * @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} |
| */ |
| @VisibleForTesting |
| void handleExpiredRequest(String processorId, String preferredHost, |
| SamzaResourceRequest request, ContainerAllocator allocator, ResourceRequestState resourceRequestState) { |
| boolean resourceAvailableOnAnyHost = allocator.hasAllocatedResource(ResourceRequestState.ANY_HOST); |
| |
| |
| // Case 1. Container placement actions can be taken in either cases of host affinity being set, in both cases |
| // mark the container placement action failed |
| if (hasActiveContainerPlacementAction(processorId)) { |
| resourceRequestState.cancelResourceRequest(request); |
| markContainerPlacementActionFailed(getPlacementActionMetadata(processorId).get(), |
| "failed the ContainerPlacement action because request for resources to ClusterManager expired"); |
| return; |
| } |
| |
| // Case 2. When host affinity is disabled wait for cluster resource manager return resources |
| // TODO: SAMZA-2330: Handle expired request for host affinity disabled case by retying request for getting ANY_HOST |
| if (!hostAffinityEnabled) { |
| return; |
| } |
| |
| // Case 2. When host affinity is enabled handle the expired requests |
| if (standbyContainerManager.isPresent()) { |
| standbyContainerManager.get() |
| .handleExpiredResourceRequest(processorId, request, |
| Optional.ofNullable(allocator.peekAllocatedResource(ResourceRequestState.ANY_HOST)), allocator, |
| resourceRequestState); |
| } else if (resourceAvailableOnAnyHost) { |
| LOG.info("Request for Processor ID: {} on host: {} has expired. Running on ANY_HOST", processorId, preferredHost); |
| allocator.runStreamProcessor(request, ResourceRequestState.ANY_HOST); |
| } else { |
| LOG.info("Request for Processor ID: {} on host: {} has expired. Requesting additional resources on ANY_HOST.", |
| processorId, preferredHost); |
| resourceRequestState.cancelResourceRequest(request); |
| allocator.requestResource(processorId, ResourceRequestState.ANY_HOST); |
| } |
| } |
| |
| /** |
| * Handles expired allocated resource by requesting the same resource again and release the expired allocated resource |
| * |
| * @param request pending request for the preferred host |
| * @param resource resource allocated from {@link ClusterResourceManager} which has expired |
| * @param preferredHost host on which container is requested to be deployed |
| * @param resourceRequestState state of request in {@link ContainerAllocator} |
| * @param allocator allocator for requesting resources |
| */ |
| void handleExpiredResource(SamzaResourceRequest request, SamzaResource resource, String preferredHost, |
| ResourceRequestState resourceRequestState, ContainerAllocator allocator) { |
| LOG.info("Allocated resource {} has expired for Processor ID: {} request: {}. Re-requesting resource again", |
| resource, request.getProcessorId(), request); |
| resourceRequestState.releaseUnstartableContainer(resource, preferredHost); |
| resourceRequestState.cancelResourceRequest(request); |
| SamzaResourceRequest newResourceRequest = allocator.getResourceRequest(request.getProcessorId(), request.getPreferredHost()); |
| if (hasActiveContainerPlacementAction(newResourceRequest.getProcessorId())) { |
| ContainerPlacementMetadata metadata = getPlacementActionMetadata(request.getProcessorId()).get(); |
| metadata.recordResourceRequest(newResourceRequest); |
| } |
| allocator.issueResourceRequest(newResourceRequest); |
| } |
| |
| /** |
| * Registers a container placement action to move the running container to destination host, if destination host is same as the |
| * host on which container is running, container placement action is treated as a restart. |
| * |
| * When host affinity is disabled a move / restart is only allowed on ANY_HOST |
| * When host affinity is enabled move / restart is allowed on specific or ANY_HOST |
| * |
| * Container placement requests are tied to deploymentId which is currently {@link org.apache.samza.config.ApplicationConfig#APP_RUN_ID} |
| * On job restarts container placement requests queued for the previous deployment are deleted using this |
| * |
| * All kinds of container placement request except for when destination host is "FORCE_RESTART_LAST_SEEN" work with |
| * a RESERVE - STOP - START policy, which means resources are accrued first before issuing a container stop, failure to |
| * do so will leave the running container untouched. Requests with destination host "FORCE_RESTART_LAST_SEEN" works with |
| * STOP - RESERVE - START policy, which means running container is stopped first then resource request are issued, this case |
| * is equivalent to doing a kill -9 on a container |
| * |
| * @param requestMessage request containing logical processor id 0,1,2 and host where container is desired to be moved, |
| * acceptable values of this param are |
| * - valid hostname |
| * - "ANY_HOST" in this case the request is sent to resource manager for any host |
| * - "LAST_SEEN" in this case request is sent to resource manager for last seen host |
| * - "FORCE_RESTART_LAST_SEEN" in this case request is sent to resource manager for last seen host |
| * @param containerAllocator to request physical resources |
| */ |
| public void registerContainerPlacementAction(ContainerPlacementRequestMessage requestMessage, ContainerAllocator containerAllocator) { |
| String processorId = requestMessage.getProcessorId(); |
| String destinationHost = requestMessage.getDestinationHost(); |
| // Is the action ready to be de-queued and taken or it needs to wait to be executed in future |
| if (!deQueueAction(requestMessage)) { |
| return; |
| } |
| LOG.info("ContainerPlacement action is de-queued metadata: {}", requestMessage); |
| Pair<ContainerPlacementMessage.StatusCode, String> actionStatus = validatePlacementAction(requestMessage); |
| // Action is de-queued upon so we record it in the cache |
| placementRequestsCache.put(requestMessage.getUuid()); |
| // Remove the request message from metastore since this message is already acted upon |
| containerPlacementMetadataStore.deleteContainerPlacementRequestMessage(requestMessage.getUuid()); |
| // Request is bad just update the response on message & return |
| if (actionStatus.getKey() == ContainerPlacementMessage.StatusCode.BAD_REQUEST) { |
| LOG.info("Status updated for ContainerPlacement action request: {} response: {}", requestMessage, actionStatus.getValue()); |
| writeContainerPlacementResponseMessage(requestMessage, actionStatus.getKey(), actionStatus.getValue()); |
| return; |
| } |
| |
| /* |
| * When destination host is {@code FORCE_RESTART_LAST_SEEN} its treated as eqvivalent to kill -9 operation for the container |
| * In this scenario container is stopped first and we fallback to normal restart path so the policy here is |
| * stop - reserve - move |
| */ |
| if (destinationHost.equals(FORCE_RESTART_LAST_SEEN)) { |
| LOG.info("Issuing a force restart for Processor ID: {} for ContainerPlacement action request {}", processorId, requestMessage); |
| clusterResourceManager.stopStreamProcessor(samzaApplicationState.runningProcessors.get(processorId)); |
| writeContainerPlacementResponseMessage(requestMessage, ContainerPlacementMessage.StatusCode.SUCCEEDED, |
| "Successfully issued a stop container request falling back to normal restart path"); |
| return; |
| } |
| |
| /** |
| * When destination host is {@code LAST_SEEN} its treated as a restart request on the host where container is running |
| * on or has been seen last, but in this policy would be reserve - stop - move, which means reserve resources first |
| * only if resources are accrued stop the active container and issue a start on it on resource acquired |
| */ |
| if (destinationHost.equals(LAST_SEEN)) { |
| String lastSeenHost = getSourceHostForContainer(requestMessage); |
| LOG.info("Changing the requested host for placement action to {} because requested host is LAST_SEEN", lastSeenHost); |
| destinationHost = lastSeenHost; |
| } |
| |
| // TODO: SAMZA-2457: Allow host affinity disabled jobs to move containers to specific host |
| if (!hostAffinityEnabled) { |
| LOG.info("Changing the requested host for placement action to {} because host affinity is disabled", ResourceRequestState.ANY_HOST); |
| destinationHost = ANY_HOST; |
| } |
| |
| // Register metadata |
| ContainerPlacementMetadata actionMetaData = new ContainerPlacementMetadata(requestMessage, getSourceHostForContainer(requestMessage)); |
| actions.put(processorId, actionMetaData); |
| |
| // If the job is running in a degraded state then the container is already stopped |
| if (samzaApplicationState.failedProcessors.containsKey(requestMessage.getProcessorId())) { |
| actionMetaData.setContainerStatus(ContainerPlacementMetadata.ContainerStatus.STOPPED); |
| } |
| |
| SamzaResourceRequest resourceRequest = containerAllocator.getResourceRequest(processorId, destinationHost); |
| // Record the resource request for monitoring |
| actionMetaData.recordResourceRequest(resourceRequest); |
| actions.put(processorId, actionMetaData); |
| updateContainerPlacementActionStatus(actionMetaData, ContainerPlacementMessage.StatusCode.IN_PROGRESS, "Preferred Resources requested"); |
| containerAllocator.issueResourceRequest(resourceRequest); |
| } |
| |
| /** |
| * This method is only used for Testing the Container Placement actions to get a hold of {@link ContainerPlacementMetadata} |
| * for assertions. Not intended to be used in src |
| */ |
| @VisibleForTesting |
| ContainerPlacementMetadata registerContainerPlacementActionForTest(ContainerPlacementRequestMessage requestMessage, |
| ContainerAllocator containerAllocator) { |
| registerContainerPlacementAction(requestMessage, containerAllocator); |
| if (hasActiveContainerPlacementAction(requestMessage.getProcessorId())) { |
| return getPlacementActionMetadata(requestMessage.getProcessorId()).get(); |
| } |
| return null; |
| } |
| |
| public Optional<Duration> getActionExpiryTimeout(SamzaResourceRequest resourceRequest) { |
| for (ContainerPlacementMetadata actionMetadata : actions.values()) { |
| if (actionMetadata.containsResourceRequest(resourceRequest) |
| && actionMetadata.getActionStatus() == ContainerPlacementMessage.StatusCode.IN_PROGRESS) { |
| return actionMetadata.getRequestActionExpiryTimeout(); |
| } |
| } |
| return Optional.empty(); |
| } |
| |
| private void markContainerPlacementActionFailed(ContainerPlacementMetadata metaData, String failureMessage) { |
| samzaApplicationState.failedContainerPlacementActions.incrementAndGet(); |
| updateContainerPlacementActionStatus(metaData, ContainerPlacementMessage.StatusCode.FAILED, failureMessage); |
| } |
| |
| /** |
| * A ContainerPlacementAction is only active if it is either CREATED, ACCEPTED or IN_PROGRESS |
| */ |
| private boolean hasActiveContainerPlacementAction(String processorId) { |
| Optional<ContainerPlacementMetadata> metadata = getPlacementActionMetadata(processorId); |
| if (metadata.isPresent()) { |
| switch (metadata.get().getActionStatus()) { |
| case ACCEPTED: |
| case IN_PROGRESS: |
| return true; |
| default: |
| return false; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Check if a activeContainerResource has control-action-metadata associated with it |
| */ |
| private Optional<ContainerPlacementMetadata> getPlacementActionMetadata(String processorId) { |
| return Optional.ofNullable(this.actions.get(processorId)); |
| } |
| |
| private void updateContainerPlacementActionStatus(ContainerPlacementMetadata metadata, |
| ContainerPlacementMessage.StatusCode statusCode, String responseMessage) { |
| metadata.setActionStatus(statusCode, responseMessage); |
| writeContainerPlacementResponseMessage(metadata.getRequestMessage(), statusCode, responseMessage); |
| LOG.info("Status updated for ContainerPlacement action: {}", metadata); |
| } |
| |
| private void writeContainerPlacementResponseMessage(ContainerPlacementRequestMessage requestMessage, |
| ContainerPlacementMessage.StatusCode statusCode, String responseMessage) { |
| containerPlacementMetadataStore.writeContainerPlacementResponseMessage( |
| ContainerPlacementResponseMessage.fromContainerPlacementRequestMessage(requestMessage, statusCode, |
| responseMessage, System.currentTimeMillis())); |
| } |
| |
| /** |
| * Gets the hostname on which container is either currently running or was last seen on if it is not running |
| * TODO SAMZA-2480: Move logic related to onResourcesCompleted from ContainerProcessManager to ContainerManager |
| */ |
| private String getSourceHostForContainer(ContainerPlacementRequestMessage requestMessage) { |
| String sourceHost = null; |
| String processorId = requestMessage.getProcessorId(); |
| if (samzaApplicationState.runningProcessors.containsKey(processorId)) { |
| SamzaResource currentResource = samzaApplicationState.runningProcessors.get(processorId); |
| LOG.info("Processor ID: {} matched a running container with containerId ID: {} is running on host: {} for ContainerPlacement action: {}", |
| processorId, currentResource.getContainerId(), currentResource.getHost(), requestMessage); |
| sourceHost = currentResource.getHost(); |
| } else { |
| sourceHost = Optional.ofNullable(localityManager.readLocality().getProcessorLocality(processorId)) |
| .map(ProcessorLocality::host) |
| .orElse(null); |
| LOG.info("Processor ID: {} is not running and was last seen on host: {} for ContainerPlacement action: {}", |
| processorId, sourceHost, requestMessage); |
| } |
| return sourceHost; |
| } |
| |
| /** |
| * These are specific scenarios in which a placement action should wait for existing action to complete before it is executed |
| * 1. If there is an placement request in progress on active container |
| * 2. If there is an placement request is progress on any of its standby container |
| * 3. If the container itself is pending a start |
| * |
| * @param requestMessage container placement request message |
| * @return true if action should be taken right now, false if it needs to wait to be taken in future |
| */ |
| private boolean deQueueAction(ContainerPlacementRequestMessage requestMessage) { |
| // Do not dequeue action wait for the in-flight action to complete |
| if (checkIfActiveOrStandbyContainerHasActivePlacementAction(requestMessage)) { |
| return false; |
| } |
| |
| if (samzaApplicationState.failedProcessors.containsKey(requestMessage.getProcessorId())) { |
| LOG.info("ContainerPlacement request: {} is de-queued, container with Processor ID: {} has exhausted all retries and is in failed state", |
| requestMessage, requestMessage.getProcessorId()); |
| return true; |
| } |
| |
| // Do not dequeue the action wait for the container to come to a running state |
| if (!samzaApplicationState.runningProcessors.containsKey(requestMessage.getProcessorId()) |
| || samzaApplicationState.pendingProcessors.containsKey(requestMessage.getProcessorId())) { |
| LOG.info("ContainerPlacement request: {} is en-queued because container is pending start", requestMessage); |
| return false; |
| } |
| return true; |
| } |
| |
| /** |
| * A valid container placement action needs a valid processor id. Duplicate actions are handled by de-duping on uuid. |
| * If standby containers are enabled destination host requested must meet standby constraints |
| * |
| * @param requestMessage container placement request message |
| * @return Pair<ContainerPlacementMessage.StatusCode, String> which is status code & response suggesting if the request is valid |
| */ |
| private Pair<ContainerPlacementMessage.StatusCode, String> validatePlacementAction(ContainerPlacementRequestMessage requestMessage) { |
| String errorMessagePrefix = ContainerPlacementMessage.StatusCode.BAD_REQUEST + " reason: %s"; |
| Boolean invalidAction = false; |
| String errorMessage = null; |
| |
| boolean isRunning = samzaApplicationState.runningProcessors.containsKey(requestMessage.getProcessorId()); |
| boolean isPending = samzaApplicationState.pendingProcessors.containsKey(requestMessage.getProcessorId()); |
| boolean isFailed = samzaApplicationState.failedProcessors.containsKey(requestMessage.getProcessorId()); |
| |
| if (!isRunning && !isPending && !isFailed) { |
| errorMessage = String.format(errorMessagePrefix, "invalid processor id neither in running, pending or failed processors"); |
| invalidAction = true; |
| } else if (placementRequestsCache.containsKey(requestMessage.getUuid())) { |
| errorMessage = String.format(errorMessagePrefix, "duplicate UUID of the request, please retry"); |
| invalidAction = true; |
| } else if (standbyContainerManager.isPresent() && !standbyContainerManager.get() |
| .checkStandbyConstraints(requestMessage.getProcessorId(), requestMessage.getDestinationHost())) { |
| errorMessage = String.format(errorMessagePrefix, "destination host does not meet standby constraints"); |
| invalidAction = true; |
| } |
| |
| if (invalidAction) { |
| return new ImmutablePair<>(ContainerPlacementMessage.StatusCode.BAD_REQUEST, errorMessage); |
| } |
| |
| return new ImmutablePair<>(ContainerPlacementMessage.StatusCode.ACCEPTED, "Request is accepted"); |
| } |
| |
| /** |
| * Checks if there are any active container placement action on container itself or on any of its standby containers |
| * (if enabled) |
| */ |
| private boolean checkIfActiveOrStandbyContainerHasActivePlacementAction(ContainerPlacementRequestMessage request) { |
| String processorId = request.getProcessorId(); |
| // Container itself has active container placement actions |
| if (hasActiveContainerPlacementAction(processorId)) { |
| LOG.info("ContainerPlacement request: {} is en-queued because container has an in-progress placement action", request); |
| return true; |
| } |
| |
| if (standbyContainerManager.isPresent()) { |
| // If requested placement action is on a standby container and its active container has a placement request, |
| // this request shall not be de-queued until in-flight action on active container is complete |
| if (StandbyTaskUtil.isStandbyContainer(processorId) && hasActiveContainerPlacementAction( |
| StandbyTaskUtil.getActiveContainerId(processorId))) { |
| LOG.info("ContainerPlacement request: {} is en-queued because its active container has an in-progress placement action", request); |
| return true; |
| } |
| // If requested placement action is on a standby container and its active container has a placement request, |
| // this request shall not be de-queued until in-flight action on active container is complete |
| if (!StandbyTaskUtil.isStandbyContainer(processorId)) { |
| for (String standby : standbyContainerManager.get().getStandbyList(processorId)) { |
| if (hasActiveContainerPlacementAction(standby)) { |
| LOG.info("ContainerPlacement request: {} is en-queued because one of its standby replica has an in-progress placement action", request); |
| return true; |
| } |
| } |
| } |
| } |
| return false; |
| } |
| |
| } |
