hadoop/hdfs/TestBalancer.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;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintStream;
import java.net.URI;

import java.security.SecureRandom;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hdfs.test.system.HDFSCluster;
import org.apache.hadoop.hdfs.test.system.NNClient;
import org.apache.hadoop.hdfs.test.system.DNClient;


import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mortbay.util.ajax.JSON;

public class TestBalancer {

    private static final Log LOG = LogFactory.getLog(TestBalancer.class);
    private static final String BALANCER_TEMP_DIR = "balancer-temp";
    private Configuration hadoopConf;
    private HDFSCluster dfsCluster;

    public TestBalancer() throws Exception {
    }

    @Before
    public void setUp() throws Exception {
        hadoopConf = new Configuration();
        dfsCluster = HDFSCluster.createCluster(hadoopConf);
        dfsCluster.setUp();
    }

    @After
    public void tearDown() throws Exception {
        dfsCluster.tearDown();
    }

    // Trivial @Test
    public void testNamenodePing() throws IOException {
        LOG.info("testing filesystem ping");
        NNClient namenode = dfsCluster.getNNClient();
        namenode.ping();
        LOG.info("done.");
    }

    // Trivial @Test
    public void testNamenodeConnectDisconnect() throws IOException {
        LOG.info("connecting to namenode");
        NNClient namenode = dfsCluster.getNNClient();
        namenode.connect();
        LOG.info("done.");
        LOG.info("disconnecting from namenode");
        namenode.disconnect();
    }

    /**
     * The basic scenario for balancer test is as follows
     *
     *  - Bring up cluster with 1 DataNode
     *  - Load DataNode to >50%
     *  - Count files/blocks on DataNode
     *  - Add new, empty DataNode to cluster
     *  - Run Balancer
     *  - Count files/blocks on DataNodes
     *  - Blocks counts from before and after Balancer run should be consistent
     *
     */
    @Test
    public void testBalancerBasicScenario() throws IOException {
        Path balancerTempDir = null;
        try {
            List<DNClient> testnodes = reserveDatanodesForTest(2);
            DNClient testnode1 = testnodes.get(0);
            DNClient testnode2 = testnodes.get(1);
            shutdownNonTestNodes(testnodes);

            LOG.info("attempting to kill both test nodes");
            stopDatanode(testnode1);
            stopDatanode(testnode2);

            LOG.info("starting up datanode ["+
            testnode1.getHostName()+
            "] and loading it with data");
            startDatanode(testnode1);

            // mkdir balancer-temp
            balancerTempDir = makeTempDir();
            // write 2 blocks to file system
            LOG.info("generating filesystem load");
            // TODO spec blocks to generate by blockCount, blockSize, # of writers
            generateFileSystemLoad(2);  // generate 2 blocks of test data

            LOG.info("measure space used on 1st node");
            long usedSpace0 = getDatanodeUsedSpace(testnode1);
            LOG.info("datanode " + testnode1.getHostName()
                    + " contains " + usedSpace0 + " bytes");

            LOG.info("bring up a 2nd node and run balancer on DFS");
            startDatanode(testnode2);
            runBalancerAndVerify(testnodes);
        } catch (Throwable t) {
            LOG.info("method testBalancer failed", t);
        } finally {
            // finally block to run cleanup
            LOG.info("clean off test data from DFS [rmr ~/balancer-temp]");
            try {
                deleteTempDir(balancerTempDir);
            } catch (Exception e) {
                LOG.warn("problem cleaning up temp dir", e);
            }

            // restart killed nodes
            Iterator<DNClient> iter = dfsCluster.getDNClients().iterator();

            while (iter.hasNext()) {
                DNClient dn = iter.next();
                startDatanode( dn );
            }
        }
    }

    private void shutdownNonTestNodes(List<DNClient> testnodes) {
        Set killSet = new HashSet(getAllDatanodes());
        killSet.removeAll(testnodes);
        LOG.info("attempting to kill/suspend all the nodes not used for this test");
        Iterator<DNClient> iter = killSet.iterator();
        DNClient dn = null;
        while (iter.hasNext()) {
            dn = iter.next();
            // kill may not work with some secure-HDFS configs,
            // so using our stopDataNode() method
            stopDatanode(dn);
        }
    }

