Merge branch 'YARN-5734' into trunk. (xgong)

This commit is contained in:
Xuan 2017-10-09 14:46:02 -07:00
commit f9ff17f510
45 changed files with 4338 additions and 48 deletions

View File

@ -47,6 +47,7 @@ function hadoop_usage
hadoop_add_subcommand "resourcemanager" daemon "run the ResourceManager"
hadoop_add_subcommand "rmadmin" admin "admin tools"
hadoop_add_subcommand "router" daemon "run the Router daemon"
hadoop_add_subcommand "schedulerconf" client "Updates scheduler configuration"
hadoop_add_subcommand "scmadmin" admin "SharedCacheManager admin tools"
hadoop_add_subcommand "sharedcachemanager" daemon "run the SharedCacheManager daemon"
hadoop_add_subcommand "timelinereader" client "run the timeline reader server"
@ -142,6 +143,9 @@ function yarncmd_case
HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
HADOOP_CLASSNAME='org.apache.hadoop.yarn.server.router.Router'
;;
schedulerconf)
HADOOP_CLASSNAME='org.apache.hadoop.yarn.client.cli.SchedConfCLI'
;;
scmadmin)
HADOOP_CLASSNAME='org.apache.hadoop.yarn.client.SCMAdmin'
;;

View File

@ -295,6 +295,11 @@ goto :eof
set YARN_OPTS=%YARN_OPTS% %YARN_CLIENT_OPTS%
goto :eof
:schedulerconf
set CLASS=org.apache.hadoop.yarn.client.cli.SchedConfCLI
set YARN_OPTS=%YARN_OPTS% %YARN_CLIENT_OPTS%
goto :eof
@rem This changes %1, %2 etc. Hence those cannot be used after calling this.
:make_command_arguments
if "%1" == "--config" (
@ -340,6 +345,7 @@ goto :eof
@echo node prints node report(s)
@echo queue prints queue information
@echo logs dump container logs
@echo schedulerconf updates scheduler configuration
@echo classpath prints the class path needed to get the
@echo Hadoop jar and the required libraries
@echo daemonlog get/set the log level for each daemon

View File

@ -674,6 +674,67 @@ public class YarnConfiguration extends Configuration {
public static final String DEFAULT_RM_CONFIGURATION_PROVIDER_CLASS =
"org.apache.hadoop.yarn.LocalConfigurationProvider";
@Private
@Unstable
public static final String SCHEDULER_CONFIGURATION_STORE_CLASS =
YARN_PREFIX + "scheduler.configuration.store.class";
@Private
@Unstable
public static final String FILE_CONFIGURATION_STORE = "file";
@Private
@Unstable
public static final String MEMORY_CONFIGURATION_STORE = "memory";
@Private
@Unstable
public static final String LEVELDB_CONFIGURATION_STORE = "leveldb";
@Private
@Unstable
public static final String ZK_CONFIGURATION_STORE = "zk";
@Private
@Unstable
public static final String DEFAULT_CONFIGURATION_STORE =
FILE_CONFIGURATION_STORE;
@Private
@Unstable
public static final String RM_SCHEDCONF_STORE_PATH = YARN_PREFIX
+ "scheduler.configuration.leveldb-store.path";
@Private
@Unstable
public static final String RM_SCHEDCONF_LEVELDB_COMPACTION_INTERVAL_SECS =
YARN_PREFIX
+ "scheduler.configuration.leveldb-store.compaction-interval-secs";
@Private
@Unstable
public static final long
DEFAULT_RM_SCHEDCONF_LEVELDB_COMPACTION_INTERVAL_SECS = 60 * 60 * 24L;
@Private
@Unstable
public static final String RM_SCHEDCONF_MAX_LOGS =
YARN_PREFIX + "scheduler.configuration.store.max-logs";
@Private
@Unstable
public static final long DEFAULT_RM_SCHEDCONF_LEVELDB_MAX_LOGS = 1000;
@Private
@Unstable
public static final long DEFAULT_RM_SCHEDCONF_ZK_MAX_LOGS = 1000;
/** Parent znode path under which ZKConfigurationStore will create znodes. */
@Private
@Unstable
public static final String RM_SCHEDCONF_STORE_ZK_PARENT_PATH = YARN_PREFIX
+ "scheduler.configuration.zk-store.parent-path";
@Private
@Unstable
public static final String DEFAULT_RM_SCHEDCONF_STORE_ZK_PARENT_PATH =
"/confstore";
@Private
@Unstable
public static final String RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS =
YARN_PREFIX + "scheduler.configuration.mutation.acl-policy.class";
public static final String YARN_AUTHORIZATION_PROVIDER = YARN_PREFIX
+ "authorization-provider";
private static final List<String> RM_SERVICES_ADDRESS_CONF_KEYS_HTTP =

View File

@ -0,0 +1,254 @@
/**
* 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.client.cli;
import com.google.common.annotations.VisibleForTesting;
import com.sun.jersey.api.client.Client;
import com.sun.jersey.api.client.ClientResponse;
import com.sun.jersey.api.client.WebResource;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.MissingArgumentException;
import org.apache.commons.cli.Options;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
import org.apache.hadoop.yarn.webapp.util.YarnWebServiceUtils;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response.Status;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* CLI for modifying scheduler configuration.
*/
@Public
@Unstable
public class SchedConfCLI extends Configured implements Tool {
private static final String ADD_QUEUES_OPTION = "addQueues";
private static final String REMOVE_QUEUES_OPTION = "removeQueues";
private static final String UPDATE_QUEUES_OPTION = "updateQueues";
private static final String GLOBAL_OPTIONS = "globalUpdates";
private static final String HELP_CMD = "help";
private static final String CONF_ERR_MSG = "Specify configuration key " +
"value as confKey=confVal.";
public SchedConfCLI() {
super(new YarnConfiguration());
}
public static void main(String[] args) throws Exception {
SchedConfCLI cli = new SchedConfCLI();
int exitCode = cli.run(args);
System.exit(exitCode);
}
@Override
public int run(String[] args) throws Exception {
Options opts = new Options();
opts.addOption("add", ADD_QUEUES_OPTION, true,
"Add queues with configurations");
opts.addOption("remove", REMOVE_QUEUES_OPTION, true,
"Remove queues");
opts.addOption("update", UPDATE_QUEUES_OPTION, true,
"Update queue configurations");
opts.addOption("global", GLOBAL_OPTIONS, true,
"Update global scheduler configurations");
opts.addOption("h", HELP_CMD, false, "Displays help for all commands.");
int exitCode = -1;
CommandLine parsedCli = null;
try {
parsedCli = new GnuParser().parse(opts, args);
} catch (MissingArgumentException ex) {
System.err.println("Missing argument for options");
printUsage();
return exitCode;
}
if (parsedCli.hasOption(HELP_CMD)) {
printUsage();
return 0;
}
boolean hasOption = false;
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
try {
if (parsedCli.hasOption(ADD_QUEUES_OPTION)) {
hasOption = true;
addQueues(parsedCli.getOptionValue(ADD_QUEUES_OPTION), updateInfo);
}
if (parsedCli.hasOption(REMOVE_QUEUES_OPTION)) {
hasOption = true;
removeQueues(parsedCli.getOptionValue(REMOVE_QUEUES_OPTION),
updateInfo);
}
if (parsedCli.hasOption(UPDATE_QUEUES_OPTION)) {
hasOption = true;
updateQueues(parsedCli.getOptionValue(UPDATE_QUEUES_OPTION),
updateInfo);
}
if (parsedCli.hasOption(GLOBAL_OPTIONS)) {
hasOption = true;
globalUpdates(parsedCli.getOptionValue(GLOBAL_OPTIONS), updateInfo);
}
} catch (IllegalArgumentException e) {
System.err.println(e.getMessage());
return -1;
}
if (!hasOption) {
System.err.println("Invalid Command Usage: ");
printUsage();
return -1;
}
Client webServiceClient = Client.create();
WebResource webResource = webServiceClient.resource(WebAppUtils.
getRMWebAppURLWithScheme(getConf()));
ClientResponse response = webResource.path("ws").path("v1").path("cluster")
.path("scheduler-conf").accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
if (response != null) {
if (response.getStatus() == Status.OK.getStatusCode()) {
System.out.println("Configuration changed successfully.");
return 0;
} else {
System.err.println("Configuration change unsuccessful: "
+ response.getEntity(String.class));
}
} else {
System.err.println("Configuration change unsuccessful: null response");
}
return -1;
}
@VisibleForTesting
void addQueues(String args, SchedConfUpdateInfo updateInfo) {
if (args == null) {
return;
}
ArrayList<QueueConfigInfo> queueConfigInfos = new ArrayList<>();
for (String arg : args.split(";")) {
queueConfigInfos.add(getQueueConfigInfo(arg));
}
updateInfo.setAddQueueInfo(queueConfigInfos);
}
@VisibleForTesting
void removeQueues(String args, SchedConfUpdateInfo updateInfo) {
if (args == null) {
return;
}
List<String> queuesToRemove = Arrays.asList(args.split(";"));
updateInfo.setRemoveQueueInfo(new ArrayList<>(queuesToRemove));
}
@VisibleForTesting
void updateQueues(String args, SchedConfUpdateInfo updateInfo) {
if (args == null) {
return;
}
ArrayList<QueueConfigInfo> queueConfigInfos = new ArrayList<>();
for (String arg : args.split(";")) {
queueConfigInfos.add(getQueueConfigInfo(arg));
}
updateInfo.setUpdateQueueInfo(queueConfigInfos);
}
@VisibleForTesting
void globalUpdates(String args, SchedConfUpdateInfo updateInfo) {
if (args == null) {
return;
}
HashMap<String, String> globalUpdates = new HashMap<>();
for (String globalUpdate : args.split(",")) {
putKeyValuePair(globalUpdates, globalUpdate);
}
updateInfo.setGlobalParams(globalUpdates);
}
private QueueConfigInfo getQueueConfigInfo(String arg) {
String[] args = arg.split(":");
String queuePath = args[0];
Map<String, String> queueConfigs = new HashMap<>();
if (args.length > 1) {
String[] queueArgs = args[1].split(",");
for (int i = 0; i < queueArgs.length; ++i) {
putKeyValuePair(queueConfigs, queueArgs[i]);
}
}
return new QueueConfigInfo(queuePath, queueConfigs);
}
private void putKeyValuePair(Map<String, String> kv, String args) {
String[] argParts = args.split("=");
if (argParts.length == 1) {
if (argParts[0].isEmpty() || !args.contains("=")) {
throw new IllegalArgumentException(CONF_ERR_MSG);
} else {
// key specified, but no value e.g. "confKey="
kv.put(argParts[0], null);
}
} else if (argParts.length > 2) {
throw new IllegalArgumentException(CONF_ERR_MSG);
} else {
if (argParts[0].isEmpty()) {
throw new IllegalArgumentException(CONF_ERR_MSG);
}
kv.put(argParts[0], argParts[1]);
}
}
private void printUsage() {
System.out.println("yarn schedulerconf [-add "
+ "\"queueAddPath1:confKey1=confVal1,confKey2=confVal2;"
+ "queueAddPath2:confKey3=confVal3\"] "
+ "[-remove \"queueRemovePath1;queueRemovePath2\"] "
+ "[-update \"queueUpdatePath1:confKey1=confVal1\"] "
+ "[-global globalConfKey1=globalConfVal1,"
+ "globalConfKey2=globalConfVal2]\n"
+ "Example (adding queues): yarn schedulerconf -add "
+ "\"root.a.a1:capacity=100,maximum-capacity=100;root.a.a2:capacity=0,"
+ "maximum-capacity=0\"\n"
+ "Example (removing queues): yarn schedulerconf -remove \"root.a.a1;"
+ "root.a.a2\"\n"
+ "Example (updating queues): yarn schedulerconf -update \"root.a.a1"
+ ":capacity=25,maximum-capacity=25;root.a.a2:capacity=75,"
+ "maximum-capacity=75\"\n"
+ "Example (global scheduler update): yarn schedulerconf "
+ "-global yarn.scheduler.capacity.maximum-applications=10000\n"
+ "Note: This is an alpha feature, the syntax/options are subject to "
+ "change, please run at your own risk.");
}
}

View File

@ -0,0 +1,158 @@
/**
* 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.client.cli;
import org.junit.Before;
import org.junit.Test;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* Class for testing {@link SchedConfCLI}.
*/
public class TestSchedConfCLI {
private ByteArrayOutputStream sysOutStream;
private PrintStream sysOut;
private ByteArrayOutputStream sysErrStream;
private PrintStream sysErr;
private SchedConfCLI cli;
@Before
public void setUp() {
sysOutStream = new ByteArrayOutputStream();
sysOut = new PrintStream(sysOutStream);
System.setOut(sysOut);
sysErrStream = new ByteArrayOutputStream();
sysErr = new PrintStream(sysErrStream);
System.setErr(sysErr);
cli = new SchedConfCLI();
}
@Test(timeout = 10000)
public void testInvalidConf() throws Exception {
// conf pair with no key should be invalid
int exitCode = cli.run(new String[] {"-add", "root.a:=confVal"});
assertTrue("Should return an error code", exitCode != 0);
assertTrue(sysErrStream.toString().contains("Specify configuration key " +
"value as confKey=confVal."));
exitCode = cli.run(new String[] {"-update", "root.a:=confVal"});
assertTrue("Should return an error code", exitCode != 0);
assertTrue(sysErrStream.toString().contains("Specify configuration key " +
"value as confKey=confVal."));
exitCode = cli.run(new String[] {"-add", "root.a:confKey=confVal=conf"});
assertTrue("Should return an error code", exitCode != 0);
assertTrue(sysErrStream.toString().contains("Specify configuration key " +
"value as confKey=confVal."));
exitCode = cli.run(new String[] {"-update", "root.a:confKey=confVal=c"});
assertTrue("Should return an error code", exitCode != 0);
assertTrue(sysErrStream.toString().contains("Specify configuration key " +
"value as confKey=confVal."));
}
@Test(timeout = 10000)
public void testAddQueues() {
SchedConfUpdateInfo schedUpdateInfo = new SchedConfUpdateInfo();
cli.addQueues("root.a:a1=aVal1,a2=aVal2,a3=", schedUpdateInfo);
QueueConfigInfo addInfo = schedUpdateInfo.getAddQueueInfo().get(0);
assertEquals("root.a", addInfo.getQueue());
Map<String, String> params = addInfo.getParams();
assertEquals(3, params.size());
assertEquals("aVal1", params.get("a1"));
assertEquals("aVal2", params.get("a2"));
assertNull(params.get("a3"));
schedUpdateInfo = new SchedConfUpdateInfo();
cli.addQueues("root.b:b1=bVal1;root.c:c1=cVal1", schedUpdateInfo);
assertEquals(2, schedUpdateInfo.getAddQueueInfo().size());
QueueConfigInfo bAddInfo = schedUpdateInfo.getAddQueueInfo().get(0);
assertEquals("root.b", bAddInfo.getQueue());
Map<String, String> bParams = bAddInfo.getParams();
assertEquals(1, bParams.size());
assertEquals("bVal1", bParams.get("b1"));
QueueConfigInfo cAddInfo = schedUpdateInfo.getAddQueueInfo().get(1);
assertEquals("root.c", cAddInfo.getQueue());
Map<String, String> cParams = cAddInfo.getParams();
assertEquals(1, cParams.size());
assertEquals("cVal1", cParams.get("c1"));
}
@Test(timeout = 10000)
public void testRemoveQueues() {
SchedConfUpdateInfo schedUpdateInfo = new SchedConfUpdateInfo();
cli.removeQueues("root.a;root.b;root.c.c1", schedUpdateInfo);
List<String> removeInfo = schedUpdateInfo.getRemoveQueueInfo();
assertEquals(3, removeInfo.size());
assertEquals("root.a", removeInfo.get(0));
assertEquals("root.b", removeInfo.get(1));
assertEquals("root.c.c1", removeInfo.get(2));
}
@Test(timeout = 10000)
public void testUpdateQueues() {
SchedConfUpdateInfo schedUpdateInfo = new SchedConfUpdateInfo();
cli.updateQueues("root.a:a1=aVal1,a2=aVal2,a3=", schedUpdateInfo);
QueueConfigInfo updateInfo = schedUpdateInfo.getUpdateQueueInfo().get(0);
assertEquals("root.a", updateInfo.getQueue());
Map<String, String> params = updateInfo.getParams();
assertEquals(3, params.size());
assertEquals("aVal1", params.get("a1"));
assertEquals("aVal2", params.get("a2"));
assertNull(params.get("a3"));
schedUpdateInfo = new SchedConfUpdateInfo();
cli.updateQueues("root.b:b1=bVal1;root.c:c1=cVal1", schedUpdateInfo);
assertEquals(2, schedUpdateInfo.getUpdateQueueInfo().size());
QueueConfigInfo bUpdateInfo = schedUpdateInfo.getUpdateQueueInfo().get(0);
assertEquals("root.b", bUpdateInfo.getQueue());
Map<String, String> bParams = bUpdateInfo.getParams();
assertEquals(1, bParams.size());
assertEquals("bVal1", bParams.get("b1"));
QueueConfigInfo cUpdateInfo = schedUpdateInfo.getUpdateQueueInfo().get(1);
assertEquals("root.c", cUpdateInfo.getQueue());
Map<String, String> cParams = cUpdateInfo.getParams();
assertEquals(1, cParams.size());
assertEquals("cVal1", cParams.get("c1"));
}
@Test(timeout = 10000)
public void testGlobalUpdate() {
SchedConfUpdateInfo schedUpdateInfo = new SchedConfUpdateInfo();
cli.globalUpdates("schedKey1=schedVal1,schedKey2=schedVal2",
schedUpdateInfo);
Map<String, String> globalInfo = schedUpdateInfo.getGlobalParams();
assertEquals(2, globalInfo.size());
assertEquals("schedVal1", globalInfo.get("schedKey1"));
assertEquals("schedVal2", globalInfo.get("schedKey2"));
}
}

View File

@ -0,0 +1,57 @@
/**
* 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.webapp.dao;
import java.util.HashMap;
import java.util.Map;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlRootElement;
/**
* Information for adding or updating a queue to scheduler configuration
* for this queue.
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
public class QueueConfigInfo {
@XmlElement(name = "queue-name")
private String queue;
private HashMap<String, String> params = new HashMap<>();
public QueueConfigInfo() { }
public QueueConfigInfo(String queue, Map<String, String> params) {
this.queue = queue;
this.params = new HashMap<>(params);
}
public String getQueue() {
return this.queue;
}
public HashMap<String, String> getParams() {
return this.params;
}
}

View File

@ -0,0 +1,85 @@
/**
* 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.webapp.dao;
import java.util.ArrayList;
import java.util.HashMap;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlElementWrapper;
import javax.xml.bind.annotation.XmlRootElement;
/**
* Information for making scheduler configuration changes (supports adding,
* removing, or updating a queue, as well as global scheduler conf changes).
*/
@XmlRootElement(name = "sched-conf")
@XmlAccessorType(XmlAccessType.FIELD)
public class SchedConfUpdateInfo {
@XmlElement(name = "add-queue")
private ArrayList<QueueConfigInfo> addQueueInfo = new ArrayList<>();
@XmlElement(name = "remove-queue")
private ArrayList<String> removeQueueInfo = new ArrayList<>();
@XmlElement(name = "update-queue")
private ArrayList<QueueConfigInfo> updateQueueInfo = new ArrayList<>();
private HashMap<String, String> global = new HashMap<>();
public SchedConfUpdateInfo() {
// JAXB needs this
}
public ArrayList<QueueConfigInfo> getAddQueueInfo() {
return addQueueInfo;
}
public void setAddQueueInfo(ArrayList<QueueConfigInfo> addQueueInfo) {
this.addQueueInfo = addQueueInfo;
}
public ArrayList<String> getRemoveQueueInfo() {
return removeQueueInfo;
}
public void setRemoveQueueInfo(ArrayList<String> removeQueueInfo) {
this.removeQueueInfo = removeQueueInfo;
}
public ArrayList<QueueConfigInfo> getUpdateQueueInfo() {
return updateQueueInfo;
}
public void setUpdateQueueInfo(ArrayList<QueueConfigInfo> updateQueueInfo) {
this.updateQueueInfo = updateQueueInfo;
}
@XmlElementWrapper(name = "global-updates")
public HashMap<String, String> getGlobalParams() {
return global;
}
public void setGlobalParams(HashMap<String, String> globalInfo) {
this.global = globalInfo;
}
}

View File

@ -0,0 +1,27 @@
/*
* 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.
*/
/**
* Data structures for scheduler configuration mutation info.
*/
@InterfaceAudience.Private
@InterfaceStability.Unstable
package org.apache.hadoop.yarn.webapp.dao;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;

View File

@ -23,9 +23,14 @@ import com.sun.jersey.api.client.ClientResponse;
import com.sun.jersey.api.client.UniformInterfaceException;
import com.sun.jersey.api.client.WebResource;
import javax.ws.rs.core.MediaType;
import com.sun.jersey.api.json.JSONJAXBContext;
import com.sun.jersey.api.json.JSONMarshaller;
import org.apache.hadoop.conf.Configuration;
import org.codehaus.jettison.json.JSONObject;
import java.io.StringWriter;
/**
* This class contains several utility function which could be used to generate
* Restful calls to RM/NM/AHS.
@ -59,4 +64,13 @@ public final class YarnWebServiceUtils {
.get(ClientResponse.class);
return response.getEntity(JSONObject.class);
}
@SuppressWarnings("rawtypes")
public static String toJson(Object nsli, Class klass) throws Exception {
StringWriter sw = new StringWriter();
JSONJAXBContext ctx = new JSONJAXBContext(klass);
JSONMarshaller jm = ctx.createJSONMarshaller();
jm.marshallToJSON(nsli, sw);
return sw.toString();
}
}

View File

@ -3367,4 +3367,68 @@
</description>
</property>
<property>
<description>
The type of configuration store to use for scheduler configurations.
Default is "file", which uses file based capacity-scheduler.xml to
retrieve and change scheduler configuration. To enable API based
scheduler configuration, use either "memory" (in memory storage, no
persistence across restarts), "leveldb" (leveldb based storage), or
"zk" (zookeeper based storage). API based configuration is only useful
when using a scheduler which supports mutable configuration. Currently
only capacity scheduler supports this.
</description>
<name>yarn.scheduler.configuration.store.class</name>
<value>file</value>
</property>
<property>
<description>
The class to use for configuration mutation ACL policy if using a mutable
configuration provider. Controls whether a mutation request is allowed.
The DefaultConfigurationMutationACLPolicy checks if the requestor is a
YARN admin.
</description>
<name>yarn.scheduler.configuration.mutation.acl-policy.class</name>
<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.DefaultConfigurationMutationACLPolicy</value>
</property>
<property>
<description>
The storage path for LevelDB implementation of configuration store,
when yarn.scheduler.configuration.store.class is configured to be
"leveldb".
</description>
<name>yarn.scheduler.configuration.leveldb-store.path</name>
<value>${hadoop.tmp.dir}/yarn/system/confstore</value>
</property>
<property>
<description>
The compaction interval for LevelDB configuration store in secs,
when yarn.scheduler.configuration.store.class is configured to be
"leveldb". Default is one day.
</description>
<name>yarn.scheduler.configuration.leveldb-store.compaction-interval-secs</name>
<value>86400</value>
</property>
<property>
<description>
The max number of configuration change log entries kept in config
store, when yarn.scheduler.configuration.store.class is configured to be
"leveldb" or "zk". Default is 1000 for either.
</description>
<name>yarn.scheduler.configuration.store.max-logs</name>
<value>1000</value>
</property>
<property>
<description>
ZK root node path for configuration store when using zookeeper-based
configuration store.
</description>
<name>yarn.scheduler.configuration.zk-store.parent-path</name>
<value>/confstore</value>
</property>
</configuration>

View File

@ -29,6 +29,7 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.ha.HAServiceProtocol;
@ -92,6 +93,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSyst
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.MutableConfScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.security.authorize.RMPolicyProvider;
import com.google.common.annotations.VisibleForTesting;
@ -384,6 +387,10 @@ public class AdminService extends CompositeService implements
RefreshQueuesResponse response =
recordFactory.newRecordInstance(RefreshQueuesResponse.class);
try {
if (isSchedulerMutable()) {
throw new IOException("Scheduler configuration is mutable. " +
operation + " is not allowed in this scenario.");
}
refreshQueues();
RMAuditLogger.logSuccess(user.getShortUserName(), operation,
"AdminService");
@ -393,7 +400,8 @@ public class AdminService extends CompositeService implements
}
}
private void refreshQueues() throws IOException, YarnException {
@Private
public void refreshQueues() throws IOException, YarnException {
rm.getRMContext().getScheduler().reinitialize(getConfig(),
this.rm.getRMContext());
// refresh the reservation system
@ -403,6 +411,12 @@ public class AdminService extends CompositeService implements
}
}
private boolean isSchedulerMutable() {
ResourceScheduler scheduler = rm.getRMContext().getScheduler();
return (scheduler instanceof MutableConfScheduler
&& ((MutableConfScheduler) scheduler).isConfigurationMutable());
}
@Override
public RefreshNodesResponse refreshNodes(RefreshNodesRequest request)
throws YarnException, StandbyException {
@ -711,6 +725,14 @@ public class AdminService extends CompositeService implements
void refreshAll() throws ServiceFailedException {
try {
checkAcls("refreshAll");
if (isSchedulerMutable()) {
try {
((MutableConfScheduler) rm.getRMContext().getScheduler())
.getMutableConfProvider().reloadConfigurationFromStore();
} catch (Exception e) {
throw new IOException("Failed to refresh configuration:", e);
}
}
refreshQueues();
refreshNodes();
refreshSuperUserGroupsConfiguration();

View File

@ -351,7 +351,7 @@ public class ResourceManager extends CompositeService implements Recoverable {
conf.getBoolean(YarnConfiguration.CURATOR_LEADER_ELECTOR,
YarnConfiguration.DEFAULT_CURATOR_LEADER_ELECTOR_ENABLED);
if (curatorEnabled) {
this.zkManager = createAndStartZKManager(conf);
this.zkManager = getAndStartZKManager(conf);
elector = new CuratorBasedElectorService(this);
} else {
elector = new ActiveStandbyElectorBasedElectorService(this);
@ -360,13 +360,16 @@ public class ResourceManager extends CompositeService implements Recoverable {
}
/**
* Create and ZooKeeper Curator manager.
* Get ZooKeeper Curator manager, creating and starting if not exists.
* @param config Configuration for the ZooKeeper curator.
* @return New ZooKeeper Curator manager.
* @return ZooKeeper Curator manager.
* @throws IOException If it cannot create the manager.
*/
public ZKCuratorManager createAndStartZKManager(Configuration config)
throws IOException {
public synchronized ZKCuratorManager getAndStartZKManager(Configuration
config) throws IOException {
if (this.zkManager != null) {
return zkManager;
}
ZKCuratorManager manager = new ZKCuratorManager(config);
// Get authentication
@ -386,15 +389,8 @@ public class ResourceManager extends CompositeService implements Recoverable {
}
manager.start(authInfos);
return manager;
}
/**
* Get the ZooKeeper Curator manager.
* @return ZooKeeper Curator manager.
*/
public ZKCuratorManager getZKManager() {
return this.zkManager;
this.zkManager = manager;
return zkManager;
}
public CuratorFramework getCurator() {

View File

@ -22,7 +22,7 @@ import org.apache.hadoop.yarn.exceptions.YarnException;
/**
* This exception is thrown by ResourceManager if it's loading an incompatible
* version of state from state store on recovery.
* version of storage on recovery.
*/
public class RMStateVersionIncompatibleException extends YarnException {

View File

@ -327,10 +327,7 @@ public class ZKRMStateStore extends RMStateStore {
amrmTokenSecretManagerRoot =
getNodePath(zkRootNodePath, AMRMTOKEN_SECRET_MANAGER_ROOT);
reservationRoot = getNodePath(zkRootNodePath, RESERVATION_SYSTEM_ROOT);
zkManager = resourceManager.getZKManager();
if (zkManager == null) {
zkManager = resourceManager.createAndStartZKManager(conf);
}
zkManager = resourceManager.getAndStartZKManager(conf);
}
@Override

View File

@ -0,0 +1,47 @@
/**
* 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.resourcemanager.scheduler;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
/**
* Interface for determining whether configuration mutations are allowed.
*/
public interface ConfigurationMutationACLPolicy {
/**
* Initialize ACL policy with configuration and RMContext.
* @param conf Configuration to initialize with.
* @param rmContext rmContext
*/
void init(Configuration conf, RMContext rmContext);
/**
* Check if mutation is allowed.
* @param user User issuing the request
* @param confUpdate configurations to be updated
* @return whether provided mutation is allowed or not
*/
boolean isMutationAllowed(UserGroupInformation user, SchedConfUpdateInfo
confUpdate);
}

View File

@ -0,0 +1,49 @@
/**
* 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.resourcemanager.scheduler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
/**
* Factory class for creating instances of
* {@link ConfigurationMutationACLPolicy}.
*/
public final class ConfigurationMutationACLPolicyFactory {
private static final Log LOG = LogFactory.getLog(
ConfigurationMutationACLPolicyFactory.class);
private ConfigurationMutationACLPolicyFactory() {
// Unused.
}
public static ConfigurationMutationACLPolicy getPolicy(Configuration conf) {
Class<? extends ConfigurationMutationACLPolicy> policyClass =
conf.getClass(YarnConfiguration.RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS,
DefaultConfigurationMutationACLPolicy.class,
ConfigurationMutationACLPolicy.class);
LOG.info("Using ConfigurationMutationACLPolicy implementation - " +
policyClass);
return ReflectionUtils.newInstance(policyClass, conf);
}
}

View File

@ -0,0 +1,45 @@
/**
* 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.resourcemanager.scheduler;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.security.YarnAuthorizationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
/**
* Default configuration mutation ACL policy. Checks if user is YARN admin.
*/
public class DefaultConfigurationMutationACLPolicy implements
ConfigurationMutationACLPolicy {
private YarnAuthorizationProvider authorizer;
@Override
public void init(Configuration conf, RMContext rmContext) {
authorizer = YarnAuthorizationProvider.getInstance(conf);
}
@Override
public boolean isMutationAllowed(UserGroupInformation user,
SchedConfUpdateInfo confUpdate) {
return authorizer.isAdmin(user);
}
}

View File

@ -0,0 +1,53 @@
/**
* 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.resourcemanager.scheduler;
import org.apache.hadoop.conf.Configuration;
/**
* Interface for a scheduler that supports changing configuration at runtime.
*
*/
public interface MutableConfScheduler extends ResourceScheduler {
/**
* Get the scheduler configuration.
* @return the scheduler configuration
*/
Configuration getConfiguration();
/**
* Get queue object based on queue name.
* @param queueName the queue name
* @return the queue object
*/
Queue getQueue(String queueName);
/**
* Return whether the scheduler configuration is mutable.
* @return whether scheduler configuration is mutable or not.
*/
boolean isConfigurationMutable();
/**
* Get scheduler's configuration provider, so other classes can directly
* call mutation APIs on configuration provider.
* @return scheduler's configuration provider
*/
MutableConfigurationProvider getMutableConfProvider();
}

View File

@ -0,0 +1,66 @@
/**
* 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.resourcemanager.scheduler;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import java.io.IOException;
/**
* Interface for allowing changing scheduler configurations.
*/
public interface MutableConfigurationProvider {
/**
* Get the acl mutation policy for this configuration provider.
* @return The acl mutation policy.
*/
ConfigurationMutationACLPolicy getAclMutationPolicy();
/**
* Called when a new ResourceManager is starting/becomes active. Ensures
* configuration is up-to-date.
* @throws Exception if configuration could not be refreshed from store
*/
void reloadConfigurationFromStore() throws Exception;
/**
* Log user's requested configuration mutation, and applies it in-memory.
* @param user User who requested the change
* @param confUpdate User's requested configuration change
* @throws Exception if logging the mutation fails
*/
void logAndApplyMutation(UserGroupInformation user, SchedConfUpdateInfo
confUpdate) throws Exception;
/**
* Confirm last logged mutation.
* @param isValid if the last logged mutation is applied to scheduler
* properly.
* @throws Exception if confirming mutation fails
*/
void confirmPendingMutation(boolean isValid) throws Exception;
/**
* Closes the configuration provider, releasing any required resources.
* @throws IOException on failure to close
*/
void close() throws IOException;
}

View File

@ -19,7 +19,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumSet;
@ -87,6 +86,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnSched
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AppSchedulingInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ContainerUpdates;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.PreemptableResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
@ -103,6 +104,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.Activi
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.AllocationState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.CSConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.FileBasedCSConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.MutableCSConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.KillableContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.PreemptionManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.AssignmentInformation;
@ -148,7 +152,7 @@ import com.google.common.util.concurrent.SettableFuture;
public class CapacityScheduler extends
AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
PreemptableResourceScheduler, CapacitySchedulerContext, Configurable,
ResourceAllocationCommitter {
ResourceAllocationCommitter, MutableConfScheduler {
private static final Log LOG = LogFactory.getLog(CapacityScheduler.class);
@ -167,6 +171,8 @@ public class CapacityScheduler extends
private int maxAssignPerHeartbeat;
private CSConfigurationProvider csConfProvider;
@Override
public void setConf(Configuration conf) {
yarnConf = conf;
@ -289,7 +295,25 @@ public class CapacityScheduler extends
IOException {
try {
writeLock.lock();
this.conf = loadCapacitySchedulerConfiguration(configuration);
String confProviderStr = configuration.get(
YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
YarnConfiguration.DEFAULT_CONFIGURATION_STORE);
switch (confProviderStr) {
case YarnConfiguration.FILE_CONFIGURATION_STORE:
this.csConfProvider =
new FileBasedCSConfigurationProvider(rmContext);
break;
case YarnConfiguration.MEMORY_CONFIGURATION_STORE:
case YarnConfiguration.LEVELDB_CONFIGURATION_STORE:
case YarnConfiguration.ZK_CONFIGURATION_STORE:
this.csConfProvider = new MutableCSConfigurationProvider(rmContext);
break;
default:
throw new IOException("Invalid configuration store class: " +
confProviderStr);
}
this.csConfProvider.init(configuration);
this.conf = this.csConfProvider.loadConfiguration(configuration);
validateConf(this.conf);
this.minimumAllocation = super.getMinimumAllocation();
initMaximumResourceCapability(super.getMaximumAllocation());
@ -389,6 +413,9 @@ public class CapacityScheduler extends
writeLock.unlock();
}
if (isConfigurationMutable()) {
((MutableConfigurationProvider) csConfProvider).close();
}
super.serviceStop();
}
@ -399,7 +426,7 @@ public class CapacityScheduler extends
writeLock.lock();
Configuration configuration = new Configuration(newConf);
CapacitySchedulerConfiguration oldConf = this.conf;
this.conf = loadCapacitySchedulerConfiguration(configuration);
this.conf = csConfProvider.loadConfiguration(configuration);
validateConf(this.conf);
try {
LOG.info("Re-initializing queues...");
@ -627,6 +654,7 @@ public class CapacityScheduler extends
preemptionManager.refreshQueues(null, this.getRootQueue());
}
@Override
public CSQueue getQueue(String queueName) {
if (queueName == null) {
return null;
@ -1831,23 +1859,6 @@ public class CapacityScheduler extends
return true;
}
private CapacitySchedulerConfiguration loadCapacitySchedulerConfiguration(
Configuration configuration) throws IOException {
try {
InputStream CSInputStream =
this.rmContext.getConfigurationProvider()
.getConfigurationInputStream(configuration,
YarnConfiguration.CS_CONFIGURATION_FILE);
if (CSInputStream != null) {
configuration.addResource(CSInputStream);
return new CapacitySchedulerConfiguration(configuration, false);
}
return new CapacitySchedulerConfiguration(configuration, true);
} catch (Exception e) {
throw new IOException(e);
}
}
private String getDefaultReservationQueueName(String planQueueName) {
return planQueueName + ReservationConstants.DEFAULT_QUEUE_SUFFIX;
}
@ -2607,4 +2618,17 @@ public class CapacityScheduler extends
// In seconds
return ((LeafQueue) queue).getMaximumApplicationLifetime();
}
@Override
public boolean isConfigurationMutable() {
return csConfProvider instanceof MutableConfigurationProvider;
}
@Override
public MutableConfigurationProvider getMutableConfProvider() {
if (isConfigurationMutable()) {
return (MutableConfigurationProvider) csConfProvider;
}
return null;
}
}

View File

@ -88,4 +88,10 @@ public interface CapacitySchedulerContext {
* @return Max Cluster level App priority.
*/
Priority getMaxClusterLevelAppPriority();
/**
* Returns if configuration is mutable.
* @return if configuration is mutable
*/
boolean isConfigurationMutable();
}

View File

@ -33,6 +33,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ha.HAServiceProtocol;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.QueueState;
@ -170,8 +171,14 @@ public class CapacitySchedulerQueueManager implements SchedulerQueueManager<
CSQueue newRoot = parseQueue(this.csContext, newConf, null,
CapacitySchedulerConfiguration.ROOT, newQueues, queues, NOOP);
// Ensure queue hiearchy in the new XML file is proper.
validateQueueHierarchy(queues, newQueues);
// When failing over, if using configuration store, don't validate queue
// hierarchy since queues can be removed without being STOPPED.
if (!csContext.isConfigurationMutable() ||
csContext.getRMContext().getHAServiceState()
!= HAServiceProtocol.HAServiceState.STANDBY) {
// Ensure queue hiearchy in the new XML file is proper.
validateQueueHierarchy(queues, newQueues);
}
// Add new queues and delete OldQeueus only after validation.
updateQueues(queues, newQueues);

View File

@ -0,0 +1,47 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import java.io.IOException;
/**
* Configuration provider for {@link CapacityScheduler}.
*/
public interface CSConfigurationProvider {
/**
* Initialize the configuration provider with given conf.
* @param conf configuration to initialize with
* @throws IOException if initialization fails due to misconfiguration
*/
void init(Configuration conf) throws IOException;
/**
* Loads capacity scheduler configuration object.
* @param conf initial bootstrap configuration
* @return CS configuration
* @throws IOException if fail to retrieve configuration
*/
CapacitySchedulerConfiguration loadConfiguration(Configuration conf)
throws IOException;
}

View File

@ -0,0 +1,67 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import java.io.IOException;
import java.io.InputStream;
/**
* {@link CapacityScheduler} configuration provider based on local
* {@code capacity-scheduler.xml} file.
*/
public class FileBasedCSConfigurationProvider implements
CSConfigurationProvider {
private RMContext rmContext;
/**
* Construct file based CS configuration provider with given context.
* @param rmContext the RM context
*/
public FileBasedCSConfigurationProvider(RMContext rmContext) {
this.rmContext = rmContext;
}
@Override
public void init(Configuration conf) {}
@Override
public CapacitySchedulerConfiguration loadConfiguration(Configuration conf)
throws IOException {
try {
InputStream csInputStream =
this.rmContext.getConfigurationProvider()
.getConfigurationInputStream(conf,
YarnConfiguration.CS_CONFIGURATION_FILE);
if (csInputStream != null) {
conf.addResource(csInputStream);
return new CapacitySchedulerConfiguration(conf, false);
}
return new CapacitySchedulerConfiguration(conf, true);
} catch (Exception e) {
throw new IOException(e);
}
}
}

View File

@ -0,0 +1,95 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.server.records.Version;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import java.util.List;
import java.util.Map;
/**
* A default implementation of {@link YarnConfigurationStore}. Doesn't offer
* persistent configuration storage, just stores the configuration in memory.
*/
public class InMemoryConfigurationStore extends YarnConfigurationStore {
private Configuration schedConf;
private LogMutation pendingMutation;
@Override
public void initialize(Configuration conf, Configuration schedConf,
RMContext rmContext) {
this.schedConf = schedConf;
}
@Override
public void logMutation(LogMutation logMutation) {
pendingMutation = logMutation;
}
@Override
public void confirmMutation(boolean isValid) {
if (isValid) {
for (Map.Entry<String, String> kv : pendingMutation.getUpdates()
.entrySet()) {
if (kv.getValue() == null) {
schedConf.unset(kv.getKey());
} else {
schedConf.set(kv.getKey(), kv.getValue());
}
}
}
pendingMutation = null;
}
@Override
public synchronized Configuration retrieve() {
return schedConf;
}
@Override
public List<LogMutation> getConfirmedConfHistory(long fromId) {
// Unimplemented.
return null;
}
@Override
public Version getConfStoreVersion() throws Exception {
// Does nothing.
return null;
}
@Override
public void storeVersion() throws Exception {
// Does nothing.
}
@Override
public Version getCurrentVersion() {
// Does nothing.
return null;
}
@Override
public void checkVersion() {
// Does nothing. (Version is always compatible since it's in memory)
}
}

View File

@ -0,0 +1,322 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.proto.YarnServerCommonProtos;
import org.apache.hadoop.yarn.server.records.Version;
import org.apache.hadoop.yarn.server.records.impl.pb.VersionPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.fusesource.leveldbjni.JniDBFactory;
import org.fusesource.leveldbjni.internal.NativeDB;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBComparator;
import org.iq80.leveldb.DBException;
import org.iq80.leveldb.DBIterator;
import org.iq80.leveldb.Options;
import org.iq80.leveldb.WriteBatch;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectInputStream;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import static org.fusesource.leveldbjni.JniDBFactory.bytes;
/**
* A LevelDB implementation of {@link YarnConfigurationStore}.
*/
public class LeveldbConfigurationStore extends YarnConfigurationStore {
public static final Log LOG =
LogFactory.getLog(LeveldbConfigurationStore.class);
private static final String DB_NAME = "yarn-conf-store";
private static final String LOG_KEY = "log";
private static final String VERSION_KEY = "version";
private DB db;
private long maxLogs;
private Configuration conf;
private LogMutation pendingMutation;
@VisibleForTesting
protected static final Version CURRENT_VERSION_INFO = Version
.newInstance(0, 1);
private Timer compactionTimer;
private long compactionIntervalMsec;
@Override
public void initialize(Configuration config, Configuration schedConf,
RMContext rmContext) throws IOException {
this.conf = config;
try {
initDatabase(schedConf);
this.maxLogs = config.getLong(
YarnConfiguration.RM_SCHEDCONF_MAX_LOGS,
YarnConfiguration.DEFAULT_RM_SCHEDCONF_LEVELDB_MAX_LOGS);
this.compactionIntervalMsec = config.getLong(
YarnConfiguration.RM_SCHEDCONF_LEVELDB_COMPACTION_INTERVAL_SECS,
YarnConfiguration
.DEFAULT_RM_SCHEDCONF_LEVELDB_COMPACTION_INTERVAL_SECS) * 1000;
startCompactionTimer();
} catch (Exception e) {
throw new IOException(e);
}
}
private void initDatabase(Configuration config) throws Exception {
Path storeRoot = createStorageDir();
Options options = new Options();
options.createIfMissing(false);
options.comparator(new DBComparator() {
@Override
public int compare(byte[] key1, byte[] key2) {
String key1Str = new String(key1, StandardCharsets.UTF_8);
String key2Str = new String(key2, StandardCharsets.UTF_8);
if (key1Str.equals(key2Str)) {
return 0;
} else if (key1Str.equals(VERSION_KEY)) {
return 1;
} else if (key2Str.equals(VERSION_KEY)) {
return -1;
} else if (key1Str.equals(LOG_KEY)) {
return 1;
} else if (key2Str.equals(LOG_KEY)) {
return -1;
}
return key1Str.compareTo(key2Str);
}
@Override
public String name() {
return "keyComparator";
}
public byte[] findShortestSeparator(byte[] start, byte[] limit) {
return start;
}
public byte[] findShortSuccessor(byte[] key) {
return key;
}
});
LOG.info("Using conf database at " + storeRoot);
File dbfile = new File(storeRoot.toString());
try {
db = JniDBFactory.factory.open(dbfile, options);
} catch (NativeDB.DBException e) {
if (e.isNotFound() || e.getMessage().contains(" does not exist ")) {
LOG.info("Creating conf database at " + dbfile);
options.createIfMissing(true);
try {
db = JniDBFactory.factory.open(dbfile, options);
// Write the initial scheduler configuration
WriteBatch initBatch = db.createWriteBatch();
for (Map.Entry<String, String> kv : config) {
initBatch.put(bytes(kv.getKey()), bytes(kv.getValue()));
}
db.write(initBatch);
} catch (DBException dbErr) {
throw new IOException(dbErr.getMessage(), dbErr);
}
} else {
throw e;
}
}
}
private Path createStorageDir() throws IOException {
Path root = getStorageDir();
FileSystem fs = FileSystem.getLocal(conf);
fs.mkdirs(root, new FsPermission((short) 0700));
return root;
}
private Path getStorageDir() throws IOException {
String storePath = conf.get(YarnConfiguration.RM_SCHEDCONF_STORE_PATH);
if (storePath == null) {
throw new IOException("No store location directory configured in " +
YarnConfiguration.RM_SCHEDCONF_STORE_PATH);
}
return new Path(storePath, DB_NAME);
}
@Override
public void close() throws IOException {
if (db != null) {
db.close();
}
}
@Override
public void logMutation(LogMutation logMutation) throws IOException {
LinkedList<LogMutation> logs = deserLogMutations(db.get(bytes(LOG_KEY)));
logs.add(logMutation);
if (logs.size() > maxLogs) {
logs.removeFirst();
}
db.put(bytes(LOG_KEY), serLogMutations(logs));
pendingMutation = logMutation;
}
@Override
public void confirmMutation(boolean isValid) throws IOException {
WriteBatch updateBatch = db.createWriteBatch();
if (isValid) {
for (Map.Entry<String, String> changes :
pendingMutation.getUpdates().entrySet()) {
if (changes.getValue() == null || changes.getValue().isEmpty()) {
updateBatch.delete(bytes(changes.getKey()));
} else {
updateBatch.put(bytes(changes.getKey()), bytes(changes.getValue()));
}
}
}
db.write(updateBatch);
pendingMutation = null;
}
private byte[] serLogMutations(LinkedList<LogMutation> mutations) throws
IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try (ObjectOutput oos = new ObjectOutputStream(baos)) {
oos.writeObject(mutations);
oos.flush();
return baos.toByteArray();
}
}
private LinkedList<LogMutation> deserLogMutations(byte[] mutations) throws
IOException {
if (mutations == null) {
return new LinkedList<>();
}
try (ObjectInput input = new ObjectInputStream(
new ByteArrayInputStream(mutations))) {
return (LinkedList<LogMutation>) input.readObject();
} catch (ClassNotFoundException e) {
throw new IOException(e);
}
}
@Override
public synchronized Configuration retrieve() {
DBIterator itr = db.iterator();
itr.seekToFirst();
Configuration config = new Configuration(false);
while (itr.hasNext()) {
Map.Entry<byte[], byte[]> entry = itr.next();
String key = new String(entry.getKey(), StandardCharsets.UTF_8);
String value = new String(entry.getValue(), StandardCharsets.UTF_8);
if (key.equals(LOG_KEY) || key.equals(VERSION_KEY)) {
break;
}
config.set(key, value);
}
return config;
}
@Override
public List<LogMutation> getConfirmedConfHistory(long fromId) {
return null; // unimplemented
}
// TODO below was taken from LeveldbRMStateStore, it can probably be
// refactored
private void startCompactionTimer() {
if (compactionIntervalMsec > 0) {
compactionTimer = new Timer(
this.getClass().getSimpleName() + " compaction timer", true);
compactionTimer.schedule(new CompactionTimerTask(),
compactionIntervalMsec, compactionIntervalMsec);
}
}
// TODO: following is taken from LeveldbRMStateStore
@Override
public Version getConfStoreVersion() throws Exception {
Version version = null;
try {
byte[] data = db.get(bytes(VERSION_KEY));
if (data != null) {
version = new VersionPBImpl(YarnServerCommonProtos.VersionProto
.parseFrom(data));
}
} catch (DBException e) {
throw new IOException(e);
}
return version;
}
@VisibleForTesting
protected LinkedList<LogMutation> getLogs() throws Exception {
return deserLogMutations(db.get(bytes(LOG_KEY)));
}
@Override
public void storeVersion() throws Exception {
String key = VERSION_KEY;
byte[] data = ((VersionPBImpl) CURRENT_VERSION_INFO).getProto()
.toByteArray();
try {
db.put(bytes(key), data);
} catch (DBException e) {
throw new IOException(e);
}
}
@Override
public Version getCurrentVersion() {
return CURRENT_VERSION_INFO;
}
private class CompactionTimerTask extends TimerTask {
@Override
public void run() {
long start = Time.monotonicNow();
LOG.info("Starting full compaction cycle");
try {
db.compactRange(null, null);
} catch (DBException e) {
LOG.error("Error compacting database", e);
}
long duration = Time.monotonicNow() - start;
LOG.info("Full compaction cycle completed in " + duration + " msec");
}
}
}

View File

@ -0,0 +1,284 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ConfigurationMutationACLPolicy;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ConfigurationMutationACLPolicyFactory;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.YarnConfigurationStore.LogMutation;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* CS configuration provider which implements
* {@link MutableConfigurationProvider} for modifying capacity scheduler
* configuration.
*/
public class MutableCSConfigurationProvider implements CSConfigurationProvider,
MutableConfigurationProvider {
public static final Log LOG =
LogFactory.getLog(MutableCSConfigurationProvider.class);
private Configuration schedConf;
private Configuration oldConf;
private YarnConfigurationStore confStore;
private ConfigurationMutationACLPolicy aclMutationPolicy;
private RMContext rmContext;
public MutableCSConfigurationProvider(RMContext rmContext) {
this.rmContext = rmContext;
}
@Override
public void init(Configuration config) throws IOException {
String store = config.get(
YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
YarnConfiguration.MEMORY_CONFIGURATION_STORE);
switch (store) {
case YarnConfiguration.MEMORY_CONFIGURATION_STORE:
this.confStore = new InMemoryConfigurationStore();
break;
case YarnConfiguration.LEVELDB_CONFIGURATION_STORE:
this.confStore = new LeveldbConfigurationStore();
break;
case YarnConfiguration.ZK_CONFIGURATION_STORE:
this.confStore = new ZKConfigurationStore();
break;
default:
this.confStore = YarnConfigurationStoreFactory.getStore(config);
break;
}
Configuration initialSchedConf = new Configuration(false);
initialSchedConf.addResource(YarnConfiguration.CS_CONFIGURATION_FILE);
this.schedConf = new Configuration(false);
// We need to explicitly set the key-values in schedConf, otherwise
// these configuration keys cannot be deleted when
// configuration is reloaded.
for (Map.Entry<String, String> kv : initialSchedConf) {
schedConf.set(kv.getKey(), kv.getValue());
}
try {
confStore.initialize(config, schedConf, rmContext);
confStore.checkVersion();
} catch (Exception e) {
throw new IOException(e);
}
// After initializing confStore, the store may already have an existing
// configuration. Use this one.
schedConf = confStore.retrieve();
this.aclMutationPolicy = ConfigurationMutationACLPolicyFactory
.getPolicy(config);
aclMutationPolicy.init(config, rmContext);
}
@Override
public void close() throws IOException {
confStore.close();
}
@VisibleForTesting
public YarnConfigurationStore getConfStore() {
return confStore;
}
@Override
public CapacitySchedulerConfiguration loadConfiguration(Configuration
configuration) throws IOException {
Configuration loadedConf = new Configuration(schedConf);
loadedConf.addResource(configuration);
return new CapacitySchedulerConfiguration(loadedConf, false);
}
@Override
public ConfigurationMutationACLPolicy getAclMutationPolicy() {
return aclMutationPolicy;
}
@Override
public void logAndApplyMutation(UserGroupInformation user,
SchedConfUpdateInfo confUpdate) throws Exception {
oldConf = new Configuration(schedConf);
Map<String, String> kvUpdate = constructKeyValueConfUpdate(confUpdate);
LogMutation log = new LogMutation(kvUpdate, user.getShortUserName());
confStore.logMutation(log);
for (Map.Entry<String, String> kv : kvUpdate.entrySet()) {
if (kv.getValue() == null) {
schedConf.unset(kv.getKey());
} else {
schedConf.set(kv.getKey(), kv.getValue());
}
}
}
@Override
public void confirmPendingMutation(boolean isValid) throws Exception {
confStore.confirmMutation(isValid);
if (!isValid) {
schedConf = oldConf;
}
}
@Override
public void reloadConfigurationFromStore() throws Exception {
schedConf = confStore.retrieve();
}
private List<String> getSiblingQueues(String queuePath, Configuration conf) {
String parentQueue = queuePath.substring(0, queuePath.lastIndexOf('.'));
String childQueuesKey = CapacitySchedulerConfiguration.PREFIX +
parentQueue + CapacitySchedulerConfiguration.DOT +
CapacitySchedulerConfiguration.QUEUES;
return new ArrayList<>(conf.getStringCollection(childQueuesKey));
}
private Map<String, String> constructKeyValueConfUpdate(
SchedConfUpdateInfo mutationInfo) throws IOException {
CapacitySchedulerConfiguration proposedConf =
new CapacitySchedulerConfiguration(schedConf, false);
Map<String, String> confUpdate = new HashMap<>();
for (String queueToRemove : mutationInfo.getRemoveQueueInfo()) {
removeQueue(queueToRemove, proposedConf, confUpdate);
}
for (QueueConfigInfo addQueueInfo : mutationInfo.getAddQueueInfo()) {
addQueue(addQueueInfo, proposedConf, confUpdate);
}
for (QueueConfigInfo updateQueueInfo : mutationInfo.getUpdateQueueInfo()) {
updateQueue(updateQueueInfo, proposedConf, confUpdate);
}
for (Map.Entry<String, String> global : mutationInfo.getGlobalParams()
.entrySet()) {
confUpdate.put(global.getKey(), global.getValue());
}
return confUpdate;
}
private void removeQueue(
String queueToRemove, CapacitySchedulerConfiguration proposedConf,
Map<String, String> confUpdate) throws IOException {
if (queueToRemove == null) {
return;
} else {
String queueName = queueToRemove.substring(
queueToRemove.lastIndexOf('.') + 1);
if (queueToRemove.lastIndexOf('.') == -1) {
throw new IOException("Can't remove queue " + queueToRemove);
} else {
List<String> siblingQueues = getSiblingQueues(queueToRemove,
proposedConf);
if (!siblingQueues.contains(queueName)) {
throw new IOException("Queue " + queueToRemove + " not found");
}
siblingQueues.remove(queueName);
String parentQueuePath = queueToRemove.substring(0, queueToRemove
.lastIndexOf('.'));
proposedConf.setQueues(parentQueuePath, siblingQueues.toArray(
new String[0]));
String queuesConfig = CapacitySchedulerConfiguration.PREFIX
+ parentQueuePath + CapacitySchedulerConfiguration.DOT
+ CapacitySchedulerConfiguration.QUEUES;
if (siblingQueues.size() == 0) {
confUpdate.put(queuesConfig, null);
} else {
confUpdate.put(queuesConfig, Joiner.on(',').join(siblingQueues));
}
for (Map.Entry<String, String> confRemove : proposedConf.getValByRegex(
".*" + queueToRemove.replaceAll("\\.", "\\.") + "\\..*")
.entrySet()) {
proposedConf.unset(confRemove.getKey());
confUpdate.put(confRemove.getKey(), null);
}
}
}
}
private void addQueue(
QueueConfigInfo addInfo, CapacitySchedulerConfiguration proposedConf,
Map<String, String> confUpdate) throws IOException {
if (addInfo == null) {
return;
} else {
String queuePath = addInfo.getQueue();
String queueName = queuePath.substring(queuePath.lastIndexOf('.') + 1);
if (queuePath.lastIndexOf('.') == -1) {
throw new IOException("Can't add invalid queue " + queuePath);
} else if (getSiblingQueues(queuePath, proposedConf).contains(
queueName)) {
throw new IOException("Can't add existing queue " + queuePath);
}
String parentQueue = queuePath.substring(0, queuePath.lastIndexOf('.'));
String[] siblings = proposedConf.getQueues(parentQueue);
List<String> siblingQueues = siblings == null ? new ArrayList<>() :
new ArrayList<>(Arrays.<String>asList(siblings));
siblingQueues.add(queuePath.substring(queuePath.lastIndexOf('.') + 1));
proposedConf.setQueues(parentQueue,
siblingQueues.toArray(new String[0]));
confUpdate.put(CapacitySchedulerConfiguration.PREFIX
+ parentQueue + CapacitySchedulerConfiguration.DOT
+ CapacitySchedulerConfiguration.QUEUES,
Joiner.on(',').join(siblingQueues));
String keyPrefix = CapacitySchedulerConfiguration.PREFIX
+ queuePath + CapacitySchedulerConfiguration.DOT;
for (Map.Entry<String, String> kv : addInfo.getParams().entrySet()) {
if (kv.getValue() == null) {
proposedConf.unset(keyPrefix + kv.getKey());
} else {
proposedConf.set(keyPrefix + kv.getKey(), kv.getValue());
}
confUpdate.put(keyPrefix + kv.getKey(), kv.getValue());
}
}
}
private void updateQueue(QueueConfigInfo updateInfo,
CapacitySchedulerConfiguration proposedConf,
Map<String, String> confUpdate) {
if (updateInfo == null) {
return;
} else {
String queuePath = updateInfo.getQueue();
String keyPrefix = CapacitySchedulerConfiguration.PREFIX
+ queuePath + CapacitySchedulerConfiguration.DOT;
for (Map.Entry<String, String> kv : updateInfo.getParams().entrySet()) {
if (kv.getValue() == null) {
proposedConf.unset(keyPrefix + kv.getKey());
} else {
proposedConf.set(keyPrefix + kv.getKey(), kv.getValue());
}
confUpdate.put(keyPrefix + kv.getKey(), kv.getValue());
}
}
}
}

View File

@ -0,0 +1,110 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.QueueACL;
import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.security.YarnAuthorizationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ConfigurationMutationACLPolicy;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
/**
* A configuration mutation ACL policy which checks that user has admin
* privileges on all queues they are changing.
*/
public class QueueAdminConfigurationMutationACLPolicy implements
ConfigurationMutationACLPolicy {
private Configuration conf;
private RMContext rmContext;
private YarnAuthorizationProvider authorizer;
@Override
public void init(Configuration config, RMContext context) {
this.conf = config;
this.rmContext = context;
this.authorizer = YarnAuthorizationProvider.getInstance(conf);
}
@Override
public boolean isMutationAllowed(UserGroupInformation user,
SchedConfUpdateInfo confUpdate) {
// If there are global config changes, check if user is admin.
Map<String, String> globalParams = confUpdate.getGlobalParams();
if (globalParams != null && globalParams.size() != 0) {
if (!authorizer.isAdmin(user)) {
return false;
}
}
// Check if user is admin of all modified queues.
Set<String> queues = new HashSet<>();
for (QueueConfigInfo addQueueInfo : confUpdate.getAddQueueInfo()) {
queues.add(addQueueInfo.getQueue());
}
for (String removeQueue : confUpdate.getRemoveQueueInfo()) {
queues.add(removeQueue);
}
for (QueueConfigInfo updateQueueInfo : confUpdate.getUpdateQueueInfo()) {
queues.add(updateQueueInfo.getQueue());
}
for (String queuePath : queues) {
String queueName = queuePath.lastIndexOf('.') != -1 ?
queuePath.substring(queuePath.lastIndexOf('.') + 1) : queuePath;
QueueInfo queueInfo = null;
try {
queueInfo = rmContext.getScheduler()
.getQueueInfo(queueName, false, false);
} catch (IOException e) {
// Queue is not found, do nothing.
}
String parentPath = queuePath;
while (queueInfo == null) {
// We are adding a queue (whose parent we are possibly also adding).
// Check ACL of lowest parent queue which already exists.
parentPath = parentPath.substring(0, parentPath.lastIndexOf('.'));
String parentName = parentPath.lastIndexOf('.') != -1 ?
parentPath.substring(parentPath.lastIndexOf('.') + 1) : parentPath;
try {
queueInfo = rmContext.getScheduler()
.getQueueInfo(parentName, false, false);
} catch (IOException e) {
// Queue is not found, do nothing.
}
}
Queue queue = ((MutableConfScheduler) rmContext.getScheduler())
.getQueue(queueInfo.getQueueName());
if (queue != null && !queue.hasAccess(QueueACL.ADMINISTER_QUEUE, user)) {
return false;
}
}
return true;
}
}

View File

@ -0,0 +1,178 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.server.records.Version;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateVersionIncompatibleException;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import java.io.IOException;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
/**
* YarnConfigurationStore exposes the methods needed for retrieving and
* persisting {@link CapacityScheduler} configuration via key-value
* using write-ahead logging. When configuration mutation is requested, caller
* should first log it with {@code logMutation}, which persists this pending
* mutation. This mutation is merged to the persisted configuration only after
* {@code confirmMutation} is called.
*
* On startup/recovery, caller should call {@code retrieve} to get all
* confirmed mutations, then get pending mutations which were not confirmed via
* {@code getPendingMutations}, and replay/confirm them via
* {@code confirmMutation} as in the normal case.
*/
public abstract class YarnConfigurationStore {
public static final Log LOG =
LogFactory.getLog(YarnConfigurationStore.class);
/**
* LogMutation encapsulates the fields needed for configuration mutation
* audit logging and recovery.
*/
static class LogMutation implements Serializable {
private Map<String, String> updates;
private String user;
/**
* Create log mutation.
* @param updates key-value configuration updates
* @param user user who requested configuration change
*/
LogMutation(Map<String, String> updates, String user) {
this.updates = updates;
this.user = user;
}
/**
* Get key-value configuration updates.
* @return map of configuration updates
*/
public Map<String, String> getUpdates() {
return updates;
}
/**
* Get user who requested configuration change.
* @return user who requested configuration change
*/
public String getUser() {
return user;
}
}
/**
* Initialize the configuration store, with schedConf as the initial
* scheduler configuration. If a persisted store already exists, use the
* scheduler configuration stored there, and ignore schedConf.
* @param conf configuration to initialize store with
* @param schedConf Initial key-value scheduler configuration to persist.
* @param rmContext RMContext for this configuration store
* @throws IOException if initialization fails
*/
public abstract void initialize(Configuration conf, Configuration schedConf,
RMContext rmContext) throws Exception;
/**
* Closes the configuration store, releasing any required resources.
* @throws IOException on failure to close
*/
public void close() throws IOException {}
/**
* Logs the configuration change to backing store.
* @param logMutation configuration change to be persisted in write ahead log
* @throws IOException if logging fails
*/
public abstract void logMutation(LogMutation logMutation) throws Exception;
/**
* Should be called after {@code logMutation}. Gets the pending mutation
* last logged by {@code logMutation} and marks the mutation as persisted (no
* longer pending). If isValid is true, merge the mutation with the persisted
* configuration.
* @param isValid if true, update persisted configuration with pending
* mutation.
* @throws Exception if mutation confirmation fails
*/
public abstract void confirmMutation(boolean isValid) throws Exception;
/**
* Retrieve the persisted configuration.
* @return configuration as key-value
*/
public abstract Configuration retrieve();
/**
* Get a list of confirmed configuration mutations starting from a given id.
* @param fromId id from which to start getting mutations, inclusive
* @return list of configuration mutations
*/
public abstract List<LogMutation> getConfirmedConfHistory(long fromId);
/**
* Get schema version of persisted conf store, for detecting compatibility
* issues when changing conf store schema.
* @return Schema version currently used by the persisted configuration store.
* @throws Exception On version fetch failure
*/
protected abstract Version getConfStoreVersion() throws Exception;
/**
* Persist the hard-coded schema version to the conf store.
* @throws Exception On storage failure
*/
protected abstract void storeVersion() throws Exception;
/**
* Get the hard-coded schema version, for comparison against the schema
* version currently persisted.
* @return Current hard-coded schema version
*/
protected abstract Version getCurrentVersion();
public void checkVersion() throws Exception {
// TODO this was taken from RMStateStore. Should probably refactor
Version loadedVersion = getConfStoreVersion();
LOG.info("Loaded configuration store version info " + loadedVersion);
if (loadedVersion != null && loadedVersion.equals(getCurrentVersion())) {
return;
}
// if there is no version info, treat it as CURRENT_VERSION_INFO;
if (loadedVersion == null) {
loadedVersion = getCurrentVersion();
}
if (loadedVersion.isCompatibleTo(getCurrentVersion())) {
LOG.info("Storing configuration store version info "
+ getCurrentVersion());
storeVersion();
} else {
throw new RMStateVersionIncompatibleException(
"Expecting configuration store version " + getCurrentVersion()
+ ", but loading version " + loadedVersion);
}
}
}

View File

@ -0,0 +1,46 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
/**
* Factory class for creating instances of {@link YarnConfigurationStore}.
*/
public final class YarnConfigurationStoreFactory {
private static final Log LOG = LogFactory.getLog(
YarnConfigurationStoreFactory.class);
private YarnConfigurationStoreFactory() {
// Unused.
}
public static YarnConfigurationStore getStore(Configuration conf) {
Class<? extends YarnConfigurationStore> storeClass =
conf.getClass(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
InMemoryConfigurationStore.class, YarnConfigurationStore.class);
LOG.info("Using YarnConfigurationStore implementation - " + storeClass);
return ReflectionUtils.newInstance(storeClass, conf);
}
}

View File

@ -0,0 +1,235 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.curator.ZKCuratorManager;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.proto.YarnServerCommonProtos;
import org.apache.hadoop.yarn.server.records.Version;
import org.apache.hadoop.yarn.server.records.impl.pb.VersionPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.ACL;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
/**
* A Zookeeper-based implementation of {@link YarnConfigurationStore}.
*/
public class ZKConfigurationStore extends YarnConfigurationStore {
public static final Log LOG =
LogFactory.getLog(ZKConfigurationStore.class);
private long maxLogs;
@VisibleForTesting
protected static final Version CURRENT_VERSION_INFO = Version
.newInstance(0, 1);
private Configuration conf;
private LogMutation pendingMutation;
private String znodeParentPath;
private static final String ZK_VERSION_PATH = "VERSION";
private static final String LOGS_PATH = "LOGS";
private static final String CONF_STORE_PATH = "CONF_STORE";
private static final String FENCING_PATH = "FENCING";
private String zkVersionPath;
private String logsPath;
private String confStorePath;
private String fencingNodePath;
@VisibleForTesting
protected ZKCuratorManager zkManager;
private List<ACL> zkAcl;
@Override
public void initialize(Configuration config, Configuration schedConf,
RMContext rmContext) throws Exception {
this.conf = config;
this.maxLogs = conf.getLong(YarnConfiguration.RM_SCHEDCONF_MAX_LOGS,
YarnConfiguration.DEFAULT_RM_SCHEDCONF_ZK_MAX_LOGS);
this.znodeParentPath =
conf.get(YarnConfiguration.RM_SCHEDCONF_STORE_ZK_PARENT_PATH,
YarnConfiguration.DEFAULT_RM_SCHEDCONF_STORE_ZK_PARENT_PATH);
this.zkManager = rmContext.getResourceManager().getAndStartZKManager(conf);
this.zkAcl = ZKCuratorManager.getZKAcls(conf);
this.zkVersionPath = getNodePath(znodeParentPath, ZK_VERSION_PATH);
this.logsPath = getNodePath(znodeParentPath, LOGS_PATH);
this.confStorePath = getNodePath(znodeParentPath, CONF_STORE_PATH);
this.fencingNodePath = getNodePath(znodeParentPath, FENCING_PATH);
zkManager.createRootDirRecursively(znodeParentPath);
zkManager.delete(fencingNodePath);
if (!zkManager.exists(logsPath)) {
zkManager.create(logsPath);
zkManager.setData(logsPath,
serializeObject(new LinkedList<LogMutation>()), -1);
}
if (!zkManager.exists(confStorePath)) {
zkManager.create(confStorePath);
HashMap<String, String> mapSchedConf = new HashMap<>();
for (Map.Entry<String, String> entry : schedConf) {
mapSchedConf.put(entry.getKey(), entry.getValue());
}
zkManager.setData(confStorePath, serializeObject(mapSchedConf), -1);
}
}
@VisibleForTesting
protected LinkedList<LogMutation> getLogs() throws Exception {
return (LinkedList<LogMutation>)
deserializeObject(zkManager.getData(logsPath));
}
// TODO: following version-related code is taken from ZKRMStateStore
@Override
public Version getCurrentVersion() {
return CURRENT_VERSION_INFO;
}
@Override
public Version getConfStoreVersion() throws Exception {
if (zkManager.exists(zkVersionPath)) {
byte[] data = zkManager.getData(zkVersionPath);
return new VersionPBImpl(YarnServerCommonProtos.VersionProto
.parseFrom(data));
}
return null;
}
@Override
public synchronized void storeVersion() throws Exception {
byte[] data =
((VersionPBImpl) CURRENT_VERSION_INFO).getProto().toByteArray();
if (zkManager.exists(zkVersionPath)) {
zkManager.safeSetData(zkVersionPath, data, -1, zkAcl, fencingNodePath);
} else {
zkManager.safeCreate(zkVersionPath, data, zkAcl, CreateMode.PERSISTENT,
zkAcl, fencingNodePath);
}
}
@Override
public void logMutation(LogMutation logMutation) throws Exception {
byte[] storedLogs = zkManager.getData(logsPath);
LinkedList<LogMutation> logs = new LinkedList<>();
if (storedLogs != null) {
logs = (LinkedList<LogMutation>) deserializeObject(storedLogs);
}
logs.add(logMutation);
if (logs.size() > maxLogs) {
logs.remove(logs.removeFirst());
}
zkManager.safeSetData(logsPath, serializeObject(logs), -1, zkAcl,
fencingNodePath);
pendingMutation = logMutation;
}
@Override
public void confirmMutation(boolean isValid)
throws Exception {
if (isValid) {
Configuration storedConfigs = retrieve();
Map<String, String> mapConf = new HashMap<>();
for (Map.Entry<String, String> storedConf : storedConfigs) {
mapConf.put(storedConf.getKey(), storedConf.getValue());
}
for (Map.Entry<String, String> confChange :
pendingMutation.getUpdates().entrySet()) {
if (confChange.getValue() == null || confChange.getValue().isEmpty()) {
mapConf.remove(confChange.getKey());
} else {
mapConf.put(confChange.getKey(), confChange.getValue());
}
}
zkManager.safeSetData(confStorePath, serializeObject(mapConf), -1,
zkAcl, fencingNodePath);
}
pendingMutation = null;
}
@Override
public synchronized Configuration retrieve() {
byte[] serializedSchedConf;
try {
serializedSchedConf = zkManager.getData(confStorePath);
} catch (Exception e) {
LOG.error("Failed to retrieve configuration from zookeeper store", e);
return null;
}
try {
Map<String, String> map =
(HashMap<String, String>) deserializeObject(serializedSchedConf);
Configuration c = new Configuration();
for (Map.Entry<String, String> e : map.entrySet()) {
c.set(e.getKey(), e.getValue());
}
return c;
} catch (Exception e) {
LOG.error("Exception while deserializing scheduler configuration " +
"from store", e);
}
return null;
}
@Override
public List<LogMutation> getConfirmedConfHistory(long fromId) {
return null; // unimplemented
}
private static String getNodePath(String root, String nodeName) {
return ZKCuratorManager.getNodePath(root, nodeName);
}
private static byte[] serializeObject(Object o) throws Exception {
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(baos);) {
oos.writeObject(o);
oos.flush();
baos.flush();
return baos.toByteArray();
}
}
private static Object deserializeObject(byte[] bytes) throws Exception {
try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
ObjectInputStream ois = new ObjectInputStream(bais);) {
return ois.readObject();
}
}
}

View File

@ -0,0 +1,29 @@
/*
* 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.resourcemanager.scheduler.capacity.conf
* contains classes related to capacity scheduler configuration management.
*/
@InterfaceAudience.Private
@InterfaceStability.Unstable
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;

View File

@ -55,7 +55,8 @@ import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import com.google.common.base.Joiner;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@ -134,6 +135,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesManager;
@ -196,6 +200,7 @@ import org.apache.hadoop.yarn.webapp.BadRequestException;
import org.apache.hadoop.yarn.webapp.ForbiddenException;
import org.apache.hadoop.yarn.webapp.NotFoundException;
import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import com.google.common.annotations.VisibleForTesting;
import com.google.inject.Inject;
@ -2454,4 +2459,63 @@ public class RMWebServices extends WebServices implements RMWebServiceProtocol {
GetContainersRequest request) throws YarnException, IOException {
return rm.getClientRMService().getContainers(request).getContainerList();
}
@PUT
@Path("/scheduler-conf")
@Produces({ MediaType.APPLICATION_JSON + "; " + JettyUtils.UTF_8,
MediaType.APPLICATION_XML + "; " + JettyUtils.UTF_8 })
@Consumes({ MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML })
public synchronized Response updateSchedulerConfiguration(SchedConfUpdateInfo
mutationInfo, @Context HttpServletRequest hsr)
throws AuthorizationException, InterruptedException {
init();
UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
ApplicationACLsManager aclsManager = rm.getApplicationACLsManager();
if (aclsManager.areACLsEnabled()) {
if (callerUGI == null || !aclsManager.isAdmin(callerUGI)) {
String msg = "Only admins can carry out this operation.";
throw new ForbiddenException(msg);
}
}
ResourceScheduler scheduler = rm.getResourceScheduler();
if (scheduler instanceof MutableConfScheduler && ((MutableConfScheduler)
scheduler).isConfigurationMutable()) {
try {
callerUGI.doAs(new PrivilegedExceptionAction<Void>() {
@Override
public Void run() throws Exception {
MutableConfigurationProvider provider = ((MutableConfScheduler)
scheduler).getMutableConfProvider();
if (!provider.getAclMutationPolicy().isMutationAllowed(callerUGI,
mutationInfo)) {
throw new org.apache.hadoop.security.AccessControlException("User"
+ " is not admin of all modified queues.");
}
provider.logAndApplyMutation(callerUGI, mutationInfo);
try {
rm.getRMContext().getRMAdminService().refreshQueues();
} catch (IOException | YarnException e) {
provider.confirmPendingMutation(false);
throw e;
}
provider.confirmPendingMutation(true);
return null;
}
});
} catch (IOException e) {
LOG.error("Exception thrown when modifying configuration.", e);
return Response.status(Status.BAD_REQUEST).entity(e.getMessage())
.build();
}
return Response.status(Status.OK).entity("Configuration change " +
"successfully applied.").build();
} else {
return Response.status(Status.BAD_REQUEST)
.entity("Configuration change only supported by " +
"MutableConfScheduler.")
.build();
}
}
}

View File

@ -197,6 +197,29 @@ public class TestRMAdminService {
Assert.assertTrue(maxAppsAfter != maxAppsBefore);
}
@Test
public void testAdminRefreshQueuesWithMutableSchedulerConfiguration() {
configuration.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
YarnConfiguration.MEMORY_CONFIGURATION_STORE);
try {
rm = new MockRM(configuration);
rm.init(configuration);
rm.start();
} catch (Exception ex) {
fail("Should not get any exceptions");
}
try {
rm.adminService.refreshQueues(RefreshQueuesRequest.newInstance());
fail("Expected exception while calling refreshQueues when scheduler" +
" configuration is mutable.");
} catch (Exception ex) {
assertTrue(ex.getMessage().endsWith("Scheduler configuration is " +
"mutable. refreshQueues is not allowed in this scenario."));
}
}
@Test
public void testAdminRefreshNodesWithoutConfiguration()
throws IOException, YarnException {

View File

@ -0,0 +1,172 @@
/**
* 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.resourcemanager.scheduler;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.QueueACL;
import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.QueueAdminConfigurationMutationACLPolicy;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class TestConfigurationMutationACLPolicies {
private ConfigurationMutationACLPolicy policy;
private RMContext rmContext;
private MutableConfScheduler scheduler;
private static final UserGroupInformation GOOD_USER = UserGroupInformation
.createUserForTesting("goodUser", new String[] {});
private static final UserGroupInformation BAD_USER = UserGroupInformation
.createUserForTesting("badUser", new String[] {});
private static final Map<String, String> EMPTY_MAP =
Collections.<String, String>emptyMap();
@Before
public void setUp() throws IOException {
rmContext = mock(RMContext.class);
scheduler = mock(MutableConfScheduler.class);
when(rmContext.getScheduler()).thenReturn(scheduler);
mockQueue("a", scheduler);
mockQueue("b", scheduler);
mockQueue("b1", scheduler);
}
private void mockQueue(String queueName, MutableConfScheduler scheduler)
throws IOException {
QueueInfo queueInfo = QueueInfo.newInstance(queueName, 0, 0, 0, null, null,
null, null, null, null, false);
when(scheduler.getQueueInfo(eq(queueName), anyBoolean(), anyBoolean()))
.thenReturn(queueInfo);
Queue queue = mock(Queue.class);
when(queue.hasAccess(eq(QueueACL.ADMINISTER_QUEUE), eq(GOOD_USER)))
.thenReturn(true);
when(queue.hasAccess(eq(QueueACL.ADMINISTER_QUEUE), eq(BAD_USER)))
.thenReturn(false);
when(scheduler.getQueue(eq(queueName))).thenReturn(queue);
}
@Test
public void testDefaultPolicy() {
Configuration conf = new Configuration();
conf.set(YarnConfiguration.YARN_ADMIN_ACL, GOOD_USER.getShortUserName());
conf.setClass(YarnConfiguration.RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS,
DefaultConfigurationMutationACLPolicy.class,
ConfigurationMutationACLPolicy.class);
policy = ConfigurationMutationACLPolicyFactory.getPolicy(conf);
policy.init(conf, rmContext);
assertTrue(policy.isMutationAllowed(GOOD_USER, null));
assertFalse(policy.isMutationAllowed(BAD_USER, null));
}
@Test
public void testQueueAdminBasedPolicy() {
Configuration conf = new Configuration();
conf.setClass(YarnConfiguration.RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS,
QueueAdminConfigurationMutationACLPolicy.class,
ConfigurationMutationACLPolicy.class);
policy = ConfigurationMutationACLPolicyFactory.getPolicy(conf);
policy.init(conf, rmContext);
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
QueueConfigInfo configInfo = new QueueConfigInfo("root.a", EMPTY_MAP);
updateInfo.getUpdateQueueInfo().add(configInfo);
assertTrue(policy.isMutationAllowed(GOOD_USER, updateInfo));
assertFalse(policy.isMutationAllowed(BAD_USER, updateInfo));
}
@Test
public void testQueueAdminPolicyAddQueue() {
Configuration conf = new Configuration();
conf.setClass(YarnConfiguration.RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS,
QueueAdminConfigurationMutationACLPolicy.class,
ConfigurationMutationACLPolicy.class);
policy = ConfigurationMutationACLPolicyFactory.getPolicy(conf);
policy.init(conf, rmContext);
// Add root.b.b1. Should check ACL of root.b queue.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
QueueConfigInfo configInfo = new QueueConfigInfo("root.b.b2", EMPTY_MAP);
updateInfo.getAddQueueInfo().add(configInfo);
assertTrue(policy.isMutationAllowed(GOOD_USER, updateInfo));
assertFalse(policy.isMutationAllowed(BAD_USER, updateInfo));
}
@Test
public void testQueueAdminPolicyAddNestedQueue() {
Configuration conf = new Configuration();
conf.setClass(YarnConfiguration.RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS,
QueueAdminConfigurationMutationACLPolicy.class,
ConfigurationMutationACLPolicy.class);
policy = ConfigurationMutationACLPolicyFactory.getPolicy(conf);
policy.init(conf, rmContext);
// Add root.b.b1.b11. Should check ACL of root.b queue.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
QueueConfigInfo configInfo = new QueueConfigInfo("root.b.b2.b21", EMPTY_MAP);
updateInfo.getAddQueueInfo().add(configInfo);
assertTrue(policy.isMutationAllowed(GOOD_USER, updateInfo));
assertFalse(policy.isMutationAllowed(BAD_USER, updateInfo));
}
@Test
public void testQueueAdminPolicyRemoveQueue() {
Configuration conf = new Configuration();
conf.setClass(YarnConfiguration.RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS,
QueueAdminConfigurationMutationACLPolicy.class,
ConfigurationMutationACLPolicy.class);
policy = ConfigurationMutationACLPolicyFactory.getPolicy(conf);
policy.init(conf, rmContext);
// Remove root.b.b1.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
updateInfo.getRemoveQueueInfo().add("root.b.b1");
assertTrue(policy.isMutationAllowed(GOOD_USER, updateInfo));
assertFalse(policy.isMutationAllowed(BAD_USER, updateInfo));
}
@Test
public void testQueueAdminPolicyGlobal() {
Configuration conf = new Configuration();
conf.set(YarnConfiguration.YARN_ADMIN_ACL, GOOD_USER.getShortUserName());
conf.setClass(YarnConfiguration.RM_SCHEDULER_MUTATION_ACL_POLICY_CLASS,
QueueAdminConfigurationMutationACLPolicy.class,
ConfigurationMutationACLPolicy.class);
policy = ConfigurationMutationACLPolicyFactory.getPolicy(conf);
policy.init(conf, rmContext);
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
assertTrue(policy.isMutationAllowed(GOOD_USER, updateInfo));
assertTrue(policy.isMutationAllowed(BAD_USER, updateInfo));
updateInfo.getGlobalParams().put("globalKey", "globalValue");
assertTrue(policy.isMutationAllowed(GOOD_USER, updateInfo));
assertFalse(policy.isMutationAllowed(BAD_USER, updateInfo));
}
}

View File

@ -241,13 +241,13 @@ public class TestCapacityScheduler {
@Test (timeout = 30000)
public void testConfValidation() throws Exception {
ResourceScheduler scheduler = new CapacityScheduler();
CapacityScheduler scheduler = new CapacityScheduler();
scheduler.setRMContext(resourceManager.getRMContext());
Configuration conf = new YarnConfiguration();
conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 2048);
conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, 1024);
try {
scheduler.reinitialize(conf, mockContext);
scheduler.init(conf);
fail("Exception is expected because the min memory allocation is" +
" larger than the max memory allocation.");
} catch (YarnRuntimeException e) {

View File

@ -0,0 +1,92 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.resourcemanager.scheduler.capacity.conf;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
/**
* Base class for {@link YarnConfigurationStore} implementations.
*/
public abstract class ConfigurationStoreBaseTest {
protected YarnConfigurationStore confStore = createConfStore();
protected abstract YarnConfigurationStore createConfStore();
protected Configuration conf;
protected Configuration schedConf;
protected RMContext rmContext;
protected static final String TEST_USER = "testUser";
@Before
public void setUp() throws Exception {
this.conf = new Configuration();
this.schedConf = new Configuration(false);
}
@Test
public void testConfigurationUpdate() throws Exception {
schedConf.set("key1", "val1");
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val1", confStore.retrieve().get("key1"));
Map<String, String> update1 = new HashMap<>();
update1.put("keyUpdate1", "valUpdate1");
YarnConfigurationStore.LogMutation mutation1 =
new YarnConfigurationStore.LogMutation(update1, TEST_USER);
confStore.logMutation(mutation1);
confStore.confirmMutation(true);
assertEquals("valUpdate1", confStore.retrieve().get("keyUpdate1"));
Map<String, String> update2 = new HashMap<>();
update2.put("keyUpdate2", "valUpdate2");
YarnConfigurationStore.LogMutation mutation2 =
new YarnConfigurationStore.LogMutation(update2, TEST_USER);
confStore.logMutation(mutation2);
confStore.confirmMutation(false);
assertNull("Configuration should not be updated",
confStore.retrieve().get("keyUpdate2"));
confStore.close();
}
@Test
public void testNullConfigurationUpdate() throws Exception {
schedConf.set("key", "val");
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val", confStore.retrieve().get("key"));
Map<String, String> update = new HashMap<>();
update.put("key", null);
YarnConfigurationStore.LogMutation mutation =
new YarnConfigurationStore.LogMutation(update, TEST_USER);
confStore.logMutation(mutation);
confStore.confirmMutation(true);
assertNull(confStore.retrieve().get("key"));
confStore.close();
}
}

View File

@ -0,0 +1,30 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
/**
* Tests {@link InMemoryConfigurationStore}.
*/
public class TestInMemoryConfigurationStore extends ConfigurationStoreBaseTest {
@Override
protected YarnConfigurationStore createConfStore() {
return new InMemoryConfigurationStore();
}
}

View File

@ -0,0 +1,211 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
/**
* Tests {@link LeveldbConfigurationStore}.
*/
public class TestLeveldbConfigurationStore extends ConfigurationStoreBaseTest {
public static final Log LOG =
LogFactory.getLog(TestLeveldbConfigurationStore.class);
private static final File TEST_DIR = new File(
System.getProperty("test.build.data",
System.getProperty("java.io.tmpdir")),
TestLeveldbConfigurationStore.class.getName());
private ResourceManager rm;
@Before
public void setUp() throws Exception {
super.setUp();
FileUtil.fullyDelete(TEST_DIR);
conf.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
YarnConfiguration.LEVELDB_CONFIGURATION_STORE);
conf.set(YarnConfiguration.RM_SCHEDCONF_STORE_PATH, TEST_DIR.toString());
}
@Test
public void testVersioning() throws Exception {
confStore.initialize(conf, schedConf, rmContext);
assertNull(confStore.getConfStoreVersion());
confStore.checkVersion();
assertEquals(LeveldbConfigurationStore.CURRENT_VERSION_INFO,
confStore.getConfStoreVersion());
confStore.close();
}
@Test
public void testPersistConfiguration() throws Exception {
schedConf.set("key", "val");
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val", confStore.retrieve().get("key"));
confStore.close();
// Create a new configuration store, and check for old configuration
confStore = createConfStore();
schedConf.set("key", "badVal");
// Should ignore passed-in scheduler configuration.
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val", confStore.retrieve().get("key"));
confStore.close();
}
@Test
public void testPersistUpdatedConfiguration() throws Exception {
confStore.initialize(conf, schedConf, rmContext);
assertNull(confStore.retrieve().get("key"));
Map<String, String> update = new HashMap<>();
update.put("key", "val");
YarnConfigurationStore.LogMutation mutation =
new YarnConfigurationStore.LogMutation(update, TEST_USER);
confStore.logMutation(mutation);
confStore.confirmMutation(true);
assertEquals("val", confStore.retrieve().get("key"));
confStore.close();
// Create a new configuration store, and check for updated configuration
confStore = createConfStore();
schedConf.set("key", "badVal");
// Should ignore passed-in scheduler configuration.
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val", confStore.retrieve().get("key"));
confStore.close();
}
@Test
public void testMaxLogs() throws Exception {
conf.setLong(YarnConfiguration.RM_SCHEDCONF_MAX_LOGS, 2);
confStore.initialize(conf, schedConf, rmContext);
LinkedList<YarnConfigurationStore.LogMutation> logs =
((LeveldbConfigurationStore) confStore).getLogs();
assertEquals(0, logs.size());
Map<String, String> update1 = new HashMap<>();
update1.put("key1", "val1");
YarnConfigurationStore.LogMutation mutation =
new YarnConfigurationStore.LogMutation(update1, TEST_USER);
confStore.logMutation(mutation);
logs = ((LeveldbConfigurationStore) confStore).getLogs();
assertEquals(1, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
confStore.confirmMutation(true);
assertEquals(1, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
Map<String, String> update2 = new HashMap<>();
update2.put("key2", "val2");
mutation = new YarnConfigurationStore.LogMutation(update2, TEST_USER);
confStore.logMutation(mutation);
logs = ((LeveldbConfigurationStore) confStore).getLogs();
assertEquals(2, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
assertEquals("val2", logs.get(1).getUpdates().get("key2"));
confStore.confirmMutation(true);
assertEquals(2, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
assertEquals("val2", logs.get(1).getUpdates().get("key2"));
// Next update should purge first update from logs.
Map<String, String> update3 = new HashMap<>();
update3.put("key3", "val3");
mutation = new YarnConfigurationStore.LogMutation(update3, TEST_USER);
confStore.logMutation(mutation);
logs = ((LeveldbConfigurationStore) confStore).getLogs();
assertEquals(2, logs.size());
assertEquals("val2", logs.get(0).getUpdates().get("key2"));
assertEquals("val3", logs.get(1).getUpdates().get("key3"));
confStore.confirmMutation(true);
assertEquals(2, logs.size());
assertEquals("val2", logs.get(0).getUpdates().get("key2"));
assertEquals("val3", logs.get(1).getUpdates().get("key3"));
confStore.close();
}
/**
* When restarting, RM should read from current state of store, including
* any updates from the previous RM instance.
* @throws Exception
*/
@Test
public void testRestartReadsFromUpdatedStore() throws Exception {
ResourceManager rm1 = new MockRM(conf);
rm1.start();
assertNull(((MutableConfScheduler) rm1.getResourceScheduler())
.getConfiguration().get("key"));
// Update configuration on RM
SchedConfUpdateInfo schedConfUpdateInfo = new SchedConfUpdateInfo();
schedConfUpdateInfo.getGlobalParams().put("key", "val");
MutableConfigurationProvider confProvider = ((MutableConfScheduler)
rm1.getResourceScheduler()).getMutableConfProvider();
UserGroupInformation user = UserGroupInformation
.createUserForTesting(TEST_USER, new String[0]);
confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
rm1.getResourceScheduler().reinitialize(conf, rm1.getRMContext());
assertEquals("val", ((MutableConfScheduler) rm1.getResourceScheduler())
.getConfiguration().get("key"));
confProvider.confirmPendingMutation(true);
assertEquals("val", ((MutableCSConfigurationProvider) confProvider)
.getConfStore().retrieve().get("key"));
// Next update is not persisted, it should not be recovered
schedConfUpdateInfo.getGlobalParams().put("key", "badVal");
confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
rm1.close();
// Start RM2 and verifies it starts with updated configuration
ResourceManager rm2 = new MockRM(conf);
rm2.start();
assertEquals("val", ((MutableCSConfigurationProvider) (
(CapacityScheduler) rm2.getResourceScheduler())
.getMutableConfProvider()).getConfStore().retrieve().get("key"));
assertEquals("val", ((MutableConfScheduler) rm2.getResourceScheduler())
.getConfiguration().get("key"));
rm2.close();
}
@Override
public YarnConfigurationStore createConfStore() {
return new LeveldbConfigurationStore();
}
}

View File

@ -0,0 +1,102 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.AdminService;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import org.junit.Before;
import org.junit.Test;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* Tests {@link MutableCSConfigurationProvider}.
*/
public class TestMutableCSConfigurationProvider {
private MutableCSConfigurationProvider confProvider;
private RMContext rmContext;
private SchedConfUpdateInfo goodUpdate;
private SchedConfUpdateInfo badUpdate;
private CapacityScheduler cs;
private AdminService adminService;
private static final UserGroupInformation TEST_USER = UserGroupInformation
.createUserForTesting("testUser", new String[] {});
@Before
public void setUp() {
cs = mock(CapacityScheduler.class);
rmContext = mock(RMContext.class);
when(rmContext.getScheduler()).thenReturn(cs);
when(cs.getConfiguration()).thenReturn(
new CapacitySchedulerConfiguration());
adminService = mock(AdminService.class);
when(rmContext.getRMAdminService()).thenReturn(adminService);
confProvider = new MutableCSConfigurationProvider(rmContext);
goodUpdate = new SchedConfUpdateInfo();
Map<String, String> goodUpdateMap = new HashMap<>();
goodUpdateMap.put("goodKey", "goodVal");
QueueConfigInfo goodUpdateInfo = new
QueueConfigInfo("root.a", goodUpdateMap);
goodUpdate.getUpdateQueueInfo().add(goodUpdateInfo);
badUpdate = new SchedConfUpdateInfo();
Map<String, String> badUpdateMap = new HashMap<>();
badUpdateMap.put("badKey", "badVal");
QueueConfigInfo badUpdateInfo = new
QueueConfigInfo("root.a", badUpdateMap);
badUpdate.getUpdateQueueInfo().add(badUpdateInfo);
}
@Test
public void testInMemoryBackedProvider() throws Exception {
Configuration conf = new Configuration();
conf.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
YarnConfiguration.MEMORY_CONFIGURATION_STORE);
confProvider.init(conf);
assertNull(confProvider.loadConfiguration(conf)
.get("yarn.scheduler.capacity.root.a.goodKey"));
confProvider.logAndApplyMutation(TEST_USER, goodUpdate);
confProvider.confirmPendingMutation(true);
assertEquals("goodVal", confProvider.loadConfiguration(conf)
.get("yarn.scheduler.capacity.root.a.goodKey"));
assertNull(confProvider.loadConfiguration(conf).get(
"yarn.scheduler.capacity.root.a.badKey"));
confProvider.logAndApplyMutation(TEST_USER, badUpdate);
confProvider.confirmPendingMutation(false);
assertNull(confProvider.loadConfiguration(conf).get(
"yarn.scheduler.capacity.root.a.badKey"));
}
}

View File

@ -0,0 +1,412 @@
/**
* 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.resourcemanager.scheduler.capacity.conf;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.test.TestingServer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.ha.HAServiceProtocol;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.conf.HAUtil;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfigurationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* Tests {@link ZKConfigurationStore}.
*/
public class TestZKConfigurationStore extends ConfigurationStoreBaseTest {
public static final Log LOG =
LogFactory.getLog(TestZKConfigurationStore.class);
private static final int ZK_TIMEOUT_MS = 10000;
private TestingServer curatorTestingServer;
private CuratorFramework curatorFramework;
private ResourceManager rm;
public static TestingServer setupCuratorServer() throws Exception {
TestingServer curatorTestingServer = new TestingServer();
curatorTestingServer.start();
return curatorTestingServer;
}
public static CuratorFramework setupCuratorFramework(
TestingServer curatorTestingServer) throws Exception {
CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
.connectString(curatorTestingServer.getConnectString())
.retryPolicy(new RetryNTimes(100, 100))
.build();
curatorFramework.start();
return curatorFramework;
}
@Before
public void setUp() throws Exception {
super.setUp();
curatorTestingServer = setupCuratorServer();
curatorFramework = setupCuratorFramework(curatorTestingServer);
conf.set(CommonConfigurationKeys.ZK_ADDRESS,
curatorTestingServer.getConnectString());
rm = new MockRM(conf);
rm.start();
rmContext = rm.getRMContext();
}
@After
public void cleanup() throws IOException {
rm.stop();
curatorFramework.close();
curatorTestingServer.stop();
}
@Test
public void testVersioning() throws Exception {
confStore.initialize(conf, schedConf, rmContext);
assertNull(confStore.getConfStoreVersion());
confStore.checkVersion();
assertEquals(ZKConfigurationStore.CURRENT_VERSION_INFO,
confStore.getConfStoreVersion());
}
@Test
public void testPersistConfiguration() throws Exception {
schedConf.set("key", "val");
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val", confStore.retrieve().get("key"));
// Create a new configuration store, and check for old configuration
confStore = createConfStore();
schedConf.set("key", "badVal");
// Should ignore passed-in scheduler configuration.
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val", confStore.retrieve().get("key"));
}
@Test
public void testPersistUpdatedConfiguration() throws Exception {
confStore.initialize(conf, schedConf, rmContext);
assertNull(confStore.retrieve().get("key"));
Map<String, String> update = new HashMap<>();
update.put("key", "val");
YarnConfigurationStore.LogMutation mutation =
new YarnConfigurationStore.LogMutation(update, TEST_USER);
confStore.logMutation(mutation);
confStore.confirmMutation(true);
assertEquals("val", confStore.retrieve().get("key"));
// Create a new configuration store, and check for updated configuration
confStore = createConfStore();
schedConf.set("key", "badVal");
// Should ignore passed-in scheduler configuration.
confStore.initialize(conf, schedConf, rmContext);
assertEquals("val", confStore.retrieve().get("key"));
}
@Test
public void testMaxLogs() throws Exception {
conf.setLong(YarnConfiguration.RM_SCHEDCONF_MAX_LOGS, 2);
confStore.initialize(conf, schedConf, rmContext);
LinkedList<YarnConfigurationStore.LogMutation> logs =
((ZKConfigurationStore) confStore).getLogs();
assertEquals(0, logs.size());
Map<String, String> update1 = new HashMap<>();
update1.put("key1", "val1");
YarnConfigurationStore.LogMutation mutation =
new YarnConfigurationStore.LogMutation(update1, TEST_USER);
confStore.logMutation(mutation);
logs = ((ZKConfigurationStore) confStore).getLogs();
assertEquals(1, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
confStore.confirmMutation(true);
assertEquals(1, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
Map<String, String> update2 = new HashMap<>();
update2.put("key2", "val2");
mutation = new YarnConfigurationStore.LogMutation(update2, TEST_USER);
confStore.logMutation(mutation);
logs = ((ZKConfigurationStore) confStore).getLogs();
assertEquals(2, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
assertEquals("val2", logs.get(1).getUpdates().get("key2"));
confStore.confirmMutation(true);
assertEquals(2, logs.size());
assertEquals("val1", logs.get(0).getUpdates().get("key1"));
assertEquals("val2", logs.get(1).getUpdates().get("key2"));
// Next update should purge first update from logs.
Map<String, String> update3 = new HashMap<>();
update3.put("key3", "val3");
mutation = new YarnConfigurationStore.LogMutation(update3, TEST_USER);
confStore.logMutation(mutation);
logs = ((ZKConfigurationStore) confStore).getLogs();
assertEquals(2, logs.size());
assertEquals("val2", logs.get(0).getUpdates().get("key2"));
assertEquals("val3", logs.get(1).getUpdates().get("key3"));
confStore.confirmMutation(true);
assertEquals(2, logs.size());
assertEquals("val2", logs.get(0).getUpdates().get("key2"));
assertEquals("val3", logs.get(1).getUpdates().get("key3"));
}
public Configuration createRMHAConf(String rmIds, String rmId,
int adminPort) {
Configuration conf = new YarnConfiguration();
conf.setBoolean(YarnConfiguration.RM_HA_ENABLED, true);
conf.set(YarnConfiguration.RM_HA_IDS, rmIds);
conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
conf.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
YarnConfiguration.ZK_CONFIGURATION_STORE);
conf.set(YarnConfiguration.RM_STORE, ZKRMStateStore.class.getName());
conf.set(YarnConfiguration.RM_ZK_ADDRESS,
curatorTestingServer.getConnectString());
conf.set(YarnConfiguration.RM_HA_ID, rmId);
conf.set(YarnConfiguration.RM_WEBAPP_ADDRESS, "localhost:0");
conf.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false);
for (String rpcAddress :
YarnConfiguration.getServiceAddressConfKeys(conf)) {
for (String id : HAUtil.getRMHAIds(conf)) {
conf.set(HAUtil.addSuffix(rpcAddress, id), "localhost:0");
}
}
conf.set(HAUtil.addSuffix(YarnConfiguration.RM_ADMIN_ADDRESS, rmId),
"localhost:" + adminPort);
return conf;
}
/**
* When failing over, new active RM should read from current state of store,
* including any updates when the new active RM was in standby.
* @throws Exception
*/
@Test
public void testFailoverReadsFromUpdatedStore() throws Exception {
HAServiceProtocol.StateChangeRequestInfo req =
new HAServiceProtocol.StateChangeRequestInfo(
HAServiceProtocol.RequestSource.REQUEST_BY_USER);
Configuration conf1 = createRMHAConf("rm1,rm2", "rm1", 1234);
ResourceManager rm1 = new MockRM(conf1);
rm1.start();
rm1.getRMContext().getRMAdminService().transitionToActive(req);
assertEquals("RM with ZKStore didn't start",
Service.STATE.STARTED, rm1.getServiceState());
assertEquals("RM should be Active",
HAServiceProtocol.HAServiceState.ACTIVE,
rm1.getRMContext().getRMAdminService().getServiceStatus().getState());
assertNull(((MutableConfScheduler) rm1.getResourceScheduler())
.getConfiguration().get("key"));
Configuration conf2 = createRMHAConf("rm1,rm2", "rm2", 5678);
ResourceManager rm2 = new MockRM(conf2);
rm2.start();
assertEquals("RM should be Standby",
HAServiceProtocol.HAServiceState.STANDBY,
rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
// Update configuration on RM1
SchedConfUpdateInfo schedConfUpdateInfo = new SchedConfUpdateInfo();
schedConfUpdateInfo.getGlobalParams().put("key", "val");
MutableConfigurationProvider confProvider = ((MutableConfScheduler)
rm1.getResourceScheduler()).getMutableConfProvider();
UserGroupInformation user = UserGroupInformation
.createUserForTesting(TEST_USER, new String[0]);
confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
rm1.getResourceScheduler().reinitialize(conf1, rm1.getRMContext());
assertEquals("val", ((MutableConfScheduler) rm1.getResourceScheduler())
.getConfiguration().get("key"));
confProvider.confirmPendingMutation(true);
assertEquals("val", ((MutableCSConfigurationProvider) confProvider)
.getConfStore().retrieve().get("key"));
// Next update is not persisted, it should not be recovered
schedConfUpdateInfo.getGlobalParams().put("key", "badVal");
confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
// Start RM2 and verifies it starts with updated configuration
rm2.getRMContext().getRMAdminService().transitionToActive(req);
assertEquals("RM with ZKStore didn't start",
Service.STATE.STARTED, rm2.getServiceState());
assertEquals("RM should be Active",
HAServiceProtocol.HAServiceState.ACTIVE,
rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
for (int i = 0; i < ZK_TIMEOUT_MS / 50; i++) {
if (HAServiceProtocol.HAServiceState.ACTIVE ==
rm1.getRMContext().getRMAdminService().getServiceStatus()
.getState()) {
Thread.sleep(100);
}
}
assertEquals("RM should have been fenced",
HAServiceProtocol.HAServiceState.STANDBY,
rm1.getRMContext().getRMAdminService().getServiceStatus().getState());
assertEquals("RM should be Active",
HAServiceProtocol.HAServiceState.ACTIVE,
rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
assertEquals("val", ((MutableCSConfigurationProvider) (
(CapacityScheduler) rm2.getResourceScheduler())
.getMutableConfProvider()).getConfStore().retrieve().get("key"));
assertEquals("val", ((MutableConfScheduler) rm2.getResourceScheduler())
.getConfiguration().get("key"));
// Transition to standby will set RM's HA status and then reinitialize in
// a separate thread. Despite asserting for STANDBY state, it's
// possible for reinitialization to be unfinished. Wait here for it to
// finish, otherwise closing rm1 will close zkManager and the unfinished
// reinitialization will throw an exception.
Thread.sleep(10000);
rm1.close();
rm2.close();
}
/**
* When failing over, if RM1 stopped and removed a queue that RM2 has in
* memory, failing over to RM2 should not throw an exception.
* @throws Exception
*/
@Test
public void testFailoverAfterRemoveQueue() throws Exception {
HAServiceProtocol.StateChangeRequestInfo req =
new HAServiceProtocol.StateChangeRequestInfo(
HAServiceProtocol.RequestSource.REQUEST_BY_USER);
Configuration conf1 = createRMHAConf("rm1,rm2", "rm1", 1234);
ResourceManager rm1 = new MockRM(conf1);
rm1.start();
rm1.getRMContext().getRMAdminService().transitionToActive(req);
assertEquals("RM with ZKStore didn't start",
Service.STATE.STARTED, rm1.getServiceState());
assertEquals("RM should be Active",
HAServiceProtocol.HAServiceState.ACTIVE,
rm1.getRMContext().getRMAdminService().getServiceStatus().getState());
Configuration conf2 = createRMHAConf("rm1,rm2", "rm2", 5678);
ResourceManager rm2 = new MockRM(conf2);
rm2.start();
assertEquals("RM should be Standby",
HAServiceProtocol.HAServiceState.STANDBY,
rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
UserGroupInformation user = UserGroupInformation
.createUserForTesting(TEST_USER, new String[0]);
MutableConfigurationProvider confProvider = ((MutableConfScheduler)
rm1.getResourceScheduler()).getMutableConfProvider();
// Add root.a
SchedConfUpdateInfo schedConfUpdateInfo = new SchedConfUpdateInfo();
Map<String, String> addParams = new HashMap<>();
addParams.put("capacity", "100");
QueueConfigInfo addInfo = new QueueConfigInfo("root.a", addParams);
schedConfUpdateInfo.getAddQueueInfo().add(addInfo);
// Stop root.default
Map<String, String> stopParams = new HashMap<>();
stopParams.put("state", "STOPPED");
stopParams.put("capacity", "0");
QueueConfigInfo stopInfo = new QueueConfigInfo("root.default", stopParams);
schedConfUpdateInfo.getUpdateQueueInfo().add(stopInfo);
confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
rm1.getResourceScheduler().reinitialize(conf1, rm1.getRMContext());
confProvider.confirmPendingMutation(true);
assertTrue(Arrays.asList(((MutableConfScheduler) rm1.getResourceScheduler())
.getConfiguration().get("yarn.scheduler.capacity.root.queues").split
(",")).contains("a"));
// Remove root.default
schedConfUpdateInfo.getUpdateQueueInfo().clear();
schedConfUpdateInfo.getAddQueueInfo().clear();
schedConfUpdateInfo.getRemoveQueueInfo().add("root.default");
confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
rm1.getResourceScheduler().reinitialize(conf1, rm1.getRMContext());
confProvider.confirmPendingMutation(true);
assertEquals("a", ((MutableConfScheduler) rm1.getResourceScheduler())
.getConfiguration().get("yarn.scheduler.capacity.root.queues"));
// Start RM2 and verifies it starts with updated configuration
rm2.getRMContext().getRMAdminService().transitionToActive(req);
assertEquals("RM with ZKStore didn't start",
Service.STATE.STARTED, rm2.getServiceState());
assertEquals("RM should be Active",
HAServiceProtocol.HAServiceState.ACTIVE,
rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
for (int i = 0; i < ZK_TIMEOUT_MS / 50; i++) {
if (HAServiceProtocol.HAServiceState.ACTIVE ==
rm1.getRMContext().getRMAdminService().getServiceStatus()
.getState()) {
Thread.sleep(100);
}
}
assertEquals("RM should have been fenced",
HAServiceProtocol.HAServiceState.STANDBY,
rm1.getRMContext().getRMAdminService().getServiceStatus().getState());
assertEquals("RM should be Active",
HAServiceProtocol.HAServiceState.ACTIVE,
rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
assertEquals("a", ((MutableCSConfigurationProvider) (
(CapacityScheduler) rm2.getResourceScheduler())
.getMutableConfProvider()).getConfStore().retrieve()
.get("yarn.scheduler.capacity.root.queues"));
assertEquals("a", ((MutableConfScheduler) rm2.getResourceScheduler())
.getConfiguration().get("yarn.scheduler.capacity.root.queues"));
// Transition to standby will set RM's HA status and then reinitialize in
// a separate thread. Despite asserting for STANDBY state, it's
// possible for reinitialization to be unfinished. Wait here for it to
// finish, otherwise closing rm1 will close zkManager and the unfinished
// reinitialization will throw an exception.
Thread.sleep(10000);
rm1.close();
rm2.close();
}
@Override
public YarnConfigurationStore createConfStore() {
return new ZKConfigurationStore();
}
}

View File

@ -0,0 +1,507 @@
/**
* 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.resourcemanager.webapp;
import com.google.inject.Guice;
import com.google.inject.servlet.ServletModule;
import com.sun.jersey.api.client.ClientResponse;
import com.sun.jersey.api.client.WebResource;
import com.sun.jersey.guice.spi.container.servlet.GuiceContainer;
import com.sun.jersey.test.framework.WebAppDescriptor;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.QueueState;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
import org.apache.hadoop.yarn.webapp.GuiceServletConfig;
import org.apache.hadoop.yarn.webapp.JerseyTestBase;
import org.apache.hadoop.yarn.webapp.dao.QueueConfigInfo;
import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import org.apache.hadoop.yarn.webapp.util.YarnWebServiceUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response.Status;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
/**
* Test scheduler configuration mutation via REST API.
*/
public class TestRMWebServicesConfigurationMutation extends JerseyTestBase {
private static final File CONF_FILE = new File(new File("target",
"test-classes"), YarnConfiguration.CS_CONFIGURATION_FILE);
private static final File OLD_CONF_FILE = new File(new File("target",
"test-classes"), YarnConfiguration.CS_CONFIGURATION_FILE + ".tmp");
private static MockRM rm;
private static String userName;
private static CapacitySchedulerConfiguration csConf;
private static YarnConfiguration conf;
private static class WebServletModule extends ServletModule {
@Override
protected void configureServlets() {
bind(JAXBContextResolver.class);
bind(RMWebServices.class);
bind(GenericExceptionHandler.class);
try {
userName = UserGroupInformation.getCurrentUser().getShortUserName();
} catch (IOException ioe) {
throw new RuntimeException("Unable to get current user name "
+ ioe.getMessage(), ioe);
}
csConf = new CapacitySchedulerConfiguration(new Configuration(false),
false);
setupQueueConfiguration(csConf);
conf = new YarnConfiguration();
conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
ResourceScheduler.class);
conf.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
YarnConfiguration.MEMORY_CONFIGURATION_STORE);
conf.set(YarnConfiguration.YARN_ADMIN_ACL, userName);
try {
if (CONF_FILE.exists()) {
if (!CONF_FILE.renameTo(OLD_CONF_FILE)) {
throw new RuntimeException("Failed to rename conf file");
}
}
FileOutputStream out = new FileOutputStream(CONF_FILE);
csConf.writeXml(out);
out.close();
} catch (IOException e) {
throw new RuntimeException("Failed to write XML file", e);
}
rm = new MockRM(conf);
bind(ResourceManager.class).toInstance(rm);
serve("/*").with(GuiceContainer.class);
filter("/*").through(TestRMWebServicesAppsModification
.TestRMCustomAuthFilter.class);
}
}
@Override
@Before
public void setUp() throws Exception {
super.setUp();
GuiceServletConfig.setInjector(
Guice.createInjector(new WebServletModule()));
}
private static void setupQueueConfiguration(
CapacitySchedulerConfiguration config) {
config.setQueues(CapacitySchedulerConfiguration.ROOT,
new String[]{"a", "b", "c"});
final String a = CapacitySchedulerConfiguration.ROOT + ".a";
config.setCapacity(a, 25f);
config.setMaximumCapacity(a, 50f);
final String a1 = a + ".a1";
final String a2 = a + ".a2";
config.setQueues(a, new String[]{"a1", "a2"});
config.setCapacity(a1, 100f);
config.setCapacity(a2, 0f);
final String b = CapacitySchedulerConfiguration.ROOT + ".b";
config.setCapacity(b, 75f);
final String c = CapacitySchedulerConfiguration.ROOT + ".c";
config.setCapacity(c, 0f);
final String c1 = c + ".c1";
config.setQueues(c, new String[] {"c1"});
config.setCapacity(c1, 0f);
}
public TestRMWebServicesConfigurationMutation() {
super(new WebAppDescriptor.Builder(
"org.apache.hadoop.yarn.server.resourcemanager.webapp")
.contextListenerClass(GuiceServletConfig.class)
.filterClass(com.google.inject.servlet.GuiceFilter.class)
.contextPath("jersey-guice-filter").servletPath("/").build());
}
@Test
public void testAddNestedQueue() throws Exception {
WebResource r = resource();
ClientResponse response;
// Add parent queue root.d with two children d1 and d2.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
Map<String, String> d1Capacity = new HashMap<>();
d1Capacity.put(CapacitySchedulerConfiguration.CAPACITY, "25");
d1Capacity.put(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY, "25");
Map<String, String> nearEmptyCapacity = new HashMap<>();
nearEmptyCapacity.put(CapacitySchedulerConfiguration.CAPACITY, "1E-4");
nearEmptyCapacity.put(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY,
"1E-4");
Map<String, String> d2Capacity = new HashMap<>();
d2Capacity.put(CapacitySchedulerConfiguration.CAPACITY, "75");
d2Capacity.put(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY, "75");
QueueConfigInfo d1 = new QueueConfigInfo("root.d.d1", d1Capacity);
QueueConfigInfo d2 = new QueueConfigInfo("root.d.d2", d2Capacity);
QueueConfigInfo d = new QueueConfigInfo("root.d", nearEmptyCapacity);
updateInfo.getAddQueueInfo().add(d1);
updateInfo.getAddQueueInfo().add(d2);
updateInfo.getAddQueueInfo().add(d);
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(4, newCSConf.getQueues("root").length);
assertEquals(2, newCSConf.getQueues("root.d").length);
assertEquals(25.0f, newCSConf.getNonLabeledQueueCapacity("root.d.d1"),
0.01f);
assertEquals(75.0f, newCSConf.getNonLabeledQueueCapacity("root.d.d2"),
0.01f);
}
@Test
public void testAddWithUpdate() throws Exception {
WebResource r = resource();
ClientResponse response;
// Add root.d with capacity 25, reducing root.b capacity from 75 to 50.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
Map<String, String> dCapacity = new HashMap<>();
dCapacity.put(CapacitySchedulerConfiguration.CAPACITY, "25");
Map<String, String> bCapacity = new HashMap<>();
bCapacity.put(CapacitySchedulerConfiguration.CAPACITY, "50");
QueueConfigInfo d = new QueueConfigInfo("root.d", dCapacity);
QueueConfigInfo b = new QueueConfigInfo("root.b", bCapacity);
updateInfo.getAddQueueInfo().add(d);
updateInfo.getUpdateQueueInfo().add(b);
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(4, newCSConf.getQueues("root").length);
assertEquals(25.0f, newCSConf.getNonLabeledQueueCapacity("root.d"), 0.01f);
assertEquals(50.0f, newCSConf.getNonLabeledQueueCapacity("root.b"), 0.01f);
}
@Test
public void testRemoveQueue() throws Exception {
WebResource r = resource();
ClientResponse response;
stopQueue("root.a.a2");
// Remove root.a.a2
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
updateInfo.getRemoveQueueInfo().add("root.a.a2");
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(1, newCSConf.getQueues("root.a").length);
assertEquals("a1", newCSConf.getQueues("root.a")[0]);
}
@Test
public void testRemoveParentQueue() throws Exception {
WebResource r = resource();
ClientResponse response;
stopQueue("root.c", "root.c.c1");
// Remove root.c (parent queue)
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
updateInfo.getRemoveQueueInfo().add("root.c");
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(2, newCSConf.getQueues("root").length);
assertNull(newCSConf.getQueues("root.c"));
}
@Test
public void testRemoveParentQueueWithCapacity() throws Exception {
WebResource r = resource();
ClientResponse response;
stopQueue("root.a", "root.a.a1", "root.a.a2");
// Remove root.a (parent queue) with capacity 25
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
updateInfo.getRemoveQueueInfo().add("root.a");
// Set root.b capacity to 100
Map<String, String> bCapacity = new HashMap<>();
bCapacity.put(CapacitySchedulerConfiguration.CAPACITY, "100");
QueueConfigInfo b = new QueueConfigInfo("root.b", bCapacity);
updateInfo.getUpdateQueueInfo().add(b);
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(2, newCSConf.getQueues("root").length);
assertEquals(100.0f, newCSConf.getNonLabeledQueueCapacity("root.b"),
0.01f);
}
@Test
public void testRemoveMultipleQueues() throws Exception {
WebResource r = resource();
ClientResponse response;
stopQueue("root.b", "root.c", "root.c.c1");
// Remove root.b and root.c
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
updateInfo.getRemoveQueueInfo().add("root.b");
updateInfo.getRemoveQueueInfo().add("root.c");
Map<String, String> aCapacity = new HashMap<>();
aCapacity.put(CapacitySchedulerConfiguration.CAPACITY, "100");
aCapacity.put(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY, "100");
QueueConfigInfo configInfo = new QueueConfigInfo("root.a", aCapacity);
updateInfo.getUpdateQueueInfo().add(configInfo);
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(1, newCSConf.getQueues("root").length);
}
private void stopQueue(String... queuePaths) throws Exception {
WebResource r = resource();
ClientResponse response;
// Set state of queues to STOPPED.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
Map<String, String> stoppedParam = new HashMap<>();
stoppedParam.put(CapacitySchedulerConfiguration.STATE,
QueueState.STOPPED.toString());
for (String queue : queuePaths) {
QueueConfigInfo stoppedInfo = new QueueConfigInfo(queue, stoppedParam);
updateInfo.getUpdateQueueInfo().add(stoppedInfo);
}
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
for (String queue : queuePaths) {
assertEquals(QueueState.STOPPED, newCSConf.getState(queue));
}
}
@Test
public void testUpdateQueue() throws Exception {
WebResource r = resource();
ClientResponse response;
// Update config value.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
Map<String, String> updateParam = new HashMap<>();
updateParam.put(CapacitySchedulerConfiguration.MAXIMUM_AM_RESOURCE_SUFFIX,
"0.2");
QueueConfigInfo aUpdateInfo = new QueueConfigInfo("root.a", updateParam);
updateInfo.getUpdateQueueInfo().add(aUpdateInfo);
CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
assertEquals(CapacitySchedulerConfiguration
.DEFAULT_MAXIMUM_APPLICATIONMASTERS_RESOURCE_PERCENT,
cs.getConfiguration()
.getMaximumApplicationMasterResourcePerQueuePercent("root.a"),
0.001f);
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf = cs.getConfiguration();
assertEquals(0.2f, newCSConf
.getMaximumApplicationMasterResourcePerQueuePercent("root.a"), 0.001f);
// Remove config. Config value should be reverted to default.
updateParam.put(CapacitySchedulerConfiguration.MAXIMUM_AM_RESOURCE_SUFFIX,
null);
aUpdateInfo = new QueueConfigInfo("root.a", updateParam);
updateInfo.getUpdateQueueInfo().clear();
updateInfo.getUpdateQueueInfo().add(aUpdateInfo);
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
newCSConf = cs.getConfiguration();
assertEquals(CapacitySchedulerConfiguration
.DEFAULT_MAXIMUM_APPLICATIONMASTERS_RESOURCE_PERCENT, newCSConf
.getMaximumApplicationMasterResourcePerQueuePercent("root.a"),
0.001f);
}
@Test
public void testUpdateQueueCapacity() throws Exception {
WebResource r = resource();
ClientResponse response;
// Update root.a and root.b capacity to 50.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
Map<String, String> updateParam = new HashMap<>();
updateParam.put(CapacitySchedulerConfiguration.CAPACITY, "50");
QueueConfigInfo aUpdateInfo = new QueueConfigInfo("root.a", updateParam);
QueueConfigInfo bUpdateInfo = new QueueConfigInfo("root.b", updateParam);
updateInfo.getUpdateQueueInfo().add(aUpdateInfo);
updateInfo.getUpdateQueueInfo().add(bUpdateInfo);
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(50.0f, newCSConf.getNonLabeledQueueCapacity("root.a"), 0.01f);
assertEquals(50.0f, newCSConf.getNonLabeledQueueCapacity("root.b"), 0.01f);
}
@Test
public void testGlobalConfChange() throws Exception {
WebResource r = resource();
ClientResponse response;
// Set maximum-applications to 30000.
SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
updateInfo.getGlobalParams().put(CapacitySchedulerConfiguration.PREFIX +
"maximum-applications", "30000");
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
CapacitySchedulerConfiguration newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(30000, newCSConf.getMaximumSystemApplications());
updateInfo.getGlobalParams().put(CapacitySchedulerConfiguration.PREFIX +
"maximum-applications", null);
// Unset maximum-applications. Should be set to default.
response =
r.path("ws").path("v1").path("cluster")
.path("scheduler-conf").queryParam("user.name", userName)
.accept(MediaType.APPLICATION_JSON)
.entity(YarnWebServiceUtils.toJson(updateInfo,
SchedConfUpdateInfo.class), MediaType.APPLICATION_JSON)
.put(ClientResponse.class);
assertEquals(Status.OK.getStatusCode(), response.getStatus());
newCSConf =
((CapacityScheduler) rm.getResourceScheduler()).getConfiguration();
assertEquals(CapacitySchedulerConfiguration
.DEFAULT_MAXIMUM_SYSTEM_APPLICATIIONS,
newCSConf.getMaximumSystemApplications());
}
@Override
@After
public void tearDown() throws Exception {
if (rm != null) {
rm.stop();
}
CONF_FILE.delete();
if (!OLD_CONF_FILE.renameTo(CONF_FILE)) {
throw new RuntimeException("Failed to re-copy old configuration file");
}
super.tearDown();
}
}

View File

@ -295,9 +295,30 @@ The `ReservationSystem` is integrated with the `CapacityScheduler` queue hierach
Changing Queue Configuration
----------------------------
Changing queue properties and adding new queues is very simple. You need to edit **conf/capacity-scheduler.xml** and run *yarn rmadmin -refreshQueues*.
Changing queue/scheduler properties and adding/removing queues can be done in two ways, via file or via API. This behavior can be changed via `yarn.scheduler.configuration.store.class` in yarn-site.xml. Possible values are *file*, which allows modifying properties via file; *memory*, which allows modifying properties via API, but does not persist changes across restart; *leveldb*, which allows modifying properties via API and stores changes in leveldb backing store; and *zk*, which allows modifying properties via API and stores changes in zookeeper backing store. The default value is *file*.
### Changing queue configuration via file
To edit by file, you need to edit **conf/capacity-scheduler.xml** and run *yarn rmadmin -refreshQueues*.
$ vi $HADOOP_CONF_DIR/capacity-scheduler.xml
$ $HADOOP_YARN_HOME/bin/yarn rmadmin -refreshQueues
**Note:** Queues cannot be *deleted*, only addition of new queues is supported - the updated queue configuration should be a valid one i.e. queue-capacity at each *level* should be equal to 100%.
### Changing queue configuration via API
Editing by API uses a backing store for the scheduler configuration. To enable this, the following parameters can be configured in yarn-site.xml.
**Note:** This feature is in alpha phase and is subject to change.
| Property | Description |
|:---- |:---- |
| `yarn.scheduler.configuration.store.class` | The type of backing store to use, as described [above](CapacityScheduler.html#Changing_Queue_Configuration). |
| `yarn.scheduler.configuration.mutation.acl-policy.class` | An ACL policy can be configured to restrict which users can modify which queues. Default value is *org.apache.hadoop.yarn.server.resourcemanager.scheduler.DefaultConfigurationMutationACLPolicy*, which only allows YARN admins to make any configuration modifications. Another value is *org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.QueueAdminConfigurationMutationACLPolicy*, which only allows queue modifications if the caller is an admin of the queue. |
| `yarn.scheduler.configuration.store.max-logs` | Configuration changes are audit logged in the backing store, if using leveldb or zookeeper. This configuration controls the maximum number of audit logs to store, dropping the oldest logs when exceeded. Default is 1000. |
| `yarn.scheduler.configuration.leveldb-store.path` | The storage path of the configuration store when using leveldb. Default value is *${hadoop.tmp.dir}/yarn/system/confstore*. |
| `yarn.scheduler.configuration.leveldb-store.compaction-interval-secs` | The interval for compacting the configuration store in seconds, when using leveldb. Default value is 86400, or one day. |
| `yarn.scheduler.configuration.zk-store.parent-path` | The zookeeper root node path for configuration store related information, when using zookeeper. Default value is */confstore*. |
**Note:** When enabling scheduler configuration mutations via `yarn.scheduler.configuration.store.class`, *yarn rmadmin -refreshQueues* will be disabled, i.e. it will no longer be possible to update configuration via file.
See the [YARN Resource Manager REST API](ResourceManagerRest.html#Scheduler_Configuration_Mutation_API) for examples on how to change scheduler configuration via REST, and [YARN Commands Reference](YarnCommands.html#schedulerconf) for examples on how to change scheduler configuration via command line.

View File

@ -4430,3 +4430,191 @@ Response Body:
<remainingTimeInSeconds>90</remainingTimeInSeconds>
</timeout>
```
Scheduler Configuration Mutation API
--------------------------------
The scheduler configuration mutation API provides a way to modify scheduler/queue configuration and queue hierarchy.
Please note that this feature is currently in the alpha stage and is subject to change.
### URI
* http://rm-http-address:port/ws/v1/cluster/scheduler-conf
### HTTP Operations Supported
* PUT
### Elements of the *sched-conf* object
| Item | Data Type | Description |
|:---- |:---- |:---- |
| update-queue | object | A queue whose configurations should be updated |
| add-queue | object | A queue to add to the scheduler along with this queue's configurations |
| remove-queue | string | Full path name of a queue to remove |
| global-updates | map | Map of key value pairs to update scheduler's global configuration |
### PUT Request Examples
Put requests are used to modify the scheduler configuration. A successful mutation results in a 200 response. A malformed request or one which resulted in an invalid scheduler configuration results in a 400 response.
**Updating queue configuration(s)**
Request for updating queue configurations.
*Elements of the* update-queue *object*
| Item | Data Type | Description |
|:---- |:---- |:---- |
| queue-name | string | Full path name of the queue to update |
| params | map | A map of key value configuration pairs to update for this queue |
Assuming we are using the capacity scheduler and the current queue configuration is a single queue *root.default*, this example sets *root.default*'s maximum applications to 100 and its minimum user limit percent to 10.
HTTP Request:
```xml
Accept: application/xml
PUT http://rm-http-address:port/ws/v1/cluster/scheduler-conf
Content-Type: application/xml
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<sched-conf>
<update-queue>
<queue-name>root.default</queue-name>
<params>
<entry>
<key>maximum-applications</key>
<value>100</value>
</entry>
<entry>
<key>minimum-user-limit-percent</key>
<value>10</value>
</entry>
</params>
</update-queue>
</sched-conf>
```
Response Header:
HTTP/1.1 200 OK
Content-Type: application/xml
Transfer-Encoding: chunked
**Adding a queue**
Request for adding queues/updating queue configurations.
*Elements of the* add-queue *object*
| Item | Data Type | Description |
|:---- |:---- |:---- |
| queue-name | string | Full path name of the queue to add |
| params | map | A map of key value configuration pairs to set for this queue |
Assuming we are using the capacity scheduler and the current queue configuration is a single queue *root.default*, this example adds a queue *root.a* with capacity/maximum-capacity 10, and adjusts *root.default*'s capacity/maximum-capacity to 90. (More complex examples include adding a queue whose parent is also being added in the same request, or adding multiple sibling queues.)
HTTP Request:
```xml
Accept: application/xml
PUT http://rm-http-address:port/ws/v1/cluster/scheduler-conf
Content-Type: application/xml
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<sched-conf>
<add-queue>
<queue-name>root.a</queue-name>
<params>
<entry>
<key>capacity</key>
<value>10</value>
</entry>
<entry>
<key>maximum-capacity</key>
<value>10</value>
</entry>
</params>
</add-queue>
<update-queue>
<queue-name>root.default</queue-name>
<params>
<entry>
<key>capacity</key>
<value>90</value>
</entry>
<entry>
<key>maximum-capacity</key>
<value>90</value>
</entry>
</params>
</update-queue>
</sched-conf>
```
Response Header:
HTTP/1.1 200 OK
Content-Type: application/xml
Transfer-Encoding: chunked
**Removing queues**
Request for removing queues from the queue hierarchy.
Assuming we are using the capacity scheduler and the current queue configuration is three queues *root.default*, *root.a*, and *root.b*, this example removes both *root.a* and *root.b*. (More complex examples include removing a parent queue and its children.)
**Note:** Queues must be put into `STOPPED` state before they are deleted. Any updated queue configuration should be a valid one i.e. queue-capacity at each *level* should be equal to 100%.
HTTP Request:
```xml
Accept: application/xml
PUT http://rm-http-address:port/ws/v1/cluster/scheduler-conf
Content-Type: application/xml
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<sched-conf>
<remove-queue>root.a</remove-queue>
<remove-queue>root.b</remove-queue>
</sched-conf>
```
Response Header:
HTTP/1.1 200 OK
Content-Type: application/xml
Transfer-Encoding: chunked
**Updating global scheduler configurations**
Request for updating global scheduler configurations. Assuming we are using the capacity scheduler, this example enables queue mappings. For global configuration updates, the full configuration key must be specified.
HTTP Request:
```xml
Accept: application/xml
PUT http://rm-http-address:port/ws/v1/cluster/scheduler-conf
Content-Type: application/xml
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<sched-conf>
<global-updates>
<entry>
<key>yarn.scheduler.capacity.queue-mappings-override.enable</key>
<value>true</value>
</entry>
</global-updates>
</sched-conf>
```
Response Header:
HTTP/1.1 200 OK
Content-Type: application/xml
Transfer-Encoding: chunked

View File

@ -237,6 +237,19 @@ Usage:
Runs ResourceManager admin client
### schedulerconf
Usage: `yarn schedulerconf [options]`
| COMMAND\_OPTIONS | Description |
|:---- |:---- |
| -add <"queuePath1:key1=val1,key2=val2;queuePath2:key3=val3"> | Semicolon separated values of queues to add and their queue configurations. This example adds queue "queuePath1" (a full path name), which has queue configurations key1=val1 and key2=val2. It also adds queue "queuePath2", which has queue configuration key3=val3. |
| -remove <"queuePath1;queuePath2"> | Semicolon separated queues to remove. This example removes queuePath1 and queuePath2 queues (full path names). **Note:** Queues must be put into `STOPPED` state before they are deleted. |
| -update <"queuePath1:key1=val1,key2=val2;queuePath2:key3=val3"> | Semicolon separated values of queues whose configurations should be updated. This example sets key1=val1 and key2=val2 for queue configuration of queuePath1 (full path name), and sets key3=val3 for queue configuration of queuePath2. |
| -global <key1=val1,key2=val2> | Update scheduler global configurations. This example sets key1=val1 and key2=val2 for scheduler's global configuration. |
Updates scheduler configuration. Note, this feature is in alpha phase and is subject to change.
### scmadmin
Usage: `yarn scmadmin [options] `