hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/FederationTestUtils.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;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.management.ManagementFactory;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import javax.management.JMX;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
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.fs.permission.FsPermission;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.NamenodeContext;
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.FederationNamenodeServiceState;
import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager;
import org.apache.hadoop.hdfs.server.federation.resolver.NamenodeStatusReport;
import org.apache.hadoop.hdfs.server.federation.resolver.order.DestinationOrder;
import org.apache.hadoop.hdfs.server.federation.router.ConnectionManager;
import org.apache.hadoop.hdfs.server.federation.router.Router;
import org.apache.hadoop.hdfs.server.federation.router.RouterClient;
import org.apache.hadoop.hdfs.server.federation.router.RouterRpcClient;
import org.apache.hadoop.hdfs.server.federation.router.RouterRpcServer;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory;
import org.apache.hadoop.hdfs.server.namenode.ha.HAContext;
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.AddMountTableEntryRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryResponse;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
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.protocol.NamespaceInfo;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.test.Whitebox;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Helper utilities for testing HDFS Federation.
 */
public final class FederationTestUtils {

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

  public final static String[] NAMESERVICES = {"ns0", "ns1"};
  public final static String[] NAMENODES = {"nn0", "nn1", "nn2", "nn3"};
  public final static String[] ROUTERS =
      {"router0", "router1", "router2", "router3"};


  private FederationTestUtils() {
    // Utility class
  }

  public static void verifyException(Object obj, String methodName,
      Class<? extends Exception> exceptionClass, Class<?>[] parameterTypes,
      Object[] arguments) {

    Throwable triggeredException = null;
    try {
      Method m = obj.getClass().getMethod(methodName, parameterTypes);
      m.invoke(obj, arguments);
    } catch (InvocationTargetException ex) {
      triggeredException = ex.getTargetException();
    } catch (Exception e) {
      triggeredException = e;
    }
    if (exceptionClass != null) {
      assertNotNull("No exception was triggered, expected exception"
          + exceptionClass.getName(), triggeredException);
      assertEquals(exceptionClass, triggeredException.getClass());
    } else {
      assertNull("Exception was triggered but no exception was expected",
          triggeredException);
    }
  }

  public static NamenodeStatusReport createNamenodeReport(String ns, String nn,
      HAServiceState state) {
    Random rand = new Random();
    return createNamenodeReport(ns, nn, "localhost:"
        + rand.nextInt(10000), state);
  }

  public static NamenodeStatusReport createNamenodeReport(String ns, String nn,
      String rpcAddress, HAServiceState state) {
    Random rand = new Random();
    NamenodeStatusReport report = new NamenodeStatusReport(ns, nn, rpcAddress,
        "localhost:" + rand.nextInt(10000),
        "localhost:" + rand.nextInt(10000), "http",
        "testwebaddress-" + ns + nn);
    if (state == null) {
      // Unavailable, no additional info
      return report;
    }
    report.setHAServiceState(state);
    NamespaceInfo nsInfo = new NamespaceInfo(
        1, "tesclusterid", ns, 0, "testbuildvesion", "testsoftwareversion");
    report.setNamespaceInfo(nsInfo);
    return report;
  }

  /**
   * Wait for a namenode to be registered with a particular state.
   * @param resolver Active namenode resolver.
   * @param nsId Nameservice identifier.
   * @param nnId Namenode identifier.
   * @param state State to check for.
   * @throws Exception Failed to verify State Store registration of namenode
   *                   nsId:nnId for state.
   */
  public static void waitNamenodeRegistered(
      final ActiveNamenodeResolver resolver,
      final String nsId, final String nnId,
      final FederationNamenodeServiceState state) throws Exception {

    GenericTestUtils.waitFor(() -> {
      try {
        List<? extends FederationNamenodeContext> namenodes =
            resolver.getNamenodesForNameserviceId(nsId, false);
        if (namenodes != null) {
          for (FederationNamenodeContext namenode : namenodes) {
            // Check if this is the Namenode we are checking
            if (namenode.getNamenodeId() == nnId  ||
                namenode.getNamenodeId().equals(nnId)) {
              return state == null || namenode.getState().equals(state);
            }
          }
        }
      } catch (IOException e) {
        // Ignore
      }
      return false;
    }, 1000, 60 * 1000);
  }

