hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeMonitoring.java

/**
 * 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.router;

import static java.util.Arrays.asList;
import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.getFileSystem;
import static org.apache.hadoop.hdfs.server.federation.MockNamenode.registerSubclusters;
import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.getStateStoreConfiguration;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;

import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.LogVerificationAppender;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo.AdminStates;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo.DatanodeInfoBuilder;
import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
import org.apache.hadoop.hdfs.server.federation.MockNamenode;
import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver;
import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeContext;
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.resolver.NamenodeStatusReport;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
import org.apache.hadoop.http.HttpConfig;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Time;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;

/**
 * Test namenodes monitor behavior in the Router.
 */
public class TestRouterNamenodeMonitoring {

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


  /** Router for the test. */
  private Router router;
  /** Namenodes in the cluster. */
  private Map<String, Map<String, MockNamenode>> nns = new HashMap<>();
  /** Nameservices in the federated cluster. */
  private List<String> nsIds = asList("ns0", "ns1");
  /** Namenodes in the cluster. */
  private List<String> nnIds = asList("nn0", "nn1");

  /** Time the test starts. */
  private long initializedTime;


  @Before
  public void setup() throws Exception {
    LOG.info("Initialize the Mock Namenodes to monitor");
    for (String nsId : nsIds) {
      nns.put(nsId, new HashMap<>());
      for (String nnId : nnIds) {
        nns.get(nsId).put(nnId, new MockNamenode(nsId));
      }
    }

    LOG.info("Set nn0 to active for all nameservices");
    for (Map<String, MockNamenode> nnNS : nns.values()) {
      nnNS.get("nn0").transitionToActive();
      nnNS.get("nn1").transitionToStandby();
    }

    initializedTime = Time.now();
  }

  @After
  public void cleanup() throws Exception {
    for (Map<String, MockNamenode> nnNS : nns.values()) {
      for (MockNamenode nn : nnNS.values()) {
        nn.stop();
      }
    }
    nns.clear();

    if (router != null) {
      router.stop();
    }
  }