    /**
     * Kill all datanodes but leave reservationCount nodes alive,
     * return a list of the reserved datanodes
     */
    private List<DNClient> reserveDatanodesForTest(int reservationCount) {
        List<DNClient> testDNs = new LinkedList<DNClient>();
        List<DNClient> dieDNs = new LinkedList<DNClient>();
        LOG.info("getting collection of live data nodes");
        List<DNClient> dnList = getAllDatanodes();
        int dnCount = dnList.size();
        //  check to make sure there is enough capacity on these nodes to run test
        Assert.assertTrue(
                String.format(
                    "not enough datanodes available to run test,"
                    + " need %d datanodes but have only %d available",
                    reservationCount, dnCount),
                ( dnCount >= reservationCount ));
        LOG.info("selecting "+reservationCount+" nodes for test");
        dieDNs = new LinkedList<DNClient>(dnList);
        testDNs = new LinkedList<DNClient>();

        final int LEN = dnCount - 1;
        int i = getRandom(LEN);
        DNClient testDN = dieDNs.get(i);
        testDNs.add(testDN);
        dieDNs.remove(testDN);
        int j = i;
        do {
            i = getRandom(LEN);
        } while (i != j);
        testDN = dieDNs.get(i);
        testDNs.add(testDN);
        dieDNs.remove(testDN);

        LOG.info("nodes reserved for test");
        printDatanodeList(testDNs);

        LOG.info("nodes not used in test");
        printDatanodeList(dieDNs);

        return testDNs;
    }

    private List<DNClient> getAllDatanodes() {
        return dfsCluster.getDNClients();
    }

    private final static DNClient[] DATANODE_ARRAY = {};
    private DNClient[] toDatanodeArray(List<DNClient> datanodeList) {
        return (DNClient[]) datanodeList.toArray(DATANODE_ARRAY);
    }

    /**
     * Return a random number between 0 and N inclusive.
     *
     * @param int n
     * @param n  max number to return
     * @return random integer between 0 and N
     */
    private int getRandom(int n) {
        return (int) (n * Math.random());
    }

    /**
     * Calculate if the error in expected and observed values is within tolerance
     *
     * @param expectedValue  expected value of experiment
     * @param observedValue  observed value of experiment
     * @param tolerance      per cent tolerance for error, represented as a int
     */
    private boolean withinTolerance(long expectedValue,
                                    long observedValue,
                                    int tolerance) {
        double diff = 1.0 * Math.abs(observedValue - expectedValue);
        double thrs = expectedValue * (tolerance/100);
        return diff > thrs;
    }

    //  emulate tolerance calculation in balancer code
    public final static int DEFAULT_TOLERANCE = 10; // 10%
    protected boolean isClusterBalanced(DNClient[] datanodes) throws IOException {
        return isClusterBalanced(datanodes, DEFAULT_TOLERANCE);
    }
    protected boolean isClusterBalanced(DNClient[] datanodes, int tolerance)
            throws IOException {

        Assert.assertFalse("empty datanode array specified",
                ArrayUtils.isEmpty(datanodes));
        boolean result = true;
        double[] utilizationByNode = new double[ datanodes.length ];
        double totalUsedSpace = 0L;
        double totalCapacity = 0L;
        Map datanodeVolumeMap = new HashMap();
        // accumulate space stored on each node
        for(int i=0; i<datanodes.length; i++) {
            DNClient datanode = datanodes[i];
            Map volumeInfoMap = getDatanodeVolumeAttributes(datanode);
            long usedSpace = (Long)volumeInfoMap.get(ATTRNAME_USED_SPACE);
            long capacity  = (Long)volumeInfoMap.get(ATTRNAME_CAPACITY  );
            utilizationByNode[i] = ( ((double)usedSpace)/capacity ) * 100;
            totalUsedSpace += usedSpace;
            totalCapacity  += capacity;
        }
        // here we are reusing previously fetched volume-info, for speed
        // an alternative is to get fresh values from the cluster here instead
        double avgUtilization = ( totalUsedSpace/totalCapacity ) * 100;
        for(int i=0; i<datanodes.length; i++) {
            double varUtilization = Math.abs(avgUtilization - utilizationByNode[i]);
            if(varUtilization > tolerance) {
                result = false;
                break;
            }
        }

        return result;
    }

