|
@@ -46,11 +46,19 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_KEYTAB_FILE_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_MAX_LOCKED_MEMORY_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_MAX_RECEIVER_THREADS_DEFAULT;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_MAX_RECEIVER_THREADS_KEY;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_DEFAULT;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_NETWORK_COUNTS_CACHE_MAX_SIZE_DEFAULT;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_NETWORK_COUNTS_CACHE_MAX_SIZE_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_OOB_TIMEOUT_DEFAULT;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_OOB_TIMEOUT_KEY;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_DEFAULT;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_KEY;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_PEER_STATS_ENABLED_DEFAULT;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_PEER_STATS_ENABLED_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_PLUGINS_KEY;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_DEFAULT;
|
|
|
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_STARTUP_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY;
|
|
|
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT;
|
|
@@ -315,7 +323,11 @@ public class DataNode extends ReconfigurableBase
|
|
|
DFS_BLOCKREPORT_SPLIT_THRESHOLD_KEY,
|
|
|
DFS_BLOCKREPORT_INITIAL_DELAY_KEY,
|
|
|
DFS_DATANODE_MAX_RECEIVER_THREADS_KEY,
|
|
|
- DFS_CACHEREPORT_INTERVAL_MSEC_KEY));
|
|
|
+ DFS_CACHEREPORT_INTERVAL_MSEC_KEY,
|
|
|
+ DFS_DATANODE_PEER_STATS_ENABLED_KEY,
|
|
|
+ DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_KEY,
|
|
|
+ DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_KEY,
|
|
|
+ DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_KEY));
|
|
|
|
|
|
public static final Log METRICS_LOG = LogFactory.getLog("DataNodeMetricsLog");
|
|
|
|
|
@@ -357,7 +369,7 @@ public class DataNode extends ReconfigurableBase
|
|
|
|
|
|
DataNodeMetrics metrics;
|
|
|
@Nullable
|
|
|
- private DataNodePeerMetrics peerMetrics;
|
|
|
+ private volatile DataNodePeerMetrics peerMetrics;
|
|
|
private DataNodeDiskMetrics diskMetrics;
|
|
|
private InetSocketAddress streamingAddr;
|
|
|
|
|
@@ -634,6 +646,11 @@ public class DataNode extends ReconfigurableBase
|
|
|
return reconfDataXceiverParameters(property, newVal);
|
|
|
case DFS_CACHEREPORT_INTERVAL_MSEC_KEY:
|
|
|
return reconfCacheReportParameters(property, newVal);
|
|
|
+ case DFS_DATANODE_PEER_STATS_ENABLED_KEY:
|
|
|
+ case DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_KEY:
|
|
|
+ case DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_KEY:
|
|
|
+ case DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_KEY:
|
|
|
+ return reconfSlowPeerParameters(property, newVal);
|
|
|
default:
|
|
|
break;
|
|
|
}
|
|
@@ -713,6 +730,53 @@ public class DataNode extends ReconfigurableBase
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ private String reconfSlowPeerParameters(String property, String newVal)
|
|
|
+ throws ReconfigurationException {
|
|
|
+ String result = null;
|
|
|
+ try {
|
|
|
+ LOG.info("Reconfiguring {} to {}", property, newVal);
|
|
|
+ if (property.equals(DFS_DATANODE_PEER_STATS_ENABLED_KEY)) {
|
|
|
+ Preconditions.checkNotNull(dnConf, "DNConf has not been initialized.");
|
|
|
+ if (newVal != null && !newVal.equalsIgnoreCase("true")
|
|
|
+ && !newVal.equalsIgnoreCase("false")) {
|
|
|
+ throw new IllegalArgumentException("Not a valid Boolean value for " + property +
|
|
|
+ " in reconfSlowPeerParameters");
|
|
|
+ }
|
|
|
+ boolean enable = (newVal == null ? DFS_DATANODE_PEER_STATS_ENABLED_DEFAULT :
|
|
|
+ Boolean.parseBoolean(newVal));
|
|
|
+ result = Boolean.toString(enable);
|
|
|
+ dnConf.setPeerStatsEnabled(enable);
|
|
|
+ if (enable) {
|
|
|
+ // Create if it doesn't exist, overwrite if it does.
|
|
|
+ peerMetrics = DataNodePeerMetrics.create(getDisplayName(), getConf());
|
|
|
+ }
|
|
|
+ } else if (property.equals(DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_KEY)) {
|
|
|
+ Preconditions.checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
|
|
|
+ long minNodes = (newVal == null ? DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_DEFAULT :
|
|
|
+ Long.parseLong(newVal));
|
|
|
+ result = Long.toString(minNodes);
|
|
|
+ peerMetrics.setMinOutlierDetectionNodes(minNodes);
|
|
|
+ } else if (property.equals(DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_KEY)) {
|
|
|
+ Preconditions.checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
|
|
|
+ long threshold = (newVal == null ? DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_DEFAULT :
|
|
|
+ Long.parseLong(newVal));
|
|
|
+ result = Long.toString(threshold);
|
|
|
+ peerMetrics.setLowThresholdMs(threshold);
|
|
|
+ } else if (property.equals(DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_KEY)) {
|
|
|
+ Preconditions.checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
|
|
|
+ long minSamples = (newVal == null ?
|
|
|
+ DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_DEFAULT :
|
|
|
+ Long.parseLong(newVal));
|
|
|
+ result = Long.toString(minSamples);
|
|
|
+ peerMetrics.setMinOutlierDetectionSamples(minSamples);
|
|
|
+ }
|
|
|
+ LOG.info("RECONFIGURE* changed {} to {}", property, newVal);
|
|
|
+ return result;
|
|
|
+ } catch (IllegalArgumentException e) {
|
|
|
+ throw new ReconfigurationException(property, newVal, getConf().get(property), e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* Get a list of the keys of the re-configurable properties in configuration.
|
|
|
*/
|
|
@@ -3872,7 +3936,7 @@ public class DataNode extends ReconfigurableBase
|
|
|
|
|
|
@Override // DataNodeMXBean
|
|
|
public String getSendPacketDownstreamAvgInfo() {
|
|
|
- return peerMetrics != null ?
|
|
|
+ return dnConf.peerStatsEnabled && peerMetrics != null ?
|
|
|
peerMetrics.dumpSendPacketDownstreamAvgInfoAsJson() : null;
|
|
|
}
|
|
|
|