  /**
   * Wait for a namenode to be registered with a particular state.
   * @param resolver Active namenode resolver.
   * @param nsId Nameservice identifier.
   * @param state State to check for.
   * @throws Exception Failed to verify State Store registration of namenode
   *                   nsId for state.
   */
  public static void waitNamenodeRegistered(
      final ActiveNamenodeResolver resolver, final String nsId,
      final FederationNamenodeServiceState state) throws Exception {

    GenericTestUtils.waitFor(() -> {
      try {
        List<? extends FederationNamenodeContext> nns =
            resolver.getNamenodesForNameserviceId(nsId, false);
        for (FederationNamenodeContext nn : nns) {
          if (nn.getState().equals(state)) {
            return true;
          }
        }
      } catch (IOException e) {
        // Ignore
      }
      return false;
    }, 1000, 20 * 1000);
  }

  public static boolean verifyDate(Date d1, Date d2, long precision) {
    return Math.abs(d1.getTime() - d2.getTime()) < precision;
  }

  public static <T> T getBean(String name, Class<T> obj)
      throws MalformedObjectNameException {
    MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
    ObjectName poolName = new ObjectName(name);
    return JMX.newMXBeanProxy(mBeanServer, poolName, obj);
  }

  public static boolean addDirectory(FileSystem context, String path)
      throws IOException {
    context.mkdirs(new Path(path), new FsPermission("777"));
    return verifyFileExists(context, path);
  }

  public static FileStatus getFileStatus(FileSystem context, String path)
      throws IOException {
    return context.getFileStatus(new Path(path));
  }

  public static boolean verifyFileExists(FileSystem context, String path) {
    try {
      FileStatus status = getFileStatus(context, path);
      if (status != null) {
        return true;
      }
    } catch (Exception e) {
      return false;
    }
    return false;
  }

  public static boolean checkForFileInDirectory(
      FileSystem context, String testPath, String targetFile)
          throws IOException, AccessControlException, FileNotFoundException,
          UnsupportedFileSystemException, IllegalArgumentException {

    FileStatus[] fileStatus = context.listStatus(new Path(testPath));
    String file = null;
    String verifyPath = testPath + "/" + targetFile;
    if (testPath.equals("/")) {
      verifyPath = testPath + targetFile;
    }

    Boolean found = false;
    for (int i = 0; i < fileStatus.length; i++) {
      FileStatus f = fileStatus[i];
      file = Path.getPathWithoutSchemeAndAuthority(f.getPath()).toString();
      if (file.equals(verifyPath)) {
        found = true;
      }
    }
    return found;
  }

  public static int countContents(FileSystem context, String testPath)
      throws IOException {
    Path path = new Path(testPath);
    FileStatus[] fileStatus = context.listStatus(path);
    return fileStatus.length;
  }

  public static void createFile(FileSystem fs, String path, long length)
      throws IOException {
    FsPermission permissions = new FsPermission("700");
    FSDataOutputStream writeStream = fs.create(new Path(path), permissions,
        true, 1000, (short) 1, DFSConfigKeys.DFS_BLOCK_SIZE_DEFAULT, null);
    for (int i = 0; i < length; i++) {
      writeStream.write(i);
    }
    writeStream.close();
  }

  public static String readFile(FileSystem fs, String path) throws IOException {
    // Read the file from the filesystem via the active namenode
    Path fileName = new Path(path);
    InputStreamReader reader = new InputStreamReader(fs.open(fileName));
    BufferedReader bufferedReader = new BufferedReader(reader);
    StringBuilder data = new StringBuilder();
    String line;

    while ((line = bufferedReader.readLine()) != null) {
      data.append(line);
    }

    bufferedReader.close();
    reader.close();
    return data.toString();
  }

