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