HADOOP-4178. Make the capacity scheduler's default values configurable.

(Sreekanth Ramakrishnan via omalley)


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@700266 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -20,6 +20,9 @@ Trunk (unreleased changes)
     HADOOP-3180. Add name of missing class to WritableName.getClass 
     IOException. (Pete Wyckoff via omalley)
 
+    HADOOP-4178. Make the capacity scheduler's default values configurable.
+    (Sreekanth Ramakrishnan via omalley)
+
   OPTIMIZATIONS
 
   BUG FIXES

conf/capacity-scheduler.xml.template

@@ -47,4 +47,31 @@
     </description>
   </property>
   
+  <!-- The default configuration settings for the capacity task scheduler -->
+  <!-- The default values would be applied to all the queues which don't have -->
+  <!-- the appropriate property for the particular queue -->
+  <property>
+    <name>mapred.capacity-scheduler.default-reclaim-time-limit</name>
+    <value>300</value>
+    <description>The amount of time, in seconds, before which 
+    resources distributed to other queues will be reclaimed by default
+    in a job queue.
+    </description>
+  </property>
+  
+  <property>
+    <name>mapred.capacity-scheduler.default-supports-priority</name>
+    <value>false</value>
+    <description>If true, priorities of jobs will be taken into 
+      account in scheduling decisions by default in a job queue.
+    </description>
+  </property>
+  
+  <property>
+    <name>mapred.capacity-scheduler.default-minimum-user-limit-percent</name>
+    <value>100</value>
+    <description>The percentage of the resources limited to a particular user
+      for the job queue at any given point of time by default.
+    </description>
+  </property>
 </configuration>

src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/CapacitySchedulerConf.java

@@ -17,6 +17,8 @@
 
 package org.apache.hadoop.mapred;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 
@@ -25,7 +27,7 @@ import org.apache.hadoop.fs.Path;
  * 
  * Resource manager configuration involves setting up queues, and defining
  * various properties for the queues. These are typically read from a file 
- * called resource-manager-conf.xml that must be in the classpath of the
+ * called capacity-scheduler.xml that must be in the classpath of the
  * application. The class provides APIs to get/set and reload the 
  * configuration for the queues.
  */
