HDFS是Hadoop的基石，对其的使用情况进行监控显的尤为重要。那么怎么来衡量一个用户对hdfs的使用情况呢，用什么来衡量呢？众所周知HDFS是由NameNode和DataNode组成，而NameNode又是HDFS的核心，NameNode中最重要的就是记录整个集群状态的元数据。了解集群的使用情况也就是目前的状态，这些状态记录在元数据中，而元数据中是通过记录各个用户对HDFS的操作来描绘集群的状态，那么是不是可以通过统计各个用户在某个时间段对HDFS的操作数来衡量他们使用HDFS的资源。答案是肯定是，而且此功能已经merge到2.7版本中，下面就来了解下这个patch。

架构

nntop的架构图如下：
nntop架构图
nntop是在NN中的audit loggers中进行拦截，audit log从NN中接收所有user的操作。在audit log中进行拦截能够很好的扩展到其余的NN中。
HDFS默认的审计日志是DefaultAuditLogger，nntop是在HDFS中添加了一个新的审计日志类TopAuditLogger，由TopAuditLogger解析审计事件并传送给TopMetrics。TopMetrics中记录每个操作的top user list，TopMetrics也向Hadoop metrics system注册此metrics，实现了getMetrics方法，以便得到当前时间段的top user list。
TopMetrics能得到多个时间区间的top users，默认是1m，5m和25m，这是为了弥补没有画图接口。

细节设计

此功能的主要就是统计每个user的操作数，但是怎么统计呢？这是个问题。

提交patch的这位大神采用的滑动窗口的思想，并把当前窗口分为n个bucket(默认是10)，分别在每个bucket中进行计数，然后对其窗口内的bucket进行求和得到总和。每个bucket有一个时间标识updatetime，当updatetime与now的差值超过了窗口的长度，则将此bucket置0，更新updatetime，窗口也就右移了一个bucket。

这样设计很巧妙没毛病，但是HDFS中的操作类型和用户都很多，统计应该怎样逐级统计呢？又为了弥补此功能没有友好的画图接口，此patch支持对多个时间间隔进行统计，则大神将统计的流程设计为树形结构，统计的入口是TopMetrics，然后TopMetrics会根据设置的需要统计的时间区间个数初始化对应的RollingWindowManager(一个区间一个RollingWindowManager)，RollingWindowManager中又有一个metricsMap，用来记录metrics与RollingWindowMap的映射，RollingWindowMap中又是user和RollingWindow的映射，RollingWindow就是所谓的滑动窗口，里面由n个bucket进行统计。树形图如下：
统计流程

代码跟读

从上图的nntop架构图中可以看出是在写入auditlog之前进行了拦截，也就是在FSNamesystem.logAuditEvent()中进行判断的，logAuditEvent是将该操作在注册的各个auditsLoggers中都记录一边，auditsLoggers是通过initAuditLoggers向nn注册的，看下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private List<AuditLogger> initAuditLoggers(Configuration conf) {
  // Initialize the custom access loggers if configured.
  Collection<String> alClasses = conf.getStringCollection(DFS_NAMENODE_AUDIT_LOGGERS_KEY);
  List<AuditLogger> auditLoggers = Lists.newArrayList();
  ...
  // Make sure there is at least one logger installed.
  if (auditLoggers.isEmpty()) {
    auditLoggers.add(new DefaultAuditLogger());
  }
  // 开启nntop功能时，向auditsLoggers里注册TopAuditLogger
  // 开启的开关由dfs.namenode.top.enabled控制
  // Add audit logger to calculate top users
  if (topConf.isEnabled) {
    topMetrics = new TopMetrics(conf, topConf.nntopReportingPeriodsMs);
    auditLoggers.add(new TopAuditLogger(topMetrics));
  }

  return Collections.unmodifiableList(auditLoggers);
}

由代码可以看出DefaultAuditLogger和TopAuditLogger都向auditsLoggers中注册，则这两个auditLogger的功能应该类似，DefaultAuditLogger只是将操作相关的信息记录在audit-log中，而TopAuditLogger是在记录之前进行了一次metrics的report，此时的metrics是TopMetrics的对象，在构造TopAuditLogger对象时被传入。