    /**
     * Make a working directory for storing temporary files
     *
     * @throws IOException
     */
    private Path makeTempDir() throws IOException {
        Path temp = new Path(BALANCER_TEMP_DIR);
        FileSystem srcFs = temp.getFileSystem(hadoopConf);
        FileStatus fstatus = null;
        try {
            fstatus = srcFs.getFileStatus(temp);
            if (fstatus.isDir()) {
                LOG.warn(BALANCER_TEMP_DIR + ": File exists");
            } else {
                LOG.warn(BALANCER_TEMP_DIR + " exists but is not a directory");
            }
            deleteTempDir(temp);
        } catch (FileNotFoundException fileNotFoundExc) {
        } finally {
            if (!srcFs.mkdirs(temp)) {
                throw new IOException("failed to create " + BALANCER_TEMP_DIR);
            }
        }
        return temp;
    }

    /**
     * Remove the working directory used to store temporary files
     *
     * @param temp
     * @throws IOException
     */
    private void deleteTempDir(Path temp) throws IOException {
        FileSystem srcFs = temp.getFileSystem(hadoopConf);
        LOG.info("attempting to delete path " + temp + "; this path exists? -> " + srcFs.exists(temp));
        srcFs.delete(temp, true);
    }

    private void printDatanodeList(List<DNClient> lis) {
        for (DNClient datanode : lis) {
            LOG.info("\t" + datanode.getHostName());
        }
    }

    private final static String CMD_STOP_DN = "sudo yinst stop hadoop_datanode_admin";
    private void stopDatanode(DNClient dn) {
        String dnHost = dn.getHostName();
        runAndWatch(dnHost, CMD_STOP_DN);
    }
    private final static String CMD_START_DN = "sudo yinst start hadoop_datanode_admin";
    private void startDatanode(DNClient dn) {
        String dnHost = dn.getHostName();
        runAndWatch(dnHost, CMD_START_DN);
    }

    /* using "old" default block size of 64M */
    private static final int DFS_BLOCK_SIZE = 67108864;
    private static final short DEFAULT_REPLICATION = 3;
    private void generateFileSystemLoad(long numBlocks) {
        generateFileSystemLoad(numBlocks, DEFAULT_REPLICATION);
    }
    private void generateFileSystemLoad(long numBlocks, short replication) {
        String destfile = "hdfs:///user/hadoopqa/";// + BALANCER_TEMP_DIR + "/LOADGEN.DAT";
        SecureRandom randgen = new SecureRandom();
        ByteArrayOutputStream dat = null;
        ByteArrayInputStream in = null;
        final int CHUNK = 4096;
        final Configuration testConf = new Configuration(hadoopConf);
        try {
            testConf.setInt("dfs.replication", replication);
            for (int i = 0; i < numBlocks; i++) {
                FileSystem fs = FileSystem.get(
                        URI.create(destfile), testConf);
                OutputStream out = fs.create(
                        new Path(destfile),
                        replication,
                        new ProgressReporter());
                dat = new ByteArrayOutputStream(DFS_BLOCK_SIZE);
                for (int z = 0; z < DFS_BLOCK_SIZE; z += CHUNK) {
                    byte[] bytes = new byte[CHUNK];
                    randgen.nextBytes(bytes);
                    dat.write(bytes, 0, CHUNK);
                }

                in = new ByteArrayInputStream(dat.toByteArray());
                IOUtils.copyBytes(in, out, CHUNK, true);
                LOG.info("wrote block " + (i + 1) + " of " + numBlocks);
            }
        } catch (IOException ioExc) {
            LOG.warn("f/s loadgen failed!", ioExc);
        } finally {
            try {
                dat.close();
            } catch (Exception e) {
            }
            try {
                in.close();
            } catch (Exception e) {
            }
        }
    }
    // TODO this should be taken from the environment
    public final static String HADOOP_HOME = "/grid/0/gs/gridre/yroot.biga/share/hadoop-current";
    public final static String CMD_SSH = "/usr/bin/ssh";
    public final static String CMD_KINIT = "/usr/kerberos/bin/kinit";
    public final static String CMD_HADOOP = HADOOP_HOME + "/bin/hadoop";
    public final static String OPT_BALANCER = "balancer";
    public final static String KERB_KEYTAB = "/homes/hadoopqa/hadoopqa.dev.headless.keytab";
    public final static String KERB_PRINCIPAL = "hadoopqa@DEV.YGRID.YAHOO.COM";

