/**
 * 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_INTERNAL_NAMESERVICES_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.DFSConfigKeys.DFS_HTTP_POLICY_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HTTPS_ADDRESS_KEY;
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.DFS_ROUTER_SAFEMODE_ENABLE;
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.StorageType;
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.client.HdfsClientConfigKeys;
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.FederationNamespaceInfo;
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.namenode.ha.ConfiguredFailoverProxyProvider;
import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
import org.apache.hadoop.http.HttpConfig;
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 MiniRouterDFSCluster {

  private static final Logger LOG =
      LoggerFactory.getLogger(MiniRouterDFSCluster.class);

  public static final String TEST_STRING = "teststring";
  public static final String TEST_DIR = "testdir";
  public static final String TEST_FILE = "testfile";

  private static final Random RND = new Random();

  /** 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;
  /** Custom storage type for each datanode. */
  private StorageType[][] storageTypes = null;
  /** Racks for datanodes. */
  private String[] racks = null;

  /** 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 initializes. */
  private Configuration routerConf;
  /** Router configuration overrides. */
  private Configuration routerOverrides;
  /** Namenode configuration overrides. */
  private Configuration namenodeOverrides;

  /** If the DNs are shared. */
  private boolean sharedDNs = true;


  /**
   * Router context.
   */
  public static 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 URI getFileSystemURI() {
      return fileSystemUri;
    }

    public String getHttpAddress() {
      InetSocketAddress httpAddress = router.getHttpServerAddress();
      return NetUtils.getHostPortString(httpAddress);
    }

    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 void resetAdminClient() {
      adminClient = null;
    }

    public DFSClient getClient() throws IOException, URISyntaxException {
      if (client == null) {
        LOG.info("Connecting to router at {}", fileSystemUri);
        client = new DFSClient(fileSystemUri, conf);
      }
      return client;
    }

    public Configuration getConf() {
      return conf;
    }
  }

  /**
   * 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 int httpsPort;
    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();
      if (nn.getHttpAddress() != null) {
        this.httpPort = nn.getHttpAddress().getPort();
      }
      if (nn.getHttpsAddress() != null) {
        this.httpsPort = nn.getHttpsAddress().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 getWebAddress() {
      if (conf.get(DFS_HTTP_POLICY_KEY)
          .equals(HttpConfig.Policy.HTTPS_ONLY.name())) {
        return getHttpsAddress();
      }
      return getHttpAddress();
    }

    public String getHttpAddress() {
      return namenode.getHttpAddress().getHostName() + ":" + httpPort;
    }

    public String getHttpsAddress() {
      return namenode.getHttpsAddress().getHostName() + ":" + httpsPort;
    }

    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 Configuration getConf() {
      return conf;
    }
  }

  public MiniRouterDFSCluster(
      boolean ha, int numNameservices, int numNamenodes,
      long heartbeatInterval, long cacheFlushInterval,
      Configuration overrideConf) {
    this.highAvailability = ha;
    this.heartbeatInterval = heartbeatInterval;
    this.cacheFlushInterval = cacheFlushInterval;
    configureNameservices(numNameservices, numNamenodes, overrideConf);
  }

  public MiniRouterDFSCluster(
      boolean ha, int numNameservices, int numNamenodes,
      long heartbeatInterval, long cacheFlushInterval) {
    this(ha, numNameservices, numNamenodes,
        heartbeatInterval, cacheFlushInterval, null);
  }

  public MiniRouterDFSCluster(boolean ha, int numNameservices) {
    this(ha, numNameservices, 2,
        DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS,
        null);
  }

  public MiniRouterDFSCluster(
      boolean ha, int numNameservices, int numNamenodes) {
    this(ha, numNameservices, numNamenodes,
        DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS,
        null);
  }

  public MiniRouterDFSCluster(boolean ha, int numNameservices,
      Configuration overrideConf) {
    this(ha, numNameservices, 2,
        DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS, overrideConf);
  }

  /**
   * 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");
        conf.set(DFS_NAMENODE_HTTPS_ADDRESS_KEY + "." + suffix,
            "127.0.0.1:" + context.httpsPort);
        conf.set(
            HdfsClientConfigKeys.Failover.PROXY_PROVIDER_KEY_PREFIX + "." + ns,
            ConfiguredFailoverProxyProvider.class.getName());

        // 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;
    if (this.routerConf == null) {
      conf = new Configuration(false);
    } else {
      conf = new Configuration(routerConf);
    }
    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);

    // Disable safemode on startup
    conf.setBoolean(DFS_ROUTER_SAFEMODE_ENABLE, false);

    // 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,
      Configuration overrideConf) {
    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 (overrideConf != null) {
        nnConf.addResource(overrideConf);
      }

      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;
  }

  /**
   * Set custom storage type configuration for each datanode.
   * If storageTypes is uninitialized or passed null then
   * StorageType.DEFAULT is used.
   */
  public void setStorageTypes(StorageType[][] storageTypes) {
    this.storageTypes = storageTypes;
  }

  /**
   * Set racks for each datanode. If racks is uninitialized or passed null then
   * default is used.
   */
  public void setRacks(String[] racks) {
    this.racks = racks;
  }

  /**
   * Set the DNs to belong to only one subcluster.
   */
  public void setIndependentDNs() {
    this.sharedDNs = false;
  }

  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() {
    int randIndex = RND.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);

      // Set independent DNs across subclusters
      int numDNs = nameservices.size() * numDatanodesPerNameservice;
      Configuration[] dnConfs = null;
      if (!sharedDNs) {
        dnConfs = new Configuration[numDNs];
        int dnId = 0;
        for (String nsId : nameservices) {
          Configuration subclusterConf = new Configuration();
          subclusterConf.set(DFS_INTERNAL_NAMESERVICES_KEY, nsId);
          for (int i = 0; i < numDatanodesPerNameservice; i++) {
            dnConfs[dnId] = subclusterConf;
            dnId++;
          }
        }
      }

      // Start mini DFS cluster
      String ns0 = nameservices.get(0);
      Configuration nnConf = generateNamenodeConfiguration(ns0);
      if (overrideConf != null) {
        nnConf.addResource(overrideConf);
        // Router also uses this configurations as initial values.
        routerConf = new Configuration(overrideConf);
      }

      cluster = new MiniDFSCluster.Builder(nnConf)
          .numDataNodes(numDNs)
          .nnTopology(topology)
          .dataNodeConfOverlays(dnConfs)
          .storageTypes(storageTypes)
          .racks(racks)
          .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(), "http", nn.getWebAddress());
        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);
  }

  /**
   * Wait for name spaces to be active.
   * @throws Exception If we cannot check the status or we timeout.
   */
  public void waitActiveNamespaces() throws Exception {
    for (RouterContext r : this.routers) {
      Router router = r.router;
      final ActiveNamenodeResolver resolver = router.getNamenodeResolver();
      for (FederationNamespaceInfo ns : resolver.getNamespaces()) {
        final String nsId = ns.getNameserviceId();
        waitNamenodeRegistered(
            resolver, nsId, FederationNamenodeServiceState.ACTIVE);
      }
    }
  }

  /**
   * 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);
    }
  }
}