YARN-6673 Add cpu cgroup configurations for opportunistic containers. (Miklos Szegedi via Haibo Chen)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/CGroupsCpuResourceHandlerImpl.java

@@ -26,8 +26,10 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperation;
 import org.apache.hadoop.yarn.server.nodemanager.util.NodeManagerHardwareUtils;
@@ -72,6 +74,7 @@ public class CGroupsCpuResourceHandlerImpl implements CpuResourceHandler {
   static final int MIN_PERIOD_US = 1000;
   @VisibleForTesting
   static final int CPU_DEFAULT_WEIGHT = 1024; // set by kernel
+  static final int CPU_DEFAULT_WEIGHT_OPPORTUNISTIC = 2;
 
   CGroupsCpuResourceHandlerImpl(CGroupsHandler cGroupsHandler) {
     this.cGroupsHandler = cGroupsHandler;
@@ -181,16 +184,23 @@ public class CGroupsCpuResourceHandlerImpl implements CpuResourceHandler {
   @Override
   public List<PrivilegedOperation> preStart(Container container)
       throws ResourceHandlerException {
-
     String cgroupId = container.getContainerId().toString();
     Resource containerResource = container.getResource();
     cGroupsHandler.createCGroup(CPU, cgroupId);
     try {
       int containerVCores = containerResource.getVirtualCores();
-      int cpuShares = CPU_DEFAULT_WEIGHT * containerVCores;
-      cGroupsHandler
-          .updateCGroupParam(CPU, cgroupId, CGroupsHandler.CGROUP_CPU_SHARES,
-              String.valueOf(cpuShares));
+      ContainerTokenIdentifier id = container.getContainerTokenIdentifier();
+      if (id != null && id.getExecutionType() ==
+          ExecutionType.OPPORTUNISTIC) {
+        cGroupsHandler
+            .updateCGroupParam(CPU, cgroupId, CGroupsHandler.CGROUP_CPU_SHARES,
+                String.valueOf(CPU_DEFAULT_WEIGHT_OPPORTUNISTIC));
+      } else {
+        int cpuShares = CPU_DEFAULT_WEIGHT * containerVCores;
+        cGroupsHandler
+            .updateCGroupParam(CPU, cgroupId, CGroupsHandler.CGROUP_CPU_SHARES,
+                String.valueOf(cpuShares));
+      }
       if (strictResourceUsageMode) {
         if (nodeVCores != containerVCores) {
           float containerCPU =

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/TestCGroupsCpuResourceHandlerImpl.java

@@ -21,8 +21,10 @@ package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resourc
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperation;
 import org.apache.hadoop.yarn.util.ResourceCalculatorPlugin;
@@ -294,4 +296,25 @@ public class TestCGroupsCpuResourceHandlerImpl {
   public void testStrictResourceUsage() throws Exception {
     Assert.assertNull(cGroupsCpuResourceHandler.teardown());
   }
+
+  @Test
+  public void testOpportunistic() throws Exception {
+    Configuration conf = new YarnConfiguration();
+
+    cGroupsCpuResourceHandler.bootstrap(plugin, conf);
+    ContainerTokenIdentifier tokenId = mock(ContainerTokenIdentifier.class);
+    when(tokenId.getExecutionType()).thenReturn(ExecutionType.OPPORTUNISTIC);
+    Container container = mock(Container.class);
+    String id = "container_01_01";
+    ContainerId mockContainerId = mock(ContainerId.class);
+    when(mockContainerId.toString()).thenReturn(id);
+    when(container.getContainerId()).thenReturn(mockContainerId);
+    when(container.getContainerTokenIdentifier()).thenReturn(tokenId);
+    when(container.getResource()).thenReturn(Resource.newInstance(1024, 2));
+    cGroupsCpuResourceHandler.preStart(container);
+    verify(mockCGroupsHandler, times(1))
+        .updateCGroupParam(CGroupsHandler.CGroupController.CPU, id,
+            CGroupsHandler.CGROUP_CPU_SHARES, "2");
+  }
+
 }