The protocol between an ApplicationMaster and a
* NodeManager to start/stop and increase resource of containers
- * and to get status of running containers.
If security is enabled the NodeManager verifies that the
* ApplicationMaster has truly been allocated the container
@@ -220,6 +222,22 @@ SignalContainerResponse signalToContainer(SignalContainerRequest request)
@Unstable
ResourceLocalizationResponse localize(ResourceLocalizationRequest request)
throws YarnException, IOException;
+
+ /**
+ * Gets container launch context for a container specified in
+ * {@link GetContainerLaunchContextRequest}.
+ * This protocol is only used by the container relocation logic, between node managers, to
+ * transfer the launch context of the container to be relocated to the target node.
+ *
+ * @param request specifies the id of the container for which the launch context is requested
+ * @return Response that contains the requested container launch context
+ * @throws YarnException
+ * @throws IOException
+ */
+ @Public
+ @Stable
+ GetContainerLaunchContextResponse getContainerLaunchContext(
+ GetContainerLaunchContextRequest request) throws YarnException, IOException;
/**
* ReInitialize the Container with a new Launch Context.
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateRequest.java
index f7ce127168cb7..087f227ce4c14 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateRequest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateRequest.java
@@ -26,6 +26,7 @@
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerMoveRequest;
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
@@ -52,6 +53,11 @@
* the ResourceManager about the change in
* requirements of running containers.
*
+ *
ResourceManager about the application's
+ * container relocation needs.
+ * ApplicationMaster.
+ *
+ * @return list of ContainerMoveRequest
+ * being sent by the ApplicationMaster.
+ */
+ @Public
+ @Unstable
+ public abstract ListResourceManager about the containers that need to be
+ * relocated.
+ *
+ * @param containerMoveRequests list of ContainerMoveRequest
+ * for containers that need to be relocated.
+ */
+ @Public
+ @Unstable
+ public abstract void setMoveAskList(ListThe request sent by one NodeManager to another
+ * NodeManager in order to get the launch context of the container with id
+ * containerId.
This request is used only for container relocation, where the launch context of + * the origin container is necessary for launching the relocated container on the target node.
+ * + * @see ContainerManagementProtocol#getContainerLaunchContext(GetContainerLaunchContextRequest) + */ +public abstract class GetContainerLaunchContextRequest { + + @Public + @Stable + public static GetContainerLaunchContextRequest newInstance(ContainerId containerId) { + GetContainerLaunchContextRequest request = + Records.newRecord(GetContainerLaunchContextRequest.class); + request.setContainerId(containerId); + return request; + } + + /** + * Gets the container id for which the launch context is requested. + * @return the container id for which the launch context is requested + */ + @Public + @Stable + public abstract ContainerId getContainerId(); + + /** + * Sets the container id for which the launch context is requested. + * @param containerId the container id for which the launch context is requested + */ + @Public + @Stable + public abstract void setContainerId(ContainerId containerId); +} \ No newline at end of file diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetContainerLaunchContextResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetContainerLaunchContextResponse.java new file mode 100644 index 0000000000000..dbed4f0ec04f0 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetContainerLaunchContextResponse.java @@ -0,0 +1,61 @@ +/** + * 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.hadoop.yarn.api.protocolrecords; + +import org.apache.hadoop.classification.InterfaceAudience.Public; +import org.apache.hadoop.classification.InterfaceStability.Stable; +import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; +import org.apache.hadoop.yarn.util.Records; + +/** + *The response for a container launch context request sent between
+ * NodeManagers. It contains the requested container launch context.
This request is used only for container relocation, where a launch context + * of the origin container is necessary for launching the relocated container on + * the target node.
+ */ +public abstract class GetContainerLaunchContextResponse { + + @Public + @Stable + public static GetContainerLaunchContextResponse newInstance(ContainerLaunchContext + containerLaunchContext) { + GetContainerLaunchContextResponse request = + Records.newRecord(GetContainerLaunchContextResponse.class); + request.setContainerLaunchContext(containerLaunchContext); + return request; + } + + /** + * Gets the launch context of the requested container. + * @return the launch context of the requested container + */ + @Public + @Stable + public abstract ContainerLaunchContext getContainerLaunchContext(); + + /** + * Sets the launch context of the requested container. + * @param containerLaunchContext the launch context of the requested container + */ + @Public + @Stable + public abstract void setContainerLaunchContext(ContainerLaunchContext containerLaunchContext); +} \ No newline at end of file diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/StartContainerRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/StartContainerRequest.java index 50179a962bad2..6af80e18c0d22 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/StartContainerRequest.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/StartContainerRequest.java @@ -20,9 +20,12 @@ import org.apache.hadoop.classification.InterfaceAudience.Public; import org.apache.hadoop.classification.InterfaceStability.Stable; +import org.apache.hadoop.classification.InterfaceStability.Unstable; import org.apache.hadoop.yarn.api.ContainerManagementProtocol; +import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; import org.apache.hadoop.yarn.api.records.NMToken; +import org.apache.hadoop.yarn.api.records.NodeId; import org.apache.hadoop.yarn.api.records.Token; import org.apache.hadoop.yarn.util.Records; @@ -36,6 +39,15 @@ * necessary binaries/jar/shared-objects etc. via the * {@link ContainerLaunchContext}. * + *The isMove flag tells whether this request corresponds to a container + * relocation. If true, then no container launch context is needed as it will be + * transferred directly from the origin node manager to the target node manager. + * The container to be relocated is identified by the originNodeId and + * originContainerId. + * The originNMToken is sent along so that the origin container can be shut down + * by the target node manager. + *
+ * * @see ContainerManagementProtocol#startContainers(StartContainersRequest) */ @Public @@ -49,6 +61,21 @@ public static StartContainerRequest newInstance( Records.newRecord(StartContainerRequest.class); request.setContainerLaunchContext(context); request.setContainerToken(container); + request.setIsMove(false); + return request; + } + + @Public + @Stable + public static StartContainerRequest newInstance( + Token container, ContainerId originContainerId, NodeId originNodeId, Token originNMToken) { + StartContainerRequest request = + Records.newRecord(StartContainerRequest.class); + request.setContainerToken(container); + request.setIsMove(true); + request.setOriginContainerId(originContainerId); + request.setOriginNodeId(originNodeId); + request.setOriginNMToken(originNMToken); return request; } @@ -91,4 +118,91 @@ public static StartContainerRequest newInstance( @Public @Stable public abstract void setContainerToken(Token container); + + /** + * Gets whether this start container request corresponds to a container relocation. + * @return whether this start container request corresponds to a container relocation + */ + @Public + @Unstable + public abstract boolean getIsMove(); + + /** + * Sets whether this start container request corresponds to a container relocation. + * @param isMove whether this start container request corresponds to a container + * relocation + */ + @Public + @Unstable + public abstract void setIsMove(boolean isMove); + + /** + * Gets the origin container id for this start container request. + * The origin container id is set if and only if this start container request + * corresponds to a container relocation. It identifies the container that should + * be relocated. + * + * @return the origin container id for this start container request + */ + @Public + @Unstable + public abstract ContainerId getOriginContainerId(); + + /** + * Sets the origin container id for this start container request. + * The origin container id should be set if and only if this start container request + * corresponds to a container relocation. It identifies the container that should + * be relocated. + * + * @param originContainerId the origin container id for this start container request + */ + @Public + @Unstable + public abstract void setOriginContainerId(ContainerId originContainerId); + + /** + * Gets the origin node id for this start container request. + * The origin node id is set if and only if this resource start container corresponds + * to a container relocation. It identifies the node of the container that should + * be relocated. + * + * @return the origin node id for this start container request + */ + @Public + @Unstable + public abstract NodeId getOriginNodeId(); + + /** + * Sets the origin node id for this start container request. + * The origin node id should be set if and only if this start container request + * corresponds to a container relocation. It identifies the node of the container + * that should be relocated. + * + * @param originNodeId the origin node id for this start container request + */ + @Public + @Unstable + public abstract void setOriginNodeId(NodeId originNodeId); + + /** + * Gets the security token for the origin node. + * The origin NM token is set if and only if this start container request corresponds + * to a container relocation. It is used for shutting down the origin container. + * + * @return the security token for the origin node + */ + @Public + @Unstable + public abstract Token getOriginNMToken(); + + /** + * Sets the security token for the origin node. + * The origin NM token should be set if and only if this start container request corresponds + * to a container relocation. It is used for shutting down the origin container. + * + * @param originNMToken the security token for the origin node + */ + @Public + @Unstable + public abstract void setOriginNMToken(Token originNMToken); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java index 4fdc8034c5707..9cc3828edce0c 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java @@ -48,6 +48,13 @@ * Container {@link Token} of the container, used to securely verify * authenticity of the allocation. * + *+ * 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.hadoop.yarn.api.records; + +import java.io.Serializable; + +import org.apache.hadoop.classification.InterfaceAudience.Public; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.api.ApplicationMasterProtocol; +import org.apache.hadoop.yarn.util.Records; + +/** + * {@code ContainerMoveRequest} represents the request made + * by an application to the {@code ResourceManager} + * to relocate a single container to another node of the yarn cluster. + *
+ * It includes: + *
Priority of the request.
+ * @return Priority of the request
+ */
+ @Public
+ @Unstable
+ public abstract Priority getPriority();
+
+ /**
+ * Set the Priority of the request.
+ * @param priority Priority of the request
+ */
+ @Public
+ @Unstable
+ public abstract void setPriority(Priority priority);
+
+ /**
+ * Gets the origin container id of this container move request.
+ * It identifies the container that should be relocated.
+ *
+ * @return the origin container id of this container move request
+ */
+ @Public
+ @Unstable
+ public abstract ContainerId getOriginContainerId();
+
+ /**
+ * Sets the origin container id of this container move request.
+ * It identifies the container that should be relocated.
+ *
+ * @param originContainerId the origin container id of this container move
+ * request
+ */
+ @Public
+ @Unstable
+ public abstract void setOriginContainerId(ContainerId originContainerId);
+
+ /**
+ * Gets the target host of this container move request, which identifies
+ * the node where the container should be relocated to.
+ *
+ * @return the target host of this container move request
+ */
+ @Public
+ @Unstable
+ public abstract String getTargetHost();
+
+ /**
+ * Sets the target host of this container move request, which identifies
+ * the node where the container should be relocated to.
+ *
+ * @param targetHost the target host of this container move request
+ */
+ @Public
+ @Unstable
+ public abstract void setTargetHost(String targetHost);
+
+ @Override
+ public int hashCode() {
+ final int prime = 2153;
+ int result = 2459;
+ Priority priority = getPriority();
+ ContainerId originContainerId = getOriginContainerId();
+ String targetHost = getTargetHost();
+
+ result = prime * result + ((priority == null) ? 0 : priority.hashCode());
+ result = prime * result +
+ ((originContainerId == null) ? 0 : originContainerId.hashCode());
+ result = prime * result + ((targetHost == null) ? 0 : targetHost.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ ContainerMoveRequest other = (ContainerMoveRequest) obj;
+ ContainerId originContainerId = getOriginContainerId();
+ ContainerId otherOriginContainerId = other.getOriginContainerId();
+ if (originContainerId == null) {
+ if (otherOriginContainerId != null)
+ return false;
+ } else if (!originContainerId.equals(otherOriginContainerId))
+ return false;
+ String targetHost = getTargetHost();
+ String otherTargetHost = other.getTargetHost();
+ if (targetHost == null) {
+ if (otherTargetHost != null)
+ return false;
+ } else if (!targetHost.equals(otherTargetHost))
+ return false;
+ Priority priority = getPriority();
+ Priority otherPriority = other.getPriority();
+ if (priority == null) {
+ if (otherPriority != null)
+ return false;
+ } else if (!priority.equals(otherPriority))
+ return false;
+ return true;
+ }
+
+ @Override
+ public int compareTo(ContainerMoveRequest other) {
+ int priorityComparison = this.getPriority().compareTo(other.getPriority());
+ if (priorityComparison == 0) {
+ int originContainerIdComparison =
+ this.getOriginContainerId().compareTo(other.getOriginContainerId());
+ if (originContainerIdComparison == 0) {
+ return this.getTargetHost().compareTo(other.getTargetHost());
+ } else {
+ return originContainerIdComparison;
+ }
+ } else {
+ return priorityComparison;
+ }
+ }
+}
+
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
index 2d6f0f460d42c..35164b66ab618 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
@@ -23,6 +23,7 @@
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Evolving;
import org.apache.hadoop.classification.InterfaceStability.Stable;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
import org.apache.hadoop.yarn.util.Records;
@@ -50,6 +51,14 @@
* locality to be loose (i.e. allows fall-through to rack or any)
* or strict (i.e. specify hard constraint on resource allocation).
*
+ * allocate
+ * @param req container move request to be sent to the resource manager
+ */
+ public abstract void addContainerMoveRequest(ContainerMoveRequest req);
/**
* Remove previous container request. The previous container request may have
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
index 32216610612ae..56ba7687e9ee0 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
@@ -52,6 +52,7 @@
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerMoveRequest;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.ContainerUpdateType;
import org.apache.hadoop.yarn.api.records.ExecutionType;
@@ -162,6 +163,8 @@ static boolean canFit(Resource arg0, Resource arg1) {
protected final SetApplicationMaster
+ * and ResourceManager
+ */
+public class TestRMContainerMove {
+ private static final Log LOG = LogFactory.getLog(TestRMContainerMove.class);
+
+ static Configuration conf = null;
+ static MiniYARNCluster yarnCluster = null;
+ static YarnClient yarnClient = null;
+ static List+ * 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.hadoop.yarn.api.records.impl.pb;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerMoveRequest;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.proto.YarnProtos.PriorityProto;
+import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
+import org.apache.hadoop.yarn.proto.YarnProtos.ContainerMoveRequestProto;
+import org.apache.hadoop.yarn.proto.YarnProtos.ContainerMoveRequestProtoOrBuilder;
+
+@Private
+@Unstable
+public class ContainerMoveRequestPBImpl extends ContainerMoveRequest {
+
+ ContainerMoveRequestProto proto = ContainerMoveRequestProto.getDefaultInstance();
+ ContainerMoveRequestProto.Builder builder = null;
+ boolean viaProto = false;
+
+ private Priority priority = null;
+ private ContainerId originContainerId = null;
+
+ public ContainerMoveRequestPBImpl() {
+ builder = ContainerMoveRequestProto.newBuilder();
+ }
+
+ public ContainerMoveRequestPBImpl(ContainerMoveRequestProto proto) {
+ this.proto = proto;
+ viaProto = true;
+ }
+
+ public ContainerMoveRequestProto getProto() {
+ mergeLocalToProto();
+ proto = viaProto ? proto : builder.build();
+ viaProto = true;
+ return proto;
+ }
+
+ private void mergeLocalToBuilder() {
+ if (this.priority != null) {
+ builder.setPriority(convertToProtoFormat(this.priority));
+ }
+ if (this.originContainerId != null) {
+ builder.setOriginContainerId(convertToProtoFormat(this.originContainerId));
+ }
+ }
+
+ private void mergeLocalToProto() {
+ if (viaProto)
+ maybeInitBuilder();
+ mergeLocalToBuilder();
+ proto = builder.build();
+ viaProto = true;
+ }
+
+ private void maybeInitBuilder() {
+ if (viaProto || builder == null) {
+ builder = ContainerMoveRequestProto.newBuilder(proto);
+ }
+ viaProto = false;
+ }
+
+ @Override
+ public Priority getPriority() {
+ ContainerMoveRequestProtoOrBuilder p = viaProto ? proto : builder;
+ if (this.priority != null) {
+ return this.priority;
+ }
+ if (!p.hasPriority()) {
+ return null;
+ }
+ this.priority = convertFromProtoFormat(p.getPriority());
+ return this.priority;
+ }
+
+ @Override
+ public void setPriority(Priority priority) {
+ maybeInitBuilder();
+ if (priority == null)
+ builder.clearPriority();
+ this.priority = priority;
+ }
+
+ @Override
+ public ContainerId getOriginContainerId() {
+ ContainerMoveRequestProtoOrBuilder p = viaProto ? proto : builder;
+ if (this.originContainerId != null) {
+ return this.originContainerId;
+ }
+ if (!p.hasOriginContainerId()) {
+ return null;
+ }
+ this.originContainerId = convertFromProtoFormat(p.getOriginContainerId());
+ return this.originContainerId;
+ }
+
+ @Override
+ public void setOriginContainerId(ContainerId originContainerId) {
+ maybeInitBuilder();
+ if (priority == null)
+ builder.clearOriginContainerId();
+ this.originContainerId = originContainerId;
+ }
+
+ @Override
+ public String getTargetHost() {
+ ContainerMoveRequestProtoOrBuilder p = viaProto ? proto : builder;
+ if (!p.hasTargetHost()) {
+ return null;
+ }
+ return (p.getTargetHost());
+ }
+
+ @Override
+ public void setTargetHost(String targetHost) {
+ maybeInitBuilder();
+ if (targetHost == null)
+ builder.clearTargetHost();
+ builder.setTargetHost(targetHost);
+ }
+
+ private PriorityPBImpl convertFromProtoFormat(PriorityProto p) {
+ return new PriorityPBImpl(p);
+ }
+
+ private PriorityProto convertToProtoFormat(Priority t) {
+ return ((PriorityPBImpl) t).getProto();
+ }
+
+ private ContainerIdPBImpl convertFromProtoFormat(ContainerIdProto p) {
+ return new ContainerIdPBImpl(p);
+ }
+
+ private ContainerIdProto convertToProtoFormat(ContainerId t) {
+ return ((ContainerIdPBImpl) t).getProto();
+ }
+
+ @Override
+ public String toString() {
+ return "{Priority: " + getPriority()
+ + ", OriginContainerId: " + getOriginContainerId()
+ + ", TargetHost: " + getTargetHost() + "}";
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ContainerPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ContainerPBImpl.java
index 6bf653dd00ab8..206088aa68bd5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ContainerPBImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ContainerPBImpl.java
@@ -49,7 +49,11 @@ public class ContainerPBImpl extends Container {
private Resource resource = null;
private Priority priority = null;
private Token containerToken = null;
-
+
+ private ContainerId originContainerId = null;
+ private NodeId originNodeId = null;
+
+
public ContainerPBImpl() {
builder = ContainerProto.newBuilder();
}
@@ -108,6 +112,16 @@ private void mergeLocalToBuilder() {
builder.getContainerToken())) {
builder.setContainerToken(convertToProtoFormat(this.containerToken));
}
+ if (this.originContainerId != null
+ && !((ContainerIdPBImpl) originContainerId).getProto().equals(
+ builder.getOriginContainerId())) {
+ builder.setOriginContainerId(convertToProtoFormat(this.originContainerId));
+ }
+ if (this.originNodeId != null
+ && !((NodeIdPBImpl) originNodeId).getProto().equals(
+ builder.getOriginNodeId())) {
+ builder.setOriginNodeId(convertToProtoFormat(this.originNodeId));
+ }
}
private void mergeLocalToProto() {
@@ -285,7 +299,61 @@ public void setVersion(int version) {
maybeInitBuilder();
builder.setVersion(version);
}
-
+
+ @Override
+ public boolean getIsMove() {
+ ContainerProtoOrBuilder p = viaProto ? proto : builder;
+ return p.getIsMove();
+ }
+
+ @Override
+ public void setIsMove(boolean isMove) {
+ maybeInitBuilder();
+ builder.setIsMove(isMove);
+ }
+
+ @Override
+ public ContainerId getOriginContainerId() {
+ ContainerProtoOrBuilder p = viaProto ? proto : builder;
+ if (this.originContainerId != null) {
+ return this.originContainerId;
+ }
+ if (!p.hasOriginContainerId()) {
+ return null;
+ }
+ this.originContainerId = convertFromProtoFormat(p.getOriginContainerId());
+ return this.originContainerId;
+ }
+
+ @Override
+ public void setOriginContainerId(ContainerId originContainerId) {
+ maybeInitBuilder();
+ if (originContainerId == null)
+ builder.clearOriginContainerId();
+ this.originContainerId = originContainerId;
+ }
+
+ @Override
+ public NodeId getOriginNodeId() {
+ ContainerProtoOrBuilder p = viaProto ? proto : builder;
+ if (this.originNodeId != null) {
+ return this.originNodeId;
+ }
+ if (!p.hasOriginNodeId()) {
+ return null;
+ }
+ this.originNodeId = convertFromProtoFormat(p.getOriginNodeId());
+ return this.originNodeId;
+ }
+
+ @Override
+ public void setOriginNodeId(NodeId originNodeId) {
+ maybeInitBuilder();
+ if (originNodeId == null)
+ builder.clearOriginNodeId();
+ this.originNodeId = originNodeId;
+ }
+
private ContainerIdPBImpl convertFromProtoFormat(ContainerIdProto p) {
return new ContainerIdPBImpl(p);
}
@@ -348,6 +416,11 @@ public String toString() {
sb.append("Priority: ").append(getPriority()).append(", ");
sb.append("Token: ").append(getContainerToken()).append(", ");
sb.append("ExecutionType: ").append(getExecutionType()).append(", ");
+ sb.append("IsMove: ").append(getIsMove()).append(", ");
+ if (getIsMove()) {
+ sb.append("OriginContainerId: ").append(getOriginContainerId()).append(", ");
+ sb.append("OriginNodeId: ").append(getOriginNodeId()).append(", ");
+ }
sb.append("]");
return sb.toString();
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ResourceRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ResourceRequestPBImpl.java
index 9890296acddd2..3acfd51ca7183 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ResourceRequestPBImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ResourceRequestPBImpl.java
@@ -21,6 +21,7 @@
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
@@ -41,6 +42,8 @@ public class ResourceRequestPBImpl extends ResourceRequest {
private Resource capability = null;
private ExecutionTypeRequest executionTypeRequest = null;
+ private boolean isMove = false;
+ private ContainerId originContainerId = null;
public ResourceRequestPBImpl() {
builder = ResourceRequestProto.newBuilder();
@@ -203,6 +206,26 @@ public void setAllocationRequestId(long allocationRequestID) {
maybeInitBuilder();
builder.setAllocationRequestId(allocationRequestID);
}
+
+ @Override
+ public boolean getIsMove() {
+ return isMove;
+ }
+
+ @Override
+ public void setIsMove(boolean isMove) {
+ this.isMove = isMove;
+ }
+
+ @Override
+ public ContainerId getOriginContainerId() {
+ return originContainerId;
+ }
+
+ @Override
+ public void setOriginContainerId(ContainerId originContainerId) {
+ this.originContainerId = originContainerId;
+ }
private PriorityPBImpl convertFromProtoFormat(PriorityProto p) {
return new PriorityPBImpl(p);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java
index 90c7573a0f343..67d0d932e558d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java
@@ -33,10 +33,12 @@
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.CommitResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceResponse;
import org.apache.hadoop.yarn.api.protocolrecords.ReInitializeContainerRequest;
import org.apache.hadoop.yarn.api.protocolrecords.ReInitializeContainerResponse;
import org.apache.hadoop.yarn.api.protocolrecords.ResourceLocalizationRequest;
@@ -211,6 +213,13 @@ public ResourceLocalizationResponse localize(
ResourceLocalizationRequest request) throws YarnException, IOException {
return null;
}
+
+ @Override
+ public GetContainerLaunchContextResponse getContainerLaunchContext(
+ GetContainerLaunchContextRequest request) throws YarnException, IOException {
+ throw new UnsupportedOperationException("getting the container launch context is not " +
+ "supported for this implementation of ContainerManagementProtocol");
+ }
@Override
public ReInitializeContainerResponse reInitializeContainer(
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerResourceIncreaseRPC.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerResourceIncreaseRPC.java
index f97f7c74df76e..81f0f0eed3540 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerResourceIncreaseRPC.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerResourceIncreaseRPC.java
@@ -27,6 +27,8 @@
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.CommitResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesResponse;
import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceRequest;
@@ -192,6 +194,13 @@ public SignalContainerResponse signalToContainer(
ResourceLocalizationRequest request) throws YarnException, IOException {
return null;
}
+
+ @Override
+ public GetContainerLaunchContextResponse getContainerLaunchContext(
+ GetContainerLaunchContextRequest request) throws YarnException, IOException {
+ throw new UnsupportedOperationException("getting the container launch context is not " +
+ "supported for this implementation of ContainerManagementProtocol");
+ }
@Override
public ReInitializeContainerResponse reInitializeContainer(
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/TestRPC.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/TestRPC.java
index e5d159b2ad6a1..f28e7f223d5b6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/TestRPC.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/TestRPC.java
@@ -34,10 +34,12 @@
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
import org.apache.hadoop.yarn.api.ContainerManagementProtocolPB;
import org.apache.hadoop.yarn.api.protocolrecords.CommitResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest;
import org.apache.hadoop.yarn.api.protocolrecords.ReInitializeContainerRequest;
import org.apache.hadoop.yarn.api.protocolrecords.ReInitializeContainerResponse;
@@ -360,6 +362,13 @@ public ResourceLocalizationResponse localize(
ResourceLocalizationRequest request) throws YarnException, IOException {
return null;
}
+
+ @Override
+ public GetContainerLaunchContextResponse getContainerLaunchContext(
+ GetContainerLaunchContextRequest request) throws YarnException, IOException {
+ throw new UnsupportedOperationException("getting the container launch context is not " +
+ "supported for this implementation of ContainerManagementProtocol");
+ }
@Override
public ReInitializeContainerResponse reInitializeContainer(
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
index e8c2b7552ca33..8c53d0e482a30 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
@@ -39,6 +39,8 @@
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.CommitResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerLaunchContextResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesResponse;
import org.apache.hadoop.yarn.api.protocolrecords.IncreaseContainersResourceRequest;
@@ -69,9 +71,11 @@
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.SerializedException;
+import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationIdPBImpl;
import org.apache.hadoop.yarn.api.records.impl.pb.LogAggregationContextPBImpl;
import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
+import org.apache.hadoop.yarn.client.NMProxy;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
@@ -144,6 +148,7 @@
import org.apache.hadoop.yarn.server.nodemanager.timelineservice.NMTimelinePublisher;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.server.utils.YarnServerSecurityUtils;
+import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.resource.Resources;
import org.apache.hadoop.yarn.util.timeline.TimelineUtils;
@@ -155,6 +160,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -861,8 +867,16 @@ public StartContainersResponse startContainers(
.equals(ContainerType.APPLICATION_MASTER)) {
this.getAMRMProxyService().processApplicationStartRequest(request);
}
- performContainerPreStartChecks(nmTokenIdentifier, request,
- containerTokenIdentifier);
+ if(!request.getIsMove()) {
+ performContainerPreStartChecks(nmTokenIdentifier, request,
+ containerTokenIdentifier);
+ } else {
+ // if the request does not have a container launch context,
+ // just authorize the container start and update the NMToken
+ authorizeStartAndResourceIncreaseRequest(
+ nmTokenIdentifier, containerTokenIdentifier, true);
+ updateNMTokenIdentifier(nmTokenIdentifier);
+ }
startContainerInternal(containerTokenIdentifier, request);
succeededContainers.add(containerId);
} catch (YarnException e) {
@@ -966,10 +980,25 @@ protected void startContainerInternal(
ContainerId containerId = containerTokenIdentifier.getContainerID();
String containerIdStr = containerId.toString();
String user = containerTokenIdentifier.getApplicationSubmitter();
+ ContainerManagementProtocol proxy = null;
LOG.info("Start request for " + containerIdStr + " by user " + user);
-
- ContainerLaunchContext launchContext = request.getContainerLaunchContext();
+
+ ContainerLaunchContext launchContext = null;
+ Configuration conf = null;
+ YarnRPC rpc = null;
+ if(request.getIsMove()) {
+ // In case of a relocation request, the launch context is borrowed from the origin container
+ conf = getConfig();
+ rpc = YarnRPC.create(conf);
+ // get the proxy to the origin NM
+ proxy = getCMClient(containerId, request.getOriginNodeId(), conf, rpc, request.getOriginNMToken());
+ // request the launch context from the origin NM
+ launchContext = proxy.getContainerLaunchContext(GetContainerLaunchContextRequest.newInstance(
+ request.getOriginContainerId())).getContainerLaunchContext();
+ } else {
+ launchContext = request.getContainerLaunchContext();
+ }
Credentials credentials =
YarnServerSecurityUtils.parseCredentials(launchContext);
@@ -1044,12 +1073,23 @@ protected void startContainerInternal(
// launch. A finished Application will not launch containers.
metrics.launchedContainer();
metrics.allocateContainer(containerTokenIdentifier.getResource());
+ LOG.debug("Started container " + containerId + " on node " + context.getNodeId() +
+ " with commands " + launchContext.getCommands());
+ // In case of a relocation request, shut down the origin container
+ if(request.getIsMove()) {
+ proxy.stopContainers(StopContainersRequest.newInstance(Collections.singletonList(request
+ .getOriginContainerId())));
+ }
} else {
throw new YarnException(
"Container start failed as the NodeManager is " +
"in the process of shutting down");
}
} finally {
+ // stop the proxy to the origin NM
+ if(proxy != null) {
+ rpc.stopProxy(proxy, conf);
+ }
this.readLock.unlock();
}
}
@@ -1129,6 +1169,24 @@ public IncreaseContainersResourceResponse increaseContainersResource(
return IncreaseContainersResourceResponse.newInstance(
successfullyIncreasedContainers, failedContainers);
}
+
+ @Override
+ public GetContainerLaunchContextResponse getContainerLaunchContext(
+ GetContainerLaunchContextRequest request) throws YarnException, IOException {
+ ContainerId containerId = request.getContainerId();
+ // Getting the launch context of the origin container locally
+ ContainerLaunchContext launchContext = context.getContainers().get(containerId)
+ .getLaunchContext();
+ LOG.debug("Got launch context of container " + containerId + " on node " + context.getNodeId());
+ // We copy the launch context in order to avoid ConcurrentModificationException
+ ContainerLaunchContext launchContextCopy;
+ try {
+ launchContextCopy = launchContext.copy();
+ } catch (Exception e) {
+ throw new YarnException("Unable to copy container launch context");
+ }
+ return GetContainerLaunchContextResponse.newInstance(launchContextCopy);
+ }
@SuppressWarnings("unchecked")
private void changeContainerResourceInternal(ContainerId containerId,
@@ -1242,6 +1300,7 @@ public StopContainersResponse stopContainers(StopContainersRequest requests)
authorizeGetAndStopContainerRequest(id, container, true, identifier);
stopContainerInternal(id);
succeededRequests.add(id);
+ LOG.debug("Stopped container " + id + " on node " + context.getNodeId());
} catch (YarnException e) {
failedRequests.put(id, SerializedException.newInstance(e));
}
@@ -1683,4 +1742,27 @@ private void internalSignalToContainer(SignalContainerRequest request,
LOG.info("Container " + containerId + " no longer exists");
}
}
+
+ /**
+ * Sets up an NM Client to the specified NM; uses the given token for authentication
+ */
+ private ContainerManagementProtocol getCMClient(ContainerId containerId, NodeId
+ originNodeId, Configuration conf, YarnRPC rpc, Token token) {
+ conf.setInt(
+ CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY, 0);
+ UserGroupInformation ugi = null;
+ try {
+ ugi =
+ UserGroupInformation.createRemoteUser(containerId
+ .getApplicationAttemptId().toString());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ InetSocketAddress serverAddress = NetUtils.createSocketAddrForHost(originNodeId.getHost(),
+ originNodeId.getPort());
+ org.apache.hadoop.security.token.Token