You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by as...@apache.org on 2015/11/26 20:52:23 UTC
[13/50] [abbrv] hadoop git commit: YARN-3980. Plumb
resource-utilization info in node heartbeat through to the scheduler. (Inigo
Goiri via kasha)
YARN-3980. Plumb resource-utilization info in node heartbeat through to the scheduler. (Inigo Goiri via kasha)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/52948bb2
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/52948bb2
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/52948bb2
Branch: refs/heads/yarn-2877
Commit: 52948bb20bd1446164df1d3920c46c96dad750ae
Parents: f80dc6f
Author: Karthik Kambatla <ka...@apache.org>
Authored: Tue Nov 24 10:05:12 2015 +0530
Committer: Karthik Kambatla <ka...@apache.org>
Committed: Tue Nov 24 13:47:17 2015 +0530
----------------------------------------------------------------------
.../hadoop/yarn/sls/nodemanager/NodeInfo.java | 11 +
.../yarn/sls/scheduler/RMNodeWrapper.java | 11 +
hadoop-yarn-project/CHANGES.txt | 3 +
.../impl/pb/ResourceUtilizationPBImpl.java | 2 +-
.../nodemanager/NodeStatusUpdaterImpl.java | 3 +-
.../resourcemanager/ResourceTrackerService.java | 5 +-
.../server/resourcemanager/rmnode/RMNode.java | 15 +-
.../resourcemanager/rmnode/RMNodeImpl.java | 55 +++++
.../rmnode/RMNodeStatusEvent.java | 53 ++--
.../scheduler/SchedulerNode.java | 38 +++
.../scheduler/capacity/CapacityScheduler.java | 5 +
.../scheduler/fair/FairScheduler.java | 5 +
.../scheduler/fifo/FifoScheduler.java | 4 +
.../yarn/server/resourcemanager/MockNodes.java | 28 ++-
.../resourcemanager/TestRMNodeTransitions.java | 16 +-
.../TestRMAppLogAggregationStatus.java | 32 +--
.../webapp/TestRMWebServicesNodes.java | 6 +-
.../hadoop/yarn/server/MiniYARNCluster.java | 54 +++-
.../TestMiniYarnClusterNodeUtilization.java | 245 +++++++++++++++++++
19 files changed, 522 insertions(+), 69 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/nodemanager/NodeInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/nodemanager/NodeInfo.java b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/nodemanager/NodeInfo.java
index dae2ce7..f5943a8 100644
--- a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/nodemanager/NodeInfo.java
+++ b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/nodemanager/NodeInfo.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.NodeState;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode
@@ -188,6 +189,16 @@ public class NodeInfo {
// TODO Auto-generated method stub
return null;
}
+
+ @Override
+ public ResourceUtilization getAggregatedContainersUtilization() {
+ return null;
+ }
+
+ @Override
+ public ResourceUtilization getNodeUtilization() {
+ return null;
+ }
}
public static RMNode newNodeInfo(String rackName, String hostName,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/RMNodeWrapper.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/RMNodeWrapper.java b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/RMNodeWrapper.java
index 8c65ccc..e778188 100644
--- a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/RMNodeWrapper.java
+++ b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/RMNodeWrapper.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.NodeState;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode
@@ -176,4 +177,14 @@ public class RMNodeWrapper implements RMNode {
// TODO Auto-generated method stub
return null;
}
+
+ @Override
+ public ResourceUtilization getAggregatedContainersUtilization() {
+ return node.getAggregatedContainersUtilization();
+ }
+
+ @Override
+ public ResourceUtilization getNodeUtilization() {
+ return node.getNodeUtilization();
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index a76c835..0532e1d 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -568,6 +568,9 @@ Release 2.8.0 - UNRELEASED
YARN-3454. Add efficient merge operation to RLESparseResourceAllocation
(Carlo Curino via asuresh)
+ YARN-3980. Plumb resource-utilization info in node heartbeat through to the
+ scheduler. (Inigo Goiri via kasha)
+
OPTIMIZATIONS
YARN-3339. TestDockerContainerExecutor should pull a single image and not
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/records/impl/pb/ResourceUtilizationPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/records/impl/pb/ResourceUtilizationPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/records/impl/pb/ResourceUtilizationPBImpl.java
index 01cda7a..29f79d4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/records/impl/pb/ResourceUtilizationPBImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/records/impl/pb/ResourceUtilizationPBImpl.java
@@ -75,7 +75,7 @@ public class ResourceUtilizationPBImpl extends ResourceUtilization {
@Override
public void setVirtualMemory(int vmem) {
maybeInitBuilder();
- builder.setPmem(vmem);
+ builder.setVmem(vmem);
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
index 3f8cf32..34267b3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
@@ -417,7 +417,8 @@ public class NodeStatusUpdaterImpl extends AbstractService implements
return appList;
}
- private NodeStatus getNodeStatus(int responseId) throws IOException {
+ @VisibleForTesting
+ protected NodeStatus getNodeStatus(int responseId) throws IOException {
NodeHealthStatus nodeHealthStatus = this.context.getNodeHealthStatus();
nodeHealthStatus.setHealthReport(healthChecker.getHealthReport());
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
index 3638a19..bd24b25 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
@@ -461,10 +461,7 @@ public class ResourceTrackerService extends AbstractService implements
// 4. Send status to RMNode, saving the latest response.
RMNodeStatusEvent nodeStatusEvent =
- new RMNodeStatusEvent(nodeId, remoteNodeStatus.getNodeHealthStatus(),
- remoteNodeStatus.getContainersStatuses(),
- remoteNodeStatus.getKeepAliveApplications(), nodeHeartBeatResponse,
- remoteNodeStatus.getIncreasedContainers());
+ new RMNodeStatusEvent(nodeId, remoteNodeStatus, nodeHeartBeatResponse);
if (request.getLogAggregationReportsForApps() != null
&& !request.getLogAggregationReportsForApps().isEmpty()) {
nodeStatusEvent.setLogAggregationReportsForApps(request
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNode.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNode.java
index f28422a..1a172e8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNode.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNode.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.NodeState;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
/**
* Node managers information on available resources
@@ -98,7 +99,19 @@ public interface RMNode {
* @return the total available resource.
*/
public Resource getTotalCapability();
-
+
+ /**
+ * the aggregated resource utilization of the containers.
+ * @return the aggregated resource utilization of the containers.
+ */
+ public ResourceUtilization getAggregatedContainersUtilization();
+
+ /**
+ * the total resource utilization of the node.
+ * @return the total resource utilization of the node.
+ */
+ public ResourceUtilization getNodeUtilization();
+
/**
* The rack name for this node manager.
* @return the rack name.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
index e0d27d6..146b0d3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
@@ -57,6 +57,7 @@ import org.apache.hadoop.yarn.server.api.protocolrecords.LogAggregationReport;
import org.apache.hadoop.yarn.server.api.protocolrecords.NMContainerStatus;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
import org.apache.hadoop.yarn.server.resourcemanager.ClusterMetrics;
import org.apache.hadoop.yarn.server.resourcemanager.NodesListManagerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.NodesListManagerEventType;
@@ -114,6 +115,11 @@ public class RMNodeImpl implements RMNode, EventHandler<RMNodeEvent> {
private long lastHealthReportTime;
private String nodeManagerVersion;
+ /* Aggregated resource utilization for the containers. */
+ private ResourceUtilization containersUtilization;
+ /* Resource utilization for the node. */
+ private ResourceUtilization nodeUtilization;
+
private final ContainerAllocationExpirer containerAllocationExpirer;
/* set of containers that have just launched */
private final Set<ContainerId> launchedContainers =
@@ -446,6 +452,49 @@ public class RMNodeImpl implements RMNode, EventHandler<RMNodeEvent> {
}
@Override
+ public ResourceUtilization getAggregatedContainersUtilization() {
+ this.readLock.lock();
+
+ try {
+ return this.containersUtilization;
+ } finally {
+ this.readLock.unlock();
+ }
+ }
+
+ public void setAggregatedContainersUtilization(
+ ResourceUtilization containersUtilization) {
+ this.writeLock.lock();
+
+ try {
+ this.containersUtilization = containersUtilization;
+ } finally {
+ this.writeLock.unlock();
+ }
+ }
+
+ @Override
+ public ResourceUtilization getNodeUtilization() {
+ this.readLock.lock();
+
+ try {
+ return this.nodeUtilization;
+ } finally {
+ this.readLock.unlock();
+ }
+ }
+
+ public void setNodeUtilization(ResourceUtilization nodeUtilization) {
+ this.writeLock.lock();
+
+ try {
+ this.nodeUtilization = nodeUtilization;
+ } finally {
+ this.writeLock.unlock();
+ }
+ }
+
+ @Override
public NodeState getState() {
this.readLock.lock();
@@ -1006,6 +1055,9 @@ public class RMNodeImpl implements RMNode, EventHandler<RMNodeEvent> {
rmNode.setHealthReport(remoteNodeHealthStatus.getHealthReport());
rmNode.setLastHealthReportTime(
remoteNodeHealthStatus.getLastHealthReportTime());
+ rmNode.setAggregatedContainersUtilization(
+ statusEvent.getAggregatedContainersUtilization());
+ rmNode.setNodeUtilization(statusEvent.getNodeUtilization());
NodeState initialState = rmNode.getState();
boolean isNodeDecommissioning =
initialState.equals(NodeState.DECOMMISSIONING);
@@ -1083,6 +1135,9 @@ public class RMNodeImpl implements RMNode, EventHandler<RMNodeEvent> {
rmNode.setHealthReport(remoteNodeHealthStatus.getHealthReport());
rmNode.setLastHealthReportTime(
remoteNodeHealthStatus.getLastHealthReportTime());
+ rmNode.setAggregatedContainersUtilization(
+ statusEvent.getAggregatedContainersUtilization());
+ rmNode.setNodeUtilization(statusEvent.getNodeUtilization());
if (remoteNodeHealthStatus.getIsNodeHealthy()) {
rmNode.context.getDispatcher().getEventHandler().handle(
new NodeAddedSchedulerEvent(rmNode));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeStatusEvent.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeStatusEvent.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeStatusEvent.java
index 0dbea1a..afc417d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeStatusEvent.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeStatusEvent.java
@@ -28,52 +28,35 @@ import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.server.api.protocolrecords.LogAggregationReport;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
+import org.apache.hadoop.yarn.server.api.records.NodeStatus;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
public class RMNodeStatusEvent extends RMNodeEvent {
- private final NodeHealthStatus nodeHealthStatus;
- private final List<ContainerStatus> containersCollection;
+ private final NodeStatus nodeStatus;
private final NodeHeartbeatResponse latestResponse;
- private final List<ApplicationId> keepAliveAppIds;
private List<LogAggregationReport> logAggregationReportsForApps;
- private final List<Container> nmReportedIncreasedContainers;
-
- // Used by tests
- public RMNodeStatusEvent(NodeId nodeId, NodeHealthStatus nodeHealthStatus,
- List<ContainerStatus> collection, List<ApplicationId> keepAliveAppIds,
- NodeHeartbeatResponse latestResponse) {
- this(nodeId, nodeHealthStatus, collection, keepAliveAppIds,
- latestResponse, null);
- }
- public RMNodeStatusEvent(NodeId nodeId, NodeHealthStatus nodeHealthStatus,
- List<ContainerStatus> collection, List<ApplicationId> keepAliveAppIds,
- NodeHeartbeatResponse latestResponse,
- List<Container> nmReportedIncreasedContainers) {
- this(nodeId, nodeHealthStatus, collection, keepAliveAppIds, latestResponse,
- null, nmReportedIncreasedContainers);
+ public RMNodeStatusEvent(NodeId nodeId, NodeStatus nodeStatus,
+ NodeHeartbeatResponse latestResponse) {
+ this(nodeId, nodeStatus, latestResponse, null);
}
- public RMNodeStatusEvent(NodeId nodeId, NodeHealthStatus nodeHealthStatus,
- List<ContainerStatus> collection, List<ApplicationId> keepAliveAppIds,
+ public RMNodeStatusEvent(NodeId nodeId, NodeStatus nodeStatus,
NodeHeartbeatResponse latestResponse,
- List<LogAggregationReport> logAggregationReportsForApps,
- List<Container> nmReportedIncreasedContainers) {
+ List<LogAggregationReport> logAggregationReportsForApps) {
super(nodeId, RMNodeEventType.STATUS_UPDATE);
- this.nodeHealthStatus = nodeHealthStatus;
- this.containersCollection = collection;
- this.keepAliveAppIds = keepAliveAppIds;
+ this.nodeStatus = nodeStatus;
this.latestResponse = latestResponse;
this.logAggregationReportsForApps = logAggregationReportsForApps;
- this.nmReportedIncreasedContainers = nmReportedIncreasedContainers;
}
public NodeHealthStatus getNodeHealthStatus() {
- return this.nodeHealthStatus;
+ return this.nodeStatus.getNodeHealthStatus();
}
public List<ContainerStatus> getContainers() {
- return this.containersCollection;
+ return this.nodeStatus.getContainersStatuses();
}
public NodeHeartbeatResponse getLatestResponse() {
@@ -81,7 +64,15 @@ public class RMNodeStatusEvent extends RMNodeEvent {
}
public List<ApplicationId> getKeepAliveAppIds() {
- return this.keepAliveAppIds;
+ return this.nodeStatus.getKeepAliveApplications();
+ }
+
+ public ResourceUtilization getAggregatedContainersUtilization() {
+ return this.nodeStatus.getContainersUtilization();
+ }
+
+ public ResourceUtilization getNodeUtilization() {
+ return this.nodeStatus.getNodeUtilization();
}
public List<LogAggregationReport> getLogAggregationReportsForApps() {
@@ -95,7 +86,7 @@ public class RMNodeStatusEvent extends RMNodeEvent {
@SuppressWarnings("unchecked")
public List<Container> getNMReportedIncreasedContainers() {
- return nmReportedIncreasedContainers == null ? Collections.EMPTY_LIST
- : nmReportedIncreasedContainers;
+ return this.nodeStatus.getIncreasedContainers() == null ?
+ Collections.EMPTY_LIST : this.nodeStatus.getIncreasedContainers();
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerNode.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerNode.java
index f3d3906..e8e1238 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerNode.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerNode.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
@@ -58,6 +59,10 @@ public abstract class SchedulerNode {
private Resource totalResourceCapability;
private RMContainer reservedContainer;
private volatile int numContainers;
+ private volatile ResourceUtilization containersUtilization =
+ ResourceUtilization.newInstance(0, 0, 0f);
+ private volatile ResourceUtilization nodeUtilization =
+ ResourceUtilization.newInstance(0, 0, 0f);
/* set of containers that are allocated containers */
@@ -339,4 +344,37 @@ public abstract class SchedulerNode {
return this.labels.iterator().next();
}
}
+
+ /**
+ * Set the resource utilization of the containers in the node.
+ * @param containersUtilization Resource utilization of the containers.
+ */
+ public void setAggregatedContainersUtilization(
+ ResourceUtilization containersUtilization) {
+ this.containersUtilization = containersUtilization;
+ }
+
+ /**
+ * Get the resource utilization of the containers in the node.
+ * @return Resource utilization of the containers.
+ */
+ public ResourceUtilization getAggregatedContainersUtilization() {
+ return this.containersUtilization;
+ }
+
+ /**
+ * Set the resource utilization of the node. This includes the containers.
+ * @param nodeUtilization Resource utilization of the node.
+ */
+ public void setNodeUtilization(ResourceUtilization nodeUtilization) {
+ this.nodeUtilization = nodeUtilization;
+ }
+
+ /**
+ * Get the resource utilization of the node.
+ * @return Resource utilization of the node.
+ */
+ public ResourceUtilization getNodeUtilization() {
+ return this.nodeUtilization;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
index e14aecf..782ed03 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
@@ -1063,6 +1063,11 @@ public class CapacityScheduler extends
releaseResources);
schedulerHealth.updateSchedulerReleaseCounts(releasedContainers);
+ // Updating node resource utilization
+ node.setAggregatedContainersUtilization(
+ nm.getAggregatedContainersUtilization());
+ node.setNodeUtilization(nm.getNodeUtilization());
+
// Now node data structures are upto date and ready for scheduling.
if(LOG.isDebugEnabled()) {
LOG.debug("Node being looked for scheduling " + nm
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
index f26e506..f1839f5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
@@ -1069,6 +1069,11 @@ public class FairScheduler extends
attemptScheduling(node);
}
+ // Updating node resource utilization
+ node.setAggregatedContainersUtilization(
+ nm.getAggregatedContainersUtilization());
+ node.setNodeUtilization(nm.getNodeUtilization());
+
long duration = getClock().getTime() - start;
fsOpDurations.addNodeUpdateDuration(duration);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
index 5999eb7..cfae3a2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
@@ -743,6 +743,10 @@ public class FifoScheduler extends
completedContainer, RMContainerEventType.FINISHED);
}
+ // Updating node resource utilization
+ node.setAggregatedContainersUtilization(
+ rmNode.getAggregatedContainersUtilization());
+ node.setNodeUtilization(rmNode.getNodeUtilization());
if (rmContext.isWorkPreservingRecoveryEnabled()
&& !rmContext.isSchedulerReadyForAllocatingContainers()) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java
index 92f3edf..96207f3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java
@@ -34,6 +34,7 @@ import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
@@ -110,11 +111,14 @@ public class MockNodes {
private long lastHealthReportTime;
private NodeState state;
private Set<String> labels;
+ private ResourceUtilization containersUtilization;
+ private ResourceUtilization nodeUtilization;
public MockRMNodeImpl(NodeId nodeId, String nodeAddr, String httpAddress,
Resource perNode, String rackName, String healthReport,
long lastHealthReportTime, int cmdPort, String hostName, NodeState state,
- Set<String> labels) {
+ Set<String> labels, ResourceUtilization containersUtilization,
+ ResourceUtilization nodeUtilization) {
this.nodeId = nodeId;
this.nodeAddr = nodeAddr;
this.httpAddress = httpAddress;
@@ -126,6 +130,8 @@ public class MockNodes {
this.hostName = hostName;
this.state = state;
this.labels = labels;
+ this.containersUtilization = containersUtilization;
+ this.nodeUtilization = nodeUtilization;
}
@Override
@@ -244,6 +250,16 @@ public class MockNodes {
public List<Container> pullNewlyIncreasedContainers() {
return Collections.emptyList();
}
+
+ @Override
+ public ResourceUtilization getAggregatedContainersUtilization() {
+ return this.containersUtilization;
+ }
+
+ @Override
+ public ResourceUtilization getNodeUtilization() {
+ return this.nodeUtilization;
+ }
};
private static RMNode buildRMNode(int rack, final Resource perNode,
@@ -254,18 +270,19 @@ public class MockNodes {
private static RMNode buildRMNode(int rack, final Resource perNode,
NodeState state, String httpAddr, Set<String> labels) {
return buildRMNode(rack, perNode, state, httpAddr, NODE_ID++, null, 123,
- labels);
+ labels, null, null);
}
private static RMNode buildRMNode(int rack, final Resource perNode,
NodeState state, String httpAddr, int hostnum, String hostName, int port) {
return buildRMNode(rack, perNode, state, httpAddr, hostnum, hostName, port,
- null);
+ null, null, null);
}
private static RMNode buildRMNode(int rack, final Resource perNode,
NodeState state, String httpAddr, int hostnum, String hostName, int port,
- Set<String> labels) {
+ Set<String> labels, ResourceUtilization containersUtilization,
+ ResourceUtilization nodeUtilization) {
final String rackName = "rack"+ rack;
final int nid = hostnum;
final String nodeAddr = hostName + ":" + nid;
@@ -277,7 +294,8 @@ public class MockNodes {
final String httpAddress = httpAddr;
String healthReport = (state == NodeState.UNHEALTHY) ? null : "HealthyMe";
return new MockRMNodeImpl(nodeID, nodeAddr, httpAddress, perNode,
- rackName, healthReport, 0, nid, hostName, state, labels);
+ rackName, healthReport, 0, nid, hostName, state, labels,
+ containersUtilization, nodeUtilization);
}
public static RMNode nodeInfo(int rack, final Resource perNode,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
index a6e1575..33a077d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
@@ -46,6 +46,7 @@ import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.event.InlineDispatcher;
import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse;
import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
+import org.apache.hadoop.yarn.server.api.records.NodeStatus;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanAppEvent;
@@ -647,8 +648,9 @@ public class TestRMNodeTransitions {
statusList.add(status);
NodeHealthStatus nodeHealth = NodeHealthStatus.newInstance(true,
"", System.currentTimeMillis());
- node.handle(new RMNodeStatusEvent(nodeId, nodeHealth,
- statusList, null, null));
+ NodeStatus nodeStatus = NodeStatus.newInstance(nodeId, 0, statusList, null,
+ nodeHealth, null, null, null);
+ node.handle(new RMNodeStatusEvent(nodeId, nodeStatus, null));
Assert.assertEquals(1, node.getRunningApps().size());
@@ -689,8 +691,9 @@ public class TestRMNodeTransitions {
RMNodeImpl node = getRunningNode();
NodeHealthStatus status = NodeHealthStatus.newInstance(false, "sick",
System.currentTimeMillis());
- node.handle(new RMNodeStatusEvent(node.getNodeID(), status,
- new ArrayList<ContainerStatus>(), null, null));
+ NodeStatus nodeStatus = NodeStatus.newInstance(node.getNodeID(), 0,
+ new ArrayList<ContainerStatus>(), null, status, null, null, null);
+ node.handle(new RMNodeStatusEvent(node.getNodeID(), nodeStatus, null));
Assert.assertEquals(NodeState.UNHEALTHY, node.getState());
return node;
}
@@ -863,8 +866,9 @@ public class TestRMNodeTransitions {
RMNodeImpl node = getDecommissioningNode();
NodeHealthStatus status = NodeHealthStatus.newInstance(false, "sick",
System.currentTimeMillis());
- node.handle(new RMNodeStatusEvent(node.getNodeID(), status,
- new ArrayList<ContainerStatus>(), null, null));
+ NodeStatus nodeStatus = NodeStatus.newInstance(node.getNodeID(), 0,
+ new ArrayList<ContainerStatus>(), null, status, null, null, null);
+ node.handle(new RMNodeStatusEvent(node.getNodeID(), nodeStatus, null));
Assert.assertEquals(NodeState.DECOMMISSIONING, node.getState());
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/logaggregationstatus/TestRMAppLogAggregationStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/logaggregationstatus/TestRMAppLogAggregationStatus.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/logaggregationstatus/TestRMAppLogAggregationStatus.java
index 1f307aa..087199d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/logaggregationstatus/TestRMAppLogAggregationStatus.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/logaggregationstatus/TestRMAppLogAggregationStatus.java
@@ -40,6 +40,7 @@ import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.event.InlineDispatcher;
import org.apache.hadoop.yarn.server.api.protocolrecords.LogAggregationReport;
import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
+import org.apache.hadoop.yarn.server.api.records.NodeStatus;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
@@ -163,9 +164,11 @@ public class TestRMAppLogAggregationStatus {
LogAggregationReport.newInstance(appId, LogAggregationStatus.RUNNING,
messageForNode1_1);
node1ReportForApp.add(report1);
- node1.handle(new RMNodeStatusEvent(node1.getNodeID(), NodeHealthStatus
- .newInstance(true, null, 0), new ArrayList<ContainerStatus>(), null,
- null, node1ReportForApp, null));
+ NodeStatus nodeStatus1 = NodeStatus.newInstance(node1.getNodeID(), 0,
+ new ArrayList<ContainerStatus>(), null,
+ NodeHealthStatus.newInstance(true, null, 0), null, null, null);
+ node1.handle(new RMNodeStatusEvent(node1.getNodeID(), nodeStatus1, null,
+ node1ReportForApp));
List<LogAggregationReport> node2ReportForApp =
new ArrayList<LogAggregationReport>();
@@ -175,9 +178,11 @@ public class TestRMAppLogAggregationStatus {
LogAggregationReport.newInstance(appId,
LogAggregationStatus.RUNNING, messageForNode2_1);
node2ReportForApp.add(report2);
- node2.handle(new RMNodeStatusEvent(node2.getNodeID(), NodeHealthStatus
- .newInstance(true, null, 0), new ArrayList<ContainerStatus>(), null,
- null, node2ReportForApp, null));
+ NodeStatus nodeStatus2 = NodeStatus.newInstance(node2.getNodeID(), 0,
+ new ArrayList<ContainerStatus>(), null,
+ NodeHealthStatus.newInstance(true, null, 0), null, null, null);
+ node2.handle(new RMNodeStatusEvent(node2.getNodeID(), nodeStatus2, null,
+ node2ReportForApp));
// node1 and node2 has updated its log aggregation status
// verify that the log aggregation status for node1, node2
// has been changed
@@ -213,9 +218,8 @@ public class TestRMAppLogAggregationStatus {
LogAggregationReport.newInstance(appId,
LogAggregationStatus.RUNNING, messageForNode1_2);
node1ReportForApp2.add(report1_2);
- node1.handle(new RMNodeStatusEvent(node1.getNodeID(), NodeHealthStatus
- .newInstance(true, null, 0), new ArrayList<ContainerStatus>(), null,
- null, node1ReportForApp2, null));
+ node1.handle(new RMNodeStatusEvent(node1.getNodeID(), nodeStatus1, null,
+ node1ReportForApp2));
// verify that the log aggregation status for node1
// has been changed
@@ -282,9 +286,8 @@ public class TestRMAppLogAggregationStatus {
LogAggregationStatus.SUCCEEDED, ""));
// For every logAggregationReport cached in memory, we can only save at most
// 10 diagnostic messages/failure messages
- node1.handle(new RMNodeStatusEvent(node1.getNodeID(), NodeHealthStatus
- .newInstance(true, null, 0), new ArrayList<ContainerStatus>(), null,
- null, node1ReportForApp3, null));
+ node1.handle(new RMNodeStatusEvent(node1.getNodeID(), nodeStatus1, null,
+ node1ReportForApp3));
logAggregationStatus = rmApp.getLogAggregationReportsForApp();
Assert.assertEquals(2, logAggregationStatus.size());
@@ -327,9 +330,8 @@ public class TestRMAppLogAggregationStatus {
LogAggregationStatus.FAILED, "");
node2ReportForApp2.add(report2_2);
node2ReportForApp2.add(report2_3);
- node2.handle(new RMNodeStatusEvent(node2.getNodeID(), NodeHealthStatus
- .newInstance(true, null, 0), new ArrayList<ContainerStatus>(), null,
- null, node2ReportForApp2, null));
+ node2.handle(new RMNodeStatusEvent(node2.getNodeID(), nodeStatus2, null,
+ node2ReportForApp2));
Assert.assertEquals(LogAggregationStatus.FAILED,
rmApp.getLogAggregationStatusForAppReport());
logAggregationStatus = rmApp.getLogAggregationReportsForApp();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodes.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodes.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodes.java
index ec20bba..206edb1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodes.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodes.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.NodeState;
import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
+import org.apache.hadoop.yarn.server.api.records.NodeStatus;
import org.apache.hadoop.yarn.server.resourcemanager.MockNM;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
@@ -142,8 +143,9 @@ public class TestRMWebServicesNodes extends JerseyTestBase {
.get(nm3.getNodeId());
NodeHealthStatus nodeHealth = NodeHealthStatus.newInstance(false,
"test health report", System.currentTimeMillis());
- node.handle(new RMNodeStatusEvent(nm3.getNodeId(), nodeHealth,
- new ArrayList<ContainerStatus>(), null, null));
+ NodeStatus nodeStatus = NodeStatus.newInstance(nm3.getNodeId(), 1,
+ new ArrayList<ContainerStatus>(), null, nodeHealth, null, null, null);
+ node.handle(new RMNodeStatusEvent(nm3.getNodeId(), nodeStatus, null));
rm.NMwaitForState(nm3.getNodeId(), NodeState.UNHEALTHY);
ClientResponse response =
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
index aa796ed..68c9efd 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
@@ -57,6 +57,7 @@ import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerRequ
import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerResponse;
import org.apache.hadoop.yarn.server.api.protocolrecords.UnRegisterNodeManagerRequest;
import org.apache.hadoop.yarn.server.api.protocolrecords.UnRegisterNodeManagerResponse;
+import org.apache.hadoop.yarn.server.api.records.NodeStatus;
import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryServer;
import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryStore;
import org.apache.hadoop.yarn.server.applicationhistoryservice.MemoryApplicationHistoryStore;
@@ -594,19 +595,66 @@ public class MiniYARNCluster extends CompositeService {
}
}
- private class CustomNodeManager extends NodeManager {
+ public class CustomNodeManager extends NodeManager {
+ protected NodeStatus nodeStatus;
+
+ public void setNodeStatus(NodeStatus status) {
+ this.nodeStatus = status;
+ }
+
+ /**
+ * Hook to allow modification/replacement of NodeStatus
+ * @param currentStatus Current status.
+ * @return New node status.
+ */
+ protected NodeStatus getSimulatedNodeStatus(NodeStatus currentStatus) {
+ if(nodeStatus == null) {
+ return currentStatus;
+ } else {
+ // Increment response ID, the RMNodeStatusEvent will not get recorded
+ // for a duplicate heartbeat
+ nodeStatus.setResponseId(nodeStatus.getResponseId() + 1);
+ return nodeStatus;
+ }
+ }
+
@Override
protected void doSecureLogin() throws IOException {
// Don't try to login using keytab in the testcase.
}
+
+ @Override
+ protected NodeStatusUpdater createNodeStatusUpdater(Context context,
+ Dispatcher dispatcher, NodeHealthCheckerService healthChecker) {
+ return new NodeStatusUpdaterImpl(context,
+ dispatcher,
+ healthChecker,
+ metrics) {
+
+ // Allow simulation of nodestatus
+ @Override
+ protected NodeStatus getNodeStatus(int responseId) throws IOException {
+ return getSimulatedNodeStatus(super.getNodeStatus(responseId));
+ }
+ };
+ }
}
private class ShortCircuitedNodeManager extends CustomNodeManager {
@Override
protected NodeStatusUpdater createNodeStatusUpdater(Context context,
Dispatcher dispatcher, NodeHealthCheckerService healthChecker) {
- return new NodeStatusUpdaterImpl(context, dispatcher,
- healthChecker, metrics) {
+ return new NodeStatusUpdaterImpl(context,
+ dispatcher,
+ healthChecker,
+ metrics) {
+
+ // Allow simulation of nodestatus
+ @Override
+ protected NodeStatus getNodeStatus(int responseId) throws IOException {
+ return getSimulatedNodeStatus(super.getNodeStatus(responseId));
+ }
+
@Override
protected ResourceTracker getRMClient() {
final ResourceTrackerService rt =
http://git-wip-us.apache.org/repos/asf/hadoop/blob/52948bb2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/TestMiniYarnClusterNodeUtilization.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/TestMiniYarnClusterNodeUtilization.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/TestMiniYarnClusterNodeUtilization.java
new file mode 100644
index 0000000..77f6492
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/TestMiniYarnClusterNodeUtilization.java
@@ -0,0 +1,245 @@
+/**
+ * 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.server;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ContainerStatus;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.server.MiniYARNCluster.CustomNodeManager;
+import org.apache.hadoop.yarn.server.api.ResourceTracker;
+import org.apache.hadoop.yarn.server.api.ServerRMProxy;
+import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatRequest;
+import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
+import org.apache.hadoop.yarn.server.api.records.NodeStatus;
+import org.apache.hadoop.yarn.server.api.records.ResourceUtilization;
+import org.apache.hadoop.yarn.server.nodemanager.NodeStatusUpdater;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestMiniYarnClusterNodeUtilization {
+ // Mini YARN cluster setup
+ private static final int NUM_RM = 1;
+ private static final int NUM_NM = 1;
+
+ // Values for the first round
+ private static final int CONTAINER_PMEM_1 = 1024;
+ private static final int CONTAINER_VMEM_1 = 2048;
+ private static final float CONTAINER_CPU_1 = 11.0f;
+
+ private static final int NODE_PMEM_1 = 10240;
+ private static final int NODE_VMEM_1 = 20480;
+ private static final float NODE_CPU_1 = 51.0f;
+
+ // Values for the second round
+ private static final int CONTAINER_PMEM_2 = 2048;
+ private static final int CONTAINER_VMEM_2 = 4096;
+ private static final float CONTAINER_CPU_2 = 22.0f;
+
+ private static final int NODE_PMEM_2 = 20480;
+ private static final int NODE_VMEM_2 = 40960;
+ private static final float NODE_CPU_2 = 61.0f;
+
+ private MiniYARNCluster cluster;
+ private CustomNodeManager nm;
+
+ private Configuration conf;
+
+ private NodeStatus nodeStatus;
+
+ @Before
+ public void setup() {
+ conf = new YarnConfiguration();
+ conf.set(YarnConfiguration.RM_WEBAPP_ADDRESS, "localhost:0");
+ conf.setInt(YarnConfiguration.RM_NM_HEARTBEAT_INTERVAL_MS, 100);
+ String name = TestMiniYarnClusterNodeUtilization.class.getName();
+ cluster = new MiniYARNCluster(name, NUM_RM, NUM_NM, 1, 1);
+ cluster.init(conf);
+ cluster.start();
+ assertFalse("RM never turned active", -1 == cluster.getActiveRMIndex());
+
+ nm = (CustomNodeManager)cluster.getNodeManager(0);
+ int responseId = 1;
+ nodeStatus = createNodeStatus(nm.getNMContext().getNodeId(), responseId,
+ CONTAINER_PMEM_1, CONTAINER_VMEM_1, CONTAINER_CPU_1,
+ NODE_PMEM_1, NODE_VMEM_1, NODE_CPU_1);
+ nm.setNodeStatus(nodeStatus);
+ }
+
+ /**
+ * Simulates a NM heartbeat using the simulated NodeStatus fixture. Verify
+ * both the RMNode and SchedulerNode have been updated with the new
+ * utilization.
+ */
+ @Test(timeout=60000)
+ public void testUpdateNodeUtilization()
+ throws InterruptedException, IOException, YarnException {
+ assertTrue("NMs fail to connect to the RM",
+ cluster.waitForNodeManagersToConnect(10000));
+
+ // Simulate heartbeat using NodeStatus fixture
+ NodeHeartbeatRequest request =
+ NodeHeartbeatRequest.newInstance(nodeStatus, null, null, null);
+ ResourceTracker tracker =
+ ServerRMProxy.createRMProxy(conf, ResourceTracker.class);
+ tracker.nodeHeartbeat(request);
+
+ // Give the heartbeat time to propagate to the RM
+ verifySimulatedUtilization();
+
+ // Alter utilization
+ int responseId = 10;
+ nodeStatus = createNodeStatus(nm.getNMContext().getNodeId(), responseId,
+ CONTAINER_PMEM_2, CONTAINER_VMEM_2, CONTAINER_CPU_2,
+ NODE_PMEM_2, NODE_VMEM_2, NODE_CPU_2);
+ nm.setNodeStatus(nodeStatus);
+ tracker.nodeHeartbeat(request);
+
+ // Give the heartbeat time to propagate to the RM
+ verifySimulatedUtilization();
+ }
+
+ /**
+ * Trigger the NM to send a heartbeat using the simulated NodeStatus fixture.
+ * Verify both the RMNode and SchedulerNode have been updated with the new
+ * utilization.
+ */
+ @Test(timeout=60000)
+ public void testMockNodeStatusHeartbeat()
+ throws InterruptedException, YarnException {
+ assertTrue("NMs fail to connect to the RM",
+ cluster.waitForNodeManagersToConnect(10000));
+
+ NodeStatusUpdater updater = nm.getNodeStatusUpdater();
+ updater.sendOutofBandHeartBeat();
+
+ // Give the heartbeat time to propagate to the RM
+ verifySimulatedUtilization();
+
+ // Alter utilization
+ int responseId = 20;
+ nodeStatus = createNodeStatus(nm.getNMContext().getNodeId(), responseId,
+ CONTAINER_PMEM_2, CONTAINER_VMEM_2, CONTAINER_CPU_2,
+ NODE_PMEM_2, NODE_VMEM_2, NODE_CPU_2);
+ nm.setNodeStatus(nodeStatus);
+ updater.sendOutofBandHeartBeat();
+
+ verifySimulatedUtilization();
+ }
+
+ /**
+ * Create a NodeStatus test vector.
+ * @param nodeId Node identifier.
+ * @param responseId Response identifier.
+ * @param containerPMem Virtual memory of the container.
+ * @param containerVMem Physical memory of the container.
+ * @param containerCPU CPU percentage of the container.
+ * @param nodePMem Physical memory of the node.
+ * @param nodeVMem Virtual memory of the node.
+ * @param nodeCPU CPU percentage of the node.
+ */
+ private NodeStatus createNodeStatus(
+ NodeId nodeId,
+ int responseId,
+ int containerPMem,
+ int containerVMem,
+ float containerCPU,
+ int nodePMem,
+ int nodeVMem,
+ float nodeCPU) {
+
+ // Fake node status with fake utilization
+ ResourceUtilization containersUtilization =
+ ResourceUtilization.newInstance(containerPMem, containerVMem,
+ containerCPU);
+ ResourceUtilization nodeUtilization =
+ ResourceUtilization.newInstance(nodePMem, nodeVMem, nodeCPU);
+ NodeStatus status = NodeStatus.newInstance(
+ nodeId,
+ responseId,
+ new ArrayList<ContainerStatus>(),
+ null,
+ NodeHealthStatus.newInstance(true, null, 0),
+ containersUtilization,
+ nodeUtilization,
+ null);
+
+ return status;
+ }
+
+ /**
+ * Verify both the RMNode and SchedulerNode have been updated with the test
+ * fixture utilization data.
+ * @param containersUtilization Utilization of the container.
+ * @param nodeUtilization Utilization of the node.
+ */
+ private void verifySimulatedUtilization() throws InterruptedException {
+ ResourceManager rm = cluster.getResourceManager(0);
+ RMContext rmContext = rm.getRMContext();
+
+ ResourceUtilization containersUtilization =
+ nodeStatus.getContainersUtilization();
+ ResourceUtilization nodeUtilization =
+ nodeStatus.getNodeUtilization();
+
+ // Give the heartbeat time to propagate to the RM (max 10 seconds)
+ // We check if the nodeUtilization is up to date
+ for (int i=0; i<100; i++) {
+ for (RMNode ni : rmContext.getRMNodes().values()) {
+ if (ni.getNodeUtilization().equals(nodeUtilization)) {
+ break;
+ }
+ }
+ Thread.sleep(100);
+ }
+
+ // Verify the data is readable from the RM and scheduler nodes
+ for (RMNode ni : rmContext.getRMNodes().values()) {
+ ResourceUtilization cu = ni.getAggregatedContainersUtilization();
+ assertEquals("Containers Utillization not propagated to RMNode",
+ containersUtilization, cu);
+
+ ResourceUtilization nu = ni.getNodeUtilization();
+ assertEquals("Node Utillization not propagated to RMNode",
+ nodeUtilization, nu);
+
+ SchedulerNode scheduler =
+ rmContext.getScheduler().getSchedulerNode(ni.getNodeID());
+ cu = scheduler.getAggregatedContainersUtilization();
+ assertEquals("Containers Utillization not propagated to SchedulerNode",
+ containersUtilization, cu);
+
+ nu = scheduler.getNodeUtilization();
+ assertEquals("Node Utillization not propagated to SchedulerNode",
+ nodeUtilization, nu);
+ }
+ }
+}