Repository: hadoop Updated Branches: refs/heads/branch-2 8df4c04c3 -> 541172f2e
HDFS-6982. nntop: topÂ-like tool for name node users. (Maysam Yabandeh via wang) (cherry picked from commit dcb8e24427b02e2f3ff9a12d2eb1eb878e3443bb) Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/541172f2 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/541172f2 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/541172f2 Branch: refs/heads/branch-2 Commit: 541172f2ee99090cb23b895e40826b9eb1c22bc7 Parents: 8df4c04 Author: Andrew Wang <[email protected]> Authored: Mon Nov 17 17:31:42 2014 -0800 Committer: Andrew Wang <[email protected]> Committed: Mon Nov 17 17:34:02 2014 -0800 ---------------------------------------------------------------------- .../hadoop/security/UserGroupInformation.java | 17 +- hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 3 + .../org/apache/hadoop/hdfs/DFSConfigKeys.java | 15 ++ .../hdfs/server/namenode/FSNamesystem.java | 21 +- .../server/namenode/top/TopAuditLogger.java | 74 ++++++ .../hdfs/server/namenode/top/TopConf.java | 61 +++++ .../server/namenode/top/metrics/TopMetrics.java | 265 +++++++++++++++++++ .../namenode/top/window/RollingWindow.java | 189 +++++++++++++ .../top/window/RollingWindowManager.java | 265 +++++++++++++++++++ .../src/main/resources/hdfs-default.xml | 28 ++ .../hdfs/server/namenode/TestAuditLogger.java | 29 +- .../namenode/metrics/TestNameNodeMetrics.java | 61 +++++ .../namenode/top/window/TestRollingWindow.java | 84 ++++++ .../top/window/TestRollingWindowManager.java | 93 +++++++ 14 files changed, 1200 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java ---------------------------------------------------------------------- diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java index 7fb036c..bb75ce8 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java @@ -775,7 +775,22 @@ public class UserGroupInformation { } return loginUser; } - + + /** + * remove the login method that is followed by a space from the username + * e.g. "jack (auth:SIMPLE)" -> "jack" + * + * @param userName + * @return userName without login method + */ + public static String trimLoginMethod(String userName) { + int spaceIndex = userName.indexOf(' '); + if (spaceIndex >= 0) { + userName = userName.substring(0, spaceIndex); + } + return userName; + } + /** * Log in a user using the given subject * @parma subject the subject to use when logging in a user, or null to http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt index c5e9f82..bce0668 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt +++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt @@ -12,6 +12,9 @@ Release 2.7.0 - UNRELEASED HDFS-6663. Admin command to track file and locations from block id. (Chen He via kihwal) + HDFS-6982. nntop: topÂ-like tool for name node users. + (Maysam Yabandeh via wang) + IMPROVEMENTS HDFS-7055. Add tracing to DFSInputStream (cmccabe) http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java index fd313bb..46b409e 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java @@ -746,4 +746,19 @@ public class DFSConfigKeys extends CommonConfigurationKeys { public static final String IGNORE_SECURE_PORTS_FOR_TESTING_KEY = "ignore.secure.ports.for.testing"; public static final boolean IGNORE_SECURE_PORTS_FOR_TESTING_DEFAULT = false; + + // nntop Configurations + public static final String NNTOP_ENABLED_KEY = + "dfs.namenode.top.enabled"; + public static final boolean NNTOP_ENABLED_DEFAULT = true; + public static final String NNTOP_BUCKETS_PER_WINDOW_KEY = + "dfs.namenode.top.window.num.buckets"; + public static final int NNTOP_BUCKETS_PER_WINDOW_DEFAULT = 10; + public static final String NNTOP_NUM_USERS_KEY = + "dfs.namenode.top.num.users"; + public static final int NNTOP_NUM_USERS_DEFAULT = 10; + // comma separated list of nntop reporting periods in minutes + public static final String NNTOP_WINDOWS_MINUTES_KEY = + "dfs.namenode.top.windows.minutes"; + public static final String[] NNTOP_WINDOWS_MINUTES_DEFAULT = {"1","5","25"}; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java index 1514245..9e51b2d 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java @@ -243,6 +243,9 @@ import org.apache.hadoop.hdfs.server.namenode.startupprogress.StartupProgress.Co import org.apache.hadoop.hdfs.server.namenode.startupprogress.Status; import org.apache.hadoop.hdfs.server.namenode.startupprogress.Step; import org.apache.hadoop.hdfs.server.namenode.startupprogress.StepType; +import org.apache.hadoop.hdfs.server.namenode.top.TopAuditLogger; +import org.apache.hadoop.hdfs.server.namenode.top.TopConf; +import org.apache.hadoop.hdfs.server.namenode.top.metrics.TopMetrics; import org.apache.hadoop.hdfs.server.namenode.web.resources.NamenodeWebHdfsMethods; import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand; import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration; @@ -872,7 +875,12 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean, throw re; } } - + + @VisibleForTesting + public List<AuditLogger> getAuditLoggers() { + return auditLoggers; + } + @VisibleForTesting public RetryCache getRetryCache() { return retryCache; @@ -961,6 +969,17 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean, if (auditLoggers.isEmpty()) { auditLoggers.add(new DefaultAuditLogger()); } + + // Add audit logger to calculate top users + if (conf.getBoolean(DFSConfigKeys.NNTOP_ENABLED_KEY, + DFSConfigKeys.NNTOP_ENABLED_DEFAULT)) { + String sessionId = conf.get(DFSConfigKeys.DFS_METRICS_SESSION_ID_KEY); + TopConf nntopConf = new TopConf(conf); + TopMetrics.initSingleton(conf, NamenodeRole.NAMENODE.name(), sessionId, + nntopConf.nntopReportingPeriodsMs); + auditLoggers.add(new TopAuditLogger()); + } + return Collections.unmodifiableList(auditLoggers); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java new file mode 100644 index 0000000..4f26b17 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java @@ -0,0 +1,74 @@ +/** + * 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; + +import java.net.InetAddress; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.hdfs.server.namenode.AuditLogger; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hadoop.hdfs.server.namenode.top.metrics.TopMetrics; + +/** + * An {@link AuditLogger} that sends logged data directly to the metrics + * systems. It is used when the top service is used directly by the name node + */ [email protected] +public class TopAuditLogger implements AuditLogger { + public static final Logger LOG = LoggerFactory.getLogger(TopAuditLogger.class); + + @Override + public void initialize(Configuration conf) { + } + + @Override + public void logAuditEvent(boolean succeeded, String userName, + InetAddress addr, String cmd, String src, String dst, FileStatus status) { + + TopMetrics instance = TopMetrics.getInstance(); + if (instance != null) { + instance.report(succeeded, userName, addr, cmd, src, dst, status); + } else { + LOG.error("TopMetrics is not initialized yet!"); + } + + 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); + } + } + +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopConf.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopConf.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopConf.java new file mode 100644 index 0000000..0f4ebac --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopConf.java @@ -0,0 +1,61 @@ +/** + * 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; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import com.google.common.base.Preconditions; + +/** + * This class is a common place for NNTop configuration. + */ [email protected] +public final class TopConf { + + public static final String TOP_METRICS_REGISTRATION_NAME = "topusers"; + public static final String TOP_METRICS_RECORD_NAME = "topparam"; + /** + * A meta command representing the total number of commands + */ + public static final String CMD_TOTAL = "total"; + /** + * A meta user representing all users + */ + public static String ALL_USERS = "ALL"; + + /** + * nntop reporting periods in milliseconds + */ + public final long[] nntopReportingPeriodsMs; + + public TopConf(Configuration conf) { + String[] periodsStr = conf.getTrimmedStrings( + DFSConfigKeys.NNTOP_WINDOWS_MINUTES_KEY, + DFSConfigKeys.NNTOP_WINDOWS_MINUTES_DEFAULT); + nntopReportingPeriodsMs = new long[periodsStr.length]; + for (int i = 0; i < periodsStr.length; i++) { + nntopReportingPeriodsMs[i] = Integer.parseInt(periodsStr[i]) * + 60L * 1000L; //min to ms + } + for (long aPeriodMs: nntopReportingPeriodsMs) { + Preconditions.checkArgument(aPeriodMs >= 60L * 1000L, + "minimum reporting period is 1 min!"); + } + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/metrics/TopMetrics.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/metrics/TopMetrics.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/metrics/TopMetrics.java new file mode 100644 index 0000000..e8a4e23 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/metrics/TopMetrics.java @@ -0,0 +1,265 @@ +/** + * 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.metrics; + +import static org.apache.hadoop.metrics2.impl.MsInfo.ProcessName; +import static org.apache.hadoop.metrics2.impl.MsInfo.SessionId; +import static org.apache.hadoop.metrics2.lib.Interns.info; + +import java.net.InetAddress; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.server.namenode.top.TopAuditLogger; +import org.apache.hadoop.hdfs.server.namenode.top.TopConf; +import org.apache.hadoop.metrics2.MetricsCollector; +import org.apache.hadoop.metrics2.MetricsRecordBuilder; +import org.apache.hadoop.metrics2.MetricsSource; +import org.apache.hadoop.metrics2.MetricsSystem; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.util.Time; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hadoop.hdfs.server.namenode.top.window.RollingWindowManager; +import org.apache.hadoop.hdfs.server.namenode.top.window.RollingWindowManager.MetricValueMap; + +/*** + * The interface to the top metrics + * <p/> + * The producers use the {@link #report} method to report events and the + * consumers use {@link #getMetrics(MetricsCollector, boolean)} to retrieve the + * current top metrics. The default consumer is JMX but it could be any other + * user interface. + * <p/> + * Thread-safe: relies on thread-safety of RollingWindowManager + */ [email protected] +public class TopMetrics implements MetricsSource { + public static final Logger LOG = LoggerFactory.getLogger(TopMetrics.class); + + enum Singleton { + INSTANCE; + + volatile TopMetrics impl = null; + + synchronized TopMetrics init(Configuration conf, String processName, + String sessionId, long[] reportingPeriods) { + if (impl == null) { + impl = + create(conf, processName, sessionId, reportingPeriods, + DefaultMetricsSystem.instance()); + } + logConf(conf); + return impl; + } + } + + private static void logConf(Configuration conf) { + LOG.info("NNTop conf: " + DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY + + " = " + conf.get(DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY)); + LOG.info("NNTop conf: " + DFSConfigKeys.NNTOP_NUM_USERS_KEY + + " = " + conf.get(DFSConfigKeys.NNTOP_NUM_USERS_KEY)); + LOG.info("NNTop conf: " + DFSConfigKeys.NNTOP_WINDOWS_MINUTES_KEY + + " = " + conf.get(DFSConfigKeys.NNTOP_WINDOWS_MINUTES_KEY)); + } + + /** + * Return only the shortest periods for default + * TODO: make it configurable + */ + final boolean smallestOnlyDefault = true; + + /** + * The smallest of reporting periods + */ + long smallestPeriod = Long.MAX_VALUE; + + /** + * processName and sessionId might later be leveraged later when we aggregate + * report from multiple federated name nodes + */ + final String processName, sessionId; + + /** + * A map from reporting periods to WindowManager. Thread-safety is provided by + * the fact that the mapping is not changed after construction. + */ + final Map<Long, RollingWindowManager> rollingWindowManagers = + new HashMap<Long, RollingWindowManager>(); + + TopMetrics(Configuration conf, String processName, String sessionId, + long[] reportingPeriods) { + this.processName = processName; + this.sessionId = sessionId; + for (int i = 0; i < reportingPeriods.length; i++) { + smallestPeriod = Math.min(smallestPeriod, reportingPeriods[i]); + rollingWindowManagers.put(reportingPeriods[i], new RollingWindowManager( + conf, reportingPeriods[i])); + } + } + + public static TopMetrics create(Configuration conf, String processName, + String sessionId, long[] reportingPeriods, MetricsSystem ms) { + return ms.register(TopConf.TOP_METRICS_REGISTRATION_NAME, + "top metrics of the namenode in a last period of time", new TopMetrics( + conf, processName, sessionId, reportingPeriods)); + } + + public static TopMetrics initSingleton(Configuration conf, + String processName, String sessionId, long[] reportingPeriods) { + return Singleton.INSTANCE.init(conf, processName, sessionId, + reportingPeriods); + } + + public static TopMetrics getInstance() { + TopMetrics topMetrics = Singleton.INSTANCE.impl; + Preconditions.checkArgument(topMetrics != null, + "The TopMetric singleton instance is not initialized." + + " Have you called initSingleton first?"); + return topMetrics; + } + + /** + * In testing, the previous initialization should be reset if the entire + * metric system is reinitialized + */ + @VisibleForTesting + public static void reset() { + Singleton.INSTANCE.impl = null; + } + + @Override + public void getMetrics(MetricsCollector collector, boolean all) { + long realTime = Time.monotonicNow(); + getMetrics(smallestOnlyDefault, realTime, collector, all); + } + + public void getMetrics(boolean smallestOnly, long currTime, + MetricsCollector collector, boolean all) { + for (Entry<Long, RollingWindowManager> entry : rollingWindowManagers + .entrySet()) { + if (!smallestOnly || smallestPeriod == entry.getKey()) { + getMetrics(currTime, collector, entry.getKey(), entry.getValue(), all); + } + } + } + + /** + * Get metrics for a particular recording period and its corresponding + * {@link RollingWindowManager} + * <p/> + * + * @param collector the metric collector + * @param period the reporting period + * @param rollingWindowManager the window manager corresponding to the + * reporting period + * @param all currently ignored + */ + void getMetrics(long currTime, MetricsCollector collector, Long period, + RollingWindowManager rollingWindowManager, boolean all) { + MetricsRecordBuilder rb = + collector.addRecord(createTopMetricsRecordName(period)) + .setContext("namenode").tag(ProcessName, processName) + .tag(SessionId, sessionId); + + MetricValueMap snapshotMetrics = rollingWindowManager.snapshot(currTime); + LOG.debug("calling snapshot, result size is: " + snapshotMetrics.size()); + for (Map.Entry<String, Number> entry : snapshotMetrics.entrySet()) { + String key = entry.getKey(); + Number value = entry.getValue(); + LOG.debug("checking an entry: key: {} value: {}", key, value); + long min = period / 1000L / 60L; //ms -> min + String desc = "top user of name node in the past " + min + " minutes"; + + if (value instanceof Integer) { + rb.addGauge(info(key, desc), (Integer) value); + } else if (value instanceof Long) { + rb.addGauge(info(key, desc), (Long) value); + } else if (value instanceof Float) { + rb.addGauge(info(key, desc), (Float) value); + } else if (value instanceof Double) { + rb.addGauge(info(key, desc), (Double) value); + } else { + LOG.warn("Unsupported metric type: " + value.getClass()); + } + } + LOG.debug("END iterating over metrics, result size is: {}", + snapshotMetrics.size()); + } + + /** + * Pick the same information that DefaultAuditLogger does before writing to a + * log file. This is to be consistent when {@link TopMetrics} is charged with + * data read back from log files instead of being invoked directly by the + * FsNamesystem + * + * @param succeeded + * @param userName + * @param addr + * @param cmd + * @param src + * @param dst + * @param status + */ + public void report(boolean succeeded, String userName, InetAddress addr, + String cmd, String src, String dst, FileStatus status) { + //currently we nntop makes use of only the username and the command + report(userName, cmd); + } + + public void report(String userName, String cmd) { + long currTime = Time.monotonicNow(); + report(currTime, userName, cmd); + } + + public void report(long currTime, String userName, String cmd) { + LOG.debug("a metric is reported: cmd: {} user: {}", cmd, userName); + userName = UserGroupInformation.trimLoginMethod(userName); + try { + for (RollingWindowManager rollingWindowManager : rollingWindowManagers + .values()) { + rollingWindowManager.recordMetric(currTime, cmd, userName, 1); + rollingWindowManager.recordMetric(currTime, + TopConf.CMD_TOTAL, userName, 1); + } + } catch (Throwable t) { + LOG.error("An error occurred while reflecting the event in top service, " + + "event: (time,cmd,userName)=(" + currTime + "," + cmd + "," + + userName); + } + } + + /*** + * + * @param period the reporting period length in ms + * @return + */ + public static String createTopMetricsRecordName(Long period) { + return TopConf.TOP_METRICS_RECORD_NAME + "-" + period; + } + +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindow.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindow.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindow.java new file mode 100644 index 0000000..63ff125 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindow.java @@ -0,0 +1,189 @@ +/** + * 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.Date; +import java.util.concurrent.atomic.AtomicLong; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A class for exposing a rolling window view on the event that occur over time. + * Events are reported based on occurrence time. The total number of events in + * the last period covered by the rolling window can be retrieved by the + * {@link #getSum(long)} method. + * <p/> + * + * Assumptions: + * <p/> + * + * (1) Concurrent invocation of {@link #incAt} method are possible + * <p/> + * + * (2) The time parameter of two consecutive invocation of {@link #incAt} could + * be in any given order + * <p/> + * + * (3) The buffering delays are not more than the window length, i.e., after two + * consecutive invocation {@link #incAt(long time1, long)} and + * {@link #incAt(long time2, long)}, time1 < time2 || time1 - time2 < windowLenMs. + * This assumption helps avoiding unnecessary synchronizations. + * <p/> + * + * Thread-safety is built in the {@link RollingWindow.Bucket} + */ [email protected] +public class RollingWindow { + private static final Logger LOG = LoggerFactory.getLogger(RollingWindow.class); + + /** + * Each window is composed of buckets, which offer a trade-off between + * accuracy and space complexity: the lower the number of buckets, the less + * memory is required by the rolling window but more inaccuracy is possible in + * reading window total values. + */ + Bucket[] buckets; + final int windowLenMs; + final int bucketSize; + + /** + * @param windowLenMs The period that is covered by the window. This period must + * be more than the buffering delays. + * @param numBuckets number of buckets in the window + */ + RollingWindow(int windowLenMs, int numBuckets) { + buckets = new Bucket[numBuckets]; + for (int i = 0; i < numBuckets; i++) { + buckets[i] = new Bucket(); + } + this.windowLenMs = windowLenMs; + this.bucketSize = windowLenMs / numBuckets; + if (this.bucketSize % bucketSize != 0) { + throw new IllegalArgumentException( + "The bucket size in the rolling window is not integer: windowLenMs= " + + windowLenMs + " numBuckets= " + numBuckets); + } + } + + /** + * When an event occurs at the specified time, this method reflects that in + * the rolling window. + * <p/> + * + * @param time the time at which the event occurred + * @param delta the delta that will be added to the window + */ + 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); + } + + private int computeBucketIndex(long time) { + int positionOnWindow = (int) (time % windowLenMs); + int bucketIndex = positionOnWindow * buckets.length / windowLenMs; + return bucketIndex; + } + + /** + * Thread-safety is provided by synchronization when resetting the update time + * as well as atomic fields. + */ + private class Bucket { + AtomicLong value = new AtomicLong(0); + AtomicLong updateTime = new AtomicLong(0); + + /** + * Check whether the last time that the bucket was updated is no longer + * covered by rolling window. + * + * @param time the current time + * @return true if the bucket state is stale + */ + boolean isStaleNow(long time) { + long utime = updateTime.get(); + return time - utime >= windowLenMs; + } + + /** + * Safely reset the bucket state considering concurrent updates (inc) and + * resets. + * + * @param time the current time + */ + 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 + } + } + + /** + * Increment the bucket. It assumes that staleness check is already + * performed. We do not need to update the {@link #updateTime} because as + * long as the {@link #updateTime} belongs to the current view of the + * rolling window, the algorithm works fine. + */ + void inc(long delta) { + value.addAndGet(delta); + } + } + + /** + * Get value represented by this window at the specified time + * <p/> + * + * If time lags behind the latest update time, the new updates are still + * included in the sum + * + * @param time + * @return number of events occurred in the past period + */ + public long getSum(long time) { + long sum = 0; + for (Bucket bucket : buckets) { + boolean stale = bucket.isStaleNow(time); + if (!stale) { + sum += bucket.value.get(); + } + if (LOG.isDebugEnabled()) { + long bucketTime = bucket.updateTime.get(); + String timeStr = new Date(bucketTime).toString(); + LOG.debug("Sum: + " + sum + " Bucket: updateTime: " + timeStr + " (" + + bucketTime + ") isStale " + stale + " at " + time); + } + } + return sum; + } + +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindowManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindowManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindowManager.java new file mode 100644 index 0000000..d818cce --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/window/RollingWindowManager.java @@ -0,0 +1,265 @@ +/** + * 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.Iterator; +import java.util.Map; +import java.util.PriorityQueue; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.server.namenode.top.TopConf; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.hadoop.conf.Configuration; + +/** + * A class to manage the set of {@link RollingWindow}s. This class is the + * interface of metrics system to the {@link RollingWindow}s to retrieve the + * current top metrics. + * <p/> + * Thread-safety is provided by each {@link RollingWindow} being thread-safe as + * well as {@link ConcurrentHashMap} for the collection of them. + */ [email protected] +public class RollingWindowManager { + public static final Logger LOG = LoggerFactory.getLogger( + RollingWindowManager.class); + + private int windowLenMs; + private int bucketsPerWindow; // e.g., 10 buckets per minute + private int topUsersCnt; // e.g., report top 10 metrics + + /** + * Create a metric name composed of the command and user + * + * @param command the command executed + * @param user the user + * @return a composed metric name + */ + @VisibleForTesting + public static String createMetricName(String command, String user) { + return command + "." + user; + } + + static private class RollingWindowMap extends + ConcurrentHashMap<String, RollingWindow> { + private static final long serialVersionUID = -6785807073237052051L; + } + + /** + * A mapping from each reported metric to its {@link RollingWindowMap} that + * maintains the set of {@link RollingWindow}s for the users that have + * operated on that metric. + */ + public ConcurrentHashMap<String, RollingWindowMap> metricMap = + new ConcurrentHashMap<String, RollingWindowMap>(); + + public RollingWindowManager(Configuration conf, long reportingPeriodMs) { + windowLenMs = (int) reportingPeriodMs; + bucketsPerWindow = + conf.getInt(DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_KEY, + DFSConfigKeys.NNTOP_BUCKETS_PER_WINDOW_DEFAULT); + Preconditions.checkArgument(bucketsPerWindow > 0, + "a window should have at least one bucket"); + Preconditions.checkArgument(bucketsPerWindow <= windowLenMs, + "the minimum size of a bucket is 1 ms"); + //same-size buckets + Preconditions.checkArgument(windowLenMs % bucketsPerWindow == 0, + "window size must be a multiplication of number of buckets"); + topUsersCnt = + conf.getInt(DFSConfigKeys.NNTOP_NUM_USERS_KEY, + DFSConfigKeys.NNTOP_NUM_USERS_DEFAULT); + Preconditions.checkArgument(topUsersCnt > 0, + "the number of requested top users must be at least 1"); + } + + /** + * Called when the metric command is changed by "delta" units at time "time" + * via user "user" + * + * @param time the time of the event + * @param command the metric that is updated, e.g., the operation name + * @param user the user that updated the metric + * @param delta the amount of change in the metric, e.g., +1 + */ + public void recordMetric(long time, String command, String user, long delta) { + RollingWindow window = getRollingWindow(command, user); + window.incAt(time, delta); + } + + /** + * Take a snapshot of current top users in the past period. + * + * @param time the current time + * @return a map between the top metrics and their values. The user is encoded + * in the metric name. Refer to {@link RollingWindowManager#createMetricName} for + * the actual format. + */ + public MetricValueMap snapshot(long time) { + MetricValueMap map = new MetricValueMap(); + Set<String> metricNames = metricMap.keySet(); + LOG.debug("iterating in reported metrics, size={} values={}", + metricNames.size(), metricNames); + for (Map.Entry<String,RollingWindowMap> rwEntry: metricMap.entrySet()) { + String metricName = rwEntry.getKey(); + RollingWindowMap rollingWindows = rwEntry.getValue(); + 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(); + 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); + topN.offer(new NameValuePair(userName, windowSum)); + } + int n = topN.size(); + LOG.info("topN size for command " + metricName + " is: " + n); + if (n == 0) { + continue; + } + String allMetricName = + createMetricName(metricName, TopConf.ALL_USERS); + map.put(allMetricName, Long.valueOf(topN.total)); + for (int i = 0; i < n; i++) { + NameValuePair userEntry = topN.poll(); + String userMetricName = + createMetricName(metricName, userEntry.name); + map.put(userMetricName, Long.valueOf(userEntry.value)); + } + } + return map; + } + + /** + * Get the rolling window specified by metric and user. + * + * @param metric the updated metric + * @param user the user that updated the metric + * @return the rolling window + */ + private RollingWindow getRollingWindow(String metric, String user) { + 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; + } + + /** + * A pair of a name and its corresponding value + */ + static private class NameValuePair implements Comparable<NameValuePair> { + String name; + long value; + + public NameValuePair(String metricName, long value) { + this.name = metricName; + this.value = value; + } + + @Override + public int compareTo(NameValuePair other) { + return (int) (value - other.value); + } + + @Override + public boolean equals(Object other) { + if (other instanceof NameValuePair) { + return compareTo((NameValuePair)other) == 0; + } + return false; + } + + @Override + public int hashCode() { + return Long.valueOf(value).hashCode(); + } + } + + /** + * A fixed-size priority queue, used to retrieve top-n of offered entries. + */ + static private class TopN extends PriorityQueue<NameValuePair> { + private static final long serialVersionUID = 5134028249611535803L; + int n; // > 0 + private long total = 0; + + TopN(int n) { + super(n); + this.n = n; + } + + @Override + public boolean offer(NameValuePair entry) { + updateTotal(entry.value); + if (size() == n) { + NameValuePair smallest = peek(); + if (smallest.value >= entry.value) { + return false; + } + poll(); // remove smallest + } + return super.offer(entry); + } + + private void updateTotal(long value) { + total += value; + } + + public long getTotal() { + return total; + } + } + + /** + * A mapping from metric names to their absolute values and their percentage + */ + @InterfaceAudience.Private + public static class MetricValueMap extends HashMap<String, Number> { + private static final long serialVersionUID = 8936732010242400171L; + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml index edd8a9f..4f489b4 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml @@ -2224,4 +2224,32 @@ </description> </property> +<property> + <name>dfs.namenode.top.enabled</name> + <value>true</value> + <description>Enable nntop: reporting top users on namenode + </description> +</property> + +<property> + <name>dfs.namenode.top.window.num.buckets</name> + <value>10</value> + <description>Number of buckets in the rolling window implementation of nntop + </description> +</property> + +<property> + <name>dfs.namenode.top.num.users</name> + <value>10</value> + <description>Number of top users returned by the top tool + </description> +</property> + +<property> + <name>dfs.namenode.top.windows.minutes</name> + <value>1,5,25</value> + <description>comma separated list of nntop reporting periods in minutes + </description> +</property> + </configuration> http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestAuditLogger.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestAuditLogger.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestAuditLogger.java index e1e1c67..c91cd75 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestAuditLogger.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestAuditLogger.java @@ -20,10 +20,9 @@ package org.apache.hadoop.hdfs.server.namenode; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_AUDIT_LOGGERS_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_ACLS_ENABLED_KEY; +import static org.apache.hadoop.hdfs.DFSConfigKeys.NNTOP_ENABLED_KEY; import static org.apache.hadoop.test.GenericTestUtils.assertExceptionContains; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; +import static org.junit.Assert.*; import static org.mockito.Matchers.anyListOf; import static org.mockito.Matchers.anyString; @@ -43,6 +42,7 @@ import org.apache.hadoop.fs.permission.AclEntry; import org.apache.hadoop.fs.permission.FsPermission; import org.apache.hadoop.hdfs.HdfsConfiguration; import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.hadoop.hdfs.server.namenode.top.TopAuditLogger; import org.apache.hadoop.hdfs.web.resources.GetOpParam; import org.apache.hadoop.ipc.RemoteException; import org.apache.hadoop.net.NetUtils; @@ -94,6 +94,29 @@ public class TestAuditLogger { } } + /** + * Tests that TopAuditLogger can be disabled + */ + @Test + public void testDisableTopAuditLogger() throws IOException { + Configuration conf = new HdfsConfiguration(); + conf.setBoolean(NNTOP_ENABLED_KEY, false); + MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).build(); + + try { + cluster.waitClusterUp(); + List<AuditLogger> auditLoggers = + cluster.getNameNode().getNamesystem().getAuditLoggers(); + for (AuditLogger auditLogger : auditLoggers) { + assertFalse( + "top audit logger is still hooked in after it is disabled", + auditLogger instanceof TopAuditLogger); + } + } finally { + cluster.shutdown(); + } + } + @Test public void testWebHdfsAuditLogger() throws IOException, URISyntaxException { Configuration conf = new HdfsConfiguration(); http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/metrics/TestNameNodeMetrics.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/metrics/TestNameNodeMetrics.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/metrics/TestNameNodeMetrics.java index 23d26d6..dea8174 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/metrics/TestNameNodeMetrics.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/metrics/TestNameNodeMetrics.java @@ -17,6 +17,7 @@ */ package org.apache.hadoop.hdfs.server.namenode.metrics; +import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_AUDIT_LOGGERS_KEY; import static org.apache.hadoop.test.MetricsAsserts.assertCounter; import static org.apache.hadoop.test.MetricsAsserts.assertGauge; import static org.apache.hadoop.test.MetricsAsserts.assertQuantileGauges; @@ -46,6 +47,10 @@ import org.apache.hadoop.hdfs.server.datanode.DataNode; import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils; import org.apache.hadoop.hdfs.server.namenode.FSNamesystem; import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter; +import org.apache.hadoop.hdfs.server.namenode.top.TopConf; +import org.apache.hadoop.hdfs.server.namenode.top.metrics.TopMetrics; +import org.apache.hadoop.hdfs.server.namenode.top.TopAuditLogger; +import org.apache.hadoop.hdfs.server.namenode.top.window.RollingWindowManager; import org.apache.hadoop.metrics2.MetricsRecordBuilder; import org.apache.hadoop.metrics2.MetricsSource; import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; @@ -53,6 +58,7 @@ import org.apache.hadoop.test.MetricsAsserts; import org.apache.hadoop.util.Time; import org.apache.log4j.Level; import org.junit.After; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -87,6 +93,11 @@ public class TestNameNodeMetrics { CONF.setBoolean(DFSConfigKeys.DFS_NAMENODE_AVOID_STALE_DATANODE_FOR_READ_KEY, true); ((Log4JLogger)LogFactory.getLog(MetricsAsserts.class)) .getLogger().setLevel(Level.DEBUG); + /** + * need it to test {@link #testTopAuditLogger} + */ + CONF.set(DFS_NAMENODE_AUDIT_LOGGERS_KEY, + TopAuditLogger.class.getName()); } private MiniDFSCluster cluster; @@ -101,6 +112,7 @@ public class TestNameNodeMetrics { @Before public void setUp() throws Exception { + TopMetrics.reset();//reset the static init done by prev test cluster = new MiniDFSCluster.Builder(CONF).numDataNodes(DATANODE_COUNT).build(); cluster.waitActive(); namesystem = cluster.getNamesystem(); @@ -454,4 +466,53 @@ public class TestNameNodeMetrics { assertQuantileGauges("Syncs1s", rb); assertQuantileGauges("BlockReport1s", rb); } + + /** + * Test whether {@link TopMetrics} is registered with metrics system + * @throws Exception + */ + @Test + public void testTopMetrics() throws Exception { + final String testUser = "NNTopTestUser"; + final String testOp = "NNTopTestOp"; + final String metricName = + RollingWindowManager.createMetricName(testOp, testUser); + TopMetrics.getInstance().report(testUser, testOp); + final String regName = TopConf.TOP_METRICS_REGISTRATION_NAME; + MetricsRecordBuilder rb = getMetrics(regName); + assertGauge(metricName, 1L, rb); + } + + /** + * Test whether {@link TopAuditLogger} is registered as an audit logger + * @throws Exception + */ + @Test + public void testTopAuditLogger() throws Exception { + //note: the top audit logger should already be set in conf + //issue one command, any command is fine + FileSystem fs = cluster.getFileSystem(); + long time = System.currentTimeMillis(); + fs.setTimes(new Path("/"), time, time); + //the command should be reflected in the total count of all users + final String testUser = TopConf.ALL_USERS; + final String testOp = TopConf.CMD_TOTAL; + final String metricName = + RollingWindowManager.createMetricName(testOp, testUser); + final String regName = TopConf.TOP_METRICS_REGISTRATION_NAME; + MetricsRecordBuilder rb = getMetrics(regName); + assertGaugeGreaterThan(metricName, 1L, rb); + } + + /** + * Assert a long gauge metric greater than + * @param name of the metric + * @param expected minimum expected value of the metric + * @param rb the record builder mock used to getMetrics + */ + public static void assertGaugeGreaterThan(String name, long expected, + MetricsRecordBuilder rb) { + Assert.assertTrue("Bad value for metric " + name, + expected <= MetricsAsserts.getLongGauge(name, rb)); + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindow.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindow.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindow.java new file mode 100644 index 0000000..804c641 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindow.java @@ -0,0 +1,84 @@ +/** + * 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 org.junit.Assert; +import org.junit.Test; + +public class TestRollingWindow { + + final int WINDOW_LEN = 60000; + final int BUCKET_CNT = 10; + final int BUCKET_LEN = WINDOW_LEN / BUCKET_CNT; + + @Test + public void testBasics() { + RollingWindow window = new RollingWindow(WINDOW_LEN, BUCKET_CNT); + long time = 1; + Assert.assertEquals("The initial sum of rolling window must be 0", 0, + window.getSum(time)); + time = WINDOW_LEN + BUCKET_LEN * 3 / 2; + Assert.assertEquals("The initial sum of rolling window must be 0", 0, + window.getSum(time)); + + window.incAt(time, 5); + Assert.assertEquals( + "The sum of rolling window does not reflect the recent update", 5, + window.getSum(time)); + + time += BUCKET_LEN; + window.incAt(time, 6); + Assert.assertEquals( + "The sum of rolling window does not reflect the recent update", 11, + window.getSum(time)); + + time += WINDOW_LEN - BUCKET_LEN; + Assert.assertEquals( + "The sum of rolling window does not reflect rolling effect", 6, + window.getSum(time)); + + time += BUCKET_LEN; + Assert.assertEquals( + "The sum of rolling window does not reflect rolling effect", 0, + window.getSum(time)); + } + + @Test + public void testReorderedAccess() { + RollingWindow window = new RollingWindow(WINDOW_LEN, BUCKET_CNT); + long time = 2 * WINDOW_LEN + BUCKET_LEN * 3 / 2; + window.incAt(time, 5); + + time++; + Assert.assertEquals( + "The sum of rolling window does not reflect the recent update", 5, + window.getSum(time)); + + long reorderedTime = time - 2 * BUCKET_LEN; + window.incAt(reorderedTime, 6); + Assert.assertEquals( + "The sum of rolling window does not reflect the reordered update", 11, + window.getSum(time)); + + time = reorderedTime + WINDOW_LEN; + Assert.assertEquals( + "The sum of rolling window does not reflect rolling effect", 5, + window.getSum(time)); + } + +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/541172f2/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindowManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindowManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindowManager.java new file mode 100644 index 0000000..de21714 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/top/window/TestRollingWindowManager.java @@ -0,0 +1,93 @@ +/** + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.junit.Before; +import org.junit.Test; + +import org.apache.hadoop.hdfs.server.namenode.top.window.RollingWindowManager.MetricValueMap; + +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() { + 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++; + MetricValueMap tops = manager.snapshot(time); + + assertEquals("The number of returned top metrics is invalid", + 2 * (N_TOP_USERS + 1), tops.size()); + int userIndex = users.length - 2; + String metricName = RollingWindowManager.createMetricName("open", + users[userIndex]); + boolean includes = tops.containsKey(metricName); + assertTrue("The order of entries in top metrics is wrong", includes); + assertEquals("The reported value by top is different from recorded one", + (userIndex + 1) * 2, ((Long) tops.get(metricName)).longValue()); + + // move the window forward not to see the "open" results + time += WINDOW_LEN_MS - 2; + // top should not include only "close" results + tops = manager.snapshot(time); + assertEquals("The number of returned top metrics is invalid", + N_TOP_USERS + 1, tops.size()); + includes = tops.containsKey(metricName); + assertFalse("After rolling, the top list still includes the stale metrics", + includes); + + metricName = RollingWindowManager.createMetricName("close", + users[userIndex]); + includes = tops.containsKey(metricName); + assertTrue("The order of entries in top metrics is wrong", includes); + assertEquals("The reported value by top is different from recorded one", + (userIndex + 1), ((Long) tops.get(metricName)).longValue()); + } +}