@@ -33,25 +35,13 @@ class CapacitySchedulerConf {
   
   /** Default file name from which the resource manager configuration is read. */ 
   public static final String SCHEDULER_CONF_FILE = "capacity-scheduler.xml";
-
-  /** Default value for guaranteed capacity of maps (as percentage).
-   * The default value is set to 100, to represent the entire queue. 
-   */ 
-  public static final float DEFAULT_GUARANTEED_CAPACITY = 100;
-
-  /** Default value for reclaiming redistributed resources.
-   * The default value is set to <code>300</code>. 
-   */ 
-  public static final int DEFAULT_RECLAIM_TIME_LIMIT = 300;
-  
-  /** Default value for minimum resource limit per user per queue, as a 
-   * percentage.
-   * The default value is set to <code>100</code>, the idea
-   * being that the default is suitable for organizations that do not
-   * require setting up any queues.
-   */ 
-  public static final int DEFAULT_MIN_USER_LIMIT_PERCENT = 100;
-
+  
+  private int defaultReclaimTime;
+  
+  private int defaultUlimitMinimum;
+  
+  private boolean defaultSupportPriority;
+  
   private static final String QUEUE_CONF_PROPERTY_NAME_PREFIX = 
     "mapred.capacity-scheduler.queue.";
 
@@ -66,6 +56,7 @@ class CapacitySchedulerConf {
   public CapacitySchedulerConf() {
     rmConf = new Configuration(false);
     rmConf.addResource(SCHEDULER_CONF_FILE);
+    initializeDefaults();
   }
 
   /**
@@ -78,22 +69,51 @@ class CapacitySchedulerConf {
   public CapacitySchedulerConf(Path configFile) {
     rmConf = new Configuration(false);
     rmConf.addResource(configFile);
+    initializeDefaults();
+  }
+  
+  /*
+   * Method used to initialize the default values and the queue list
+   * which is used by the Capacity Scheduler.
+   */
+  private void initializeDefaults() {
+    defaultReclaimTime = rmConf.getInt(
+        "mapred.capacity-scheduler.default-reclaim-time-limit",300);
+    defaultUlimitMinimum = rmConf.getInt(
+        "mapred.capacity-scheduler.default-minimum-user-limit-percent", 100);
+    defaultSupportPriority = rmConf.getBoolean(
+        "mapred.capacity-scheduler.default-supports-priority", false);
   }
   
   /**
    * Get the guaranteed percentage of the cluster for the specified queue.
    * 
-   * This method defaults to {@link #DEFAULT_GUARANTEED_CAPACITY} if
-   * no value is specified in the configuration for this queue. If the queue
-   * name is unknown, this method throws a {@link IllegalArgumentException}
+   * This method defaults to configured default Guaranteed Capacity if
+   * no value is specified in the configuration for this queue. 
+   * If the configured capacity is negative value or greater than 100 an
+   * {@link IllegalArgumentException} is thrown.
+   * 
+   * If default Guaranteed capacity is not configured for a queue, then
+   * system allocates capacity based on what is free at the time of 
+   * capacity scheduler start
+   * 
+   * 
    * @param queue name of the queue
    * @return guaranteed percent of the cluster for the queue.
    */
   public float getGuaranteedCapacity(String queue) {
-    checkQueue(queue);
+    //Check done in order to return default GC which can be negative
+    //In case of both GC and default GC not configured.
+    //Last check is if the configuration is specified and is marked as
+    //negative we throw exception
+    String raw = rmConf.getRaw(toFullPropertyName(queue, 
+        "guaranteed-capacity"));
+    if(raw == null) {
+      return -1;
+    }
     float result = rmConf.getFloat(toFullPropertyName(queue, 
                                    "guaranteed-capacity"), 
-                                   DEFAULT_GUARANTEED_CAPACITY);
+                                   -1);
     if (result < 0.0 || result > 100.0) {
       throw new IllegalArgumentException("Illegal capacity for queue " + queue +
                                          " of " + result);
@@ -101,6 +121,17 @@ class CapacitySchedulerConf {
     return result;
   }
   
+  /**
+   * Sets the Guaranteed capacity of the given queue.
+   * 
+   * @param queue name of the queue
+   * @param gc guaranteed percent of the cluster for the queue.
+   */
+  public void setGuaranteedCapacity(String queue,float gc) {
+    rmConf.setFloat(toFullPropertyName(queue, "guaranteed-capacity"),gc);
+  }
+  
+  
   /**
    * Get the amount of time before which redistributed resources must be
    * reclaimed for the specified queue.
@@ -110,16 +141,23 @@ class CapacitySchedulerConf {
    * submitted to the first queue requires back the resources, they must
    * be reclaimed within the specified configuration time limit.
    * 
-   * This method defaults to {@link #DEFAULT_RECLAIM_TIME_LIMIT} if
-   * no value is specified in the configuration for this queue. If the queue
-   * name is unknown, this method throws a {@link IllegalArgumentException}
+   * This method defaults to configured default reclaim time limit if
+   * no value is specified in the configuration for this queue.
+   * 
+   * Throws an {@link IllegalArgumentException} when invalid value is 
+   * configured.
+   * 
    * @param queue name of the queue
    * @return reclaim time limit for this queue.
    */
   public int getReclaimTimeLimit(String queue) {
-    checkQueue(queue);
-    return rmConf.getInt(toFullPropertyName(queue, "reclaim-time-limit"), 
-                          DEFAULT_RECLAIM_TIME_LIMIT);
+    int reclaimTimeLimit = rmConf.getInt(toFullPropertyName(queue, "reclaim-time-limit"), 
+        defaultReclaimTime);
+    if(reclaimTimeLimit <= 0) {
+      throw new IllegalArgumentException("Invalid reclaim time limit : " 
+          + reclaimTimeLimit + " for queue : " + queue);
+    }
+    return reclaimTimeLimit;
   }
   
   /**
@@ -130,7 +168,6 @@ class CapacitySchedulerConf {
    * must be retained.
    */
   public void setReclaimTimeLimit(String queue, int value) {
-    checkQueue(queue);
     rmConf.setInt(toFullPropertyName(queue, "reclaim-time-limit"), value);
   }
   
@@ -139,27 +176,23 @@ class CapacitySchedulerConf {
    * 
    * If this value is false, then job priorities will be ignored in 
    * scheduling decisions. This method defaults to <code>false</code> if 
-   * the property is not configured for this queue. If the queue name is 
-   * unknown, this method throws a {@link IllegalArgumentException}
+   * the property is not configured for this queue. 
    * @param queue name of the queue
    * @return Whether this queue supports priority or not.
    */
   public boolean isPrioritySupported(String queue) {
-    checkQueue(queue);
     return rmConf.getBoolean(toFullPropertyName(queue, "supports-priority"),
-                              false);  
+        defaultSupportPriority);  
   }
   
   /**
    * Set whether priority is supported for this queue.
    * 
-   * If the queue name is unknown, this method throws a 
-   * {@link IllegalArgumentException}
+   * 
    * @param queue name of the queue
    * @param value true, if the queue must support priorities, false otherwise.
    */
   public void setPrioritySupported(String queue, boolean value) {
-    checkQueue(queue);
     rmConf.setBoolean(toFullPropertyName(queue, "supports-priority"), value);
   }
   
@@ -167,33 +200,36 @@ class CapacitySchedulerConf {
    * Get the minimum limit of resources for any user submitting jobs in 
    * this queue, in percentage.
    * 
-   * This method defaults to {@link #DEFAULT_MIN_USER_LIMIT_PERCENT} if
-   * no value is specified in the configuration for this queue. If the queue
-   * name is unknown, this method throws a {@link IllegalArgumentException}
+   * This method defaults to default user limit configured if
+   * no value is specified in the configuration for this queue.
+   * 
+   * Throws an {@link IllegalArgumentException} when invalid value is 
+   * configured.
+   * 
    * @param queue name of the queue
    * @return minimum limit of resources, in percentage, that will be 
    * available for a user.
    * 
    */
   public int getMinimumUserLimitPercent(String queue) {
-    checkQueue(queue);
-    return rmConf.getInt(toFullPropertyName(queue, 
-                          "minimum-user-limit-percent"), 
-                          DEFAULT_MIN_USER_LIMIT_PERCENT);
+    int userLimit = rmConf.getInt(toFullPropertyName(queue,
+        "minimum-user-limit-percent"), defaultUlimitMinimum);
+    if(userLimit <= 0 || userLimit > 100) {
+      throw new IllegalArgumentException("Invalid user limit : "
+          + userLimit + " for queue : " + queue);
+    }
+    return userLimit;
   }
   
   /**
    * Set the minimum limit of resources for any user submitting jobs in
    * this queue, in percentage.
    * 
-   * If the queue name is unknown, this method throws a 
-   * {@link IllegalArgumentException}
    * @param queue name of the queue
    * @param value minimum limit of resources for any user submitting jobs
    * in this queue
    */
   public void setMinimumUserLimitPercent(String queue, int value) {
-    checkQueue(queue);
     rmConf.setInt(toFullPropertyName(queue, "minimum-user-limit-percent"), 
                     value);
   }
@@ -204,19 +240,12 @@ class CapacitySchedulerConf {
    */
   public synchronized void reloadConfiguration() {
     rmConf.reloadConfiguration();
+    initializeDefaults();
   }
   
-  private synchronized void checkQueue(String queue) {
-    /*if (queues == null) {
-      queues = getQueues();
-    }
-    if (!queues.contains(queue)) {
-      throw new IllegalArgumentException("Queue " + queue + " is undefined.");
-    }*/
-  }
-
   private static final String toFullPropertyName(String queue, 
                                                   String property) {
       return QUEUE_CONF_PROPERTY_NAME_PREFIX + queue + "." + property;
   }
+  
 }

src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred/CapacityTaskScheduler.java

@@ -36,6 +36,7 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobTracker.IllegalStateException;
 import org.apache.hadoop.util.StringUtils;
 
+
 /**
  * A {@link TaskScheduler} that implements the requirements in HADOOP-3421
  * and provides a HOD-less way to share large clusters. This scheduler 
@@ -971,10 +972,15 @@ class CapacityTaskScheduler extends TaskScheduler {
     // read queue info from config file
     QueueManager queueManager = taskTrackerManager.getQueueManager();
     Set<String> queues = queueManager.getQueues();
+    Set<String> queuesWithoutConfiguredGC = new HashSet<String>();
     float totalCapacity = 0.0f;
     for (String queueName: queues) {
       float gc = rmConf.getGuaranteedCapacity(queueName); 
-      totalCapacity += gc;
+      if(gc == -1.0) {
+        queuesWithoutConfiguredGC.add(queueName);
+      }else {
+        totalCapacity += gc;
+      }
       int ulMin = rmConf.getMinimumUserLimitPercent(queueName); 
       long reclaimTimeLimit = rmConf.getReclaimTimeLimit(queueName);
       // reclaimTimeLimit is the time(in millisec) within which we need to
@@ -990,15 +996,26 @@ class CapacityTaskScheduler extends TaskScheduler {
       jobQueuesManager.createQueue(queueName, supportsPrio);
       
       SchedulingInfo schedulingInfo = new SchedulingInfo(
-          mapScheduler.getQueueSchedulingInfo(queueName),reduceScheduler.getQueueSchedulingInfo(queueName),supportsPrio);
+          mapScheduler.getQueueSchedulingInfo(queueName),
+          reduceScheduler.getQueueSchedulingInfo(queueName),supportsPrio);
       queueManager.setSchedulerInfo(queueName, schedulingInfo);
       
     }
+    float remainingQuantityToAllocate = 100 - totalCapacity;
+    float quantityToAllocate = 
+      remainingQuantityToAllocate/queuesWithoutConfiguredGC.size();
+    for(String queue: queuesWithoutConfiguredGC) {
+      QueueSchedulingInfo schedulingInfo = 
+        mapScheduler.getQueueSchedulingInfo(queue);
+      schedulingInfo.guaranteedCapacityPercent = quantityToAllocate;
+      schedulingInfo = reduceScheduler.getQueueSchedulingInfo(queue);
+      schedulingInfo.guaranteedCapacityPercent = quantityToAllocate;
+      rmConf.setGuaranteedCapacity(queue, quantityToAllocate);
+    }
     if (totalCapacity > 100.0) {
       throw new IllegalArgumentException("Sum of queue capacities over 100% at "
                                          + totalCapacity);
-    }
-    
+    }    
     // Sanity check: there should be at least one queue. 
     if (0 == mapScheduler.getNumQueues()) {
       throw new IllegalStateException("System has no queue configured");

src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java

@@ -343,6 +343,9 @@ public class TestCapacityScheduler extends TestCase {
     }*/
     
     public float getGuaranteedCapacity(String queue) {
+      if(queueMap.get(queue).gc == -1) {
+        return super.getGuaranteedCapacity(queue);
+      }
       return queueMap.get(queue).gc;
     }
     
@@ -477,6 +480,28 @@ public class TestCapacityScheduler extends TestCase {
                         getWaitingJobQueue("default").size());
   }
   
+  //Basic test to test GC allocation across the queues which have no
+  //GC configured.
+  
+  public void testGCAllocationToQueues() throws Exception {
+    String[] qs = {"default","q1","q2","q3","q4"};
+    taskTrackerManager.addQueues(qs);
+    resConf = new FakeResourceManagerConf();
+    ArrayList<FakeQueueInfo> queues = new ArrayList<FakeQueueInfo>();
+    queues.add(new FakeQueueInfo("default",25.0f,5000,true,25));
+    queues.add(new FakeQueueInfo("q1",-1.0f,5000,true,25));
+    queues.add(new FakeQueueInfo("q2",-1.0f,5000,true,25));
+    queues.add(new FakeQueueInfo("q3",-1.0f,5000,true,25));
+    queues.add(new FakeQueueInfo("q4",-1.0f,5000,true,25));
+    resConf.setFakeQueues(queues);
+    scheduler.setResourceManagerConf(resConf);
+    scheduler.start(); 
+    assertEquals(18.75f, resConf.getGuaranteedCapacity("q1"));
+    assertEquals(18.75f, resConf.getGuaranteedCapacity("q2"));
+    assertEquals(18.75f, resConf.getGuaranteedCapacity("q3"));
+    assertEquals(18.75f, resConf.getGuaranteedCapacity("q4"));
+  }
+  
   // test capacity transfer
   public void testCapacityTransfer() throws Exception {
     // set up some queues

src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacitySchedulerConf.java

@@ -172,12 +172,11 @@ public class TestCapacitySchedulerConf extends TestCase {
 
     openFile();
     startConfig();
+    writeDefaultConfiguration();
     writeQueueDetails("default", q1Props);
     writeQueueDetails("production", q2Props);
     endConfig();
-    
     testConf.reloadConfiguration();
-    
     Map<String, Map<String, String>> queueDetails 
                       = new HashMap<String, Map<String, String>>();
     queueDetails.put("default", q1Props);
@@ -185,6 +184,138 @@ public class TestCapacitySchedulerConf extends TestCase {
     checkQueueProperties(testConf, queueDetails);
   }
 
+  public void testQueueWithUserDefinedDefaultProperties() throws IOException {
+    openFile();
+    startConfig();
+    writeUserDefinedDefaultConfiguration();
+    endConfig();
+
+    Map<String, String> q1Props = setupQueueProperties(
+        new String[] { "guaranteed-capacity", 
+                       "reclaim-time-limit",
+                       "supports-priority",
+                       "minimum-user-limit-percent" }, 
+        new String[] { "-1", 
+                        "800",
+                        "true", 
+                        "50" }
+                      );
+
+    Map<String, String> q2Props = setupQueueProperties(
+        new String[] { "guaranteed-capacity", 
+                       "reclaim-time-limit",
+                       "supports-priority",
+                       "minimum-user-limit-percent" }, 
+        new String[] { "-1", 
+                        "800",
+                        "true",
+                        "50" }
+                      );
+    
+    testConf = new CapacitySchedulerConf(new Path(testConfFile));
+
+    Map<String, Map<String, String>> queueDetails
+              = new HashMap<String, Map<String,String>>();
+    
+    queueDetails.put("default", q1Props);
+    queueDetails.put("production", q2Props);
+    
+    checkQueueProperties(testConf, queueDetails);
+  }
+  
+  public void testQueueWithDefaultPropertiesOverriden() throws IOException {
+    openFile();
+    startConfig();
+    writeUserDefinedDefaultConfiguration();
+    Map<String, String> q1Props = setupQueueProperties(
+        new String[] { "guaranteed-capacity", 
+                       "reclaim-time-limit",
+                       "supports-priority",
+                       "minimum-user-limit-percent" }, 
+        new String[] { "-1", 
+                        "800",
+                        "true", 
+                        "50" }
+                      );
+
+    Map<String, String> q2Props = setupQueueProperties(
+        new String[] { "guaranteed-capacity", 
+                       "supports-priority",
+                       "minimum-user-limit-percent" }, 
+        new String[] { "40", 
+                        "true",
+                        "50" }
+                      );
+    Map<String, String> q3Props = setupQueueProperties(
+        new String[] { "guaranteed-capacity", 
+                       "reclaim-time-limit",
+                       "supports-priority",
+                       "minimum-user-limit-percent" }, 
+        new String[] { "40", 
+                       "500",
+                        "true",
+                        "50" }
+                      );
+    writeQueueDetails("production", q2Props);
+    writeQueueDetails("test", q3Props);
+    endConfig();
+    testConf = new CapacitySchedulerConf(new Path(testConfFile));
+    Map<String, Map<String, String>> queueDetails
+              = new HashMap<String, Map<String,String>>();
+    q2Props.put("reclaim-time-limit", "800");
+    queueDetails.put("default", q1Props);
+    queueDetails.put("production", q2Props);
+    queueDetails.put("test", q3Props);
+    checkQueueProperties(testConf, queueDetails);
+  }
+  
+  public void testInvalidUserLimit() throws IOException {
+    openFile();
+    startConfig();
+    Map<String, String> q1Props = setupQueueProperties(
+        new String[] { "guaranteed-capacity", 
+                       "reclaim-time-limit",
+                       "supports-priority",
+                       "minimum-user-limit-percent" }, 
+        new String[] { "-1", 
+                        "800",
+                        "true", 
+                        "-50" }
+                      );
+    writeQueueDetails("default", q1Props);
+    endConfig();
+    try {
+      testConf = new CapacitySchedulerConf(new Path(testConfFile));
+      testConf.getMinimumUserLimitPercent("default");
+      fail("Expect Invalid user limit to raise Exception");
+    }catch(IllegalArgumentException e) {
+      assertTrue(true);
+    }
+  }
+  public void testInvalidReclaimTimeLimit() throws IOException {
+    openFile();
+    startConfig();
+    Map<String, String> q1Props = setupQueueProperties(
+        new String[] { "guaranteed-capacity", 
+                       "reclaim-time-limit",
+                       "supports-priority",
+                       "minimum-user-limit-percent" }, 
+        new String[] { "-1", 
+                        "-800",
+                        "true", 
+                        "50" }
+                      );
+    writeQueueDetails("default", q1Props);
+    endConfig();
+    try {
+      testConf = new CapacitySchedulerConf(new Path(testConfFile));
+      testConf.getReclaimTimeLimit("default");
+      fail("Expect Invalid reclaim time limit to raise Exception");
+    }catch(IllegalArgumentException e) {
+      assertTrue(true);
+    }
+  }
+  
   private void checkQueueProperties(
                         CapacitySchedulerConf testConf,
                         Map<String, Map<String, String>> queueDetails) {
@@ -237,6 +368,44 @@ public class TestCapacitySchedulerConf extends TestCase {
     }
   }
   
+  
+  private void writeDefaultConfiguration() {
+    writeProperty("mapred.capacity-scheduler.default-reclaim-time-limit"
+        , "300");
+    writeProperty("mapred.capacity-scheduler.default-supports-priority"
+        , "false");
+    writeProperty("mapred.capacity-scheduler.default-minimum-user-limit-percent"
+        , "100");  
+  }
+  
+  
+  private void writeUserDefinedDefaultConfiguration() {
+    writeProperty("mapred.capacity-scheduler.default-reclaim-time-limit"
+        , "800");
+    writeProperty("mapred.capacity-scheduler.default-supports-priority"
+        , "true");
+    writeProperty("mapred.capacity-scheduler.default-minimum-user-limit-percent"
+        , "50"); 
+  }
+  
+  
+  private void writeUserDefinedDefaultConfigurationWithoutGC() {
+    writeProperty("mapred.capacity-scheduler.default-reclaim-time-limit"
+        , "800");
+    writeProperty("mapred.capacity-scheduler.default-supports-priority"
+        , "true");
+    writeProperty("mapred.capacity-scheduler.default-minimum-user-limit-percent"
+        , "50");    
+  }
+  
+  private void writeProperty(String name, String value) {
+    writer.println("<property>");
+    writer.println("<name> " + name + "</name>");
+    writer.println("<value>"+ value+"</value>");
+    writer.println("</property>");
+    
+  }
+  
   private void endConfig() {
     writer.println("</configuration>");
     writer.close();

src/core/org/apache/hadoop/conf/Configuration.java

@@ -470,7 +470,16 @@ public class Configuration implements Iterable<Map.Entry<String,String>> {
       return defaultValue;
     }
   }
-
+  /**
+   * Set the value of the <code>name</code> property to a <code>float</code>.
+   * 
+   * @param name property name.
+   * @param value property value.
+   */
+  public void setFloat(String name, float value) {
+    set(name,Float.toString(value));
+  }
+ 
   /** 
    * Get the value of the <code>name</code> property as a <code>boolean</code>.  
    * If no such property is specified, or if the specified value is not a valid