    public final static int DEFAULT_THRESHOLD = 10;
    private int runBalancer() throws IOException {
        return runBalancer(DEFAULT_THRESHOLD);
    }

    private int runBalancer(int threshold) throws IOException {
        return runBalancer(""+threshold);
    }
    /*
     * TODO change the heap size balancer uses so it can run on gateways
     * i.e., 14G heap is too big for gateways
     */
    private int runBalancer(String threshold)
            throws IOException {

        String balancerCommand = String.format("\"%s -k -t %s %s; %s %s -threshold %s",
                CMD_KINIT,
                KERB_KEYTAB,
                KERB_PRINCIPAL,
                CMD_HADOOP,
                OPT_BALANCER,
                threshold);
        String nnHost = dfsCluster.getNNClient().getHostName();
        return runAndWatch(nnHost, balancerCommand);
    }
    private void runBalancerAndVerify(List<DNClient> testnodes)
            throws IOException {
        runBalancerAndVerify(testnodes, DEFAULT_THRESHOLD);
    }
    private void runBalancerAndVerify(List<DNClient> testnodes, int threshold)
            throws IOException {
        runBalancerAndVerify(testnodes, ""+DEFAULT_THRESHOLD);
    }
      private void runBalancerAndVerify(List<DNClient> testnodes, String threshold)
            throws IOException {
        int exitStatus = runBalancer(threshold);
        // assert balancer exits with status SUCCESSe
        Assert.assertTrue(
                String.format("balancer returned non-success exit code: %d",
                exitStatus),
                (exitStatus == SUCCESS));
        DNClient[] testnodeArr = toDatanodeArray(testnodes);
        Assert.assertTrue(
                "cluster is not balanced",
                isClusterBalanced(testnodeArr));
    }

    private int runAndWatch(String remoteHost, String remoteCommand) {
        int exitStatus = -1;
        try {
            Process proc = new ProcessBuilder(CMD_SSH, remoteHost, remoteCommand).start();
            watchProcStream(proc.getInputStream(), System.out);
            watchProcStream(proc.getErrorStream(), System.err);
            exitStatus = proc.waitFor();
        } catch(InterruptedException intExc) {
            LOG.warn("got thread interrupt error", intExc);
        } catch(IOException ioExc) {
            LOG.warn("got i/o error", ioExc);
        }
        return exitStatus;
    }