TopAuditLogger初始化并向auditsLoggers中注册之后，当有hdfs操作时，会调用FSNamesystem.logAuditEvent()方法对操作进行记录，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private void logAuditEvent(boolean succeeded,
    UserGroupInformation ugi, InetAddress addr, String cmd, String src,
    String dst, HdfsFileStatus stat) {
  FileStatus status = null;
  ...
  for (AuditLogger logger : auditLoggers) {
  	// DefaultAuditLogger extends HdfsAuditLogger
  	// HdfsAuditLogger implements AuditLogger
    if (logger instanceof HdfsAuditLogger) {
      HdfsAuditLogger hdfsLogger = (HdfsAuditLogger) logger;
      hdfsLogger.logAuditEvent(succeeded, ugi.toString(), addr, cmd, src, dst,
          status, ugi, dtSecretManager);
    } else {
      // TopAuditLogger implements AuditLogger 
      // 则此处调用TopAuditLogger的logAuditEvent方法
      logger.logAuditEvent(succeeded, ugi.toString(), addr,
          cmd, src, dst, status);
    }
  }
}

FSNamesystem.logAuditEvent()中会对auditsLoggers进行遍历，将操作记录在所有的auditsLogger中，通过调用auditsLogger的logAuditEvent方法，看下TopAuditLogger.logAuditEvent方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public void logAuditEvent(boolean succeeded, String userName,
    InetAddress addr, String cmd, String src, String dst, FileStatus status) {
  // 核心代码
  try {
    topMetrics.report(succeeded, userName, addr, cmd, src, dst, status);
  } catch (Throwable t) {
    LOG.error("An error occurred while reflecting the event in top service, "
        + "event: (cmd={},userName={})", cmd, userName);
  }
  // 与DefaultAuditLogger中的logAuditEvent类似
  if (LOG.isDebugEnabled()) {
    final StringBuilder sb = new StringBuilder();
    sb.append("allowed=").append(succeeded).append("\t");
    sb.append("ugi=").append(userName).append("\t");
    sb.append("ip=").append(addr).append("\t");
    sb.append("cmd=").append(cmd).append("\t");
    sb.append("src=").append(src).append("\t");
    sb.append("dst=").append(dst).append("\t");
    if (null == status) {
      sb.append("perm=null");
    } else {
      sb.append("perm=");
      sb.append(status.getOwner()).append(":");
      sb.append(status.getGroup()).append(":");
      sb.append(status.getPermission());
    }
    LOG.debug("------------------- logged event for top service: " + sb);
  }
}

logAuditEvent在写入audit-log之前先向metrics系统report，report代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
public void report(long currTime, String userName, String cmd) {
  LOG.debug("a metric is reported: cmd: {} user: {}", cmd, userName);
  userName = UserGroupInformation.trimLoginMethod(userName);
  // 一个时间区间一个rollingWindowManager
  // 遍历所有的manager，让所有区间中的metrics加1
  for (RollingWindowManager rollingWindowManager : rollingWindowManagers
      .values()) {
    rollingWindowManager.recordMetric(currTime, cmd, userName, 1);
    rollingWindowManager.recordMetric(currTime,
        TopConf.ALL_CMDS, userName, 1);
  }
}

其中有一个重要的属性需要说一下rollingWindowManagers，rollingWindowManagers是一个key为Integer，value为RollingWindowManager的map，map的长度由dfs.namenode.top.windows.minutes属性的元素个数控制，key的值就是该属性的元素，该属性的值是一个数组，默认是{"1","5","25"}。此属性代表这我要统计三个时间区间的值，三个区间分别为1分钟、5分钟和25分钟。这样做是为了观察一个趋势。

每个时间区间对应一个RollingWindowManager，而rollingWindowManagers是存放这些manager的map，各个区间具体的metrics统计由各自的manager控制。manager通过调用recordMetric对metrics进行更新，代码如下：

1
2
3
4
public void recordMetric(long time, String command, String user, long delta) {
  RollingWindow window = getRollingWindow(command, user);
  window.incAt(time, delta);
}

RollingWindow是通过getRollingWindow得到的，从上面统计流程的分析中得知RollingWindow位于统计中的最底层，由RollingWindow进行滑动窗口进行统计，下面看下getRollingWindow代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
private RollingWindow getRollingWindow(String metric, String user) {
  // ConcurrentHashMap<String, RollingWindowMap> metricMap
  RollingWindowMap rwMap = metricMap.get(metric);
  if (rwMap == null) {
    rwMap = new RollingWindowMap();
    RollingWindowMap prevRwMap = metricMap.putIfAbsent(metric, rwMap);
    if (prevRwMap != null) {
      rwMap = prevRwMap;
    }
  }
  RollingWindow window = rwMap.get(user);
  if (window != null) {
    return window;
  }
  window = new RollingWindow(windowLenMs, bucketsPerWindow);
  RollingWindow prevWindow = rwMap.putIfAbsent(user, window);
  if (prevWindow != null) {
    window = prevWindow;
  }
  return window;
}