  /**
   * Get the configuration of the cluster which contains all the Namenodes and
   * their addresses.
   * @return Configuration containing all the Namenodes.
   */
  private Configuration getNamenodesConfig() {
    final Configuration conf = new HdfsConfiguration();
    conf.set(DFSConfigKeys.DFS_NAMESERVICES,
        StringUtils.join(",", nns.keySet()));
    for (String nsId : nns.keySet()) {
      Set<String> nsNnIds = nns.get(nsId).keySet();

      StringBuilder sb = new StringBuilder();
      sb.append(DFSConfigKeys.DFS_HA_NAMENODES_KEY_PREFIX);
      sb.append(".").append(nsId);
      conf.set(sb.toString(), StringUtils.join(",", nsNnIds));

      for (String nnId : nsNnIds) {
        final MockNamenode nn = nns.get(nsId).get(nnId);

        sb = new StringBuilder();
        sb.append(DFSConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY);
        sb.append(".").append(nsId);
        sb.append(".").append(nnId);
        conf.set(sb.toString(), "localhost:" + nn.getRPCPort());

        sb = new StringBuilder();
        sb.append(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
        sb.append(".").append(nsId);
        sb.append(".").append(nnId);
        conf.set(sb.toString(), "localhost:" + nn.getHTTPPort());
      }
    }
    return conf;
  }

  @Test
  public void testNamenodeMonitoring() throws Exception {
    Configuration nsConf = getNamenodesConfig();

    // Setup the State Store for the Router to use
    Configuration stateStoreConfig = getStateStoreConfiguration();
    stateStoreConfig.setClass(
        RBFConfigKeys.FEDERATION_NAMENODE_RESOLVER_CLIENT_CLASS,
        MembershipNamenodeResolver.class, ActiveNamenodeResolver.class);
    stateStoreConfig.setClass(
        RBFConfigKeys.FEDERATION_FILE_RESOLVER_CLIENT_CLASS,
        MountTableResolver.class, FileSubclusterResolver.class);

    Configuration routerConf = new RouterConfigBuilder(nsConf)
        .enableLocalHeartbeat(true)
        .heartbeat()
        .stateStore()
        .rpc()
        .build();

    // Specify namenodes (ns1.nn0,ns1.nn1) to monitor
    routerConf.set(RBFConfigKeys.DFS_ROUTER_RPC_ADDRESS_KEY, "0.0.0.0:0");
    routerConf.set(RBFConfigKeys.DFS_ROUTER_MONITOR_NAMENODE,
        "ns1.nn0,ns1.nn1");
    routerConf.addResource(stateStoreConfig);

    // Specify local node (ns0.nn1) to monitor
    routerConf.set(DFSConfigKeys.DFS_NAMESERVICE_ID, "ns0");
    routerConf.set(DFSConfigKeys.DFS_HA_NAMENODE_ID_KEY, "nn1");

    // Start the Router with the namenodes to monitor
    router = new Router();
    router.init(routerConf);
    router.start();

    // Manually trigger the heartbeat and update the values
    Collection<NamenodeHeartbeatService> heartbeatServices =
        router.getNamenodeHeartbeatServices();
    for (NamenodeHeartbeatService service : heartbeatServices) {
      service.periodicInvoke();
    }
    MembershipNamenodeResolver resolver =
        (MembershipNamenodeResolver) router.getNamenodeResolver();
    resolver.loadCache(true);

    // Check that the monitored values are expected
    final List<FederationNamenodeContext> namespaceInfo = new ArrayList<>();
    for (String nsId : nns.keySet()) {
      List<? extends FederationNamenodeContext> nnReports =
          resolver.getNamenodesForNameserviceId(nsId, false);
      namespaceInfo.addAll(nnReports);
    }
    for (FederationNamenodeContext nnInfo : namespaceInfo) {
      long modTime = nnInfo.getDateModified();
      long diff = modTime - initializedTime;
      if ("ns0".equals(nnInfo.getNameserviceId()) &&
          "nn0".equals(nnInfo.getNamenodeId())) {
        // The modified date won't be updated in ns0.nn0
        // since it isn't monitored by the Router.
        assertTrue(nnInfo + " shouldn't be updated: " + diff,
            modTime < initializedTime);
      } else {
        // other namnodes should be updated as expected
        assertTrue(nnInfo + " should be updated: " + diff,
            modTime > initializedTime);
      }
    }
  }

  @Test
  public void testNamenodeMonitoringConfig() throws Exception {
    testConfig(asList(), "");
    testConfig(asList("ns1.nn0"), "ns1.nn0");
    testConfig(asList("ns1.nn0", "ns1.nn1"), "ns1.nn0,ns1.nn1");
    testConfig(asList("ns1.nn0", "ns1.nn1"), "ns1.nn0, ns1.nn1");
    testConfig(asList("ns1.nn0", "ns1.nn1"), " ns1.nn0,ns1.nn1");
    testConfig(asList("ns1.nn0", "ns1.nn1"), "ns1.nn0,ns1.nn1,");
  }

  /**
   * Test if configuring a Router to monitor particular Namenodes actually
   * takes effect.
   * @param expectedNNs Namenodes that should be monitored.
   * @param confNsIds Router configuration setting for Namenodes to monitor.
   */
  private void testConfig(
      Collection<String> expectedNNs, String confNsIds) {

    // Setup and start the Router
    Configuration conf = getNamenodesConfig();
    Configuration routerConf = new RouterConfigBuilder(conf)
        .heartbeat(true)
        .build();
    routerConf.set(RBFConfigKeys.DFS_ROUTER_RPC_ADDRESS_KEY, "0.0.0.0:0");
    routerConf.set(RBFConfigKeys.DFS_ROUTER_MONITOR_NAMENODE, confNsIds);
    router = new Router();
    router.init(routerConf);

    // Test the heartbeat services of the Router
    Collection<NamenodeHeartbeatService> heartbeatServices =
        router.getNamenodeHeartbeatServices();
    assertNamenodeHeartbeatService(expectedNNs, heartbeatServices);
  }

  /**
   * Assert that the namenodes monitored by the Router are the expected.
   * @param expected Expected namenodes.
   * @param actual Actual heartbeat services for the Router
   */
  private static void assertNamenodeHeartbeatService(
      Collection<String> expected,
      Collection<NamenodeHeartbeatService> actual) {

    final Set<String> actualSet = new TreeSet<>();
    for (NamenodeHeartbeatService heartbeatService : actual) {
      NamenodeStatusReport report = heartbeatService.getNamenodeStatusReport();
      StringBuilder sb = new StringBuilder();
      sb.append(report.getNameserviceId());
      sb.append(".");
      sb.append(report.getNamenodeId());
      actualSet.add(sb.toString());
    }
    assertTrue(expected + " does not contain all " + actualSet,
        expected.containsAll(actualSet));
    assertTrue(actualSet + " does not contain all " + expected,
        actualSet.containsAll(expected));
  }

  @Test
  public void testJmxUrlHTTP() {
    verifyUrlSchemes(HttpConfig.Policy.HTTP_ONLY.name());
  }

  @Test
  public void testJmxUrlHTTPs() {
    verifyUrlSchemes(HttpConfig.Policy.HTTPS_ONLY.name());
  }

  @Test
  public void testJmxRequestFrequency() {
    // Disable JMX requests
    Configuration conf = getNamenodesConfig();
    conf.setLong(RBFConfigKeys.DFS_ROUTER_NAMENODE_HEARTBEAT_JMX_INTERVAL_MS, -1);
    verifyUrlSchemes(HttpConfig.Policy.HTTPS_ONLY.name(), conf, 0, 0, 1);

    // Set JMX requests to lower frequency
    conf = getNamenodesConfig();
    conf.setLong(RBFConfigKeys.DFS_ROUTER_NAMENODE_HEARTBEAT_JMX_INTERVAL_MS,
        TimeUnit.MINUTES.toMillis(5));
    verifyUrlSchemes(HttpConfig.Policy.HTTPS_ONLY.name(), conf, 0, 1, 2);

    // Set JMX requests to default frequency
    conf = getNamenodesConfig();
    verifyUrlSchemes(HttpConfig.Policy.HTTPS_ONLY.name(), conf, 0, 2, 2);
  }

  private void verifyUrlSchemes(String scheme) {
    int httpRequests = HttpConfig.Policy.HTTP_ONLY.name().equals(scheme) ? 1 : 0;
    int httpsRequests = HttpConfig.Policy.HTTPS_ONLY.name().equals(scheme) ? 1 : 0;
    verifyUrlSchemes(scheme, getNamenodesConfig(), httpRequests, httpsRequests, 1);
  }

  private void verifyUrlSchemes(String scheme, Configuration conf, int httpRequests,
      int httpsRequests, int requestsPerService) {

    // Attach our own log appender so we can verify output
    final LogVerificationAppender appender =
        new LogVerificationAppender();
    final org.apache.log4j.Logger logger =
        org.apache.log4j.Logger.getRootLogger();
    logger.addAppender(appender);
    GenericTestUtils.setRootLogLevel(Level.DEBUG);

    // Setup and start the Router
    conf.set(DFSConfigKeys.DFS_HTTP_POLICY_KEY, scheme);
    Configuration routerConf = new RouterConfigBuilder(conf)
        .heartbeat(true)
        .build();
    routerConf.set(RBFConfigKeys.DFS_ROUTER_RPC_ADDRESS_KEY, "0.0.0.0:0");
    routerConf.set(RBFConfigKeys.DFS_ROUTER_MONITOR_NAMENODE, "ns1.nn0");
    router = new Router();
    router.init(routerConf);

    // Test the heartbeat services of the Router
    Collection<NamenodeHeartbeatService> heartbeatServices =
        router.getNamenodeHeartbeatServices();
    for (NamenodeHeartbeatService heartbeatService : heartbeatServices) {
      for (int request = 0; request < requestsPerService; request++) {
        heartbeatService.getNamenodeStatusReport();
      }
    }
    assertEquals(httpsRequests * 2, appender.countLinesWithMessage("JMX URL: https://"));
    assertEquals(httpRequests * 2, appender.countLinesWithMessage("JMX URL: http://"));
  }

  /**
   * Test the view of the Datanodes that the Router sees. If a Datanode is
   * registered in two subclusters, it should return the most up to date
   * information.
   * @throws IOException If the test cannot run.
   */
  @Test
  public void testDatanodesView() throws IOException {

    // Setup the router
    Configuration routerConf = new RouterConfigBuilder()
        .stateStore()
        .rpc()
        .build();
    router = new Router();
    router.init(routerConf);
    router.start();

    // Setup the namenodes
    for (String nsId : nsIds) {
      registerSubclusters(router, nns.get(nsId).values());
      for (String nnId : nnIds) {
        MockNamenode nn = nns.get(nsId).get(nnId);
        if ("nn0".equals(nnId)) {
          nn.transitionToActive();
        }
        nn.addDatanodeMock();
      }
    }

    // Set different states for the DNs in each namespace
    long time = Time.now();
    for (String nsId : nsIds) {
      for (String nnId : nnIds) {
        // dn0 is DECOMMISSIONED in the most recent (ns1)
        DatanodeInfoBuilder dn0Builder = new DatanodeInfoBuilder()
            .setDatanodeUuid("dn0")
            .setHostName("dn0")
            .setIpAddr("dn0")
            .setXferPort(10000);
        if ("ns0".equals(nsId)) {
          dn0Builder.setLastUpdate(time - 1000);
          dn0Builder.setAdminState(AdminStates.NORMAL);
        } else if ("ns1".equals(nsId)) {
          dn0Builder.setLastUpdate(time - 500);
          dn0Builder.setAdminState(AdminStates.DECOMMISSIONED);
        }

        // dn1 is NORMAL in the most recent (ns0)
        DatanodeInfoBuilder dn1Builder = new DatanodeInfoBuilder()
            .setDatanodeUuid("dn1")
            .setHostName("dn1")
            .setIpAddr("dn1")
            .setXferPort(10000);
        if ("ns0".equals(nsId)) {
          dn1Builder.setLastUpdate(time - 1000);
          dn1Builder.setAdminState(AdminStates.NORMAL);
        } else if ("ns1".equals(nsId)) {
          dn1Builder.setLastUpdate(time - 5 * 1000);
          dn1Builder.setAdminState(AdminStates.DECOMMISSION_INPROGRESS);
        }

        // Update the mock NameNode with the DN views
        MockNamenode nn = nns.get(nsId).get(nnId);
        List<DatanodeInfo> dns = nn.getDatanodes();
        dns.add(dn0Builder.build());
        dns.add(dn1Builder.build());
      }
    }

    // Get the datanodes from the Router and check we get the right view
    DistributedFileSystem dfs = (DistributedFileSystem)getFileSystem(router);
    DFSClient dfsClient = dfs.getClient();
    DatanodeStorageReport[] dns = dfsClient.getDatanodeStorageReport(
        DatanodeReportType.ALL);
    assertEquals(2, dns.length);
    for (DatanodeStorageReport dn : dns) {
      DatanodeInfo dnInfo = dn.getDatanodeInfo();
      if ("dn0".equals(dnInfo.getHostName())) {
        assertEquals(AdminStates.DECOMMISSIONED, dnInfo.getAdminState());
      } else if ("dn1".equals(dnInfo.getHostName())) {
        assertEquals(AdminStates.NORMAL, dnInfo.getAdminState());
      } else {
        fail("Unexpected DN: " + dnInfo.getHostName());
      }
    }
  }
}