    private void watchProcStream(InputStream in, PrintStream out) {
        new Thread(new StreamWatcher(in, out)).start();
    }
    private static final String DATANODE_VOLUME_INFO = "VolumeInfo";
    private static final String ATTRNAME_USED_SPACE  = "usedSpace";
    private static final String ATTRNAME_FREE_SPACE  = "freeSpace";
    // pseudo attribute, JMX doesn't really provide this
    private static final String ATTRNAME_CAPACITY    = "capacity";
    // TODO maybe the static methods below belong in some utility class...
    private static long getDatanodeUsedSpace(DNClient datanode)
            throws IOException {
        return (Long)getDatanodeVolumeAttributes(datanode).get(ATTRNAME_USED_SPACE);
    }/*
    private static long getDatanodeFreeSpace(DNClient datanode)
            throws IOException {
        return (Long)getDatanodeVolumeAttributes(datanode).get(ATTRNAME_FREE_SPACE);
    }*/
    private static Map getDatanodeVolumeAttributes(DNClient datanode)
            throws IOException {
        Map result = new HashMap();
        long usedSpace = getVolumeAttribute(datanode, ATTRNAME_USED_SPACE);
        long freeSpace = getVolumeAttribute(datanode, ATTRNAME_FREE_SPACE);
        result.put(ATTRNAME_USED_SPACE, usedSpace);
        result.put(ATTRNAME_CAPACITY,   usedSpace+freeSpace);
        return result;
    }

    private static long getVolumeAttribute(DNClient datanode,
                                                 String attribName)
        throws IOException {

        Object volInfo = datanode.getDaemonAttribute(DATANODE_VOLUME_INFO);
        Assert
        .assertNotNull( String
                        .format( "Attribute \"%s\" should be non-null",
                                 DATANODE_VOLUME_INFO ),
                        volInfo );
        String strVolInfo = volInfo.toString();
        LOG.debug( String.format("Value of %s: %s",
                   DATANODE_VOLUME_INFO,
                   strVolInfo) );
        Map volInfoMap = (Map) JSON.parse(strVolInfo);
        long attrVal = 0L;
        for(Object key: volInfoMap.keySet()) {
            Map attrMap = (Map) volInfoMap.get(key);
            long val = (Long) attrMap.get(attribName);
            attrVal += val;
        }
        return attrVal;

    }
    /** simple utility to watch streams from an exec'ed process */
    static class StreamWatcher implements Runnable {

        private BufferedReader reader;
        private PrintStream printer;

        StreamWatcher(InputStream in, PrintStream out) {
            reader = getReader(in);
            printer = out;
        }

        private static BufferedReader getReader(InputStream in) {
            return new BufferedReader(new InputStreamReader(in));
        }

        public void run() {
            try {
                if (reader.ready()) {
                    printer.println(reader.readLine());
                }
            } catch (IOException ioExc) {
            }
        }
    }

    /** simple utility to report progress in generating data */
    static class ProgressReporter implements Progressable {

        StringBuffer buf = null;

        public void progress() {
            if (buf == null) {
                buf = new StringBuffer();
            }
            buf.append(".");
            if (buf.length() == 10000) {
                LOG.info("..........");
                buf = null;
            }
        }
    }

    // A constant for SUCCESS exit code
     static final int SUCCESS = 1;

     /**
     * Balancer_01
     * Start balancer and check if the cluster is balanced after the run.
     * Cluster should end up in balanced state.
     */
    @Test
    public void testBalancerSimple() throws IOException {

        DNClient[] datanodes = toDatanodeArray( getAllDatanodes() );
        int exitStatus = runBalancer();
        // assert on successful exit code here
        Assert.assertTrue(
                String.format("balancer returned non-success exit code: %d",
                              exitStatus),
                (exitStatus == SUCCESS));
        Assert.assertTrue( "cluster is not balanced", isClusterBalanced(datanodes) );

    }

    /**
     * Balancer_02
     * Test a cluster with even distribution, then a new empty node is
     * added to the cluster. Here, even distribution effectively means the
     * cluster is in "balanced" state, as bytes consumed for block allocation
     * are evenly distributed throughout the cluster.
     */
    @Test
    public void testBalancerEvenDistributionWithNewNodeAdded() throws IOException {
        throw new UnsupportedOperationException("not implemented yet!");

        // get all nodes
        // need to get an external reserve of nodes we can boot up
        // to add to this cluster?
        // HOW?

        // IDEA try to steal some nodes from omega-M for now.....
        // hmmm also need a way to give an alternate "empty-node" config
        // to "hide" the data that may already exist on this node
    }