由此代码可以看出RollingWindow、RollingWindowMap、metricsMap和RollingWindowManager的关系，metricsMap是RollingWindowManager的一个属性，metricsMap是ConcurrentHashMap(线程安全)类型，key是操作名，value是RollingWindowMap，而RollingWindowMap也是ConcurrentHashMap类型的map，key是user name，value是RollingWindow。

recordMetric时通过command和user定位到RollingWindow，然后调用RollingWindow.incAt进行累加，代码如下：

1
2
3
4
5
6
7
8
9
10
public void incAt(long time, long delta) {
  int bi = computeBucketIndex(time);
  Bucket bucket = buckets[bi];
  // If the last time the bucket was updated is out of the scope of the
  // rolling window, reset the bucket.
  if (bucket.isStaleNow(time)) {
    bucket.safeReset(time);
  }
  bucket.inc(delta);
}

RollingWindow进行metrics累加时，通过当前time定位到某个bucket中，让bucket进行累加。bucket进行累加时先判断当前bucket是否过时，也就是说当前bucket是否还在以当前time为结尾的时间窗口中，不在则进行重置。

定位bucket的代码是在computeBucketIndex中，

1
2
3
4
5
6
private int computeBucketIndex(long time) {
  // time%windwoLenMs 得到time所在window中的偏移量
  int positionOnWindow = (int) (time % windowLenMs);
  int bucketIndex = positionOnWindow * buckets.length / windowLenMs;
  return bucketIndex;
}

计算time所在的bucket时，先计算time所在window的偏移量positionOnWindow*(time%windowLenMs)，然后计算每个bucket的时间长度time_length(windowLenMs/buckets.length)*，然后让positionOnWindow/time_length就得到了bucketIndex，也就是上面的代码bucketIndex = positionOnWindow * buckets.length / windowLenMs

然后调用bucket.isStaleNow判断是否过时，过时则重置。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
boolean isStaleNow(long time) {
  long utime = updateTime.get();
  return time - utime >= windowLenMs;
}

void safeReset(long time) {
  // At any point in time, only one thread is allowed to reset the
  // bucket
  synchronized (this) {
    if (isStaleNow(time)) {
      // reset the value before setting the time, it allows other
      // threads to safely assume that the value is updated if the
      // time is not stale
      value.set(0);
      updateTime.set(time);
    }
    // else a concurrent thread has already reset it: do nothing
  }
}

Bucket是RollingWindow的一个内部类，有两个属性value和updateTime，都是AtomicLong类型的，这里需要注意的是updateTime并不是在value每次递增的时候改变，updateTime只改变一次，其实相当于bucket开始计数的startTime，当次startTime与当前时间time的差值大于windowLenMs时，就认为bucket过时了，需要重置。

get统计数据

通过jmx获得数据的入口函数是getTopUserOpCounts，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public String getTopUserOpCounts() {
  ...
  Date now = new Date();
  // get 统计结果
  // private TopMetrics topMetrics;
  final List<RollingWindowManager.TopWindow> topWindows =
      topMetrics.getTopWindows();
  Map<String, Object> topMap = new TreeMap<String, Object>();
  topMap.put("windows", topWindows);
  topMap.put("timestamp", DFSUtil.dateToIso8601String(now));
  ObjectMapper mapper = new ObjectMapper();
  try {
    return mapper.writeValueAsString(topMap);
  } catch (IOException e) {
    LOG.warn("Failed to fetch TopUser metrics", e);
  }
  return null;
}

通过TopMetrics.getTopWindows得到

1
2
3
4
5
6
7
8
9
10
11
public List<TopWindow> getTopWindows() {
  long monoTime = Time.monotonicNow();
  List<TopWindow> windows = Lists.newArrayListWithCapacity
      (rollingWindowManagers.size());
  for (Entry<Integer, RollingWindowManager> entry : rollingWindowManagers
      .entrySet()) {
    TopWindow window = entry.getValue().snapshot(monoTime);
    windows.add(window);
  }
  return windows;
}

getTopWindows只是遍历rollingWindowManagers得到多个时间段的统计结果，而真正取得topuser的是各个时间区间对应的RollingWindowManager.snapshot。

在跟读snapshot代码之前，先跟读下RollingWindowManager类，此类中有一些内部类，

首先看下TopWindow、Op和User，这三者的关系是TopWindow中有个属性top，类型List<Op>，Op有一个属性topUsers，类型是List<User>。TopWindow是当前window中各个metrics的一个快照，各个metrics以Op为对象存放在top的list中，list中一个Op就是一个metrics，Op中存放的是操作此metrics的所有user及其操作总次数，Op中的各个user信息是存放在User对象中，User对象中记录着当前user对Op的操作次数。

