hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindowManager.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.namenode.top.window;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.server.namenode.top.TopConf;

import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import static org.apache.hadoop.hdfs.server.namenode.top.window.RollingWindowManager.Op;
import static org.apache.hadoop.hdfs.server.namenode.top.window.RollingWindowManager.TopWindow;
import static org.apache.hadoop.hdfs.server.namenode.top.window.RollingWindowManager.User;
import static org.junit.Assert.assertEquals;

public class TestRollingWindowManager {

  Configuration conf;
  RollingWindowManager manager;
  String[] users;
  final static int MIN_2_MS = 60000;

  final int WINDOW_LEN_MS = 1 * MIN_2_MS;
  final int BUCKET_CNT = 10;
  final int N_TOP_USERS = 10;
  final int BUCKET_LEN = WINDOW_LEN_MS / BUCKET_CNT;

  @Before
  public void init() {
    conf = new Configuration();
    conf.setInt(DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY, BUCKET_CNT);
    conf.setInt(DFSConfigKeys.NNTOP_NUM_USERS_KEY, N_TOP_USERS);
    manager = new RollingWindowManager(conf, WINDOW_LEN_MS);
    users = new String[2 * N_TOP_USERS];
    for (int i = 0; i < users.length; i++) {
      users[i] = "user" + i;
    }
  }

  @Test
  public void testTops() throws Exception {
    long time = WINDOW_LEN_MS + BUCKET_LEN * 3 / 2;
    for (int i = 0; i < users.length; i++)
      manager.recordMetric(time, "open", users[i], (i + 1) * 2);
    time++;
    for (int i = 0; i < users.length; i++)
      manager.recordMetric(time, "close", users[i], i + 1);
    time++;
    TopWindow tops = manager.snapshot(time);

    assertEquals("Unexpected number of ops", 3, tops.getOps().size());
    assertEquals(TopConf.ALL_CMDS, tops.getOps().get(0).getOpType());
    for (Op op : tops.getOps()) {
      final List<User> topUsers = op.getTopUsers();
      assertEquals("Unexpected number of users", N_TOP_USERS, topUsers.size());
      if (op.getOpType().equals("open")) {
        for (int i = 0; i < topUsers.size(); i++) {
          User user = topUsers.get(i);
          assertEquals("Unexpected count for user " + user.getUser(),
              (users.length-i)*2, user.getCount());
        }
        // Closed form of sum(range(2,42,2))
        assertEquals("Unexpected total count for op", 
            (2+(users.length*2))*(users.length/2),
            op.getTotalCount());
      }
    }

    // move the window forward not to see the "open" results
    time += WINDOW_LEN_MS - 2;
    tops = manager.snapshot(time);
    assertEquals("Unexpected number of ops", 2, tops.getOps().size());
    assertEquals(TopConf.ALL_CMDS, tops.getOps().get(0).getOpType());
    final Op op = tops.getOps().get(1);
    assertEquals("Should only see close ops", "close", op.getOpType());
    final List<User> topUsers = op.getTopUsers();
    for (int i = 0; i < topUsers.size(); i++) {
      User user = topUsers.get(i);
      assertEquals("Unexpected count for user " + user.getUser(),
          (users.length-i), user.getCount());
    }
    // Closed form of sum(range(1,21))
    assertEquals("Unexpected total count for op",
        (1 + users.length) * (users.length / 2), op.getTotalCount());
  }

  @Test
  public void windowReset() throws Exception {
    Configuration config = new Configuration();
    config.setInt(DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY, 1);
    config.setInt(DFSConfigKeys.NNTOP_NUM_USERS_KEY, N_TOP_USERS);
    int period = 2;
    RollingWindowManager rollingWindowManager =
        new RollingWindowManager(config, period);
    rollingWindowManager.recordMetric(0, "op1", users[0], 3);
    checkValues(rollingWindowManager, 0, "op1", 3, 3);
    checkValues(rollingWindowManager, period - 1, "op1", 3, 3);
    checkValues(rollingWindowManager, period, "op1", 0, 0);
  }