  public static boolean deleteFile(FileSystem fs, String path)
      throws IOException {
    return fs.delete(new Path(path), true);
  }

  /**
   * Simulate that a Namenode is slow by adding a sleep to the check operation
   * in the NN.
   * @param nn Namenode to simulate slow.
   * @param seconds Number of seconds to add to the Namenode.
   * @throws Exception If we cannot add the sleep time.
   */
  public static void simulateSlowNamenode(final NameNode nn, final int seconds)
      throws Exception {
    FSNamesystem namesystem = nn.getNamesystem();
    HAContext haContext = namesystem.getHAContext();
    HAContext spyHAContext = spy(haContext);
    doAnswer(new Answer<Object>() {
      @Override
      public Object answer(InvocationOnMock invocation) throws Throwable {
        LOG.info("Simulating slow namenode {}", invocation.getMock());
        try {
          Thread.sleep(seconds * 1000);
        } catch(InterruptedException e) {
          LOG.error("Simulating a slow namenode aborted");
        }
        return null;
      }
    }).when(spyHAContext).checkOperation(any(OperationCategory.class));
    Whitebox.setInternalState(namesystem, "haContext", spyHAContext);
  }

  /**
   * Wait for a number of routers to be registered in state store.
   *
   * @param stateManager number of routers to be registered.
   * @param routerCount number of routers to be registered.
   * @param timeout max wait time in ms
   */
  public static void waitRouterRegistered(RouterStore stateManager,
      long routerCount, int timeout) throws Exception {
    GenericTestUtils.waitFor(() -> {
      try {
        List<RouterState> cachedRecords = stateManager.getCachedRecords();
        if (cachedRecords.size() == routerCount) {
          return true;
        }
      } catch (IOException e) {
        // Ignore
      }
      return false;
    }, 100, timeout);
  }

  /**
   * Simulate that a RouterRpcServer, the ConnectionManager of its
   * RouterRpcClient throws IOException when call getConnection. So the
   * RouterRpcClient will get a null Connection.
   * @param server RouterRpcServer
   * @throws IOException
   */
  public static void simulateThrowExceptionRouterRpcServer(
      final RouterRpcServer server) throws IOException {
    RouterRpcClient rpcClient = server.getRPCClient();
    ConnectionManager connectionManager =
        new ConnectionManager(server.getConfig());
    ConnectionManager spyConnectionManager = spy(connectionManager);
    doAnswer(invocation -> {
      LOG.info("Simulating connectionManager throw IOException {}",
          invocation.getMock());
      throw new IOException("Simulate connectionManager throw IOException");
    }).when(spyConnectionManager).getConnection(
        any(UserGroupInformation.class), any(String.class), any(Class.class),
        any(String.class));

    Whitebox.setInternalState(rpcClient, "connectionManager",
        spyConnectionManager);
  }

  /**
   * Switch namenodes of all hdfs name services to standby.
   * @param cluster a federated HDFS cluster
   */
  public static void transitionClusterNSToStandby(
      StateStoreDFSCluster cluster) {
    // Name services of the cluster
    List<String> nameServiceList = cluster.getNameservices();

    // Change namenodes of each name service to standby
    for (String nameService : nameServiceList) {
      List<NamenodeContext>  nnList = cluster.getNamenodes(nameService);
      for(NamenodeContext namenodeContext : nnList) {
        cluster.switchToStandby(nameService, namenodeContext.getNamenodeId());
      }
    }
  }

  /**
   * Switch the index namenode of all hdfs name services to active.
   * @param cluster a federated HDFS cluster
   * @param index the index of namenodes
   */
  public static void transitionClusterNSToActive(
      StateStoreDFSCluster cluster, int index) {
    // Name services of the cluster
    List<String> nameServiceList = cluster.getNameservices();

    // Change the index namenode of each name service to active
    for (String nameService : nameServiceList) {
      List<NamenodeContext> listNamenodeContext =
          cluster.getNamenodes(nameService);
      cluster.switchToActive(nameService,
          listNamenodeContext.get(index).getNamenodeId());
    }
  }

