You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ha...@apache.org on 2018/03/21 23:48:05 UTC
[29/50] [abbrv] hadoop git commit: HDFS-13215. RBF: Move Router to
its own module. Contributed by Wei Yan
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java
new file mode 100644
index 0000000..9788683
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java
@@ -0,0 +1,1005 @@
+/**
+ * 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.hdfs.server.federation;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_NAMENODES_KEY_PREFIX;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_NAMENODE_ID_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RPC_BIND_HOST_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_SERVICE_RPC_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_SERVICE_RPC_BIND_HOST_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICES;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICE_ID;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMENODES;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.addDirectory;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.waitNamenodeRegistered;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ADMIN_BIND_HOST_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_DEFAULT_NAMESERVICE;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_HEARTBEAT_INTERVAL_MS;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_HTTPS_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_HTTP_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_HTTP_BIND_HOST_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_MONITOR_NAMENODE;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_RPC_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_RPC_BIND_HOST_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.FEDERATION_FILE_RESOLVER_CLIENT_CLASS;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.FEDERATION_NAMENODE_RESOLVER_CLIENT_CLASS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.UnsupportedFileSystemException;
+import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
+import org.apache.hadoop.hdfs.DFSClient;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.MiniDFSCluster.NameNodeInfo;
+import org.apache.hadoop.hdfs.MiniDFSNNTopology;
+import org.apache.hadoop.hdfs.MiniDFSNNTopology.NNConf;
+import org.apache.hadoop.hdfs.MiniDFSNNTopology.NSConf;
+import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver;
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
+import org.apache.hadoop.hdfs.server.federation.resolver.FileSubclusterResolver;
+import org.apache.hadoop.hdfs.server.federation.resolver.NamenodeStatusReport;
+import org.apache.hadoop.hdfs.server.federation.router.Router;
+import org.apache.hadoop.hdfs.server.federation.router.RouterClient;
+import org.apache.hadoop.hdfs.server.namenode.FSImage;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
+import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.service.Service.STATE;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test utility to mimic a federated HDFS cluster with multiple routers.
+ */
+public class RouterDFSCluster {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(RouterDFSCluster.class);
+
+ public static final String TEST_STRING = "teststring";
+ public static final String TEST_DIR = "testdir";
+ public static final String TEST_FILE = "testfile";
+
+
+ /** Nameservices in the federated cluster. */
+ private List<String> nameservices;
+ /** Namenodes in the federated cluster. */
+ private List<NamenodeContext> namenodes;
+ /** Routers in the federated cluster. */
+ private List<RouterContext> routers;
+ /** If the Namenodes are in high availability.*/
+ private boolean highAvailability;
+ /** Number of datanodes per nameservice. */
+ private int numDatanodesPerNameservice = 2;
+
+ /** Mini cluster. */
+ private MiniDFSCluster cluster;
+
+ protected static final long DEFAULT_HEARTBEAT_INTERVAL_MS =
+ TimeUnit.SECONDS.toMillis(5);
+ protected static final long DEFAULT_CACHE_INTERVAL_MS =
+ TimeUnit.SECONDS.toMillis(5);
+ /** Heartbeat interval in milliseconds. */
+ private long heartbeatInterval;
+ /** Cache flush interval in milliseconds. */
+ private long cacheFlushInterval;
+
+ /** Router configuration overrides. */
+ private Configuration routerOverrides;
+ /** Namenode configuration overrides. */
+ private Configuration namenodeOverrides;
+
+
+ /**
+ * Router context.
+ */
+ public class RouterContext {
+ private Router router;
+ private FileContext fileContext;
+ private String nameserviceId;
+ private String namenodeId;
+ private int rpcPort;
+ private int httpPort;
+ private DFSClient client;
+ private Configuration conf;
+ private RouterClient adminClient;
+ private URI fileSystemUri;
+
+ public RouterContext(Configuration conf, String nsId, String nnId) {
+ this.conf = conf;
+ this.nameserviceId = nsId;
+ this.namenodeId = nnId;
+
+ this.router = new Router();
+ this.router.init(conf);
+ }
+
+ public Router getRouter() {
+ return this.router;
+ }
+
+ public String getNameserviceId() {
+ return this.nameserviceId;
+ }
+
+ public String getNamenodeId() {
+ return this.namenodeId;
+ }
+
+ public int getRpcPort() {
+ return this.rpcPort;
+ }
+
+ public int getHttpPort() {
+ return this.httpPort;
+ }
+
+ public FileContext getFileContext() {
+ return this.fileContext;
+ }
+
+ public void initRouter() throws URISyntaxException {
+ // Store the bound points for the router interfaces
+ InetSocketAddress rpcAddress = router.getRpcServerAddress();
+ if (rpcAddress != null) {
+ this.rpcPort = rpcAddress.getPort();
+ this.fileSystemUri =
+ URI.create("hdfs://" + NetUtils.getHostPortString(rpcAddress));
+ // Override the default FS to point to the router RPC
+ DistributedFileSystem.setDefaultUri(conf, fileSystemUri);
+ try {
+ this.fileContext = FileContext.getFileContext(conf);
+ } catch (UnsupportedFileSystemException e) {
+ this.fileContext = null;
+ }
+ }
+ InetSocketAddress httpAddress = router.getHttpServerAddress();
+ if (httpAddress != null) {
+ this.httpPort = httpAddress.getPort();
+ }
+ }
+
+ public FileSystem getFileSystem() throws IOException {
+ return DistributedFileSystem.get(conf);
+ }
+
+ public DFSClient getClient(UserGroupInformation user)
+ throws IOException, URISyntaxException, InterruptedException {
+
+ LOG.info("Connecting to router at {}", fileSystemUri);
+ return user.doAs(new PrivilegedExceptionAction<DFSClient>() {
+ @Override
+ public DFSClient run() throws IOException {
+ return new DFSClient(fileSystemUri, conf);
+ }
+ });
+ }
+
+ public RouterClient getAdminClient() throws IOException {
+ if (adminClient == null) {
+ InetSocketAddress routerSocket = router.getAdminServerAddress();
+ LOG.info("Connecting to router admin at {}", routerSocket);
+ adminClient = new RouterClient(routerSocket, conf);
+ }
+ return adminClient;
+ }
+
+ public DFSClient getClient() throws IOException, URISyntaxException {
+ if (client == null) {
+ LOG.info("Connecting to router at {}", fileSystemUri);
+ client = new DFSClient(fileSystemUri, conf);
+ }
+ return client;
+ }
+ }
+
+ /**
+ * Namenode context in the federated cluster.
+ */
+ public class NamenodeContext {
+ private Configuration conf;
+ private NameNode namenode;
+ private String nameserviceId;
+ private String namenodeId;
+ private FileContext fileContext;
+ private int rpcPort;
+ private int servicePort;
+ private int lifelinePort;
+ private int httpPort;
+ private URI fileSystemUri;
+ private int index;
+ private DFSClient client;
+
+ public NamenodeContext(
+ Configuration conf, String nsId, String nnId, int index) {
+ this.conf = conf;
+ this.nameserviceId = nsId;
+ this.namenodeId = nnId;
+ this.index = index;
+ }
+
+ public NameNode getNamenode() {
+ return this.namenode;
+ }
+
+ public String getNameserviceId() {
+ return this.nameserviceId;
+ }
+
+ public String getNamenodeId() {
+ return this.namenodeId;
+ }
+
+ public FileContext getFileContext() {
+ return this.fileContext;
+ }
+
+ public void setNamenode(NameNode nn) throws URISyntaxException {
+ this.namenode = nn;
+
+ // Store the bound ports and override the default FS with the local NN RPC
+ this.rpcPort = nn.getNameNodeAddress().getPort();
+ this.servicePort = nn.getServiceRpcAddress().getPort();
+ this.lifelinePort = nn.getServiceRpcAddress().getPort();
+ this.httpPort = nn.getHttpAddress().getPort();
+ this.fileSystemUri = new URI("hdfs://" + namenode.getHostAndPort());
+ DistributedFileSystem.setDefaultUri(this.conf, this.fileSystemUri);
+
+ try {
+ this.fileContext = FileContext.getFileContext(this.conf);
+ } catch (UnsupportedFileSystemException e) {
+ this.fileContext = null;
+ }
+ }
+
+ public String getRpcAddress() {
+ return namenode.getNameNodeAddress().getHostName() + ":" + rpcPort;
+ }
+
+ public String getServiceAddress() {
+ return namenode.getServiceRpcAddress().getHostName() + ":" + servicePort;
+ }
+
+ public String getLifelineAddress() {
+ return namenode.getServiceRpcAddress().getHostName() + ":" + lifelinePort;
+ }
+
+ public String getHttpAddress() {
+ return namenode.getHttpAddress().getHostName() + ":" + httpPort;
+ }
+
+ public FileSystem getFileSystem() throws IOException {
+ return DistributedFileSystem.get(conf);
+ }
+
+ public void resetClient() {
+ client = null;
+ }
+
+ public DFSClient getClient(UserGroupInformation user)
+ throws IOException, URISyntaxException, InterruptedException {
+
+ LOG.info("Connecting to namenode at {}", fileSystemUri);
+ return user.doAs(new PrivilegedExceptionAction<DFSClient>() {
+ @Override
+ public DFSClient run() throws IOException {
+ return new DFSClient(fileSystemUri, conf);
+ }
+ });
+ }
+
+ public DFSClient getClient() throws IOException, URISyntaxException {
+ if (client == null) {
+ LOG.info("Connecting to namenode at {}", fileSystemUri);
+ client = new DFSClient(fileSystemUri, conf);
+ }
+ return client;
+ }
+
+ public String getConfSuffix() {
+ String suffix = nameserviceId;
+ if (highAvailability) {
+ suffix += "." + namenodeId;
+ }
+ return suffix;
+ }
+ }
+
+ public RouterDFSCluster(boolean ha, int numNameservices, int numNamenodes,
+ long heartbeatInterval, long cacheFlushInterval) {
+ this.highAvailability = ha;
+ this.heartbeatInterval = heartbeatInterval;
+ this.cacheFlushInterval = cacheFlushInterval;
+ configureNameservices(numNameservices, numNamenodes);
+ }
+
+ public RouterDFSCluster(boolean ha, int numNameservices) {
+ this(ha, numNameservices, 2,
+ DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS);
+ }
+
+ public RouterDFSCluster(boolean ha, int numNameservices, int numNamenodes) {
+ this(ha, numNameservices, numNamenodes,
+ DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS);
+ }
+
+ /**
+ * Add configuration settings to override default Router settings.
+ *
+ * @param conf Router configuration overrides.
+ */
+ public void addRouterOverrides(Configuration conf) {
+ if (this.routerOverrides == null) {
+ this.routerOverrides = conf;
+ } else {
+ this.routerOverrides.addResource(conf);
+ }
+ }
+
+ /**
+ * Add configuration settings to override default Namenode settings.
+ *
+ * @param conf Namenode configuration overrides.
+ */
+ public void addNamenodeOverrides(Configuration conf) {
+ if (this.namenodeOverrides == null) {
+ this.namenodeOverrides = conf;
+ } else {
+ this.namenodeOverrides.addResource(conf);
+ }
+ }
+
+ /**
+ * Generate the configuration for a client.
+ *
+ * @param nsId Nameservice identifier.
+ * @return New namenode configuration.
+ */
+ public Configuration generateNamenodeConfiguration(String nsId) {
+ Configuration conf = new HdfsConfiguration();
+
+ conf.set(DFS_NAMESERVICES, getNameservicesKey());
+ conf.set(FS_DEFAULT_NAME_KEY, "hdfs://" + nsId);
+
+ for (String ns : nameservices) {
+ if (highAvailability) {
+ conf.set(
+ DFS_HA_NAMENODES_KEY_PREFIX + "." + ns,
+ NAMENODES[0] + "," + NAMENODES[1]);
+ }
+
+ for (NamenodeContext context : getNamenodes(ns)) {
+ String suffix = context.getConfSuffix();
+
+ conf.set(DFS_NAMENODE_RPC_ADDRESS_KEY + "." + suffix,
+ "127.0.0.1:" + context.rpcPort);
+ conf.set(DFS_NAMENODE_HTTP_ADDRESS_KEY + "." + suffix,
+ "127.0.0.1:" + context.httpPort);
+ conf.set(DFS_NAMENODE_RPC_BIND_HOST_KEY + "." + suffix,
+ "0.0.0.0");
+
+ // If the service port is enabled by default, we need to set them up
+ boolean servicePortEnabled = false;
+ if (servicePortEnabled) {
+ conf.set(DFS_NAMENODE_SERVICE_RPC_ADDRESS_KEY + "." + suffix,
+ "127.0.0.1:" + context.servicePort);
+ conf.set(DFS_NAMENODE_SERVICE_RPC_BIND_HOST_KEY + "." + suffix,
+ "0.0.0.0");
+ }
+ }
+ }
+
+ if (this.namenodeOverrides != null) {
+ conf.addResource(this.namenodeOverrides);
+ }
+ return conf;
+ }
+
+ /**
+ * Generate the configuration for a client.
+ *
+ * @return New configuration for a client.
+ */
+ public Configuration generateClientConfiguration() {
+ Configuration conf = new HdfsConfiguration(false);
+ String ns0 = getNameservices().get(0);
+ conf.addResource(generateNamenodeConfiguration(ns0));
+ return conf;
+ }
+
+ /**
+ * Generate the configuration for a Router.
+ *
+ * @param nsId Nameservice identifier.
+ * @param nnId Namenode identifier.
+ * @return New configuration for a Router.
+ */
+ public Configuration generateRouterConfiguration(String nsId, String nnId) {
+
+ Configuration conf = new HdfsConfiguration(false);
+ conf.addResource(generateNamenodeConfiguration(nsId));
+
+ conf.setInt(DFS_ROUTER_HANDLER_COUNT_KEY, 10);
+ conf.set(DFS_ROUTER_RPC_ADDRESS_KEY, "127.0.0.1:0");
+ conf.set(DFS_ROUTER_RPC_BIND_HOST_KEY, "0.0.0.0");
+
+ conf.set(DFS_ROUTER_ADMIN_ADDRESS_KEY, "127.0.0.1:0");
+ conf.set(DFS_ROUTER_ADMIN_BIND_HOST_KEY, "0.0.0.0");
+
+ conf.set(DFS_ROUTER_HTTP_ADDRESS_KEY, "127.0.0.1:0");
+ conf.set(DFS_ROUTER_HTTPS_ADDRESS_KEY, "127.0.0.1:0");
+ conf.set(DFS_ROUTER_HTTP_BIND_HOST_KEY, "0.0.0.0");
+
+ conf.set(DFS_ROUTER_DEFAULT_NAMESERVICE, nameservices.get(0));
+ conf.setLong(DFS_ROUTER_HEARTBEAT_INTERVAL_MS, heartbeatInterval);
+ conf.setLong(DFS_ROUTER_CACHE_TIME_TO_LIVE_MS, cacheFlushInterval);
+
+ // Use mock resolver classes
+ conf.setClass(FEDERATION_NAMENODE_RESOLVER_CLIENT_CLASS,
+ MockResolver.class, ActiveNamenodeResolver.class);
+ conf.setClass(FEDERATION_FILE_RESOLVER_CLIENT_CLASS,
+ MockResolver.class, FileSubclusterResolver.class);
+
+ // Set the nameservice ID for the default NN monitor
+ conf.set(DFS_NAMESERVICE_ID, nsId);
+ if (nnId != null) {
+ conf.set(DFS_HA_NAMENODE_ID_KEY, nnId);
+ }
+
+ // Namenodes to monitor
+ StringBuilder sb = new StringBuilder();
+ for (String ns : this.nameservices) {
+ for (NamenodeContext context : getNamenodes(ns)) {
+ String suffix = context.getConfSuffix();
+ if (sb.length() != 0) {
+ sb.append(",");
+ }
+ sb.append(suffix);
+ }
+ }
+ conf.set(DFS_ROUTER_MONITOR_NAMENODE, sb.toString());
+
+ // Add custom overrides if available
+ if (this.routerOverrides != null) {
+ for (Entry<String, String> entry : this.routerOverrides) {
+ String confKey = entry.getKey();
+ String confValue = entry.getValue();
+ conf.set(confKey, confValue);
+ }
+ }
+ return conf;
+ }
+
+ public void configureNameservices(int numNameservices, int numNamenodes) {
+ this.nameservices = new ArrayList<>();
+ this.namenodes = new ArrayList<>();
+
+ NamenodeContext context = null;
+ int nnIndex = 0;
+ for (int i=0; i<numNameservices; i++) {
+ String ns = "ns" + i;
+ this.nameservices.add("ns" + i);
+
+ Configuration nnConf = generateNamenodeConfiguration(ns);
+ if (!highAvailability) {
+ context = new NamenodeContext(nnConf, ns, null, nnIndex++);
+ this.namenodes.add(context);
+ } else {
+ for (int j=0; j<numNamenodes; j++) {
+ context = new NamenodeContext(nnConf, ns, NAMENODES[j], nnIndex++);
+ this.namenodes.add(context);
+ }
+ }
+ }
+ }
+
+ public void setNumDatanodesPerNameservice(int num) {
+ this.numDatanodesPerNameservice = num;
+ }
+
+ public String getNameservicesKey() {
+ StringBuilder sb = new StringBuilder();
+ for (String nsId : this.nameservices) {
+ if (sb.length() > 0) {
+ sb.append(",");
+ }
+ sb.append(nsId);
+ }
+ return sb.toString();
+ }
+
+ public String getRandomNameservice() {
+ Random r = new Random();
+ int randIndex = r.nextInt(nameservices.size());
+ return nameservices.get(randIndex);
+ }
+
+ public List<String> getNameservices() {
+ return nameservices;
+ }
+
+ public List<NamenodeContext> getNamenodes(String nameservice) {
+ List<NamenodeContext> nns = new ArrayList<>();
+ for (NamenodeContext c : namenodes) {
+ if (c.nameserviceId.equals(nameservice)) {
+ nns.add(c);
+ }
+ }
+ return nns;
+ }
+
+ public NamenodeContext getRandomNamenode() {
+ Random rand = new Random();
+ int i = rand.nextInt(this.namenodes.size());
+ return this.namenodes.get(i);
+ }
+
+ public List<NamenodeContext> getNamenodes() {
+ return this.namenodes;
+ }
+
+ public boolean isHighAvailability() {
+ return highAvailability;
+ }
+
+ public NamenodeContext getNamenode(String nameservice, String namenode) {
+ for (NamenodeContext c : this.namenodes) {
+ if (c.nameserviceId.equals(nameservice)) {
+ if (namenode == null || namenode.isEmpty() ||
+ c.namenodeId == null || c.namenodeId.isEmpty()) {
+ return c;
+ } else if (c.namenodeId.equals(namenode)) {
+ return c;
+ }
+ }
+ }
+ return null;
+ }
+
+ public List<RouterContext> getRouters(String nameservice) {
+ List<RouterContext> nns = new ArrayList<>();
+ for (RouterContext c : routers) {
+ if (c.nameserviceId.equals(nameservice)) {
+ nns.add(c);
+ }
+ }
+ return nns;
+ }
+
+ public RouterContext getRouterContext(String nsId, String nnId) {
+ for (RouterContext c : routers) {
+ if (nnId == null) {
+ return c;
+ }
+ if (c.namenodeId.equals(nnId) &&
+ c.nameserviceId.equals(nsId)) {
+ return c;
+ }
+ }
+ return null;
+ }
+
+ public RouterContext getRandomRouter() {
+ Random rand = new Random();
+ return routers.get(rand.nextInt(routers.size()));
+ }
+
+ public List<RouterContext> getRouters() {
+ return routers;
+ }
+
+ public RouterContext buildRouter(String nsId, String nnId)
+ throws URISyntaxException, IOException {
+ Configuration config = generateRouterConfiguration(nsId, nnId);
+ RouterContext rc = new RouterContext(config, nsId, nnId);
+ return rc;
+ }
+
+ public void startCluster() {
+ startCluster(null);
+ }
+
+ public void startCluster(Configuration overrideConf) {
+ try {
+ MiniDFSNNTopology topology = new MiniDFSNNTopology();
+ for (String ns : nameservices) {
+ NSConf conf = new MiniDFSNNTopology.NSConf(ns);
+ if (highAvailability) {
+ for (int i=0; i<namenodes.size()/nameservices.size(); i++) {
+ NNConf nnConf = new MiniDFSNNTopology.NNConf("nn" + i);
+ conf.addNN(nnConf);
+ }
+ } else {
+ NNConf nnConf = new MiniDFSNNTopology.NNConf(null);
+ conf.addNN(nnConf);
+ }
+ topology.addNameservice(conf);
+ }
+ topology.setFederation(true);
+
+ // Start mini DFS cluster
+ String ns0 = nameservices.get(0);
+ Configuration nnConf = generateNamenodeConfiguration(ns0);
+ if (overrideConf != null) {
+ nnConf.addResource(overrideConf);
+ }
+ cluster = new MiniDFSCluster.Builder(nnConf)
+ .numDataNodes(nameservices.size() * numDatanodesPerNameservice)
+ .nnTopology(topology)
+ .build();
+ cluster.waitActive();
+
+ // Store NN pointers
+ for (int i = 0; i < namenodes.size(); i++) {
+ NameNode nn = cluster.getNameNode(i);
+ namenodes.get(i).setNamenode(nn);
+ }
+
+ } catch (Exception e) {
+ LOG.error("Cannot start Router DFS cluster: {}", e.getMessage(), e);
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+ }
+
+ public void startRouters()
+ throws InterruptedException, URISyntaxException, IOException {
+
+ // Create one router per nameservice
+ this.routers = new ArrayList<>();
+ for (String ns : this.nameservices) {
+ for (NamenodeContext context : getNamenodes(ns)) {
+ RouterContext router = buildRouter(ns, context.namenodeId);
+ this.routers.add(router);
+ }
+ }
+
+ // Start all routers
+ for (RouterContext router : this.routers) {
+ router.router.start();
+ }
+
+ // Wait until all routers are active and record their ports
+ for (RouterContext router : this.routers) {
+ waitActive(router);
+ router.initRouter();
+ }
+ }
+
+ public void waitActive(NamenodeContext nn) throws IOException {
+ cluster.waitActive(nn.index);
+ }
+
+ public void waitActive(RouterContext router)
+ throws InterruptedException {
+ for (int loopCount = 0; loopCount < 20; loopCount++) {
+ // Validate connection of routers to NNs
+ if (router.router.getServiceState() == STATE.STARTED) {
+ return;
+ }
+ Thread.sleep(1000);
+ }
+ fail("Timeout waiting for " + router.router + " to activate");
+ }
+
+ public void registerNamenodes() throws IOException {
+ for (RouterContext r : this.routers) {
+ ActiveNamenodeResolver resolver = r.router.getNamenodeResolver();
+ for (NamenodeContext nn : this.namenodes) {
+ // Generate a report
+ NamenodeStatusReport report = new NamenodeStatusReport(
+ nn.nameserviceId, nn.namenodeId,
+ nn.getRpcAddress(), nn.getServiceAddress(),
+ nn.getLifelineAddress(), nn.getHttpAddress());
+ FSImage fsImage = nn.namenode.getNamesystem().getFSImage();
+ NamespaceInfo nsInfo = fsImage.getStorage().getNamespaceInfo();
+ report.setNamespaceInfo(nsInfo);
+
+ // Determine HA state from nn public state string
+ String nnState = nn.namenode.getState();
+ HAServiceState haState = HAServiceState.ACTIVE;
+ for (HAServiceState state : HAServiceState.values()) {
+ if (nnState.equalsIgnoreCase(state.name())) {
+ haState = state;
+ break;
+ }
+ }
+ report.setHAServiceState(haState);
+
+ // Register with the resolver
+ resolver.registerNamenode(report);
+ }
+ }
+ }
+
+ public void waitNamenodeRegistration() throws Exception {
+ for (RouterContext r : this.routers) {
+ Router router = r.router;
+ for (NamenodeContext nn : this.namenodes) {
+ ActiveNamenodeResolver nnResolver = router.getNamenodeResolver();
+ waitNamenodeRegistered(
+ nnResolver, nn.nameserviceId, nn.namenodeId, null);
+ }
+ }
+ }
+
+ public void waitRouterRegistrationQuorum(RouterContext router,
+ FederationNamenodeServiceState state, String nsId, String nnId)
+ throws Exception {
+ LOG.info("Waiting for NN {} {} to transition to {}", nsId, nnId, state);
+ ActiveNamenodeResolver nnResolver = router.router.getNamenodeResolver();
+ waitNamenodeRegistered(nnResolver, nsId, nnId, state);
+ }
+
+ /**
+ * Get the federated path for a nameservice.
+ * @param nsId Nameservice identifier.
+ * @return Path in the Router.
+ */
+ public String getFederatedPathForNS(String nsId) {
+ return "/" + nsId;
+ }
+
+ /**
+ * Get the namenode path for a nameservice.
+ * @param nsId Nameservice identifier.
+ * @return Path in the Namenode.
+ */
+ public String getNamenodePathForNS(String nsId) {
+ return "/target-" + nsId;
+ }
+
+ /**
+ * Get the federated test directory for a nameservice.
+ * @param nsId Nameservice identifier.
+ * @return Example:
+ * <ul>
+ * <li>/ns0/testdir which maps to ns0->/target-ns0/testdir
+ * </ul>
+ */
+ public String getFederatedTestDirectoryForNS(String nsId) {
+ return getFederatedPathForNS(nsId) + "/" + TEST_DIR;
+ }
+
+ /**
+ * Get the namenode test directory for a nameservice.
+ * @param nsId Nameservice identifier.
+ * @return example:
+ * <ul>
+ * <li>/target-ns0/testdir
+ * </ul>
+ */
+ public String getNamenodeTestDirectoryForNS(String nsId) {
+ return getNamenodePathForNS(nsId) + "/" + TEST_DIR;
+ }
+
+ /**
+ * Get the federated test file for a nameservice.
+ * @param nsId Nameservice identifier.
+ * @return example:
+ * <ul>
+ * <li>/ns0/testfile which maps to ns0->/target-ns0/testfile
+ * </ul>
+ */
+ public String getFederatedTestFileForNS(String nsId) {
+ return getFederatedPathForNS(nsId) + "/" + TEST_FILE;
+ }
+
+ /**
+ * Get the namenode test file for a nameservice.
+ * @param nsId Nameservice identifier.
+ * @return example:
+ * <ul>
+ * <li>/target-ns0/testfile
+ * </ul>
+ */
+ public String getNamenodeTestFileForNS(String nsId) {
+ return getNamenodePathForNS(nsId) + "/" + TEST_FILE;
+ }
+
+ /**
+ * Switch a namenode in a nameservice to be the active.
+ * @param nsId Nameservice identifier.
+ * @param nnId Namenode identifier.
+ */
+ public void switchToActive(String nsId, String nnId) {
+ try {
+ int total = cluster.getNumNameNodes();
+ NameNodeInfo[] nns = cluster.getNameNodeInfos();
+ for (int i = 0; i < total; i++) {
+ NameNodeInfo nn = nns[i];
+ if (nn.getNameserviceId().equals(nsId) &&
+ nn.getNamenodeId().equals(nnId)) {
+ cluster.transitionToActive(i);
+ }
+ }
+ } catch (Throwable e) {
+ LOG.error("Cannot transition to active", e);
+ }
+ }
+
+ /**
+ * Switch a namenode in a nameservice to be in standby.
+ * @param nsId Nameservice identifier.
+ * @param nnId Namenode identifier.
+ */
+ public void switchToStandby(String nsId, String nnId) {
+ try {
+ int total = cluster.getNumNameNodes();
+ NameNodeInfo[] nns = cluster.getNameNodeInfos();
+ for (int i = 0; i < total; i++) {
+ NameNodeInfo nn = nns[i];
+ if (nn.getNameserviceId().equals(nsId) &&
+ nn.getNamenodeId().equals(nnId)) {
+ cluster.transitionToStandby(i);
+ }
+ }
+ } catch (Throwable e) {
+ LOG.error("Cannot transition to standby", e);
+ }
+ }
+
+ /**
+ * Stop the federated HDFS cluster.
+ */
+ public void shutdown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ if (routers != null) {
+ for (RouterContext context : routers) {
+ stopRouter(context);
+ }
+ }
+ }
+
+ /**
+ * Stop a router.
+ * @param router Router context.
+ */
+ public void stopRouter(RouterContext router) {
+ try {
+ router.router.shutDown();
+
+ int loopCount = 0;
+ while (router.router.getServiceState() != STATE.STOPPED) {
+ loopCount++;
+ Thread.sleep(1000);
+ if (loopCount > 20) {
+ LOG.error("Cannot shutdown router {}", router.rpcPort);
+ break;
+ }
+ }
+ } catch (InterruptedException e) {
+ }
+ }
+
+ /////////////////////////////////////////////////////////////////////////////
+ // Namespace Test Fixtures
+ /////////////////////////////////////////////////////////////////////////////
+
+ /**
+ * Creates test directories via the namenode.
+ * 1) /target-ns0/testfile
+ * 2) /target-ns1/testfile
+ * @throws IOException
+ */
+ public void createTestDirectoriesNamenode() throws IOException {
+ // Add a test dir to each NS and verify
+ for (String ns : getNameservices()) {
+ NamenodeContext context = getNamenode(ns, null);
+ if (!createTestDirectoriesNamenode(context)) {
+ throw new IOException("Cannot create test directory for ns " + ns);
+ }
+ }
+ }
+
+ public boolean createTestDirectoriesNamenode(NamenodeContext nn)
+ throws IOException {
+ FileSystem fs = nn.getFileSystem();
+ String testDir = getNamenodeTestDirectoryForNS(nn.nameserviceId);
+ return addDirectory(fs, testDir);
+ }
+
+ public void deleteAllFiles() throws IOException {
+ // Delete all files via the NNs and verify
+ for (NamenodeContext context : getNamenodes()) {
+ FileSystem fs = context.getFileSystem();
+ FileStatus[] status = fs.listStatus(new Path("/"));
+ for (int i = 0; i <status.length; i++) {
+ Path p = status[i].getPath();
+ fs.delete(p, true);
+ }
+ status = fs.listStatus(new Path("/"));
+ assertEquals(status.length, 0);
+ }
+ }
+
+ /////////////////////////////////////////////////////////////////////////////
+ // MockRouterResolver Test Fixtures
+ /////////////////////////////////////////////////////////////////////////////
+
+ /**
+ * <ul>
+ * <li>/ -> [ns0->/].
+ * <li>/nso -> ns0->/target-ns0.
+ * <li>/ns1 -> ns1->/target-ns1.
+ * </ul>
+ */
+ public void installMockLocations() {
+ for (RouterContext r : routers) {
+ MockResolver resolver =
+ (MockResolver) r.router.getSubclusterResolver();
+ // create table entries
+ for (String nsId : nameservices) {
+ // Direct path
+ String routerPath = getFederatedPathForNS(nsId);
+ String nnPath = getNamenodePathForNS(nsId);
+ resolver.addLocation(routerPath, nsId, nnPath);
+ }
+
+ // Root path points to both first nameservice
+ String ns0 = nameservices.get(0);
+ resolver.addLocation("/", ns0, "/");
+ }
+ }
+
+ public MiniDFSCluster getCluster() {
+ return cluster;
+ }
+
+ /**
+ * Wait until the federated cluster is up and ready.
+ * @throws IOException If we cannot wait for the cluster to be up.
+ */
+ public void waitClusterUp() throws IOException {
+ cluster.waitClusterUp();
+ registerNamenodes();
+ try {
+ waitNamenodeRegistration();
+ } catch (Exception e) {
+ throw new IOException("Cannot wait for the namenodes", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/StateStoreDFSCluster.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/StateStoreDFSCluster.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/StateStoreDFSCluster.java
new file mode 100644
index 0000000..aa1906f
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/StateStoreDFSCluster.java
@@ -0,0 +1,148 @@
+/**
+ * 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.hdfs.server.federation;
+
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.createMockRegistrationForNamenode;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.getStateStoreConfiguration;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.synchronizeRecords;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver;
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
+import org.apache.hadoop.hdfs.server.federation.resolver.FileSubclusterResolver;
+import org.apache.hadoop.hdfs.server.federation.resolver.MembershipNamenodeResolver;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver;
+import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+
+/**
+ * Test utility to mimic a federated HDFS cluster with a router and a state
+ * store.
+ */
+public class StateStoreDFSCluster extends RouterDFSCluster {
+
+ private static final Class<?> DEFAULT_FILE_RESOLVER =
+ MountTableResolver.class;
+ private static final Class<?> DEFAULT_NAMENODE_RESOLVER =
+ MembershipNamenodeResolver.class;
+
+ public StateStoreDFSCluster(boolean ha, int numNameservices, int numNamenodes,
+ long heartbeatInterval, long cacheFlushInterval)
+ throws IOException, InterruptedException {
+ this(ha, numNameservices, numNamenodes, heartbeatInterval,
+ cacheFlushInterval, DEFAULT_FILE_RESOLVER);
+ }
+
+ public StateStoreDFSCluster(boolean ha, int numNameservices, int numNamenodes,
+ long heartbeatInterval, long cacheFlushInterval, Class<?> fileResolver)
+ throws IOException, InterruptedException {
+ super(ha, numNameservices, numNamenodes, heartbeatInterval,
+ cacheFlushInterval);
+
+ // Attach state store and resolvers to router
+ Configuration stateStoreConfig = getStateStoreConfiguration();
+ // Use state store backed resolvers
+ stateStoreConfig.setClass(
+ RBFConfigKeys.FEDERATION_NAMENODE_RESOLVER_CLIENT_CLASS,
+ DEFAULT_NAMENODE_RESOLVER, ActiveNamenodeResolver.class);
+ stateStoreConfig.setClass(
+ RBFConfigKeys.FEDERATION_FILE_RESOLVER_CLIENT_CLASS,
+ fileResolver, FileSubclusterResolver.class);
+ this.addRouterOverrides(stateStoreConfig);
+ }
+
+ public StateStoreDFSCluster(boolean ha, int numNameservices,
+ Class<?> fileResolver) throws IOException, InterruptedException {
+ this(ha, numNameservices, 2,
+ DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS, fileResolver);
+ }
+
+ public StateStoreDFSCluster(boolean ha, int numNameservices)
+ throws IOException, InterruptedException {
+ this(ha, numNameservices, 2,
+ DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS);
+ }
+
+ public StateStoreDFSCluster(boolean ha, int numNameservices,
+ int numNamnodes) throws IOException, InterruptedException {
+ this(ha, numNameservices, numNamnodes,
+ DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS);
+ }
+
+ /////////////////////////////////////////////////////////////////////////////
+ // State Store Test Fixtures
+ /////////////////////////////////////////////////////////////////////////////
+
+ /**
+ * Adds test fixtures for NN registation for each NN nameservice -> NS
+ * namenode -> NN rpcAddress -> 0.0.0.0:0 webAddress -> 0.0.0.0:0 state ->
+ * STANDBY safeMode -> false blockPool -> test.
+ *
+ * @param stateStore State Store.
+ * @throws IOException If it cannot register.
+ */
+ public void createTestRegistration(StateStoreService stateStore)
+ throws IOException {
+ List<MembershipState> entries = new ArrayList<MembershipState>();
+ for (NamenodeContext nn : this.getNamenodes()) {
+ MembershipState entry = createMockRegistrationForNamenode(
+ nn.getNameserviceId(), nn.getNamenodeId(),
+ FederationNamenodeServiceState.STANDBY);
+ entries.add(entry);
+ }
+ synchronizeRecords(
+ stateStore, entries, MembershipState.class);
+ }
+
+ public void createTestMountTable(StateStoreService stateStore)
+ throws IOException {
+ List<MountTable> mounts = generateMockMountTable();
+ synchronizeRecords(stateStore, mounts, MountTable.class);
+ stateStore.refreshCaches();
+ }
+
+ public List<MountTable> generateMockMountTable() throws IOException {
+ // create table entries
+ List<MountTable> entries = new ArrayList<>();
+ for (String ns : this.getNameservices()) {
+ Map<String, String> destMap = new HashMap<>();
+ destMap.put(ns, getNamenodePathForNS(ns));
+
+ // Direct path
+ String fedPath = getFederatedPathForNS(ns);
+ MountTable entry = MountTable.newInstance(fedPath, destMap);
+ entries.add(entry);
+ }
+
+ // Root path goes to nameservice 1
+ Map<String, String> destMap = new HashMap<>();
+ String ns0 = this.getNameservices().get(0);
+ destMap.put(ns0, "/");
+ MountTable entry = MountTable.newInstance("/", destMap);
+ entries.add(entry);
+ return entries;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestFederationMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestFederationMetrics.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestFederationMetrics.java
new file mode 100644
index 0000000..94799f3
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestFederationMetrics.java
@@ -0,0 +1,293 @@
+/**
+ * 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.hdfs.server.federation.metrics;
+
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.getBean;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+import javax.management.MalformedObjectNameException;
+
+import org.apache.commons.collections.ListUtils;
+import org.apache.hadoop.hdfs.server.federation.router.Router;
+import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState;
+import org.apache.hadoop.hdfs.server.federation.store.records.MembershipStats;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
+import org.apache.hadoop.hdfs.server.federation.store.records.StateStoreVersion;
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+import org.junit.Test;
+
+/**
+ * Test the JMX interface for the {@link Router}.
+ */
+public class TestFederationMetrics extends TestMetricsBase {
+
+ public static final String FEDERATION_BEAN =
+ "Hadoop:service=Router,name=FederationState";
+ public static final String STATE_STORE_BEAN =
+ "Hadoop:service=Router,name=StateStore";
+ public static final String RPC_BEAN =
+ "Hadoop:service=Router,name=FederationRPC";
+
+ @Test
+ public void testClusterStatsJMX()
+ throws MalformedObjectNameException, IOException {
+
+ FederationMBean bean = getBean(FEDERATION_BEAN, FederationMBean.class);
+ validateClusterStatsBean(bean);
+ }
+
+ @Test
+ public void testClusterStatsDataSource() throws IOException {
+ FederationMetrics metrics = getRouter().getMetrics();
+ validateClusterStatsBean(metrics);
+ }
+
+ @Test
+ public void testMountTableStatsDataSource()
+ throws IOException, JSONException {
+
+ FederationMetrics metrics = getRouter().getMetrics();
+ String jsonString = metrics.getMountTable();
+ JSONArray jsonArray = new JSONArray(jsonString);
+ assertEquals(jsonArray.length(), getMockMountTable().size());
+
+ int match = 0;
+ for (int i = 0; i < jsonArray.length(); i++) {
+ JSONObject json = jsonArray.getJSONObject(i);
+ String src = json.getString("sourcePath");
+
+ for (MountTable entry : getMockMountTable()) {
+ if (entry.getSourcePath().equals(src)) {
+ assertEquals(entry.getDefaultLocation().getNameserviceId(),
+ json.getString("nameserviceId"));
+ assertEquals(entry.getDefaultLocation().getDest(),
+ json.getString("path"));
+ assertEquals(entry.getOwnerName(), json.getString("ownerName"));
+ assertEquals(entry.getGroupName(), json.getString("groupName"));
+ assertEquals(entry.getMode().toString(), json.getString("mode"));
+ assertEquals(entry.getQuota().toString(), json.getString("quota"));
+ assertNotNullAndNotEmpty(json.getString("dateCreated"));
+ assertNotNullAndNotEmpty(json.getString("dateModified"));
+ match++;
+ }
+ }
+ }
+ assertEquals(match, getMockMountTable().size());
+ }
+
+ private MembershipState findMockNamenode(String nsId, String nnId) {
+
+ @SuppressWarnings("unchecked")
+ List<MembershipState> namenodes =
+ ListUtils.union(getActiveMemberships(), getStandbyMemberships());
+ for (MembershipState nn : namenodes) {
+ if (nn.getNamenodeId().equals(nnId)
+ && nn.getNameserviceId().equals(nsId)) {
+ return nn;
+ }
+ }
+ return null;
+ }
+
+ @Test
+ public void testNamenodeStatsDataSource() throws IOException, JSONException {
+
+ FederationMetrics metrics = getRouter().getMetrics();
+ String jsonString = metrics.getNamenodes();
+ JSONObject jsonObject = new JSONObject(jsonString);
+ Iterator<?> keys = jsonObject.keys();
+ int nnsFound = 0;
+ while (keys.hasNext()) {
+ // Validate each entry against our mocks
+ JSONObject json = jsonObject.getJSONObject((String) keys.next());
+ String nameserviceId = json.getString("nameserviceId");
+ String namenodeId = json.getString("namenodeId");
+
+ MembershipState mockEntry =
+ this.findMockNamenode(nameserviceId, namenodeId);
+ assertNotNull(mockEntry);
+
+ assertEquals(json.getString("state"), mockEntry.getState().toString());
+ MembershipStats stats = mockEntry.getStats();
+ assertEquals(json.getLong("numOfActiveDatanodes"),
+ stats.getNumOfActiveDatanodes());
+ assertEquals(json.getLong("numOfDeadDatanodes"),
+ stats.getNumOfDeadDatanodes());
+ assertEquals(json.getLong("numOfDecommissioningDatanodes"),
+ stats.getNumOfDecommissioningDatanodes());
+ assertEquals(json.getLong("numOfDecomActiveDatanodes"),
+ stats.getNumOfDecomActiveDatanodes());
+ assertEquals(json.getLong("numOfDecomDeadDatanodes"),
+ stats.getNumOfDecomDeadDatanodes());
+ assertEquals(json.getLong("numOfBlocks"), stats.getNumOfBlocks());
+ assertEquals(json.getString("rpcAddress"), mockEntry.getRpcAddress());
+ assertEquals(json.getString("webAddress"), mockEntry.getWebAddress());
+ nnsFound++;
+ }
+ // Validate all memberships are present
+ assertEquals(getActiveMemberships().size() + getStandbyMemberships().size(),
+ nnsFound);
+ }
+
+ @Test
+ public void testNameserviceStatsDataSource()
+ throws IOException, JSONException {
+
+ FederationMetrics metrics = getRouter().getMetrics();
+ String jsonString = metrics.getNameservices();
+ JSONObject jsonObject = new JSONObject(jsonString);
+ Iterator<?> keys = jsonObject.keys();
+ int nameservicesFound = 0;
+ while (keys.hasNext()) {
+ JSONObject json = jsonObject.getJSONObject((String) keys.next());
+ String nameserviceId = json.getString("nameserviceId");
+ String namenodeId = json.getString("namenodeId");
+
+ MembershipState mockEntry =
+ this.findMockNamenode(nameserviceId, namenodeId);
+ assertNotNull(mockEntry);
+
+ // NS should report the active NN
+ assertEquals(mockEntry.getState().toString(), json.getString("state"));
+ assertEquals("ACTIVE", json.getString("state"));
+
+ // Stats in the NS should reflect the stats for the most active NN
+ MembershipStats stats = mockEntry.getStats();
+ assertEquals(stats.getNumOfFiles(), json.getLong("numOfFiles"));
+ assertEquals(stats.getTotalSpace(), json.getLong("totalSpace"));
+ assertEquals(stats.getAvailableSpace(),
+ json.getLong("availableSpace"));
+ assertEquals(stats.getNumOfBlocksMissing(),
+ json.getLong("numOfBlocksMissing"));
+ assertEquals(stats.getNumOfActiveDatanodes(),
+ json.getLong("numOfActiveDatanodes"));
+ assertEquals(stats.getNumOfDeadDatanodes(),
+ json.getLong("numOfDeadDatanodes"));
+ assertEquals(stats.getNumOfDecommissioningDatanodes(),
+ json.getLong("numOfDecommissioningDatanodes"));
+ assertEquals(stats.getNumOfDecomActiveDatanodes(),
+ json.getLong("numOfDecomActiveDatanodes"));
+ assertEquals(stats.getNumOfDecomDeadDatanodes(),
+ json.getLong("numOfDecomDeadDatanodes"));
+ assertEquals(stats.getProvidedSpace(),
+ json.getLong("providedSpace"));
+ nameservicesFound++;
+ }
+ assertEquals(getNameservices().size(), nameservicesFound);
+ }
+
+ @Test
+ public void testRouterStatsDataSource() throws IOException, JSONException {
+
+ FederationMetrics metrics = getRouter().getMetrics();
+ String jsonString = metrics.getRouters();
+ JSONObject jsonObject = new JSONObject(jsonString);
+ Iterator<?> keys = jsonObject.keys();
+ int routersFound = 0;
+ while (keys.hasNext()) {
+ JSONObject json = jsonObject.getJSONObject((String) keys.next());
+ String address = json.getString("address");
+ assertNotNullAndNotEmpty(address);
+ RouterState router = findMockRouter(address);
+ assertNotNull(router);
+
+ assertEquals(router.getStatus().toString(), json.getString("status"));
+ assertEquals(router.getCompileInfo(), json.getString("compileInfo"));
+ assertEquals(router.getVersion(), json.getString("version"));
+ assertEquals(router.getDateStarted(), json.getLong("dateStarted"));
+ assertEquals(router.getDateCreated(), json.getLong("dateCreated"));
+ assertEquals(router.getDateModified(), json.getLong("dateModified"));
+
+ StateStoreVersion version = router.getStateStoreVersion();
+ assertEquals(
+ FederationMetrics.getDateString(version.getMembershipVersion()),
+ json.get("lastMembershipUpdate"));
+ assertEquals(
+ FederationMetrics.getDateString(version.getMountTableVersion()),
+ json.get("lastMountTableUpdate"));
+ assertEquals(version.getMembershipVersion(),
+ json.get("membershipVersion"));
+ assertEquals(version.getMountTableVersion(),
+ json.get("mountTableVersion"));
+ routersFound++;
+ }
+
+ assertEquals(getMockRouters().size(), routersFound);
+ }
+
+ private void assertNotNullAndNotEmpty(String field) {
+ assertNotNull(field);
+ assertTrue(field.length() > 0);
+ }
+
+ private RouterState findMockRouter(String routerId) {
+ for (RouterState router : getMockRouters()) {
+ if (router.getAddress().equals(routerId)) {
+ return router;
+ }
+ }
+ return null;
+ }
+
+ private void validateClusterStatsBean(FederationMBean bean)
+ throws IOException {
+
+ // Determine aggregates
+ long numBlocks = 0;
+ long numLive = 0;
+ long numDead = 0;
+ long numDecom = 0;
+ long numDecomLive = 0;
+ long numDecomDead = 0;
+ long numFiles = 0;
+ for (MembershipState mock : getActiveMemberships()) {
+ MembershipStats stats = mock.getStats();
+ numBlocks += stats.getNumOfBlocks();
+ numLive += stats.getNumOfActiveDatanodes();
+ numDead += stats.getNumOfDeadDatanodes();
+ numDecom += stats.getNumOfDecommissioningDatanodes();
+ numDecomLive += stats.getNumOfDecomActiveDatanodes();
+ numDecomDead += stats.getNumOfDecomDeadDatanodes();
+ }
+
+ assertEquals(numBlocks, bean.getNumBlocks());
+ assertEquals(numLive, bean.getNumLiveNodes());
+ assertEquals(numDead, bean.getNumDeadNodes());
+ assertEquals(numDecom, bean.getNumDecommissioningNodes());
+ assertEquals(numDecomLive, bean.getNumDecomLiveNodes());
+ assertEquals(numDecomDead, bean.getNumDecomDeadNodes());
+ assertEquals(numFiles, bean.getNumFiles());
+ assertEquals(getActiveMemberships().size() + getStandbyMemberships().size(),
+ bean.getNumNamenodes());
+ assertEquals(getNameservices().size(), bean.getNumNameservices());
+ assertTrue(bean.getVersion().length() > 0);
+ assertTrue(bean.getCompiledDate().length() > 0);
+ assertTrue(bean.getCompileInfo().length() > 0);
+ assertTrue(bean.getRouterStarted().length() > 0);
+ assertTrue(bean.getHostAndPort().length() > 0);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestMetricsBase.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestMetricsBase.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestMetricsBase.java
new file mode 100644
index 0000000..2169b21
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestMetricsBase.java
@@ -0,0 +1,196 @@
+/**
+ * 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.hdfs.server.federation.metrics;
+
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMENODES;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMESERVICES;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.clearAllRecords;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.createMockMountTable;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.createMockRegistrationForNamenode;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.synchronizeRecords;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.waitStateStore;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
+import org.apache.hadoop.hdfs.server.federation.router.Router;
+import org.apache.hadoop.hdfs.server.federation.router.RouterServiceState;
+import org.apache.hadoop.hdfs.server.federation.store.MembershipStore;
+import org.apache.hadoop.hdfs.server.federation.store.RouterStore;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.NamenodeHeartbeatRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatRequest;
+import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
+import org.apache.hadoop.hdfs.server.federation.store.records.StateStoreVersion;
+import org.apache.hadoop.util.Time;
+import org.junit.After;
+import org.junit.Before;
+
+/**
+ * Test the basic metrics functionality.
+ */
+public class TestMetricsBase {
+
+ private StateStoreService stateStore;
+ private MembershipStore membershipStore;
+ private RouterStore routerStore;
+ private Router router;
+ private Configuration routerConfig;
+
+ private List<MembershipState> activeMemberships;
+ private List<MembershipState> standbyMemberships;
+ private List<MountTable> mockMountTable;
+ private List<RouterState> mockRouters;
+ private List<String> nameservices;
+
+ @Before
+ public void setupBase() throws Exception {
+
+ if (router == null) {
+ routerConfig = new RouterConfigBuilder()
+ .stateStore()
+ .metrics()
+ .http()
+ .build();
+ router = new Router();
+ router.init(routerConfig);
+ router.setRouterId("routerId");
+ router.start();
+ stateStore = router.getStateStore();
+
+ membershipStore =
+ stateStore.getRegisteredRecordStore(MembershipStore.class);
+ routerStore = stateStore.getRegisteredRecordStore(RouterStore.class);
+
+ // Read all data and load all caches
+ waitStateStore(stateStore, 10000);
+ createFixtures();
+ stateStore.refreshCaches(true);
+ Thread.sleep(1000);
+ }
+ }
+
+ @After
+ public void tearDownBase() throws IOException {
+ if (router != null) {
+ router.stop();
+ router.close();
+ router = null;
+ }
+ }
+
+ private void createFixtures() throws IOException {
+ // Clear all records
+ clearAllRecords(stateStore);
+
+ nameservices = new ArrayList<>();
+ nameservices.add(NAMESERVICES[0]);
+ nameservices.add(NAMESERVICES[1]);
+
+ // 2 NNs per NS
+ activeMemberships = new ArrayList<>();
+ standbyMemberships = new ArrayList<>();
+
+ for (String nameservice : nameservices) {
+ MembershipState namenode1 = createMockRegistrationForNamenode(
+ nameservice, NAMENODES[0], FederationNamenodeServiceState.ACTIVE);
+ NamenodeHeartbeatRequest request1 =
+ NamenodeHeartbeatRequest.newInstance(namenode1);
+ assertTrue(membershipStore.namenodeHeartbeat(request1).getResult());
+ activeMemberships.add(namenode1);
+
+ MembershipState namenode2 = createMockRegistrationForNamenode(
+ nameservice, NAMENODES[1], FederationNamenodeServiceState.STANDBY);
+ NamenodeHeartbeatRequest request2 =
+ NamenodeHeartbeatRequest.newInstance(namenode2);
+ assertTrue(membershipStore.namenodeHeartbeat(request2).getResult());
+ standbyMemberships.add(namenode2);
+ }
+
+ // Add 2 mount table memberships
+ mockMountTable = createMockMountTable(nameservices);
+ synchronizeRecords(stateStore, mockMountTable, MountTable.class);
+
+ // Add 2 router memberships in addition to the running router.
+ long t1 = Time.now();
+ mockRouters = new ArrayList<>();
+ RouterState router1 = RouterState.newInstance(
+ "router1", t1, RouterServiceState.RUNNING);
+ router1.setStateStoreVersion(StateStoreVersion.newInstance(
+ t1 - 1000, t1 - 2000));
+ RouterHeartbeatRequest heartbeatRequest =
+ RouterHeartbeatRequest.newInstance(router1);
+ assertTrue(routerStore.routerHeartbeat(heartbeatRequest).getStatus());
+
+ GetRouterRegistrationRequest getRequest =
+ GetRouterRegistrationRequest.newInstance("router1");
+ GetRouterRegistrationResponse getResponse =
+ routerStore.getRouterRegistration(getRequest);
+ RouterState routerState1 = getResponse.getRouter();
+ mockRouters.add(routerState1);
+
+ long t2 = Time.now();
+ RouterState router2 = RouterState.newInstance(
+ "router2", t2, RouterServiceState.RUNNING);
+ router2.setStateStoreVersion(StateStoreVersion.newInstance(
+ t2 - 6000, t2 - 7000));
+ heartbeatRequest.setRouter(router2);
+ assertTrue(routerStore.routerHeartbeat(heartbeatRequest).getStatus());
+ getRequest.setRouterId("router2");
+ getResponse = routerStore.getRouterRegistration(getRequest);
+ RouterState routerState2 = getResponse.getRouter();
+ mockRouters.add(routerState2);
+ }
+
+ protected Router getRouter() {
+ return router;
+ }
+
+ protected List<MountTable> getMockMountTable() {
+ return mockMountTable;
+ }
+
+ protected List<MembershipState> getActiveMemberships() {
+ return activeMemberships;
+ }
+
+ protected List<MembershipState> getStandbyMemberships() {
+ return standbyMemberships;
+ }
+
+ protected List<String> getNameservices() {
+ return nameservices;
+ }
+
+ protected List<RouterState> getMockRouters() {
+ return mockRouters;
+ }
+
+ protected StateStoreService getStateStore() {
+ return stateStore;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java
new file mode 100644
index 0000000..cb3b472
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java
@@ -0,0 +1,528 @@
+/**
+ * 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.hdfs.server.federation.resolver;
+
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.FEDERATION_MOUNT_TABLE_MAX_CACHE_SIZE;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_DEFAULT_NAMESERVICE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.router.Router;
+import org.apache.hadoop.hdfs.server.federation.store.MountTableStore;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the {@link MountTableStore} from the {@link Router}.
+ */
+public class TestMountTableResolver {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(TestMountTableResolver.class);
+
+ private static final int TEST_MAX_CACHE_SIZE = 10;
+
+ private MountTableResolver mountTable;
+
+ private Map<String, String> getMountTableEntry(
+ String subcluster, String path) {
+ Map<String, String> ret = new HashMap<>();
+ ret.put(subcluster, path);
+ return ret;
+ }
+
+ /**
+ * Setup the mount table.
+ * / -> 1:/
+ * __tmp -> 2:/tmp
+ * __user -> 3:/user
+ * ____a -> 2:/user/test
+ * ______demo
+ * ________test
+ * __________a -> 1:/user/test
+ * __________b -> 3:/user/test
+ * ____b
+ * ______file1.txt -> 4:/user/file1.txt
+ * __usr
+ * ____bin -> 2:/bin
+ * __readonly -> 2:/tmp
+ *
+ * @throws IOException If it cannot set the mount table.
+ */
+ private void setupMountTable() throws IOException {
+ Configuration conf = new Configuration();
+ conf.setInt(
+ FEDERATION_MOUNT_TABLE_MAX_CACHE_SIZE, TEST_MAX_CACHE_SIZE);
+ conf.setStrings(DFS_ROUTER_DEFAULT_NAMESERVICE, "0");
+ mountTable = new MountTableResolver(conf);
+
+ // Root mount point
+ Map<String, String> map = getMountTableEntry("1", "/");
+ mountTable.addEntry(MountTable.newInstance("/", map));
+
+ // /tmp
+ map = getMountTableEntry("2", "/");
+ mountTable.addEntry(MountTable.newInstance("/tmp", map));
+
+ // /user
+ map = getMountTableEntry("3", "/user");
+ mountTable.addEntry(MountTable.newInstance("/user", map));
+
+ // /usr/bin
+ map = getMountTableEntry("2", "/bin");
+ mountTable.addEntry(MountTable.newInstance("/usr/bin", map));
+
+ // /user/a
+ map = getMountTableEntry("2", "/user/test");
+ mountTable.addEntry(MountTable.newInstance("/user/a", map));
+
+ // /user/b/file1.txt
+ map = getMountTableEntry("4", "/user/file1.txt");
+ mountTable.addEntry(MountTable.newInstance("/user/b/file1.txt", map));
+
+ // /user/a/demo/test/a
+ map = getMountTableEntry("1", "/user/test");
+ mountTable.addEntry(MountTable.newInstance("/user/a/demo/test/a", map));
+
+ // /user/a/demo/test/b
+ map = getMountTableEntry("3", "/user/test");
+ mountTable.addEntry(MountTable.newInstance("/user/a/demo/test/b", map));
+
+ // /readonly
+ map = getMountTableEntry("2", "/tmp");
+ MountTable readOnlyEntry = MountTable.newInstance("/readonly", map);
+ readOnlyEntry.setReadOnly(true);
+ mountTable.addEntry(readOnlyEntry);
+ }
+
+ @Before
+ public void setup() throws IOException {
+ setupMountTable();
+ }
+
+ @Test
+ public void testDestination() throws IOException {
+
+ // Check files
+ assertEquals("1->/tesfile1.txt",
+ mountTable.getDestinationForPath("/tesfile1.txt").toString());
+
+ assertEquals("3->/user/testfile2.txt",
+ mountTable.getDestinationForPath("/user/testfile2.txt").toString());
+
+ assertEquals("2->/user/test/testfile3.txt",
+ mountTable.getDestinationForPath("/user/a/testfile3.txt").toString());
+
+ assertEquals("3->/user/b/testfile4.txt",
+ mountTable.getDestinationForPath("/user/b/testfile4.txt").toString());
+
+ assertEquals("1->/share/file5.txt",
+ mountTable.getDestinationForPath("/share/file5.txt").toString());
+
+ assertEquals("2->/bin/file7.txt",
+ mountTable.getDestinationForPath("/usr/bin/file7.txt").toString());
+
+ assertEquals("1->/usr/file8.txt",
+ mountTable.getDestinationForPath("/usr/file8.txt").toString());
+
+ assertEquals("2->/user/test/demo/file9.txt",
+ mountTable.getDestinationForPath("/user/a/demo/file9.txt").toString());
+
+ // Check folders
+ assertEquals("3->/user/testfolder",
+ mountTable.getDestinationForPath("/user/testfolder").toString());
+
+ assertEquals("2->/user/test/b",
+ mountTable.getDestinationForPath("/user/a/b").toString());
+
+ assertEquals("3->/user/test/a",
+ mountTable.getDestinationForPath("/user/test/a").toString());
+
+ assertEquals("2->/tmp/tesfile1.txt",
+ mountTable.getDestinationForPath("/readonly/tesfile1.txt").toString());
+
+ }
+
+ private void compareLists(List<String> list1, String[] list2) {
+ assertEquals(list1.size(), list2.length);
+ for (String item : list2) {
+ assertTrue(list1.contains(item));
+ }
+ }
+
+ @Test
+ public void testGetMountPoint() throws IOException {
+ // Check get the mount table entry for a path
+ MountTable mtEntry;
+ mtEntry = mountTable.getMountPoint("/");
+ assertTrue(mtEntry.getSourcePath().equals("/"));
+
+ mtEntry = mountTable.getMountPoint("/user");
+ assertTrue(mtEntry.getSourcePath().equals("/user"));
+
+ mtEntry = mountTable.getMountPoint("/user/a");
+ assertTrue(mtEntry.getSourcePath().equals("/user/a"));
+
+ mtEntry = mountTable.getMountPoint("/user/a/");
+ assertTrue(mtEntry.getSourcePath().equals("/user/a"));
+
+ mtEntry = mountTable.getMountPoint("/user/a/11");
+ assertTrue(mtEntry.getSourcePath().equals("/user/a"));
+
+ mtEntry = mountTable.getMountPoint("/user/a1");
+ assertTrue(mtEntry.getSourcePath().equals("/user"));
+ }
+
+ @Test
+ public void testGetMountPoints() throws IOException {
+
+ // Check getting all mount points (virtual and real) beneath a path
+ List<String> mounts = mountTable.getMountPoints("/");
+ assertEquals(4, mounts.size());
+ compareLists(mounts, new String[] {"tmp", "user", "usr", "readonly"});
+
+ mounts = mountTable.getMountPoints("/user");
+ assertEquals(2, mounts.size());
+ compareLists(mounts, new String[] {"a", "b"});
+
+ mounts = mountTable.getMountPoints("/user/a");
+ assertEquals(1, mounts.size());
+ compareLists(mounts, new String[] {"demo"});
+
+ mounts = mountTable.getMountPoints("/user/a/demo");
+ assertEquals(1, mounts.size());
+ compareLists(mounts, new String[] {"test"});
+
+ mounts = mountTable.getMountPoints("/user/a/demo/test");
+ assertEquals(2, mounts.size());
+ compareLists(mounts, new String[] {"a", "b"});
+
+ mounts = mountTable.getMountPoints("/tmp");
+ assertEquals(0, mounts.size());
+
+ mounts = mountTable.getMountPoints("/t");
+ assertNull(mounts);
+
+ mounts = mountTable.getMountPoints("/unknownpath");
+ assertNull(mounts);
+ }
+
+ private void compareRecords(List<MountTable> list1, String[] list2) {
+ assertEquals(list1.size(), list2.length);
+ for (String item : list2) {
+ for (MountTable record : list1) {
+ if (record.getSourcePath().equals(item)) {
+ return;
+ }
+ }
+ }
+ fail();
+ }
+
+ @Test
+ public void testGetMounts() throws IOException {
+
+ // Check listing the mount table records at or beneath a path
+ List<MountTable> records = mountTable.getMounts("/");
+ assertEquals(9, records.size());
+ compareRecords(records, new String[] {"/", "/tmp", "/user", "/usr/bin",
+ "user/a", "/user/a/demo/a", "/user/a/demo/b", "/user/b/file1.txt",
+ "readonly"});
+
+ records = mountTable.getMounts("/user");
+ assertEquals(5, records.size());
+ compareRecords(records, new String[] {"/user", "/user/a/demo/a",
+ "/user/a/demo/b", "user/a", "/user/b/file1.txt"});
+
+ records = mountTable.getMounts("/user/a");
+ assertEquals(3, records.size());
+ compareRecords(records,
+ new String[] {"/user/a/demo/a", "/user/a/demo/b", "/user/a"});
+
+ records = mountTable.getMounts("/tmp");
+ assertEquals(1, records.size());
+ compareRecords(records, new String[] {"/tmp"});
+
+ records = mountTable.getMounts("/readonly");
+ assertEquals(1, records.size());
+ compareRecords(records, new String[] {"/readonly"});
+ assertTrue(records.get(0).isReadOnly());
+ }
+
+ @Test
+ public void testRemoveSubTree()
+ throws UnsupportedOperationException, IOException {
+
+ // 3 mount points are present /tmp, /user, /usr
+ compareLists(mountTable.getMountPoints("/"),
+ new String[] {"user", "usr", "tmp", "readonly"});
+
+ // /tmp currently points to namespace 2
+ assertEquals("2", mountTable.getDestinationForPath("/tmp/testfile.txt")
+ .getDefaultLocation().getNameserviceId());
+
+ // Remove tmp
+ mountTable.removeEntry("/tmp");
+
+ // Now 2 mount points are present /user, /usr
+ compareLists(mountTable.getMountPoints("/"),
+ new String[] {"user", "usr", "readonly"});
+
+ // /tmp no longer exists, uses default namespace for mapping /
+ assertEquals("1", mountTable.getDestinationForPath("/tmp/testfile.txt")
+ .getDefaultLocation().getNameserviceId());
+ }
+
+ @Test
+ public void testRemoveVirtualNode()
+ throws UnsupportedOperationException, IOException {
+
+ // 3 mount points are present /tmp, /user, /usr
+ compareLists(mountTable.getMountPoints("/"),
+ new String[] {"user", "usr", "tmp", "readonly"});
+
+ // /usr is virtual, uses namespace 1->/
+ assertEquals("1", mountTable.getDestinationForPath("/usr/testfile.txt")
+ .getDefaultLocation().getNameserviceId());
+
+ // Attempt to remove /usr
+ mountTable.removeEntry("/usr");
+
+ // Verify the remove failed
+ compareLists(mountTable.getMountPoints("/"),
+ new String[] {"user", "usr", "tmp", "readonly"});
+ }
+
+ @Test
+ public void testRemoveLeafNode()
+ throws UnsupportedOperationException, IOException {
+
+ // /user/a/demo/test/a currently points to namespace 1
+ assertEquals("1", mountTable.getDestinationForPath("/user/a/demo/test/a")
+ .getDefaultLocation().getNameserviceId());
+
+ // Remove /user/a/demo/test/a
+ mountTable.removeEntry("/user/a/demo/test/a");
+
+ // Now /user/a/demo/test/a points to namespace 2 using the entry for /user/a
+ assertEquals("2", mountTable.getDestinationForPath("/user/a/demo/test/a")
+ .getDefaultLocation().getNameserviceId());
+
+ // Verify the virtual node at /user/a/demo still exists and was not deleted
+ compareLists(mountTable.getMountPoints("/user/a"), new String[] {"demo"});
+
+ // Verify the sibling node was unaffected and still points to ns 3
+ assertEquals("3", mountTable.getDestinationForPath("/user/a/demo/test/b")
+ .getDefaultLocation().getNameserviceId());
+ }
+
+ @Test
+ public void testRefreshEntries()
+ throws UnsupportedOperationException, IOException {
+
+ // Initial table loaded
+ testDestination();
+ assertEquals(9, mountTable.getMounts("/").size());
+
+ // Replace table with /1 and /2
+ List<MountTable> records = new ArrayList<>();
+ Map<String, String> map1 = getMountTableEntry("1", "/");
+ records.add(MountTable.newInstance("/1", map1));
+ Map<String, String> map2 = getMountTableEntry("2", "/");
+ records.add(MountTable.newInstance("/2", map2));
+ mountTable.refreshEntries(records);
+
+ // Verify addition
+ PathLocation destination1 = mountTable.getDestinationForPath("/1");
+ RemoteLocation defaultLoc1 = destination1.getDefaultLocation();
+ assertEquals("1", defaultLoc1.getNameserviceId());
+
+ PathLocation destination2 = mountTable.getDestinationForPath("/2");
+ RemoteLocation defaultLoc2 = destination2.getDefaultLocation();
+ assertEquals("2", defaultLoc2.getNameserviceId());
+
+ // Verify existing entries were removed
+ assertEquals(2, mountTable.getMounts("/").size());
+ boolean assertionThrown = false;
+ try {
+ testDestination();
+ fail();
+ } catch (AssertionError e) {
+ // The / entry was removed, so it triggers an exception
+ assertionThrown = true;
+ }
+ assertTrue(assertionThrown);
+ }
+
+ @Test
+ public void testMountTableScalability() throws IOException {
+
+ List<MountTable> emptyList = new ArrayList<>();
+ mountTable.refreshEntries(emptyList);
+
+ // Add 100,000 entries in flat list
+ for (int i = 0; i < 100000; i++) {
+ Map<String, String> map = getMountTableEntry("1", "/" + i);
+ MountTable record = MountTable.newInstance("/" + i, map);
+ mountTable.addEntry(record);
+ if (i % 10000 == 0) {
+ LOG.info("Adding flat mount record {}: {}", i, record);
+ }
+ }
+
+ assertEquals(100000, mountTable.getMountPoints("/").size());
+ assertEquals(100000, mountTable.getMounts("/").size());
+
+ // Add 1000 entries in deep list
+ mountTable.refreshEntries(emptyList);
+ String parent = "/";
+ for (int i = 0; i < 1000; i++) {
+ final int index = i;
+ Map<String, String> map = getMountTableEntry("1", "/" + index);
+ if (i > 0) {
+ parent = parent + "/";
+ }
+ parent = parent + i;
+ MountTable record = MountTable.newInstance(parent, map);
+ mountTable.addEntry(record);
+ }
+
+ assertEquals(1, mountTable.getMountPoints("/").size());
+ assertEquals(1000, mountTable.getMounts("/").size());
+
+ // Add 100,000 entries in deep and wide tree
+ mountTable.refreshEntries(emptyList);
+ Random rand = new Random();
+ parent = "/" + Integer.toString(rand.nextInt());
+ int numRootTrees = 1;
+ for (int i = 0; i < 100000; i++) {
+ final int index = i;
+ Map<String, String> map = getMountTableEntry("1", "/" + index);
+ parent = parent + "/" + i;
+ if (parent.length() > 2000) {
+ // Start new tree
+ parent = "/" + Integer.toString(rand.nextInt());
+ numRootTrees++;
+ }
+ MountTable record = MountTable.newInstance(parent, map);
+ mountTable.addEntry(record);
+ }
+
+ assertEquals(numRootTrees, mountTable.getMountPoints("/").size());
+ assertEquals(100000, mountTable.getMounts("/").size());
+ }
+
+ @Test
+ public void testUpdate() throws IOException {
+
+ // Add entry to update later
+ Map<String, String> map = getMountTableEntry("1", "/");
+ mountTable.addEntry(MountTable.newInstance("/testupdate", map));
+
+ MountTable entry = mountTable.getMountPoint("/testupdate");
+ List<RemoteLocation> dests = entry.getDestinations();
+ assertEquals(1, dests.size());
+ RemoteLocation dest = dests.get(0);
+ assertEquals("1", dest.getNameserviceId());
+
+ // Update entry
+ Collection<MountTable> entries = Collections.singletonList(
+ MountTable.newInstance("/testupdate", getMountTableEntry("2", "/")));
+ mountTable.refreshEntries(entries);
+
+ MountTable entry1 = mountTable.getMountPoint("/testupdate");
+ List<RemoteLocation> dests1 = entry1.getDestinations();
+ assertEquals(1, dests1.size());
+ RemoteLocation dest1 = dests1.get(0);
+ assertEquals("2", dest1.getNameserviceId());
+
+ // Remove the entry to test updates and check
+ mountTable.removeEntry("/testupdate");
+ MountTable entry2 = mountTable.getMountPoint("/testupdate");
+ assertNull(entry2);
+ }
+
+ @Test
+ public void testCacheCleaning() throws Exception {
+ for (int i = 0; i < 1000; i++) {
+ String filename = String.format("/user/a/file-%04d.txt", i);
+ mountTable.getDestinationForPath(filename);
+ }
+ long cacheSize = mountTable.getCacheSize();
+ assertTrue(cacheSize <= TEST_MAX_CACHE_SIZE);
+ }
+
+ @Test
+ public void testLocationCache() throws Exception {
+ List<MountTable> entries = new ArrayList<>();
+
+ // Add entry and test location cache
+ Map<String, String> map1 = getMountTableEntry("1", "/testlocationcache");
+ MountTable entry1 = MountTable.newInstance("/testlocationcache", map1);
+ entries.add(entry1);
+
+ Map<String, String> map2 = getMountTableEntry("2",
+ "/anothertestlocationcache");
+ MountTable entry2 = MountTable.newInstance("/anothertestlocationcache",
+ map2);
+ entries.add(entry2);
+ mountTable.refreshEntries(entries);
+ assertEquals("1->/testlocationcache",
+ mountTable.getDestinationForPath("/testlocationcache").toString());
+ assertEquals("2->/anothertestlocationcache",
+ mountTable.getDestinationForPath("/anothertestlocationcache")
+ .toString());
+
+ // Remove the entry1
+ entries.remove(entry1);
+ mountTable.refreshEntries(entries);
+
+ // Add the default location and test location cache
+ assertEquals("0->/testlocationcache",
+ mountTable.getDestinationForPath("/testlocationcache").toString());
+
+ // Add the entry again but mount to another ns
+ Map<String, String> map3 = getMountTableEntry("3", "/testlocationcache");
+ MountTable entry3 = MountTable.newInstance("/testlocationcache", map3);
+ entries.add(entry3);
+ mountTable.refreshEntries(entries);
+
+ // Ensure location cache update correctly
+ assertEquals("3->/testlocationcache",
+ mountTable.getDestinationForPath("/testlocationcache").toString());
+
+ // Cleanup before exit
+ mountTable.removeEntry("/testlocationcache");
+ mountTable.removeEntry("/anothertestlocationcache");
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org