YARN-11097. Support refresh load limit on ResourceManager

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/ResourceManagerAdministrationProtocol.java

@@ -38,6 +38,8 @@
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshAdminAclsResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshClusterMaxPriorityRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshClusterMaxPriorityResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesResourcesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesResourcesResponse;
@@ -70,6 +72,11 @@ public RefreshQueuesResponse refreshQueues(RefreshQueuesRequest request)
   public RefreshNodesResponse refreshNodes(RefreshNodesRequest request)
   throws StandbyException, YarnException, IOException;
 
+  @Private
+  @Idempotent
+  public RefreshLoadConfResponse refreshLoadConf(RefreshLoadConfRequest request)
+  throws StandbyException, YarnException, IOException;
+
   @Private
   @Idempotent
   public RefreshSuperUserGroupsConfigurationResponse 

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

@@ -0,0 +1,34 @@
+/**
+* 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.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.util.Records;
+
+@Private
+@Unstable
+public abstract class RefreshLoadConfRequest {
+  @Private
+  @Unstable
+  public static RefreshLoadConfRequest newInstance() {
+    RefreshLoadConfRequest request = Records.newRecord(RefreshLoadConfRequest.class);
+    return request;
+  }
+}

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

@@ -0,0 +1,35 @@
+/**
+* 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.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.util.Records;
+
+@Private
+@Unstable
+public abstract class RefreshLoadConfResponse {
+  @Private
+  @Unstable
+  public static RefreshLoadConfResponse newInstance() {
+    RefreshLoadConfResponse response =
+        Records.newRecord(RefreshLoadConfResponse.class);
+    return response;
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/server/resourcemanager_administration_protocol.proto

@@ -33,6 +33,7 @@ import "server/yarn_server_resourcemanager_service_protos.proto";
 
 service ResourceManagerAdministrationProtocolService {
   rpc refreshQueues(RefreshQueuesRequestProto) returns (RefreshQueuesResponseProto);
+  rpc refreshLoadConf(RefreshLoadConfRequestProto) returns (RefreshLoadConfResponseProto);
   rpc refreshNodes(RefreshNodesRequestProto) returns (RefreshNodesResponseProto);
   rpc refreshSuperUserGroupsConfiguration(RefreshSuperUserGroupsConfigurationRequestProto) returns (RefreshSuperUserGroupsConfigurationResponseProto);
   rpc refreshUserToGroupsMappings(RefreshUserToGroupsMappingsRequestProto) returns (RefreshUserToGroupsMappingsResponseProto);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/server/yarn_server_resourcemanager_service_protos.proto

@@ -43,6 +43,11 @@ message RefreshNodesRequestProto {
 message RefreshNodesResponseProto {
 }
 
+message RefreshLoadConfRequestProto {
+}
+message RefreshLoadConfResponseProto {
+}
+
 message RefreshSuperUserGroupsConfigurationRequestProto {
 }
 message RefreshSuperUserGroupsConfigurationResponseProto {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RMAdminCLI.java

@@ -67,6 +67,7 @@
 import org.apache.hadoop.yarn.server.api.protocolrecords.CheckForDecommissioningNodesResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshAdminAclsRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshClusterMaxPriorityRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesResourcesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshQueuesRequest;
@@ -120,6 +121,8 @@ public class RMAdminCLI extends HAAdmin {
               + " will immediately decommission if an RM HA failover occurs."))
           .put("-refreshNodesResources", new UsageInfo("",
               "Refresh resources of NodeManagers at the ResourceManager."))
+          .put("-refreshLoadLimitConfiguration", new UsageInfo("",
+              "Refresh the NodeManager load limit information at the ResourceManager."))
           .put("-refreshSuperUserGroupsConfiguration", new UsageInfo("",
               "Refresh superuser proxy groups mappings"))
           .put("-refreshUserToGroupsMappings", new UsageInfo("",
@@ -271,6 +274,7 @@ private static void printHelp(String cmd, boolean isHAEnabled) {
         + " [-refreshQueues]"
         + " [-refreshNodes [-g|graceful [timeout in seconds] -client|server]]"
         + " [-refreshNodesResources]"
+        + " [-refreshLoadLimitConfiguration]"
         + " [-refreshSuperUserGroupsConfiguration]"
         + " [-refreshUserToGroupsMappings]"
         + " [-refreshAdminAcls]"
@@ -347,6 +351,15 @@ private int refreshQueues() throws IOException, YarnException {
     return 0;
   }
 
+  private int refreshLoadLimitConfiguration() throws IOException, YarnException {
+    // Refresh the queue properties
+    ResourceManagerAdministrationProtocol adminProtocol = createAdminProtocol();
+    RefreshLoadConfRequest request =
+        recordFactory.newRecordInstance(RefreshLoadConfRequest.class);
+    adminProtocol.refreshLoadConf(request);
+    return 0;
+  }
+
   private int refreshNodes(boolean graceful) throws IOException, YarnException {
     // Refresh the nodes
     ResourceManagerAdministrationProtocol adminProtocol = createAdminProtocol();
@@ -753,6 +766,8 @@ public int run(String[] args) throws Exception {
     try {
       if ("-refreshQueues".equals(cmd)) {
         exitCode = refreshQueues();
+      } else if ("-refreshLoadLimitConfiguration".equals(cmd)) {
+        exitCode = refreshLoadLimitConfiguration();
       } else if ("-refreshNodes".equals(cmd)) {
         exitCode = handleRefreshNodes(args, cmd, isHAEnabled);
       } else if ("-refreshNodesResources".equals(cmd)) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/ResourceManagerAdministrationProtocolPBClientImpl.java

@@ -36,6 +36,7 @@
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.NodesToAttributesMappingRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshAdminAclsRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshClusterMaxPriorityRequestProto;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshLoadConfRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshNodesRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshNodesResourcesRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshQueuesRequestProto;
@@ -57,6 +58,8 @@
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshAdminAclsResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshClusterMaxPriorityRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshClusterMaxPriorityResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesResourcesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesResourcesResponse;
@@ -85,6 +88,8 @@
 import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshAdminAclsResponsePBImpl;
 import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshClusterMaxPriorityRequestPBImpl;
 import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshClusterMaxPriorityResponsePBImpl;
+import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshLoadConfRequestPBImpl;
+import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshLoadConfResponsePBImpl;
 import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshNodesRequestPBImpl;
 import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshNodesResourcesRequestPBImpl;
 import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.RefreshNodesResourcesResponsePBImpl;
@@ -140,6 +145,20 @@ public RefreshQueuesResponse refreshQueues(RefreshQueuesRequest request)
     }
   }
 
+  @Override
+  public RefreshLoadConfResponse refreshLoadConf(RefreshLoadConfRequest request)
+      throws YarnException, IOException {
+    RefreshLoadConfRequestProto requestProto =
+        ((RefreshLoadConfRequestPBImpl)request).getProto();
+    try {
+      return new RefreshLoadConfResponsePBImpl(
+          proxy.refreshLoadConf(null, requestProto));
+    } catch (ServiceException e) {
+      RPCUtil.unwrapAndThrowException(e);
+      return null;
+    }
+  }
+
   @Override
   public RefreshNodesResponse refreshNodes(RefreshNodesRequest request)
   throws YarnException, IOException {

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

@@ -0,0 +1,69 @@
+/**
+* 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.api.protocolrecords.impl.pb;
+
+import com.google.protobuf.TextFormat;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshLoadConfRequestProto;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfRequest;
+
+@Private
+@Unstable
+public class RefreshLoadConfRequestPBImpl extends RefreshLoadConfRequest {
+
+  RefreshLoadConfRequestProto proto = RefreshLoadConfRequestProto.getDefaultInstance();
+  RefreshLoadConfRequestProto.Builder builder = null;
+  boolean viaProto = false;
+
+  public RefreshLoadConfRequestPBImpl() {
+    builder = RefreshLoadConfRequestProto.newBuilder();
+  }
+
+  public RefreshLoadConfRequestPBImpl(RefreshLoadConfRequestProto proto) {
+    this.proto = proto;
+    viaProto = true;
+  }
+
+  public RefreshLoadConfRequestProto getProto() {
+    proto = viaProto ? proto : builder.build();
+    viaProto = true;
+    return proto;
+  }
+
+  @Override
+  public int hashCode() {
+    return getProto().hashCode();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (other == null)
+      return false;
+    if (other.getClass().isAssignableFrom(this.getClass())) {
+      return this.getProto().equals(this.getClass().cast(other).getProto());
+    }
+    return false;
+  }
+
+  @Override
+  public String toString() {
+    return TextFormat.shortDebugString(getProto());
+  }
+}

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

@@ -0,0 +1,69 @@
+/**
+* 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.api.protocolrecords.impl.pb;
+
+import com.google.protobuf.TextFormat;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerServiceProtos.RefreshLoadConfResponseProto;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfResponse;
+
+@Private
+@Unstable
+public class RefreshLoadConfResponsePBImpl extends RefreshLoadConfResponse {
+
+  RefreshLoadConfResponseProto proto = RefreshLoadConfResponseProto.getDefaultInstance();
+  RefreshLoadConfResponseProto.Builder builder = null;
+  boolean viaProto = false;
+
+  public RefreshLoadConfResponsePBImpl() {
+    builder = RefreshLoadConfResponseProto.newBuilder();
+  }
+
+  public RefreshLoadConfResponsePBImpl(RefreshLoadConfResponseProto proto) {
+    this.proto = proto;
+    viaProto = true;
+  }
+
+  public RefreshLoadConfResponseProto getProto() {
+    proto = viaProto ? proto : builder.build();
+    viaProto = true;
+    return proto;
+  }
+
+  @Override
+  public int hashCode() {
+    return getProto().hashCode();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (other == null)
+      return false;
+    if (other.getClass().isAssignableFrom(this.getClass())) {
+      return this.getProto().equals(this.getClass().cast(other).getProto());
+    }
+    return false;
+  }
+
+  @Override
+  public String toString() {
+    return TextFormat.shortDebugString(getProto());
+  }
+}

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

@@ -77,6 +77,8 @@
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshAdminAclsResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshClusterMaxPriorityRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshClusterMaxPriorityResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshLoadConfResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesResourcesRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesResourcesResponse;
@@ -100,6 +102,7 @@
 import org.apache.hadoop.yarn.server.resourcemanager.resource.DynamicResourceConfiguration;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeResourceUpdateEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.security.authorize.RMPolicyProvider;
@@ -450,6 +453,33 @@ private boolean isSchedulerMutable() {
         && ((MutableConfScheduler) scheduler).isConfigurationMutable());
   }
 
+  @Override
+  public RefreshLoadConfResponse refreshLoadConf(RefreshLoadConfRequest request)
+      throws IOException, YarnException {
+    String argName = "refreshLoadConf";
+    final String msg = "refresh load configuration.";
+    UserGroupInformation user = checkAcls("refreshLoadConf");
+
+    checkRMStatus(user.getShortUserName(), argName, msg);
+
+    try {
+      if (rm.getRMContext().getScheduler() instanceof FairScheduler) {
+        Configuration conf = getConfiguration(new Configuration(false),
+            YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);
+        FairScheduler scheduler = (FairScheduler) rm.getRMContext().getScheduler();
+        scheduler.refreshLoadConf(conf);
+        RMAuditLogger.logSuccess(user.getShortUserName(), argName,
+            "AdminService");
+      } else {
+        LOG.warn("Scheduler " + rm.getRMContext().getScheduler().getClass()
+            + " is not supported for load conf reload.");
+      }
+      return recordFactory.newRecordInstance(RefreshLoadConfResponse.class);
+    } catch (IOException ioe) {
+      throw logAndWrapException(ioe, user.getShortUserName(), argName, msg);
+    }
+  }
+
   @Override
   public RefreshNodesResponse refreshNodes(RefreshNodesRequest request)
       throws YarnException, StandbyException {