    /**
     * Balancer_03
     * Bring up a 1-node DFS cluster. Set files replication factor to be 1
     * and fill up the node to 30% full. Then add an empty datanode.
     */
     @Test
     public void testBalancerSingleNodeClusterWithNewNodeAdded() throws IOException {
        // empty datanode: mod config to point to non-default blocks dir.
        // limit capacity to available storage space
        throw new UnsupportedOperationException("not implemented yet!");
     }

    /**
     * Balancer_04
     * The same as _03 except that the empty new data node is on a
     * different rack.
     */
     @Test
     public void testBalancerSingleNodeClusterWithNewNodeAddedFromDifferentRack()
             throws IOException {
         // need rack awareness
         throw new UnsupportedOperationException("not implemented yet!");
     }

    /**
     * Balancer_05
     * The same as _03 except that the empty new data node is half the
     * capacity as the old one.
     */
     @Test
     public void testBalancerSingleNodeClusterWithHalfCapacityNewNode() {
         // how to limit node capacity?
         throw new UnsupportedOperationException("not implemented yet!");
     }

    /**
     * Balancer_06
     * Bring up a 2-node cluster and fill one node to be 60% and the
     * other to be 10% full. All nodes are on different racks.
     */
     @Test
    public void testBalancerTwoNodeMultiRackCluster() {
         // need rack awareness
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_07
     * Bring up a dfs cluster with nodes A and B. Set file replication
     * factor to be 2 and fill up the cluster to 30% full. Then add an
     * empty data node C. All three nodes are on the same rack.
     */
     @Test
     public void testBalancerTwoNodeSingleRackClusterWuthNewNodeAdded()
             throws IOException {

        final short TEST_REPLICATION_FACTOR = 3;
        List<DNClient> testnodes = reserveDatanodesForTest(3);
        DNClient dnA = testnodes.get(0);
        DNClient dnB = testnodes.get(1);

        DNClient dnC = testnodes.get(2);
        stopDatanode(dnC);

        // change test: 30% full-er (ie, 30% over pre-test capacity),
        // use most heavily node as baseline
        long targetLoad = (long) (
                (1/DFS_BLOCK_SIZE) *
                0.30 *
            Math.max( getDatanodeUsedSpace(dnA), getDatanodeUsedSpace(dnB) ) );
        generateFileSystemLoad(targetLoad, TEST_REPLICATION_FACTOR);
        startDatanode(dnC);
        runBalancerAndVerify(testnodes);
     }

    /**
     * Balancer_08
     * The same as _07 except that A, B and C are on different racks.
     */
     @Test
     public void testBalancerTwoNodeMultiRackClusterWithNewNodeAdded()
             throws IOException {
         // need rack awareness
         throw new UnsupportedOperationException("not implemented yet!");
     }

     /**
     * Balancer_09
     * The same as _07 except that interrupt balancing.
     */
     @Test
     public void testBalancerTwoNodeSingleRackClusterInterruptingRebalance()
             throws IOException {
         // interrupt thread
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_10
     * Restart rebalancing until it is done.
     */
    @Test
    public void testBalancerRestartInterruptedBalancerUntilDone()
            throws IOException {
        // need kill-restart thread
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_11
     * The same as _07 except that the namenode is shutdown while rebalancing.
     */
    @Test
    public void testBalancerTwoNodeSingleRackShutdownNameNodeDuringRebalance()
            throws IOException {
        // need NN shutdown thread in addition
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_12
     * The same as _05 except that FS writes occur during rebalancing.
     */
    @Test
    public void
    testBalancerSingleNodeClusterWithHalfCapacityNewNodeRebalanceWithConcurrentFSWrites()
            throws IOException {
        // writer thread
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_13
     * The same as _05 except that FS deletes occur during rebalancing.
     */
    @Test
    public void testBalancerSingleNodeClusterWithHalfCapacityNewNodeRebalanceWithConcurrentFSDeletes()
            throws IOException {
        // eraser thread
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_14
     * The same as _05 except that FS deletes AND writes occur during
     * rebalancing.
     */
    @Test
    public void testBalancerSingleNodeClusterWithHalfCapacityNewNodeRebalanceWithConcurrentFSDeletesAndWrites()
            throws IOException {
        // writer & eraser threads
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_15
     * Scalability test: Populate a 750-node cluster, then
     *    1. Run rebalancing after 3 nodes are added
     *    2. Run rebalancing after 2 racks of nodes (60 nodes) are added
     *    3. Run rebalancing after 2 racks of nodes are added and concurrently
     *       executing file writing and deleting at the same time
     */
    @Test
    public void testBalancerScalability() throws IOException {
        /* work in progress->
         *
         *
        List<DNClient> dnList = getAllDatanodes();
        int dnCount = dnList.size();

        Assert.assertTrue(
                String.format(
                    "not enough datanodes available to run test,"
                    + " need 2 datanodes but have only %d available",
                    dnCount),
                ( dnCount == (875 - 2) ));

        List<DNClient> datanodes = reserveDatanodesForTest(750);
        shutdownNonTestNodes(datanodes);
        */
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_16
     * Start balancer with a negative threshold value.
     */
    @Test
    public void testBalancerConfiguredWithThresholdValueNegative()
            throws IOException {
        List<DNClient> testnodes = getAllDatanodes();
        final int TRIALS=5;
        for(int i=0; i<TRIALS; i++) {
            int negThreshold = (int)(-1 * 100 * Math.random());
            runBalancerAndVerify(testnodes, negThreshold);
        }
    }

    /**
     * Balancer_17
     * Start balancer with out-of-range threshold value
     *  (e.g. -123, 0, -324, 100000, -12222222, 1000000000, -10000, 345, 989)
     */
    @Test
    public void testBalancerConfiguredWithThresholdValueOutOfRange()
            throws IOException {
        List<DNClient> testnodes = getAllDatanodes();
        final int[] THRESHOLD_OUT_OF_RANGE_DATA = {
            -123, 0, -324, 100000, -12222222, 1000000000, -10000, 345, 989
        };
        for(int threshold: THRESHOLD_OUT_OF_RANGE_DATA) {
            runBalancerAndVerify(testnodes, threshold);
        }
    }

    /**
     * Balancer_18
     * Start balancer with alpha-numeric threshold value
     *  (e.g., 103dsf, asd234, asfd, ASD, #$asd, 2345&, $35, %34)
     */
    @Test
    public void testBalancerConfiguredWithThresholdValueAlphanumeric()
            throws IOException {
        List<DNClient> testnodes = getAllDatanodes();
        final String[] THRESHOLD_ALPHA_DATA = {
            "103dsf", "asd234", "asfd", "ASD", "#$asd", "2345&", "$35", "%34",
            "0x64", "0xde", "0xad", "0xbe", "0xef"
        };
        for(String threshold: THRESHOLD_ALPHA_DATA) {
            runBalancerAndVerify(testnodes,threshold);
        }
    }

    /**
     * Balancer_19
     * Start 2 instances of balancer on the same gateway
     */
    @Test
    public void testBalancerRunTwoConcurrentInstancesOnSingleGateway()
            throws IOException {
        // do on gateway logic with small balancer heap
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_20
     * Start 2 instances of balancer on two different gateways
     */
    @Test
    public void testBalancerRunTwoConcurrentInstancesOnDistinctGateways()
            throws IOException {
            // do on gateway logic with small balancer heap
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * Balancer_21
     * Start balancer when the cluster is already balanced
     */
    @Test
    public void testBalancerOnBalancedCluster() throws IOException {
        // run balancer twice
        testBalancerSimple();
        testBalancerSimple();
    }

    /**
     * Balancer_22
     * Running the balancer with half the data nodes not running
     */
     @Test
     public void testBalancerWithOnlyHalfOfDataNodesRunning()
            throws IOException {
        List<DNClient> datanodes = getAllDatanodes();
        int testnodeCount = (int)Math.floor(datanodes.size() * 0.5);
        List<DNClient> testnodes = reserveDatanodesForTest(testnodeCount);
        runBalancerAndVerify(testnodes);
    }

    /**
     * Balancer_23
     * Running the balancer and simultaneously simulating load on the
     * cluster with half the data nodes not running.
     */
     @Test
     public void testBalancerOnBusyClusterWithOnlyHalfOfDatanodesRunning()
             throws IOException {
         // load thread
         throw new UnsupportedOperationException("not implemented yet!");
     }

    /**
     * Protocol Test Prelude
     *
     * First set up 3 node cluster with nodes NA, NB and NC, which are on
     * different racks. Then create a file with one block B with a replication
     * factor 3. Finally add a new node ND to the cluster on the same rack as NC.
     */

    /**
     * ProtocolTest_01
     * Copy block B from ND to NA with del hint NC
     */
    @Test
    public void
    testBlockReplacementProtocolFailWhenCopyBlockSourceDoesNotHaveBlockToCopy()
            throws IOException {
         throw new UnsupportedOperationException("not implemented yet!");
    }

    /*
     * ProtocolTest_02
     * Copy block B from NA to NB with del hint NB
     */
    @Test
    public void
    testBlockReplacementProtocolFailWhenCopyBlockDestinationContainsBlockCopy()
            throws IOException {
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * ProtocolTest_03
     * Copy block B from NA to ND with del hint NB
     */
    @Test
    public void testBlockReplacementProtocolCopyBlock() throws IOException {
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * ProtocolTest_04
     * Copy block B from NB to NC with del hint NA
     */
    @Test
    public void testBlockReplacementProtocolWithInvalidHint()
            throws IOException {
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * ThrottleTest_01
     * Create a throttler with 1MB/s bandwidth. Send 6MB data, and throttle
     * at 0.5MB, 0.75MB, and in the end [1MB/s?].
     */

    /**
     * NamenodeProtocolTest_01
     * Get blocks from datanode 0 with a size of 2 blocks.
     */
    @Test
    public void testNamenodeProtocolGetBlocksCheckThroughput()
            throws IOException {
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * NamenodeProtocolTest_02
     * Get blocks from datanode 0 with a size of 1 block.
     */
    @Test
    public void testNamenodeProtocolGetSingleBlock()
            throws IOException {
        throw new UnsupportedOperationException("not implemented yet!");
    }

    /**
     * NamenodeProtocolTest_03
     * Get blocks from datanode 0 with a size of 0.
     */
    @Test
    public void testNamenodeProtocolGetZeroBlocks() throws IOException {
        throw new UnsupportedOperationException("not implemented yet!");
    }
    /**
     * NamenodeProtocolTest_04
     * Get blocks from datanode 0 with a size of -1.
     */
    @Test
    public void testNamenodeProtocolGetMinusOneBlocks() throws Exception {

    }

    /**
     * NamenodeProtocolTest_05
     * Get blocks from a non-existent datanode.
     */
    @Test
    public void testNamenodeProtocolGetBlocksFromNonexistentDatanode()
            throws IOException {
        final short replication = 1;
        Path balancerTempDir = null;
        try {
        // reserve 2 nodes for test
        List<DNClient> testnodes = reserveDatanodesForTest(2);
        shutdownNonTestNodes(testnodes);

        DNClient testnode1 = testnodes.get(0);
        DNClient testnode2 = testnodes.get(1);

        // write some blocks with replication factor of 1
        balancerTempDir = makeTempDir();
        generateFileSystemLoad(20, replication);

        // get block locations from NN
        NNClient namenode = dfsCluster.getNNClient();
        // TODO extend namenode to get block locations
        //namenode.get

        // shutdown 1 node
        stopDatanode(testnode1);

        // attempt to retrieve blocks from the dead node
        // we should fail
        } finally {
            // cleanup
                        // finally block to run cleanup
            LOG.info("clean off test data from DFS [rmr ~/balancer-temp]");
            try {
                deleteTempDir(balancerTempDir);
            } catch (Exception e) {
                LOG.warn("problem cleaning up temp dir", e);
            }
        }
    }
}