还有两个内部类TopN和NameValuePair，TopN是一个优先级队列(PriorityQueue，关于优先级队列会在随后的blog中进行分析)，存放的元素是NameValuePair对象，NameValuePair实现了Comparable接口，重写compareTo方法，实现NameValuePair对象以value的大小作为比较的依据*(NameValuePair中的name是user，value是user对Op的操作次数)*。NameValuePair实现了自己定义的compareTo方法之后在TopN的优先级队列中就可以依靠value的值进行排序，则TopN的对象就是操作过Op的所有user的topN排序的结果。

RollingWindowManager中的内部类的关系屡清楚之后就来看下snapshot代码，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public TopWindow snapshot(long time) {
  TopWindow window = new TopWindow(windowLenMs);
  ...
  // metricMap 存放的是 metrics -> RollingWindowMap
  for (Map.Entry<String, RollingWindowMap> entry : metricMap.entrySet()) {
    String metricName = entry.getKey();
    // RollingWindowMap 存放的是 username -> RollingWindow
    RollingWindowMap rollingWindows = entry.getValue();
    // 通过metricName和time得到当前window中user的topN
    // 顺序是从小到大，升序
    TopN topN = getTopUsersForMetric(time, metricName, rollingWindows);
    final int size = topN.size();
    if (size == 0) {
      continue;
    }
    Op op = new Op(metricName, topN.getTotal());
    window.addOp(op);
    // Reverse the users from the TopUsers using a stack, 
    // since we'd like them sorted in descending rather than ascending order
    // 栈，利用栈先进后出的特点，将topN中的升序变为降序
    Stack<NameValuePair> reverse = new Stack<NameValuePair>();
    for (int i = 0; i < size; i++) {
      reverse.push(topN.poll());
    }
    for (int i = 0; i < size; i++) {
      NameValuePair userEntry = reverse.pop();
      // topN中的NameValuePair以User的形态存放到Op的topUsers list中
      User user = new User(userEntry.name, Long.valueOf(userEntry.value));
      op.addUser(user);
    }
  }
  return window;
}

snapshot将当前状态存放在TopWindow中返回，每个Op的topUsers是通过方法getTopUsersForMetric得到的，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
private TopN getTopUsersForMetric(long time, String metricName, 
    RollingWindowMap rollingWindows) {
  TopN topN = new TopN(topUsersCnt);
  Iterator<Map.Entry<String, RollingWindow>> iterator =
      rollingWindows.entrySet().iterator();
  while (iterator.hasNext()) {
    Map.Entry<String, RollingWindow> entry = iterator.next();
    String userName = entry.getKey();
    RollingWindow aWindow = entry.getValue();
    // getSum 得到当前window中各个bucket计数的总和
    long windowSum = aWindow.getSum(time);
    // do the gc here
    if (windowSum == 0) {
      LOG.debug("gc window of metric: {} userName: {}",
          metricName, userName);
      iterator.remove();
      continue;
    }
    LOG.debug("offer window of metric: {} userName: {} sum: {}",
        metricName, userName, windowSum);
    // 将NameValuePair放入TopN中，重写offer，对NameValuePair进行排序放入堆中
    topN.offer(new NameValuePair(userName, windowSum));
  }
  LOG.info("topN size for command {} is: {}", metricName, topN.size());
  return topN;
}

附加

  • window中bucket的个数关系到统计的精度，window中bucket的个数越多，精度越高。
    解析：window中bucket中的统计次数是以bucket的updateTime为开始时间，则如果window的长度是60s，bucket的个数是6个，则每个bucket的长度是10s。如果从时间0开始统计，则bucket0的updateTime是第0s，bucket1的updateTime是第10s，bucket9的updateTime是第50s，在65时，用户调用getTopUserOpCounts方法统计第65s时的topN user，则当前window应该是5s到65s，可是bucket0统计的时间区间是0-10s，此时bucket0的updateTime已经超过了统计的阈值，bucket0已经过时，则此时统计的区间只能从bucket1，也就是从10s处开始统计，从10s-65s处，也就是少统计了5s，即5s-10s区间中的值，误差是5s，如果从bucket的长度为5s，则统计的结果值就是正确的，不会出现误差，则得出结论，bucket个数越多，则精度越高。但是bucket也不能太多，会损耗性能。

  • 为什么采用滑动窗口

参考

Patch地址
HDFS-6982设计文档
HDFS nnTop统计功能