  @Test
  public void testTotal() throws Exception {
    Configuration config = new Configuration();
    config.setInt(DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY, 1);
    config.setInt(DFSConfigKeys.NNTOP_NUM_USERS_KEY, N_TOP_USERS);
    int period = 10;
    RollingWindowManager rollingWindowManager =
        new RollingWindowManager(config, period);

    long t = 0;
    rollingWindowManager.recordMetric(t, "op1", users[0], 3);
    checkValues(rollingWindowManager, t, "op1", 3, 3);

    // both should have a value.
    t = (long)(period * .5);
    rollingWindowManager.recordMetric(t, "op2", users[0], 4);
    checkValues(rollingWindowManager, t, "op1", 3, 7);
    checkValues(rollingWindowManager, t, "op2", 4, 7);

    // neither should reset.
    t = period - 1;
    checkValues(rollingWindowManager, t, "op1", 3, 7);
    checkValues(rollingWindowManager, t, "op2", 4, 7);

    // op1 should reset in its next period, but not op2.
    t = period;
    rollingWindowManager.recordMetric(10, "op1", users[0], 10);
    checkValues(rollingWindowManager, t, "op1", 10, 14);
    checkValues(rollingWindowManager, t, "op2", 4, 14);

    // neither should reset.
    t = (long)(period * 1.25);
    rollingWindowManager.recordMetric(t, "op2", users[0], 7);
    checkValues(rollingWindowManager, t, "op1", 10, 21);
    checkValues(rollingWindowManager, t, "op2", 11, 21);

    // op2 should reset.
    t = (long)(period * 1.5);
    rollingWindowManager.recordMetric(t, "op2", users[0], 13);
    checkValues(rollingWindowManager, t, "op1", 10, 23);
    checkValues(rollingWindowManager, t, "op2", 13, 23);
  }

  @Test
  public void testWithFuzzing() throws Exception {
    Configuration config = new Configuration();
    config.setInt(DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY, 1);
    config.setInt(DFSConfigKeys.NNTOP_NUM_USERS_KEY, N_TOP_USERS);
    int period = users.length/2;
    RollingWindowManager rollingWindowManager =
        new RollingWindowManager(config, period);

    String[] ops = {"op1", "op2", "op3", "op4"};
    Random rand = new Random();
    for (int i=0; i < 10000; i++) {
      rollingWindowManager.recordMetric(i, ops[rand.nextInt(ops.length)],
          users[rand.nextInt(users.length)],
          rand.nextInt(100));
      TopWindow window = rollingWindowManager.snapshot(i);
      checkTotal(window);
    }
  }

  @Test
  public void testOpTotal() throws Exception {
    int numTopUsers = 2;
    Configuration config = new Configuration();
    config.setInt(DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY, 1);
    config.setInt(DFSConfigKeys.NNTOP_NUM_USERS_KEY, numTopUsers);
    int period = users.length/2;
    RollingWindowManager rollingWindowManager =
        new RollingWindowManager(config, period);

    int numOps = 3;
    rollingWindowManager.recordMetric(0, "op1", "user1", 10);
    rollingWindowManager.recordMetric(0, "op1", "user2", 20);
    rollingWindowManager.recordMetric(0, "op1", "user3", 30);

    rollingWindowManager.recordMetric(0, "op2", "user1", 1);
    rollingWindowManager.recordMetric(0, "op2", "user4", 40);
    rollingWindowManager.recordMetric(0, "op2", "user5", 50);

    rollingWindowManager.recordMetric(0, "op3", "user6", 1);
    rollingWindowManager.recordMetric(0, "op3", "user7", 11);
    rollingWindowManager.recordMetric(0, "op3", "user8", 1);

    TopWindow window = rollingWindowManager.snapshot(0);
    Assert.assertEquals(numOps + 1, window.getOps().size());

    Op allOp = window.getOps().get(0);
    Assert.assertEquals(TopConf.ALL_CMDS, allOp.getOpType());
    List<User> topUsers = allOp.getTopUsers();
    Assert.assertEquals(numTopUsers * numOps, topUsers.size());
    // ensure all the top users for each op are present in the total op.
    for (int i = 1; i < numOps; i++) {
      Assert.assertTrue(
          topUsers.containsAll(window.getOps().get(i).getTopUsers()));
    }
  }

  private void checkValues(RollingWindowManager rwManager, long time,
      String opType, long value, long expectedTotal) throws Exception {
    TopWindow window = rwManager.snapshot(time);
    for (Op windowOp : window.getOps()) {
      if (opType.equals(windowOp.getOpType())) {
        assertEquals(value, windowOp.getTotalCount());
        break;
      }
    }
    assertEquals(expectedTotal, checkTotal(window));
  }

  private long checkTotal(TopWindow window) throws Exception {
    long allOpTotal = 0;
    long computedOpTotal = 0;

    Map<String, User> userOpTally = new HashMap<>();
    for (String user : users) {
      userOpTally.put(user, new User(user, 0));
    }
    for (Op windowOp : window.getOps()) {
      int multiplier;
      if (TopConf.ALL_CMDS.equals(windowOp.getOpType())) {
        multiplier = -1;
        allOpTotal += windowOp.getTotalCount();
      } else {
        multiplier = 1;
        computedOpTotal += windowOp.getTotalCount();
      }
      for (User user : windowOp.getAllUsers()) {
        userOpTally.get(user.getUser()).add((int)(multiplier*user.getCount()));
      }
    }
    assertEquals(allOpTotal, computedOpTotal);
    for (String user : userOpTally.keySet()) {
      assertEquals(0, userOpTally.get(user).getCount());
    }
    return computedOpTotal;
  }
}