  /**
   * Get the file system for HDFS in an RPC port.
   * @param rpcPort RPC port.
   * @return HDFS file system.
   * @throws IOException If it cannot create the file system.
   */
  public static FileSystem getFileSystem(int rpcPort) throws IOException {
    Configuration conf = new HdfsConfiguration();
    URI uri = URI.create("hdfs://localhost:" + rpcPort);
    return DistributedFileSystem.get(uri, conf);
  }

  /**
   * Get the file system for HDFS for a Router.
   * @param router Router.
   * @return HDFS file system.
   * @throws IOException If it cannot create the file system.
   */
  public static FileSystem getFileSystem(final Router router)
      throws IOException {
    InetSocketAddress rpcAddress = router.getRpcServerAddress();
    int rpcPort = rpcAddress.getPort();
    return getFileSystem(rpcPort);
  }

  /**
   * Get the admin interface for a Router.
   * @param router Router.
   * @return Admin interface.
   * @throws IOException If it cannot create the admin interface.
   */
  public static RouterClient getAdminClient(
      final Router router) throws IOException {
    Configuration conf = new HdfsConfiguration();
    InetSocketAddress routerSocket = router.getAdminServerAddress();
    return new RouterClient(routerSocket, conf);
  }

  /**
   * Add a mount table entry in some name services and wait until it is
   * available. If there are multiple routers,
   * {@link #createMountTableEntry(List, String, DestinationOrder, Collection)}
   * should be used instead because the method does not refresh
   * the mount tables of the other routers.
   * @param router Router to change.
   * @param mountPoint Name of the mount point.
   * @param order Order of the mount table entry.
   * @param nsIds Name service identifiers.
   * @throws Exception If the entry could not be created.
   */
  public static void createMountTableEntry(
      final Router router,
      final String mountPoint, final DestinationOrder order,
      Collection<String> nsIds) throws Exception {
    createMountTableEntry(
        Collections.singletonList(router), mountPoint, order, nsIds);
  }

  /**
   * Add a mount table entry in some name services and wait until it is
   * available.
   * @param routers List of routers.
   * @param mountPoint Name of the mount point.
   * @param order Order of the mount table entry.
   * @param nsIds Name service identifiers.
   * @throws Exception If the entry could not be created.
   */
  public static void createMountTableEntry(
      final List<Router> routers,
      final String mountPoint,
      final DestinationOrder order,
      final Collection<String> nsIds) throws Exception {
    Router router = routers.get(0);
    RouterClient admin = getAdminClient(router);
    MountTableManager mountTable = admin.getMountTableManager();
    Map<String, String> destMap = new HashMap<>();
    for (String nsId : nsIds) {
      destMap.put(nsId, mountPoint);
    }
    MountTable newEntry = MountTable.newInstance(mountPoint, destMap);
    newEntry.setDestOrder(order);
    AddMountTableEntryRequest addRequest =
        AddMountTableEntryRequest.newInstance(newEntry);
    AddMountTableEntryResponse addResponse =
        mountTable.addMountTableEntry(addRequest);
    boolean created = addResponse.getStatus();
    assertTrue(created);

    refreshRoutersCaches(routers);

    // Check for the path
    GetMountTableEntriesRequest getRequest =
        GetMountTableEntriesRequest.newInstance(mountPoint);
    GetMountTableEntriesResponse getResponse =
        mountTable.getMountTableEntries(getRequest);
    List<MountTable> entries = getResponse.getEntries();
    assertEquals("Too many entries: " + entries, 1, entries.size());
    assertEquals(mountPoint, entries.get(0).getSourcePath());
  }

  /**
   * Refresh the caches of a set of Routers.
   * @param routers List of Routers.
   */
  public static void refreshRoutersCaches(final List<Router> routers) {
    for (final Router router : routers) {
      StateStoreService stateStore = router.getStateStore();
      stateStore.refreshCaches(true);
    }
  }
}