YARN-1387. RMWebServices should use ClientRMService for filtering applications (Karthik Kambatla via Sandy Ryza)

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

Author: sryza

hadoop-yarn-project/CHANGES.txt

@@ -91,6 +91,9 @@ Release 2.3.0 - UNRELEASED
     YARN-1121. Changed ResourceManager's state-store to drain all events on
     shut-down. (Jian He via vinodkv)
 
+    YARN-1387. RMWebServices should use ClientRMService for filtering
+    applications (Karthik Kambatla via Sandy Ryza)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetApplicationsRequest.java

@@ -21,6 +21,8 @@
 import java.util.EnumSet;
 import java.util.Set;
 
+import org.apache.commons.collections.buffer.UnboundedFifoBuffer;
+import org.apache.commons.lang.math.LongRange;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Stable;
@@ -150,4 +152,109 @@ public static GetApplicationsRequest newInstance(
   @Unstable
   public abstract void
       setApplicationStates(EnumSet<YarnApplicationState> applicationStates);
+
+  /**
+   * Set the application states to filter applications on
+   *
+   * @param applicationStates all lower-case string representation of the
+   *                          application states to filter on
+   */
+  @Private
+  @Unstable
+  public abstract void setApplicationStates(Set<String> applicationStates);
+
+  /**
+   * Get the users to filter applications on
+   *
+   * @return set of users to filter applications on
+   */
+  @Private
+  @Unstable
+  public abstract Set<String> getUsers();
+
+  /**
+   * Set the users to filter applications on
+   *
+   * @param users set of users to filter applications on
+   */
+  @Private
+  @Unstable
+  public abstract void setUsers(Set<String> users);
+
+  /**
+   * Get the queues to filter applications on
+   *
+   * @return set of queues to filter applications on
+   */
+  @Private
+  @Unstable
+  public abstract Set<String> getQueues();
+
+  /**
+   * Set the queue to filter applications on
+   *
+   * @param queue user to filter applications on
+   */
+  @Private
+  @Unstable
+  public abstract void setQueues(Set<String> queue);
+
+  /**
+   * Get the limit on the number applications to return
+   *
+   * @return number of applications to limit to
+   */
+  @Private
+  @Unstable
+  public abstract long getLimit();
+
+  /**
+   * Limit the number applications to return
+   *
+   * @param limit number of applications to limit to
+   */
+  @Private
+  @Unstable
+  public abstract void setLimit(long limit);
+
+  /**
+   * Get the range of start times to filter applications on
+   *
+   * @return {@link LongRange} of start times to filter applications on
+   */
+  @Private
+  @Unstable
+  public abstract LongRange getStartRange();
+
+  /**
+   * Set the range of start times to filter applications on
+   *
+   * @param begin beginning of the range
+   * @param end end of the range
+   * @throws IllegalArgumentException
+   */
+  @Private
+  @Unstable
+  public abstract void setStartRange(long begin, long end)
+      throws IllegalArgumentException;
+
+  /**
+   * Get the range of finish times to filter applications on
+   *
+   * @return {@link LongRange} of finish times to filter applications on
+   */
+  @Private
+  @Unstable
+  public abstract LongRange getFinishRange();
+
+  /**
+   * Set the range of finish times to filter applications on
+   *
+   * @param begin beginning of the range
+   * @param end end of the range
+   * @throws IllegalArgumentException
+   */
+  @Private
+  @Unstable
+  public abstract void setFinishRange(long begin, long end);
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto

@@ -125,6 +125,13 @@ message GetClusterMetricsResponseProto {
 message GetApplicationsRequestProto {
   repeated string application_types = 1;
   repeated YarnApplicationStateProto application_states = 2;
+  repeated string users = 3;
+  repeated string queues = 4;
+  optional int64 limit = 5;
+  optional int64 start_begin = 6;
+  optional int64 start_end = 7;
+  optional int64 finish_begin = 8;
+  optional int64 finish_end = 9;
 }
 
 message GetApplicationsResponseProto {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetApplicationsRequestPBImpl.java

@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.Set;
 
+import org.apache.commons.lang.math.LongRange;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
@@ -44,6 +45,10 @@ public class GetApplicationsRequestPBImpl extends GetApplicationsRequest {
 
   Set<String> applicationTypes = null;
   EnumSet<YarnApplicationState> applicationStates = null;
+  Set<String> users = null;
+  Set<String> queues = null;
+  long limit = Long.MAX_VALUE;
+  LongRange start = null, finish = null;
 
   public GetApplicationsRequestPBImpl() {
     builder = GetApplicationsRequestProto.newBuilder();
@@ -148,6 +153,26 @@ private void initApplicationStates() {
     }
   }
 
+  private void initUsers() {
+    if (this.users != null) {
+      return;
+    }
+    GetApplicationsRequestProtoOrBuilder p = viaProto ? proto : builder;
+    List<String> usersList = p.getUsersList();
+    this.users = new HashSet<String>();
+    this.users.addAll(usersList);
+  }
+
+  private void initQueues() {
+    if (this.queues != null) {
+      return;
+    }
+    GetApplicationsRequestProtoOrBuilder p = viaProto ? proto : builder;
+    List<String> queuesList = p.getQueuesList();
+    this.queues = new HashSet<String>();
+    this.queues.addAll(queuesList);
+  }
+
   @Override
   public Set<String> getApplicationTypes() {
     initApplicationTypes();
@@ -177,6 +202,111 @@ public void setApplicationStates(EnumSet<YarnApplicationState> applicationStates
     this.applicationStates = applicationStates;
   }
 
+  @Override
+  public void setApplicationStates(Set<String> applicationStates) {
+    EnumSet<YarnApplicationState> appStates = null;
+    for (YarnApplicationState state : YarnApplicationState.values()) {
+      if (applicationStates.contains(state.name().toLowerCase())) {
+        if (appStates == null) {
+          appStates = EnumSet.of(state);
+        } else {
+          appStates.add(state);
+        }
+      }
+    }
+    setApplicationStates(appStates);
+  }
+
+  @Override
+  public Set<String> getUsers() {
+    initUsers();
+    return this.users;
+  }
+
+  @Override
+  public void setUsers(Set<String> users) {
+    maybeInitBuilder();
+    if (users == null) {
+      builder.clearUsers();
+    }
+    this.users = users;
+  }
+
+  @Override
+  public Set<String> getQueues() {
+    initQueues();
+    return this.queues;
+  }
+
+  @Override
+  public void setQueues(Set<String> queues) {
+    maybeInitBuilder();
+    if (queues == null) {
+      builder.clearQueues();
+    }
+    this.queues = queues;
+  }
+
+  @Override
+  public long getLimit() {
+    if (this.limit == Long.MAX_VALUE) {
+      GetApplicationsRequestProtoOrBuilder p = viaProto ? proto : builder;
+      this.limit = p.hasLimit() ? p.getLimit() : Long.MAX_VALUE;
+    }
+    return this.limit;
+  }
+
+  @Override
+  public void setLimit(long limit) {
+    maybeInitBuilder();
+    this.limit = limit;
+  }
+
+  @Override
+  public LongRange getStartRange() {
+    if (this.start == null) {
+      GetApplicationsRequestProtoOrBuilder p = viaProto ? proto: builder;
+      if (p.hasStartBegin() || p.hasFinishBegin()) {
+        long begin = p.hasStartBegin() ? p.getStartBegin() : 0L;
+        long end = p.hasStartEnd() ? p.getStartEnd() : Long.MAX_VALUE;
+        this.start = new LongRange(begin, end);
+      }
+    }
+    return this.start;
+  }
+
+  @Override
+  public void setStartRange(long begin, long end)
+      throws IllegalArgumentException {
+    if (begin > end) {
+      throw new IllegalArgumentException("begin > end in range (begin, " +
+          "end): (" + begin + ", " + end + ")");
+    }
+    this.start = new LongRange(begin, end);
+  }
+
+  @Override
+  public LongRange getFinishRange() {
+    if (this.finish == null) {
+      GetApplicationsRequestProtoOrBuilder p = viaProto ? proto: builder;
+      if (p.hasFinishBegin() || p.hasFinishEnd()) {
+        long begin = p.hasFinishBegin() ? p.getFinishBegin() : 0L;
+        long end = p.hasFinishEnd() ? p.getFinishEnd() : Long.MAX_VALUE;
+        this.finish = new LongRange(begin, end);
+      }
+    }
+    return this.finish;
+  }
+
+  @Override
+  public void setFinishRange(long begin, long end) {
+    if (begin > end) {
+      throw new IllegalArgumentException("begin > end in range (begin, " +
+          "end): (" + begin + ", " + end + ")");
+    }
+    this.finish = new LongRange(begin, end);
+  }
+
   @Override
   public int hashCode() {
     return getProto().hashCode();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java

@@ -28,6 +28,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.commons.lang.math.LongRange;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -401,6 +402,18 @@ public GetClusterMetricsResponse getClusterMetrics(
   @Override
   public GetApplicationsResponse getApplications(
       GetApplicationsRequest request) throws YarnException {
+    return getApplications(request, true);
+  }
+
+  /**
+   * Get applications matching the {@link GetApplicationsRequest}. If
+   * caseSensitive is set to false, applicationTypes in
+   * GetApplicationRequest are expected to be in all-lowercase
+   */
+  @Private
+  public GetApplicationsResponse getApplications(
+      GetApplicationsRequest request, boolean caseSensitive)
+      throws YarnException {
     UserGroupInformation callerUGI;
     try {
       callerUGI = UserGroupInformation.getCurrentUser();
@@ -412,11 +425,23 @@ public GetApplicationsResponse getApplications(
     Set<String> applicationTypes = request.getApplicationTypes();
     EnumSet<YarnApplicationState> applicationStates =
         request.getApplicationStates();
+    Set<String> users = request.getUsers();
+    Set<String> queues = request.getQueues();
+    long limit = request.getLimit();
+    LongRange start = request.getStartRange();
+    LongRange finish = request.getFinishRange();
 
     List<ApplicationReport> reports = new ArrayList<ApplicationReport>();
+    long count = 0;
     for (RMApp application : this.rmContext.getRMApps().values()) {
+      if (++count > limit) {
+        break;
+      }
       if (applicationTypes != null && !applicationTypes.isEmpty()) {
-        if (!applicationTypes.contains(application.getApplicationType())) {
+        String appTypeToMatch = caseSensitive
+            ? application.getApplicationType()
+            : application.getApplicationType().toLowerCase();
+        if (!applicationTypes.contains(appTypeToMatch)) {
           continue;
         }
       }
@@ -427,6 +452,25 @@ public GetApplicationsResponse getApplications(
           continue;
         }
       }
+
+      if (users != null && !users.isEmpty() &&
+          !users.contains(application.getUser())) {
+        continue;
+      }
+
+      if (queues != null && !queues.isEmpty() &&
+          !queues.contains(application.getQueue())) {
+        continue;
+      }
+
+      if (start != null && !start.containsLong(application.getStartTime())) {
+        continue;
+      }
+
+      if (finish != null && !finish.containsLong(application.getFinishTime())) {
+        continue;
+      }
+
       boolean allowAccess = checkAccess(callerUGI, application.getUser(),
           ApplicationAccessType.VIEW_APP, application);
       reports.add(application.createAndGetApplicationReport(