You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sz...@apache.org on 2014/09/25 22:51:45 UTC
[6/6] git commit: Merge branch 'HDFS-6584' into trunk
Merge branch 'HDFS-6584' into trunk
Conflicts:
hadoop-common-project/hadoop-common/src/main/conf/hadoop-env.sh
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogLoader.java
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicy.java
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicyConsiderLoad.java
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/022474c8
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/022474c8
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/022474c8
Branch: refs/heads/branch-2
Commit: 022474c807170e127d5d3adcf2a41fd652295b8e
Parents: fd3aa38
Author: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Authored: Fri Sep 19 07:11:19 2014 +0800
Committer: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Committed: Fri Sep 26 04:47:25 2014 +0800
----------------------------------------------------------------------
.../hadoop-common/src/main/conf/hadoop-env.sh | 13 +
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 76 ++
.../hadoop-hdfs/src/main/bin/hdfs | 5 +
.../hadoop-hdfs/src/main/bin/hdfs.cmd | 8 +-
.../src/main/conf/blockStoragePolicy-site.xml | 21 +
.../hadoop-hdfs/src/main/conf/hdfs-site.xml | 3 +-
.../apache/hadoop/hdfs/BlockStoragePolicy.java | 419 +++++++
.../java/org/apache/hadoop/hdfs/DFSClient.java | 19 +
.../org/apache/hadoop/hdfs/DFSConfigKeys.java | 15 +
.../hadoop/hdfs/DistributedFileSystem.java | 34 +-
.../org/apache/hadoop/hdfs/StorageType.java | 6 +-
.../hadoop/hdfs/protocol/ClientProtocol.java | 14 +
.../hadoop/hdfs/protocol/HdfsConstants.java | 5 +-
.../hadoop/hdfs/protocol/HdfsFileStatus.java | 10 +-
.../hdfs/protocol/HdfsLocatedFileStatus.java | 10 +-
.../protocol/SnapshottableDirectoryStatus.java | 3 +-
...tNamenodeProtocolServerSideTranslatorPB.java | 16 +
.../ClientNamenodeProtocolTranslatorPB.java | 16 +
.../apache/hadoop/hdfs/protocolPB/PBHelper.java | 38 +-
.../hadoop/hdfs/server/balancer/Balancer.java | 19 +-
.../hadoop/hdfs/server/balancer/Dispatcher.java | 242 ++--
.../hadoop/hdfs/server/balancer/Matcher.java | 15 +
.../hdfs/server/balancer/MovedBlocks.java | 2 +-
.../hdfs/server/balancer/NameNodeConnector.java | 76 +-
.../server/blockmanagement/BlockCollection.java | 5 +
.../server/blockmanagement/BlockManager.java | 89 +-
.../blockmanagement/BlockPlacementPolicy.java | 22 +-
.../BlockPlacementPolicyDefault.java | 377 +++---
.../BlockPlacementPolicyWithNodeGroup.java | 73 +-
.../blockmanagement/DatanodeDescriptor.java | 52 +-
.../blockmanagement/DatanodeStorageInfo.java | 24 +-
.../hadoop/hdfs/server/datanode/DataNode.java | 4 +-
.../apache/hadoop/hdfs/server/mover/Mover.java | 644 +++++++++++
.../hdfs/server/namenode/FSDirectory.java | 123 +-
.../hadoop/hdfs/server/namenode/FSEditLog.java | 12 +-
.../hdfs/server/namenode/FSEditLogLoader.java | 12 +-
.../hdfs/server/namenode/FSEditLogOp.java | 67 ++
.../hdfs/server/namenode/FSEditLogOpCodes.java | 1 +
.../hdfs/server/namenode/FSImageFormat.java | 4 +-
.../server/namenode/FSImageFormatPBINode.java | 6 +-
.../server/namenode/FSImageSerialization.java | 2 +-
.../hdfs/server/namenode/FSNamesystem.java | 89 +-
.../hadoop/hdfs/server/namenode/INode.java | 17 +-
.../hdfs/server/namenode/INodeDirectory.java | 27 +
.../namenode/INodeDirectoryAttributes.java | 4 +-
.../hadoop/hdfs/server/namenode/INodeFile.java | 49 +-
.../server/namenode/INodeFileAttributes.java | 14 +-
.../hadoop/hdfs/server/namenode/INodeMap.java | 11 +
.../hdfs/server/namenode/INodeReference.java | 10 +
.../hdfs/server/namenode/INodeSymlink.java | 12 +
.../server/namenode/NameNodeLayoutVersion.java | 3 +-
.../hdfs/server/namenode/NameNodeRpcServer.java | 8 +-
.../snapshot/FSImageFormatPBSnapshot.java | 3 +-
.../web/resources/NamenodeWebHdfsMethods.java | 8 +-
.../org/apache/hadoop/hdfs/tools/DFSAdmin.java | 75 +-
.../apache/hadoop/hdfs/util/EnumCounters.java | 9 +
.../org/apache/hadoop/hdfs/web/JsonUtil.java | 10 +-
.../src/main/proto/ClientNamenodeProtocol.proto | 10 +
.../hadoop-hdfs/src/main/proto/fsimage.proto | 1 +
.../hadoop-hdfs/src/main/proto/hdfs.proto | 3 +
.../resources/blockStoragePolicy-default.xml | 118 ++
.../src/main/resources/hdfs-default.xml | 3 +-
.../src/site/apt/ArchivalStorage.apt.vm | 302 +++++
.../src/site/apt/HDFSCommands.apt.vm | 43 +-
.../org/apache/hadoop/hdfs/DFSTestUtil.java | 119 +-
.../org/apache/hadoop/hdfs/MiniDFSCluster.java | 12 +-
.../hadoop/hdfs/TestBlockStoragePolicy.java | 1075 ++++++++++++++++++
.../hadoop/hdfs/TestDFSClientRetries.java | 4 +-
.../hdfs/TestDFSInotifyEventInputStream.java | 2 +-
.../java/org/apache/hadoop/hdfs/TestLease.java | 8 +-
.../hadoop/hdfs/TestSnapshotCommands.java | 152 +--
.../hadoop/hdfs/TestStoragePolicyCommands.java | 81 ++
.../hdfs/server/balancer/TestBalancer.java | 7 +
.../balancer/TestBalancerWithHANameNodes.java | 2 +-
.../TestBalancerWithMultipleNameNodes.java | 2 +-
.../balancer/TestBalancerWithNodeGroup.java | 2 +-
.../blockmanagement/TestBlockManager.java | 20 +-
.../blockmanagement/TestReplicationPolicy.java | 29 +-
.../TestReplicationPolicyConsiderLoad.java | 5 +-
.../TestReplicationPolicyWithNodeGroup.java | 17 +-
.../hadoop/hdfs/server/mover/TestMover.java | 222 ++++
.../hdfs/server/mover/TestStorageMover.java | 766 +++++++++++++
.../hdfs/server/namenode/CreateEditsLog.java | 4 +-
.../hdfs/server/namenode/TestAddBlockRetry.java | 4 +-
.../hdfs/server/namenode/TestDeleteRace.java | 6 +-
.../hdfs/server/namenode/TestEditLog.java | 2 +-
.../namenode/TestFSPermissionChecker.java | 35 +-
.../hadoop/hdfs/server/namenode/TestFsck.java | 3 +-
.../hdfs/server/namenode/TestINodeFile.java | 35 +-
.../hdfs/server/namenode/ha/TestDNFencing.java | 4 +-
.../apache/hadoop/hdfs/web/TestJsonUtil.java | 2 +-
.../hadoop-hdfs/src/test/resources/editsStored | Bin 5252 -> 4999 bytes
.../src/test/resources/editsStored.xml | 422 ++++---
hadoop-project/src/site/site.xml | 1 +
94 files changed, 5662 insertions(+), 812 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-common-project/hadoop-common/src/main/conf/hadoop-env.sh
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/conf/hadoop-env.sh b/hadoop-common-project/hadoop-common/src/main/conf/hadoop-env.sh
index 05872a2..e964ec0 100644
--- a/hadoop-common-project/hadoop-common/src/main/conf/hadoop-env.sh
+++ b/hadoop-common-project/hadoop-common/src/main/conf/hadoop-env.sh
@@ -67,6 +67,19 @@ export HADOOP_SECURE_DN_USER=${HADOOP_SECURE_DN_USER}
# Where log files are stored in the secure data environment.
export HADOOP_SECURE_DN_LOG_DIR=${HADOOP_LOG_DIR}/${HADOOP_HDFS_USER}
+###
+# HDFS Mover specific parameters
+###
+# Specify the JVM options to be used when starting the HDFS Mover.
+# These options will be appended to the options specified as HADOOP_OPTS
+# and therefore may override any similar flags set in HADOOP_OPTS
+#
+# export HADOOP_MOVER_OPTS=""
+
+###
+# Advanced Users Only!
+###
+
# The directory where pid files are stored. /tmp by default.
# NOTE: this should be set to a directory that can only be written to by
# the user that will run the hadoop daemons. Otherwise there is the
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/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 0d9c501..69cc187 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -1,5 +1,81 @@
Hadoop HDFS Change Log
+HDFS-6584: Archival Storage
+
+ HDFS-6677. Change INodeFile and FSImage to support storage policy ID.
+ (szetszwo)
+
+ HDFS-6670. Add block storage policy support with default HOT, WARM and COLD
+ policies. (szetszwo)
+
+ HDFS-6671. Change BlockPlacementPolicy to consider block storage policy
+ in replicaiton. (szetszwo)
+
+ HDFS-6710. Change BlockPlacementPolicy to consider block storage policy
+ in replica deletion. (szetszwo)
+
+ HDFS-6679. Bump NameNodeLayoutVersion and update editsStored test files.
+ (vinayakumarb via szetszwo)
+
+ HDFS-6686. Change BlockPlacementPolicy to use fallback when some storage
+ types are unavailable. (szetszwo)
+
+ HDFS-6835. Add a new API to set storage policy. (jing9)
+
+ HDFS-6847. Support storage policy on directories and include storage policy
+ in HdfsFileStatus. (Jing Zhao via szetszwo)
+
+ HDFS-6801. Add a new data migration tool, Mover, for archiving data.
+ (szetszwo via jing9)
+
+ HDFS-6863. Support migration for snapshot paths. (jing9)
+
+ HDFS-6906. Add more tests for BlockStoragePolicy. (szetszwo via jing9)
+
+ HDFS-6911. check if a block is already scheduled in Mover.
+ (szetszwo via jing9)
+
+ HDFS-6920. Check the storage type of delNodeHintStorage when deleting
+ a replica. (szetszwo via jing9)
+
+ HDFS-6944. Add retry and termination logic for Mover. (jing9)
+
+ HDFS-6969. INode#getStoragePolicyID should always return the latest
+ storage policy. (jing9)
+
+ HDFS-6961. BlockPlacementPolicy#chooseTarget should check each valid
+ storage type in each choosing round. (jing9)
+
+ HDFS-6876. support set/get storage policy in DFSAdmin. (jing9)
+
+ HDFS-6997. Add more tests for data migration and replicaion. (szetszwo)
+
+ HDFS-6875. Support migration for a list of specified paths. (jing9)
+
+ HDFS-7027. Mover does not terminate when some storage type is out of space.
+ (szetszwo via jing9)
+
+ HDFS-7029. Fix TestDFSInotifyEventInputStream and TestDistributedFileSystem.
+ (szetszwo via jing9)
+
+ HDFS-7028. FSDirectory should not get storage policy id from symlinks.
+ (szetszwo)
+
+ HDFS-7034. Fix TestBlockPlacement and TestStorageMover. (jing9)
+
+ HDFS-7039. Fix Balancer tests. (szetszwo via jing9)
+
+ HDFS-7062. Skip under construction block for migration. (jing9)
+
+ HDFS-7052. Add Mover into hdfs script. (jing9)
+
+ HDFS-7072. Fix TestBlockManager and TestStorageMover. (jing9 via szetszwo)
+
+ HDFS-6864. Archival Storage: add user documentation. (szetszwo via jing9)
+
+ HDFS-7088. Archival Storage: fix TestBalancer and
+ TestBalancerWithMultipleNameNodes. (szetszwo via jing9)
+
Release 2.6.0 - UNRELEASED
INCOMPATIBLE CHANGES
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
index 5fbb3db..046163c 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
@@ -48,6 +48,8 @@ function print_usage(){
echo " fsck run a DFS filesystem checking utility"
echo " balancer run a cluster balancing utility"
echo " jmxget get JMX exported values from NameNode or DataNode."
+ echo " mover run a utility to move block replicas across"
+ echo " storage types"
echo " oiv apply the offline fsimage viewer to an fsimage"
echo " oiv_legacy apply the offline fsimage viewer to an legacy fsimage"
echo " oev apply the offline edits viewer to an edits file"
@@ -159,6 +161,9 @@ elif [ "$COMMAND" = "fsck" ] ; then
elif [ "$COMMAND" = "balancer" ] ; then
CLASS=org.apache.hadoop.hdfs.server.balancer.Balancer
HADOOP_OPTS="$HADOOP_OPTS $HADOOP_BALANCER_OPTS"
+elif [ "$COMMAND" = "mover" ] ; then
+ CLASS=org.apache.hadoop.hdfs.server.mover.Mover
+ HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_MOVER_OPTS}"
elif [ "$COMMAND" = "jmxget" ] ; then
CLASS=org.apache.hadoop.hdfs.tools.JMXGet
elif [ "$COMMAND" = "oiv" ] ; then
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd
index f5f77f0..9fb8426 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs.cmd
@@ -47,7 +47,7 @@ if "%1" == "--config" (
goto print_usage
)
- set hdfscommands=dfs namenode secondarynamenode journalnode zkfc datanode dfsadmin haadmin fsck balancer jmxget oiv oev fetchdt getconf groups snapshotDiff lsSnapshottableDir cacheadmin
+ set hdfscommands=dfs namenode secondarynamenode journalnode zkfc datanode dfsadmin haadmin fsck balancer jmxget oiv oev fetchdt getconf groups snapshotDiff lsSnapshottableDir cacheadmin mover
for %%i in ( %hdfscommands% ) do (
if %hdfs-command% == %%i set hdfscommand=true
)
@@ -150,6 +150,11 @@ goto :eof
set CLASS=org.apache.hadoop.hdfs.tools.CacheAdmin
goto :eof
+:mover
+ set CLASS=org.apache.hadoop.hdfs.server.mover.Mover
+ set HADOOP_OPTS=%HADOOP_OPTS% %HADOOP_MOVER_OPTS%
+ goto :eof
+
@rem This changes %1, %2 etc. Hence those cannot be used after calling this.
:make_command_arguments
if "%1" == "--config" (
@@ -198,6 +203,7 @@ goto :eof
@echo lsSnapshottableDir list all snapshottable dirs owned by the current user
@echo Use -help to see options
@echo cacheadmin configure the HDFS cache
+ @echo mover run a utility to move block replicas across storage types
@echo.
@echo Most commands print help when invoked w/o parameters.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/blockStoragePolicy-site.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/blockStoragePolicy-site.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/blockStoragePolicy-site.xml
new file mode 100644
index 0000000..04142ad
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/blockStoragePolicy-site.xml
@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+<!--
+ Licensed 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. See accompanying LICENSE file.
+-->
+
+<!-- Put site-specific property overrides in this file. -->
+
+<configuration>
+
+</configuration>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/hdfs-site.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/hdfs-site.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/hdfs-site.xml
index 50ec146..3a0b0ed 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/hdfs-site.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/conf/hdfs-site.xml
@@ -16,6 +16,7 @@
<!-- Put site-specific property overrides in this file. -->
-<configuration>
+<configuration xmlns:xi="http://www.w3.org/2001/XInclude">
+ <xi:include href="blockStoragePolicy-site.xml" />
</configuration>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockStoragePolicy.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockStoragePolicy.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockStoragePolicy.java
new file mode 100644
index 0000000..efbf8a0
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockStoragePolicy.java
@@ -0,0 +1,419 @@
+/**
+ * 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;
+
+import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.List;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.XAttr;
+import org.apache.hadoop.fs.XAttr.NameSpace;
+
+/**
+ * A block storage policy describes how to select the storage types
+ * for the replicas of a block.
+ */
+@InterfaceAudience.Private
+public class BlockStoragePolicy {
+ public static final Log LOG = LogFactory.getLog(BlockStoragePolicy.class);
+
+ public static final String DFS_BLOCK_STORAGE_POLICIES_KEY
+ = "dfs.block.storage.policies";
+ public static final String DFS_BLOCK_STORAGE_POLICY_KEY_PREFIX
+ = "dfs.block.storage.policy.";
+ public static final String DFS_BLOCK_STORAGE_POLICY_CREATION_FALLBACK_KEY_PREFIX
+ = "dfs.block.storage.policy.creation-fallback.";
+ public static final String DFS_BLOCK_STORAGE_POLICY_REPLICATION_FALLBACK_KEY_PREFIX
+ = "dfs.block.storage.policy.replication-fallback.";
+ public static final String STORAGE_POLICY_XATTR_NAME = "bsp";
+ /** set the namespace to TRUSTED so that only privilege users can access */
+ public static final NameSpace XAttrNS = NameSpace.TRUSTED;
+
+ public static final int ID_BIT_LENGTH = 4;
+ public static final int ID_MAX = (1 << ID_BIT_LENGTH) - 1;
+ public static final byte ID_UNSPECIFIED = 0;
+
+ private static final Suite DEFAULT_SUITE = createDefaultSuite();
+
+ private static Suite createDefaultSuite() {
+ final BlockStoragePolicy[] policies = new BlockStoragePolicy[1 << ID_BIT_LENGTH];
+ final StorageType[] storageTypes = {StorageType.DISK};
+ final byte defaultPolicyId = 12;
+ policies[defaultPolicyId] = new BlockStoragePolicy(defaultPolicyId, "HOT",
+ storageTypes, StorageType.EMPTY_ARRAY, StorageType.EMPTY_ARRAY);
+ return new Suite(defaultPolicyId, policies);
+ }
+
+ /** A block storage policy suite. */
+ public static class Suite {
+ private final byte defaultPolicyID;
+ private final BlockStoragePolicy[] policies;
+
+ private Suite(byte defaultPolicyID, BlockStoragePolicy[] policies) {
+ this.defaultPolicyID = defaultPolicyID;
+ this.policies = policies;
+ }
+
+ /** @return the corresponding policy. */
+ public BlockStoragePolicy getPolicy(byte id) {
+ // id == 0 means policy not specified.
+ return id == 0? getDefaultPolicy(): policies[id];
+ }
+
+ /** @return the default policy. */
+ public BlockStoragePolicy getDefaultPolicy() {
+ return getPolicy(defaultPolicyID);
+ }
+
+ public BlockStoragePolicy getPolicy(String policyName) {
+ if (policies != null) {
+ for (BlockStoragePolicy policy : policies) {
+ if (policy != null && policy.name.equals(policyName)) {
+ return policy;
+ }
+ }
+ }
+ return null;
+ }
+ }
+
+ /** A 4-bit policy ID */
+ private final byte id;
+ /** Policy name */
+ private final String name;
+
+ /** The storage types to store the replicas of a new block. */
+ private final StorageType[] storageTypes;
+ /** The fallback storage type for block creation. */
+ private final StorageType[] creationFallbacks;
+ /** The fallback storage type for replication. */
+ private final StorageType[] replicationFallbacks;
+
+ @VisibleForTesting
+ public BlockStoragePolicy(byte id, String name, StorageType[] storageTypes,
+ StorageType[] creationFallbacks, StorageType[] replicationFallbacks) {
+ this.id = id;
+ this.name = name;
+ this.storageTypes = storageTypes;
+ this.creationFallbacks = creationFallbacks;
+ this.replicationFallbacks = replicationFallbacks;
+ }
+
+ /**
+ * @return a list of {@link StorageType}s for storing the replicas of a block.
+ */
+ public List<StorageType> chooseStorageTypes(final short replication) {
+ final List<StorageType> types = new LinkedList<StorageType>();
+ int i = 0;
+ for(; i < replication && i < storageTypes.length; i++) {
+ types.add(storageTypes[i]);
+ }
+ final StorageType last = storageTypes[storageTypes.length - 1];
+ for(; i < replication; i++) {
+ types.add(last);
+ }
+ return types;
+ }
+
+ /**
+ * Choose the storage types for storing the remaining replicas, given the
+ * replication number and the storage types of the chosen replicas.
+ *
+ * @param replication the replication number.
+ * @param chosen the storage types of the chosen replicas.
+ * @return a list of {@link StorageType}s for storing the replicas of a block.
+ */
+ public List<StorageType> chooseStorageTypes(final short replication,
+ final Iterable<StorageType> chosen) {
+ return chooseStorageTypes(replication, chosen, null);
+ }
+
+ private List<StorageType> chooseStorageTypes(final short replication,
+ final Iterable<StorageType> chosen, final List<StorageType> excess) {
+ final List<StorageType> types = chooseStorageTypes(replication);
+ diff(types, chosen, excess);
+ return types;
+ }
+
+ /**
+ * Choose the storage types for storing the remaining replicas, given the
+ * replication number, the storage types of the chosen replicas and
+ * the unavailable storage types. It uses fallback storage in case that
+ * the desired storage type is unavailable.
+ *
+ * @param replication the replication number.
+ * @param chosen the storage types of the chosen replicas.
+ * @param unavailables the unavailable storage types.
+ * @param isNewBlock Is it for new block creation?
+ * @return a list of {@link StorageType}s for storing the replicas of a block.
+ */
+ public List<StorageType> chooseStorageTypes(final short replication,
+ final Iterable<StorageType> chosen,
+ final EnumSet<StorageType> unavailables,
+ final boolean isNewBlock) {
+ final List<StorageType> excess = new LinkedList<StorageType>();
+ final List<StorageType> storageTypes = chooseStorageTypes(
+ replication, chosen, excess);
+ final int expectedSize = storageTypes.size() - excess.size();
+ final List<StorageType> removed = new LinkedList<StorageType>();
+ for(int i = storageTypes.size() - 1; i >= 0; i--) {
+ // replace/remove unavailable storage types.
+ final StorageType t = storageTypes.get(i);
+ if (unavailables.contains(t)) {
+ final StorageType fallback = isNewBlock?
+ getCreationFallback(unavailables)
+ : getReplicationFallback(unavailables);
+ if (fallback == null) {
+ removed.add(storageTypes.remove(i));
+ } else {
+ storageTypes.set(i, fallback);
+ }
+ }
+ }
+ // remove excess storage types after fallback replacement.
+ diff(storageTypes, excess, null);
+ if (storageTypes.size() < expectedSize) {
+ LOG.warn("Failed to place enough replicas: expected size is " + expectedSize
+ + " but only " + storageTypes.size() + " storage types can be selected "
+ + "(replication=" + replication
+ + ", selected=" + storageTypes
+ + ", unavailable=" + unavailables
+ + ", removed=" + removed
+ + ", policy=" + this + ")");
+ }
+ return storageTypes;
+ }
+
+ /**
+ * Compute the list difference t = t - c.
+ * Further, if e is not null, set e = e + c - t;
+ */
+ private static void diff(List<StorageType> t, Iterable<StorageType> c,
+ List<StorageType> e) {
+ for(StorageType storagetype : c) {
+ final int i = t.indexOf(storagetype);
+ if (i >= 0) {
+ t.remove(i);
+ } else if (e != null) {
+ e.add(storagetype);
+ }
+ }
+ }
+
+ /**
+ * Choose excess storage types for deletion, given the
+ * replication number and the storage types of the chosen replicas.
+ *
+ * @param replication the replication number.
+ * @param chosen the storage types of the chosen replicas.
+ * @return a list of {@link StorageType}s for deletion.
+ */
+ public List<StorageType> chooseExcess(final short replication,
+ final Iterable<StorageType> chosen) {
+ final List<StorageType> types = chooseStorageTypes(replication);
+ final List<StorageType> excess = new LinkedList<StorageType>();
+ diff(types, chosen, excess);
+ return excess;
+ }
+
+ /** @return the fallback {@link StorageType} for creation. */
+ public StorageType getCreationFallback(EnumSet<StorageType> unavailables) {
+ return getFallback(unavailables, creationFallbacks);
+ }
+
+ /** @return the fallback {@link StorageType} for replication. */
+ public StorageType getReplicationFallback(EnumSet<StorageType> unavailables) {
+ return getFallback(unavailables, replicationFallbacks);
+ }
+
+ @Override
+ public int hashCode() {
+ return Byte.valueOf(id).hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ } else if (obj == null || !(obj instanceof BlockStoragePolicy)) {
+ return false;
+ }
+ final BlockStoragePolicy that = (BlockStoragePolicy)obj;
+ return this.id == that.id;
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + "{" + name + ":" + id
+ + ", storageTypes=" + Arrays.asList(storageTypes)
+ + ", creationFallbacks=" + Arrays.asList(creationFallbacks)
+ + ", replicationFallbacks=" + Arrays.asList(replicationFallbacks);
+ }
+
+ public byte getId() {
+ return id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ private static StorageType getFallback(EnumSet<StorageType> unavailables,
+ StorageType[] fallbacks) {
+ for(StorageType fb : fallbacks) {
+ if (!unavailables.contains(fb)) {
+ return fb;
+ }
+ }
+ return null;
+ }
+
+ private static byte parseID(String idString, String element, Configuration conf) {
+ byte id = 0;
+ try {
+ id = Byte.parseByte(idString);
+ } catch(NumberFormatException nfe) {
+ throwIllegalArgumentException("Failed to parse policy ID \"" + idString
+ + "\" to a " + ID_BIT_LENGTH + "-bit integer", conf);
+ }
+ if (id < 0) {
+ throwIllegalArgumentException("Invalid policy ID: id = " + id
+ + " < 1 in \"" + element + "\"", conf);
+ } else if (id == 0) {
+ throw new IllegalArgumentException("Policy ID 0 is reserved: " + element);
+ } else if (id > ID_MAX) {
+ throwIllegalArgumentException("Invalid policy ID: id = " + id
+ + " > MAX = " + ID_MAX + " in \"" + element + "\"", conf);
+ }
+ return id;
+ }
+
+ private static StorageType[] parseStorageTypes(String[] strings) {
+ if (strings == null || strings.length == 0) {
+ return StorageType.EMPTY_ARRAY;
+ }
+ final StorageType[] types = new StorageType[strings.length];
+ for(int i = 0; i < types.length; i++) {
+ types[i] = StorageType.valueOf(strings[i].trim().toUpperCase());
+ }
+ return types;
+ }
+
+ private static StorageType[] readStorageTypes(byte id, String keyPrefix,
+ Configuration conf) {
+ final String key = keyPrefix + id;
+ final String[] values = conf.getStrings(key);
+ try {
+ return parseStorageTypes(values);
+ } catch(Exception e) {
+ throw new IllegalArgumentException("Failed to parse " + key
+ + " \"" + conf.get(key), e);
+ }
+ }
+
+ private static BlockStoragePolicy readBlockStoragePolicy(byte id, String name,
+ Configuration conf) {
+ final StorageType[] storageTypes = readStorageTypes(id,
+ DFS_BLOCK_STORAGE_POLICY_KEY_PREFIX, conf);
+ if (storageTypes.length == 0) {
+ throw new IllegalArgumentException(
+ DFS_BLOCK_STORAGE_POLICY_KEY_PREFIX + id + " is missing or is empty.");
+ }
+ final StorageType[] creationFallbacks = readStorageTypes(id,
+ DFS_BLOCK_STORAGE_POLICY_CREATION_FALLBACK_KEY_PREFIX, conf);
+ final StorageType[] replicationFallbacks = readStorageTypes(id,
+ DFS_BLOCK_STORAGE_POLICY_REPLICATION_FALLBACK_KEY_PREFIX, conf);
+ return new BlockStoragePolicy(id, name, storageTypes, creationFallbacks,
+ replicationFallbacks);
+ }
+
+ /** Read {@link Suite} from conf. */
+ public static Suite readBlockStorageSuite(Configuration conf) {
+ final BlockStoragePolicy[] policies = new BlockStoragePolicy[1 << ID_BIT_LENGTH];
+ final String[] values = conf.getStrings(DFS_BLOCK_STORAGE_POLICIES_KEY);
+ if (values == null) {
+ // conf property is missing, use default suite.
+ return DEFAULT_SUITE;
+ }
+ byte firstID = -1;
+ for(String v : values) {
+ v = v.trim();
+ final int i = v.indexOf(':');
+ if (i < 0) {
+ throwIllegalArgumentException("Failed to parse element \"" + v
+ + "\" (expected format is NAME:ID)", conf);
+ } else if (i == 0) {
+ throwIllegalArgumentException("Policy name is missing in \"" + v + "\"", conf);
+ } else if (i == v.length() - 1) {
+ throwIllegalArgumentException("Policy ID is missing in \"" + v + "\"", conf);
+ }
+ final String name = v.substring(0, i).trim();
+ for(int j = 1; j < policies.length; j++) {
+ if (policies[j] != null && policies[j].name.equals(name)) {
+ throwIllegalArgumentException("Policy name duplication: \""
+ + name + "\" appears more than once", conf);
+ }
+ }
+
+ final byte id = parseID(v.substring(i + 1).trim(), v, conf);
+ if (policies[id] != null) {
+ throwIllegalArgumentException("Policy duplication: ID " + id
+ + " appears more than once", conf);
+ }
+ policies[id] = readBlockStoragePolicy(id, name, conf);
+ String prefix = "";
+ if (firstID == -1) {
+ firstID = id;
+ prefix = "(default) ";
+ }
+ LOG.info(prefix + policies[id]);
+ }
+ if (firstID == -1) {
+ throwIllegalArgumentException("Empty list is not allowed", conf);
+ }
+ return new Suite(firstID, policies);
+ }
+
+ public static String buildXAttrName() {
+ return XAttrNS.toString().toLowerCase() + "." + STORAGE_POLICY_XATTR_NAME;
+ }
+
+ public static XAttr buildXAttr(byte policyId) {
+ final String name = buildXAttrName();
+ return XAttrHelper.buildXAttr(name, new byte[] { policyId });
+ }
+
+ public static boolean isStoragePolicyXAttr(XAttr xattr) {
+ return xattr != null && xattr.getNameSpace() == BlockStoragePolicy.XAttrNS
+ && xattr.getName().equals(BlockStoragePolicy.STORAGE_POLICY_XATTR_NAME);
+ }
+
+ private static void throwIllegalArgumentException(String message,
+ Configuration conf) {
+ throw new IllegalArgumentException(message + " in "
+ + DFS_BLOCK_STORAGE_POLICIES_KEY + " \""
+ + conf.get(DFS_BLOCK_STORAGE_POLICIES_KEY) + "\".");
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
index da19291..6f94370 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -1761,6 +1761,25 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
}
/**
+ * Set storage policy for an existing file/directory
+ * @param src file/directory name
+ * @param policyName name of the storage policy
+ */
+ public void setStoragePolicy(String src, String policyName)
+ throws IOException {
+ try {
+ namenode.setStoragePolicy(src, policyName);
+ } catch (RemoteException e) {
+ throw e.unwrapRemoteException(AccessControlException.class,
+ FileNotFoundException.class,
+ SafeModeException.class,
+ NSQuotaExceededException.class,
+ UnresolvedPathException.class,
+ SnapshotAccessControlException.class);
+ }
+ }
+
+ /**
* Rename file or directory.
* @see ClientProtocol#rename(String, String)
* @deprecated Use {@link #rename(String, String, Options.Rename...)} instead.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/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 365cf7a..878a741 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
@@ -371,6 +371,12 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
public static final int DFS_BALANCER_MOVERTHREADS_DEFAULT = 1000;
public static final String DFS_BALANCER_DISPATCHERTHREADS_KEY = "dfs.balancer.dispatcherThreads";
public static final int DFS_BALANCER_DISPATCHERTHREADS_DEFAULT = 200;
+
+ public static final String DFS_MOVER_MOVEDWINWIDTH_KEY = "dfs.mover.movedWinWidth";
+ public static final long DFS_MOVER_MOVEDWINWIDTH_DEFAULT = 5400*1000L;
+ public static final String DFS_MOVER_MOVERTHREADS_KEY = "dfs.mover.moverThreads";
+ public static final int DFS_MOVER_MOVERTHREADS_DEFAULT = 1000;
+
public static final String DFS_DATANODE_ADDRESS_KEY = "dfs.datanode.address";
public static final int DFS_DATANODE_DEFAULT_PORT = 50010;
public static final String DFS_DATANODE_ADDRESS_DEFAULT = "0.0.0.0:" + DFS_DATANODE_DEFAULT_PORT;
@@ -438,6 +444,15 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
public static final Class<BlockPlacementPolicyDefault> DFS_BLOCK_REPLICATOR_CLASSNAME_DEFAULT = BlockPlacementPolicyDefault.class;
public static final String DFS_REPLICATION_MAX_KEY = "dfs.replication.max";
public static final int DFS_REPLICATION_MAX_DEFAULT = 512;
+ public static final String DFS_BLOCK_STORAGE_POLICIES_KEY
+ = BlockStoragePolicy.DFS_BLOCK_STORAGE_POLICIES_KEY;
+ public static final String DFS_BLOCK_STORAGE_POLICY_KEY_PREFIX
+ = BlockStoragePolicy.DFS_BLOCK_STORAGE_POLICY_KEY_PREFIX;
+ public static final String DFS_BLOCK_STORAGE_POLICY_CREATION_FALLBACK_KEY_PREFIX
+ = BlockStoragePolicy.DFS_BLOCK_STORAGE_POLICY_CREATION_FALLBACK_KEY_PREFIX;
+ public static final String DFS_BLOCK_STORAGE_POLICY_REPLICATION_FALLBACK_KEY_PREFIX
+ = BlockStoragePolicy.DFS_BLOCK_STORAGE_POLICY_REPLICATION_FALLBACK_KEY_PREFIX;
+
public static final String DFS_DF_INTERVAL_KEY = "dfs.df.interval";
public static final int DFS_DF_INTERVAL_DEFAULT = 60000;
public static final String DFS_BLOCKREPORT_INTERVAL_MSEC_KEY = "dfs.blockreport.intervalMsec";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
index bb671ce..79f2ef5 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
@@ -471,7 +471,39 @@ public class DistributedFileSystem extends FileSystem {
}
}.resolve(this, absF);
}
-
+
+ /**
+ * Set the source path to the specified storage policy.
+ *
+ * @param src The source path referring to either a directory or a file.
+ * @param policyName The name of the storage policy.
+ */
+ public void setStoragePolicy(final Path src, final String policyName)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ Path absF = fixRelativePart(src);
+ new FileSystemLinkResolver<Void>() {
+ @Override
+ public Void doCall(final Path p)
+ throws IOException, UnresolvedLinkException {
+ dfs.setStoragePolicy(getPathName(p), policyName);
+ return null;
+ }
+ @Override
+ public Void next(final FileSystem fs, final Path p)
+ throws IOException {
+ if (fs instanceof DistributedFileSystem) {
+ ((DistributedFileSystem) fs).setStoragePolicy(p, policyName);
+ return null;
+ } else {
+ throw new UnsupportedOperationException(
+ "Cannot perform setStoragePolicy on a non-DistributedFileSystem: "
+ + src + " -> " + p);
+ }
+ }
+ }.resolve(this, absF);
+ }
+
/**
* Move blocks from srcs to trg and delete srcs afterwards.
* The file block sizes must be the same.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/StorageType.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/StorageType.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/StorageType.java
index 3d8133c..7ca9e00 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/StorageType.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/StorageType.java
@@ -32,9 +32,11 @@ import org.apache.hadoop.classification.InterfaceStability;
@InterfaceStability.Unstable
public enum StorageType {
DISK,
- SSD;
+ SSD,
+ ARCHIVE;
public static final StorageType DEFAULT = DISK;
+
public static final StorageType[] EMPTY_ARRAY = {};
private static final StorageType[] VALUES = values();
@@ -42,4 +44,4 @@ public enum StorageType {
public static List<StorageType> asList() {
return Arrays.asList(VALUES);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
index d8b7a79..4be83f2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
@@ -260,6 +260,20 @@ public interface ClientProtocol {
SnapshotAccessControlException, IOException;
/**
+ * Set the storage policy for a file/directory
+ * @param src Path of an existing file/directory.
+ * @param policyName The name of the storage policy
+ * @throws SnapshotAccessControlException If access is denied
+ * @throws UnresolvedLinkException if <code>src</code> contains a symlink
+ * @throws FileNotFoundException If file/dir <code>src</code> is not found
+ * @throws QuotaExceededException If changes violate the quota restriction
+ */
+ @Idempotent
+ public void setStoragePolicy(String src, String policyName)
+ throws SnapshotAccessControlException, UnresolvedLinkException,
+ FileNotFoundException, QuotaExceededException, IOException;
+
+ /**
* Set permissions for an existing file/directory.
*
* @throws AccessControlException If access is denied
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java
index 2588167..20685a7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java
@@ -161,5 +161,8 @@ public class HdfsConstants {
= DFSUtil.string2Bytes(DOT_SNAPSHOT_DIR);
public static final String SEPARATOR_DOT_SNAPSHOT_DIR
- = Path.SEPARATOR + DOT_SNAPSHOT_DIR;
+ = Path.SEPARATOR + DOT_SNAPSHOT_DIR;
+
+ public static final String SEPARATOR_DOT_SNAPSHOT_DIR_SEPARATOR
+ = Path.SEPARATOR + DOT_SNAPSHOT_DIR + Path.SEPARATOR;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java
index 3d05639..94d9a92 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java
@@ -50,6 +50,7 @@ public class HdfsFileStatus {
// Used by dir, not including dot and dotdot. Always zero for a regular file.
private final int childrenNum;
+ private final byte storagePolicy;
public static final byte[] EMPTY_NAME = new byte[0];
@@ -71,7 +72,8 @@ public class HdfsFileStatus {
public HdfsFileStatus(long length, boolean isdir, int block_replication,
long blocksize, long modification_time, long access_time,
FsPermission permission, String owner, String group, byte[] symlink,
- byte[] path, long fileId, int childrenNum, FileEncryptionInfo feInfo) {
+ byte[] path, long fileId, int childrenNum, FileEncryptionInfo feInfo,
+ byte storagePolicy) {
this.length = length;
this.isdir = isdir;
this.block_replication = (short)block_replication;
@@ -90,6 +92,7 @@ public class HdfsFileStatus {
this.fileId = fileId;
this.childrenNum = childrenNum;
this.feInfo = feInfo;
+ this.storagePolicy = storagePolicy;
}
/**
@@ -251,6 +254,11 @@ public class HdfsFileStatus {
return childrenNum;
}
+ /** @return the storage policy id */
+ public final byte getStoragePolicy() {
+ return storagePolicy;
+ }
+
public final FileStatus makeQualified(URI defaultUri, Path path) {
return new FileStatus(getLen(), isDir(), getReplication(),
getBlockSize(), getModificationTime(),
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsLocatedFileStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsLocatedFileStatus.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsLocatedFileStatus.java
index a78b8bc..7e602bf 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsLocatedFileStatus.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsLocatedFileStatus.java
@@ -35,7 +35,7 @@ import org.apache.hadoop.hdfs.DFSUtil;
@InterfaceStability.Evolving
public class HdfsLocatedFileStatus extends HdfsFileStatus {
private final LocatedBlocks locations;
-
+
/**
* Constructor
*
@@ -58,13 +58,13 @@ public class HdfsLocatedFileStatus extends HdfsFileStatus {
int block_replication, long blocksize, long modification_time,
long access_time, FsPermission permission, String owner, String group,
byte[] symlink, byte[] path, long fileId, LocatedBlocks locations,
- int childrenNum, FileEncryptionInfo feInfo) {
+ int childrenNum, FileEncryptionInfo feInfo, byte storagePolicy) {
super(length, isdir, block_replication, blocksize, modification_time,
- access_time, permission, owner, group, symlink, path, fileId,
- childrenNum, feInfo);
+ access_time, permission, owner, group, symlink, path, fileId,
+ childrenNum, feInfo, storagePolicy);
this.locations = locations;
}
-
+
public LocatedBlocks getBlockLocations() {
return locations;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/SnapshottableDirectoryStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/SnapshottableDirectoryStatus.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/SnapshottableDirectoryStatus.java
index d395283..13acc7a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/SnapshottableDirectoryStatus.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/SnapshottableDirectoryStatus.java
@@ -24,6 +24,7 @@ import java.util.Date;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.BlockStoragePolicy;
import org.apache.hadoop.hdfs.DFSUtil;
/**
@@ -61,7 +62,7 @@ public class SnapshottableDirectoryStatus {
int snapshotNumber, int snapshotQuota, byte[] parentFullPath) {
this.dirStatus = new HdfsFileStatus(0, true, 0, 0, modification_time,
access_time, permission, owner, group, null, localName, inodeId,
- childrenNum, null);
+ childrenNum, null, BlockStoragePolicy.ID_UNSPECIFIED);
this.snapshotNumber = snapshotNumber;
this.snapshotQuota = snapshotQuota;
this.parentFullPath = parentFullPath;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
index 5af349e..9d0d13c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
@@ -173,6 +173,8 @@ import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetRep
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetReplicationResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetSafeModeRequestProto;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetSafeModeResponseProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetStoragePolicyRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetStoragePolicyResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetTimesRequestProto;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetTimesResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.UpdateBlockForPipelineRequestProto;
@@ -236,6 +238,8 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
static final GetSnapshottableDirListingResponseProto
NULL_GET_SNAPSHOTTABLE_DIR_LISTING_RESPONSE =
GetSnapshottableDirListingResponseProto.newBuilder().build();
+ static final SetStoragePolicyResponseProto VOID_SET_STORAGE_POLICY_RESPONSE =
+ SetStoragePolicyResponseProto.newBuilder().build();
private static final CreateResponseProto VOID_CREATE_RESPONSE =
CreateResponseProto.newBuilder().build();
@@ -1413,6 +1417,18 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
return VOID_CHECKACCESS_RESPONSE;
}
+ @Override
+ public SetStoragePolicyResponseProto setStoragePolicy(
+ RpcController controller, SetStoragePolicyRequestProto request)
+ throws ServiceException {
+ try {
+ server.setStoragePolicy(request.getSrc(), request.getPolicyName());
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ return VOID_SET_STORAGE_POLICY_RESPONSE;
+ }
+
public GetCurrentEditLogTxidResponseProto getCurrentEditLogTxid(RpcController controller,
GetCurrentEditLogTxidRequestProto req) throws ServiceException {
try {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
index 17e4439..1279f7c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
@@ -64,7 +64,9 @@ import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
import org.apache.hadoop.hdfs.protocol.NSQuotaExceededException;
+import org.apache.hadoop.hdfs.protocol.QuotaExceededException;
import org.apache.hadoop.hdfs.protocol.RollingUpgradeInfo;
+import org.apache.hadoop.hdfs.protocol.SnapshotAccessControlException;
import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport;
import org.apache.hadoop.hdfs.protocol.SnapshottableDirectoryStatus;
import org.apache.hadoop.hdfs.protocol.proto.AclProtos.GetAclStatusRequestProto;
@@ -152,6 +154,7 @@ import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetTim
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.UpdateBlockForPipelineRequestProto;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.UpdatePipelineRequestProto;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.CheckAccessRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.SetStoragePolicyRequestProto;
import org.apache.hadoop.hdfs.protocol.proto.EncryptionZonesProtos;
import org.apache.hadoop.hdfs.protocol.proto.EncryptionZonesProtos.CreateEncryptionZoneRequestProto;
import org.apache.hadoop.hdfs.protocol.proto.EncryptionZonesProtos.GetEZForPathRequestProto;
@@ -1435,6 +1438,19 @@ public class ClientNamenodeProtocolTranslatorPB implements
}
}
+ @Override
+ public void setStoragePolicy(String src, String policyName)
+ throws SnapshotAccessControlException, UnresolvedLinkException,
+ FileNotFoundException, QuotaExceededException, IOException {
+ SetStoragePolicyRequestProto req = SetStoragePolicyRequestProto
+ .newBuilder().setSrc(src).setPolicyName(policyName).build();
+ try {
+ rpcProxy.setStoragePolicy(null, req);
+ } catch (ServiceException e) {
+ throw ProtobufHelper.getRemoteException(e);
+ }
+ }
+
public long getCurrentEditLogTxid() throws IOException {
GetCurrentEditLogTxidRequestProto req = GetCurrentEditLogTxidRequestProto
.getDefaultInstance();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
index dab2466..81bb231 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
@@ -43,6 +43,8 @@ import org.apache.hadoop.fs.permission.AclStatus;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
+import org.apache.hadoop.ha.proto.HAServiceProtocolProtos;
+import org.apache.hadoop.hdfs.BlockStoragePolicy;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.StorageType;
import org.apache.hadoop.hdfs.inotify.Event;
@@ -175,6 +177,7 @@ import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifie
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NodeType;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState;
+import org.apache.hadoop.hdfs.server.common.Storage;
import org.apache.hadoop.hdfs.server.common.StorageInfo;
import org.apache.hadoop.hdfs.server.namenode.CheckpointSignature;
import org.apache.hadoop.hdfs.server.namenode.INodeId;
@@ -892,9 +895,25 @@ public class PBHelper {
}
builder.addAllTargets(convert(cmd.getTargets()))
.addAllTargetStorageUuids(convert(cmd.getTargetStorageIDs()));
+ StorageType[][] types = cmd.getTargetStorageTypes();
+ if (types != null) {
+ builder.addAllTargetStorageTypes(convert(types));
+ }
return builder.build();
}
-
+
+ private static List<StorageTypesProto> convert(StorageType[][] types) {
+ List<StorageTypesProto> list = Lists.newArrayList();
+ if (types != null) {
+ for (StorageType[] ts : types) {
+ StorageTypesProto.Builder builder = StorageTypesProto.newBuilder();
+ builder.addAllStorageTypes(convertStorageTypes(ts));
+ list.add(builder.build());
+ }
+ }
+ return list;
+ }
+
public static BlockIdCommandProto convert(BlockIdCommand cmd) {
BlockIdCommandProto.Builder builder = BlockIdCommandProto.newBuilder()
.setBlockPoolId(cmd.getBlockPoolId());
@@ -1023,7 +1042,7 @@ public class PBHelper {
} else {
for(int i = 0; i < targetStorageTypes.length; i++) {
List<StorageTypeProto> p = targetStorageTypesList.get(i).getStorageTypesList();
- targetStorageTypes[i] = p.toArray(new StorageType[p.size()]);
+ targetStorageTypes[i] = convertStorageTypes(p, targets[i].length);
}
}
@@ -1329,8 +1348,9 @@ public class PBHelper {
fs.hasFileId()? fs.getFileId(): INodeId.GRANDFATHER_INODE_ID,
fs.hasLocations() ? PBHelper.convert(fs.getLocations()) : null,
fs.hasChildrenNum() ? fs.getChildrenNum() : -1,
- fs.hasFileEncryptionInfo() ? convert(fs.getFileEncryptionInfo()) :
- null);
+ fs.hasFileEncryptionInfo() ? convert(fs.getFileEncryptionInfo()) : null,
+ fs.hasStoragePolicy() ? (byte) fs.getStoragePolicy()
+ : BlockStoragePolicy.ID_UNSPECIFIED);
}
public static SnapshottableDirectoryStatus convert(
@@ -1376,7 +1396,8 @@ public class PBHelper {
setGroup(fs.getGroup()).
setFileId(fs.getFileId()).
setChildrenNum(fs.getChildrenNum()).
- setPath(ByteString.copyFrom(fs.getLocalNameInBytes()));
+ setPath(ByteString.copyFrom(fs.getLocalNameInBytes())).
+ setStoragePolicy(fs.getStoragePolicy());
if (fs.isSymlink()) {
builder.setSymlink(ByteString.copyFrom(fs.getSymlinkInBytes()));
}
@@ -1384,7 +1405,8 @@ public class PBHelper {
builder.setFileEncryptionInfo(convert(fs.getFileEncryptionInfo()));
}
if (fs instanceof HdfsLocatedFileStatus) {
- LocatedBlocks locations = ((HdfsLocatedFileStatus)fs).getBlockLocations();
+ final HdfsLocatedFileStatus lfs = (HdfsLocatedFileStatus) fs;
+ LocatedBlocks locations = lfs.getBlockLocations();
if (locations != null) {
builder.setLocations(PBHelper.convert(locations));
}
@@ -1697,6 +1719,8 @@ public class PBHelper {
return StorageTypeProto.DISK;
case SSD:
return StorageTypeProto.SSD;
+ case ARCHIVE:
+ return StorageTypeProto.ARCHIVE;
default:
throw new IllegalStateException(
"BUG: StorageType not found, type=" + type);
@@ -1725,6 +1749,8 @@ public class PBHelper {
return StorageType.DISK;
case SSD:
return StorageType.SSD;
+ case ARCHIVE:
+ return StorageType.ARCHIVE;
default:
throw new IllegalStateException(
"BUG: StorageTypeProto not found, type=" + type);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
index 53e78c7..d77da7c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
@@ -23,7 +23,6 @@ import java.io.IOException;
import java.io.PrintStream;
import java.net.URI;
import java.text.DateFormat;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -54,6 +53,7 @@ import org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault
import org.apache.hadoop.hdfs.server.namenode.UnsupportedActionException;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
import org.apache.hadoop.hdfs.server.protocol.StorageReport;
+import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.util.Tool;
@@ -271,7 +271,7 @@ public class Balancer {
// over-utilized, above-average, below-average and under-utilized.
long overLoadedBytes = 0L, underLoadedBytes = 0L;
for(DatanodeStorageReport r : reports) {
- final DDatanode dn = dispatcher.newDatanode(r);
+ final DDatanode dn = dispatcher.newDatanode(r.getDatanodeInfo());
for(StorageType t : StorageType.asList()) {
final Double utilization = policy.getUtilization(r, t);
if (utilization == null) { // datanode does not have such storage type
@@ -295,7 +295,7 @@ public class Balancer {
}
g = s;
} else {
- g = dn.addStorageGroup(t, maxSize2Move);
+ g = dn.addTarget(t, maxSize2Move);
if (thresholdDiff <= 0) { // within threshold
belowAvgUtilized.add(g);
} else {
@@ -549,15 +549,10 @@ public class Balancer {
final Formatter formatter = new Formatter(System.out);
System.out.println("Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved");
- final List<NameNodeConnector> connectors
- = new ArrayList<NameNodeConnector>(namenodes.size());
+ List<NameNodeConnector> connectors = Collections.emptyList();
try {
- for (URI uri : namenodes) {
- final NameNodeConnector nnc = new NameNodeConnector(
- Balancer.class.getSimpleName(), uri, BALANCER_ID_PATH, conf);
- nnc.getKeyManager().startBlockKeyUpdater();
- connectors.add(nnc);
- }
+ connectors = NameNodeConnector.newNameNodeConnectors(namenodes,
+ Balancer.class.getSimpleName(), BALANCER_ID_PATH, conf);
boolean done = false;
for(int iteration = 0; !done; iteration++) {
@@ -582,7 +577,7 @@ public class Balancer {
}
} finally {
for(NameNodeConnector nnc : connectors) {
- nnc.close();
+ IOUtils.cleanup(LOG, nnc);
}
}
return ExitStatus.SUCCESS.getExitCode();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
index 35fdae5..5c0ea51 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
@@ -43,11 +43,15 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.StorageType;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
@@ -85,7 +89,11 @@ public class Dispatcher {
private static final long MAX_BLOCKS_SIZE_TO_FETCH = 2 * GB;
private static final int MAX_NO_PENDING_MOVE_ITERATIONS = 5;
- private static final long DELAY_AFTER_ERROR = 10 * 1000L; // 10 seconds
+ /**
+ * the period of time to delay the usage of a DataNode after hitting
+ * errors when using it for migrating data
+ */
+ private static long delayAfterErrors = 10 * 1000;
private final NameNodeConnector nnc;
private final SaslDataTransferClient saslClient;
@@ -102,12 +110,14 @@ public class Dispatcher {
private final MovedBlocks<StorageGroup> movedBlocks;
/** Map (datanodeUuid,storageType -> StorageGroup) */
- private final StorageGroupMap storageGroupMap = new StorageGroupMap();
+ private final StorageGroupMap<StorageGroup> storageGroupMap
+ = new StorageGroupMap<StorageGroup>();
private NetworkTopology cluster;
private final ExecutorService moveExecutor;
private final ExecutorService dispatchExecutor;
+
/** The maximum number of concurrent blocks moves at a datanode */
private final int maxConcurrentMovesPerNode;
@@ -139,18 +149,18 @@ public class Dispatcher {
}
}
- static class StorageGroupMap {
+ public static class StorageGroupMap<G extends StorageGroup> {
private static String toKey(String datanodeUuid, StorageType storageType) {
return datanodeUuid + ":" + storageType;
}
- private final Map<String, StorageGroup> map = new HashMap<String, StorageGroup>();
+ private final Map<String, G> map = new HashMap<String, G>();
- StorageGroup get(String datanodeUuid, StorageType storageType) {
+ public G get(String datanodeUuid, StorageType storageType) {
return map.get(toKey(datanodeUuid, storageType));
}
- void put(StorageGroup g) {
+ public void put(G g) {
final String key = toKey(g.getDatanodeInfo().getDatanodeUuid(), g.storageType);
final StorageGroup existing = map.put(key, g);
Preconditions.checkState(existing == null);
@@ -163,24 +173,32 @@ public class Dispatcher {
void clear() {
map.clear();
}
+
+ public Collection<G> values() {
+ return map.values();
+ }
}
/** This class keeps track of a scheduled block move */
- private class PendingMove {
+ public class PendingMove {
private DBlock block;
private Source source;
private DDatanode proxySource;
private StorageGroup target;
- private PendingMove() {
+ private PendingMove(Source source, StorageGroup target) {
+ this.source = source;
+ this.target = target;
}
@Override
public String toString() {
- final Block b = block.getBlock();
- return b + " with size=" + b.getNumBytes() + " from "
- + source.getDisplayName() + " to " + target.getDisplayName()
- + " through " + proxySource.datanode;
+ final Block b = block != null ? block.getBlock() : null;
+ String bStr = b != null ? (b + " with size=" + b.getNumBytes() + " ")
+ : " ";
+ return bStr + "from " + source.getDisplayName() + " to " + target
+ .getDisplayName() + " through " + (proxySource != null ? proxySource
+ .datanode : "");
}
/**
@@ -190,9 +208,11 @@ public class Dispatcher {
* @return true if a block and its proxy are chosen; false otherwise
*/
private boolean chooseBlockAndProxy() {
+ // source and target must have the same storage type
+ final StorageType t = source.getStorageType();
// iterate all source's blocks until find a good one
for (Iterator<DBlock> i = source.getBlockIterator(); i.hasNext();) {
- if (markMovedIfGoodBlock(i.next())) {
+ if (markMovedIfGoodBlock(i.next(), t)) {
i.remove();
return true;
}
@@ -203,10 +223,10 @@ public class Dispatcher {
/**
* @return true if the given block is good for the tentative move.
*/
- private boolean markMovedIfGoodBlock(DBlock block) {
+ private boolean markMovedIfGoodBlock(DBlock block, StorageType targetStorageType) {
synchronized (block) {
synchronized (movedBlocks) {
- if (isGoodBlockCandidate(source, target, block)) {
+ if (isGoodBlockCandidate(source, target, targetStorageType, block)) {
this.block = block;
if (chooseProxySource()) {
movedBlocks.put(block);
@@ -299,12 +319,13 @@ public class Dispatcher {
LOG.info("Successfully moved " + this);
} catch (IOException e) {
LOG.warn("Failed to move " + this + ": " + e.getMessage());
+ target.getDDatanode().setHasFailure();
// Proxy or target may have some issues, delay before using these nodes
// further in order to avoid a potential storm of "threads quota
// exceeded" warnings when the dispatcher gets out of sync with work
// going on in datanodes.
- proxySource.activateDelay(DELAY_AFTER_ERROR);
- target.getDDatanode().activateDelay(DELAY_AFTER_ERROR);
+ proxySource.activateDelay(delayAfterErrors);
+ target.getDDatanode().activateDelay(delayAfterErrors);
} finally {
IOUtils.closeStream(out);
IOUtils.closeStream(in);
@@ -355,10 +376,23 @@ public class Dispatcher {
}
/** A class for keeping track of block locations in the dispatcher. */
- private static class DBlock extends MovedBlocks.Locations<StorageGroup> {
- DBlock(Block block) {
+ public static class DBlock extends MovedBlocks.Locations<StorageGroup> {
+ public DBlock(Block block) {
super(block);
}
+
+ @Override
+ public synchronized boolean isLocatedOn(StorageGroup loc) {
+ // currently we only check if replicas are located on the same DataNodes
+ // since we do not have the capability to store two replicas in the same
+ // DataNode even though they are on two different storage types
+ for (StorageGroup existing : locations) {
+ if (existing.getDatanodeInfo().equals(loc.getDatanodeInfo())) {
+ return true;
+ }
+ }
+ return false;
+ }
}
/** The class represents a desired move. */
@@ -377,10 +411,10 @@ public class Dispatcher {
}
/** A class that keeps track of a datanode. */
- static class DDatanode {
+ public static class DDatanode {
/** A group of storages in a datanode with the same storage type. */
- class StorageGroup {
+ public class StorageGroup {
final StorageType storageType;
final long maxSize2Move;
private long scheduledSize = 0L;
@@ -389,18 +423,26 @@ public class Dispatcher {
this.storageType = storageType;
this.maxSize2Move = maxSize2Move;
}
+
+ public StorageType getStorageType() {
+ return storageType;
+ }
private DDatanode getDDatanode() {
return DDatanode.this;
}
- DatanodeInfo getDatanodeInfo() {
+ public DatanodeInfo getDatanodeInfo() {
return DDatanode.this.datanode;
}
/** Decide if still need to move more bytes */
- synchronized boolean hasSpaceForScheduling() {
- return availableSizeToMove() > 0L;
+ boolean hasSpaceForScheduling() {
+ return hasSpaceForScheduling(0L);
+ }
+
+ synchronized boolean hasSpaceForScheduling(long size) {
+ return availableSizeToMove() > size;
}
/** @return the total number of bytes that need to be moved */
@@ -409,7 +451,7 @@ public class Dispatcher {
}
/** increment scheduled size */
- synchronized void incScheduledSize(long size) {
+ public synchronized void incScheduledSize(long size) {
scheduledSize += size;
}
@@ -423,6 +465,18 @@ public class Dispatcher {
scheduledSize = 0L;
}
+ private PendingMove addPendingMove(DBlock block, final PendingMove pm) {
+ if (getDDatanode().addPendingBlock(pm)) {
+ if (pm.markMovedIfGoodBlock(block, getStorageType())) {
+ incScheduledSize(pm.block.getNumBytes());
+ return pm;
+ } else {
+ getDDatanode().removePendingBlock(pm);
+ }
+ }
+ return null;
+ }
+
/** @return the name for display */
String getDisplayName() {
return datanode + ":" + storageType;
@@ -435,38 +489,46 @@ public class Dispatcher {
}
final DatanodeInfo datanode;
- final EnumMap<StorageType, StorageGroup> storageMap
+ private final EnumMap<StorageType, Source> sourceMap
+ = new EnumMap<StorageType, Source>(StorageType.class);
+ private final EnumMap<StorageType, StorageGroup> targetMap
= new EnumMap<StorageType, StorageGroup>(StorageType.class);
protected long delayUntil = 0L;
/** blocks being moved but not confirmed yet */
private final List<PendingMove> pendings;
+ private volatile boolean hasFailure = false;
private final int maxConcurrentMoves;
@Override
public String toString() {
- return getClass().getSimpleName() + ":" + datanode + ":" + storageMap.values();
+ return getClass().getSimpleName() + ":" + datanode;
}
- private DDatanode(DatanodeStorageReport r, int maxConcurrentMoves) {
- this.datanode = r.getDatanodeInfo();
+ private DDatanode(DatanodeInfo datanode, int maxConcurrentMoves) {
+ this.datanode = datanode;
this.maxConcurrentMoves = maxConcurrentMoves;
this.pendings = new ArrayList<PendingMove>(maxConcurrentMoves);
}
- private void put(StorageType storageType, StorageGroup g) {
- final StorageGroup existing = storageMap.put(storageType, g);
+ public DatanodeInfo getDatanodeInfo() {
+ return datanode;
+ }
+
+ private static <G extends StorageGroup> void put(StorageType storageType,
+ G g, EnumMap<StorageType, G> map) {
+ final StorageGroup existing = map.put(storageType, g);
Preconditions.checkState(existing == null);
}
- StorageGroup addStorageGroup(StorageType storageType, long maxSize2Move) {
+ public StorageGroup addTarget(StorageType storageType, long maxSize2Move) {
final StorageGroup g = new StorageGroup(storageType, maxSize2Move);
- put(storageType, g);
+ put(storageType, g, targetMap);
return g;
}
- Source addSource(StorageType storageType, long maxSize2Move, Dispatcher d) {
+ public Source addSource(StorageType storageType, long maxSize2Move, Dispatcher d) {
final Source s = d.new Source(storageType, maxSize2Move, this);
- put(storageType, s);
+ put(storageType, s, sourceMap);
return s;
}
@@ -504,10 +566,14 @@ public class Dispatcher {
synchronized boolean removePendingBlock(PendingMove pendingBlock) {
return pendings.remove(pendingBlock);
}
+
+ void setHasFailure() {
+ this.hasFailure = true;
+ }
}
/** A node that can be the sources of a block move */
- class Source extends DDatanode.StorageGroup {
+ public class Source extends DDatanode.StorageGroup {
private final List<Task> tasks = new ArrayList<Task>(2);
private long blocksToReceive = 0L;
@@ -575,8 +641,11 @@ public class Dispatcher {
/** Decide if the given block is a good candidate to move or not */
private boolean isGoodBlockCandidate(DBlock block) {
+ // source and target must have the same storage type
+ final StorageType sourceStorageType = getStorageType();
for (Task t : tasks) {
- if (Dispatcher.this.isGoodBlockCandidate(this, t.target, block)) {
+ if (Dispatcher.this.isGoodBlockCandidate(this, t.target,
+ sourceStorageType, block)) {
return true;
}
}
@@ -596,11 +665,9 @@ public class Dispatcher {
for (Iterator<Task> i = tasks.iterator(); i.hasNext();) {
final Task task = i.next();
final DDatanode target = task.target.getDDatanode();
- PendingMove pendingBlock = new PendingMove();
+ final PendingMove pendingBlock = new PendingMove(this, task.target);
if (target.addPendingBlock(pendingBlock)) {
// target is not busy, so do a tentative block allocation
- pendingBlock.source = this;
- pendingBlock.target = task.target;
if (pendingBlock.chooseBlockAndProxy()) {
long blockSize = pendingBlock.block.getNumBytes();
incScheduledSize(-blockSize);
@@ -617,6 +684,11 @@ public class Dispatcher {
}
return null;
}
+
+ /** Add a pending move */
+ public PendingMove addPendingMove(DBlock block, StorageGroup target) {
+ return target.addPendingMove(block, new PendingMove(this, target));
+ }
/** Iterate all source's blocks to remove moved ones */
private void removeMovedBlocks() {
@@ -655,13 +727,7 @@ public class Dispatcher {
if (p != null) {
// Reset no pending move counter
noPendingMoveIteration=0;
- // move the block
- moveExecutor.execute(new Runnable() {
- @Override
- public void run() {
- p.dispatch();
- }
- });
+ executePendingMove(p);
continue;
}
@@ -717,7 +783,8 @@ public class Dispatcher {
this.cluster = NetworkTopology.getInstance(conf);
this.moveExecutor = Executors.newFixedThreadPool(moverThreads);
- this.dispatchExecutor = Executors.newFixedThreadPool(dispatcherThreads);
+ this.dispatchExecutor = dispatcherThreads == 0? null
+ : Executors.newFixedThreadPool(dispatcherThreads);
this.maxConcurrentMovesPerNode = maxConcurrentMovesPerNode;
this.saslClient = new SaslDataTransferClient(
@@ -725,11 +792,15 @@ public class Dispatcher {
TrustedChannelResolver.getInstance(conf), nnc.fallbackToSimpleAuth);
}
- StorageGroupMap getStorageGroupMap() {
+ public DistributedFileSystem getDistributedFileSystem() {
+ return nnc.getDistributedFileSystem();
+ }
+
+ public StorageGroupMap<StorageGroup> getStorageGroupMap() {
return storageGroupMap;
}
- NetworkTopology getCluster() {
+ public NetworkTopology getCluster() {
return cluster;
}
@@ -777,7 +848,7 @@ public class Dispatcher {
}
/** Get live datanode storage reports and then build the network topology. */
- List<DatanodeStorageReport> init() throws IOException {
+ public List<DatanodeStorageReport> init() throws IOException {
final DatanodeStorageReport[] reports = nnc.getLiveDatanodeStorageReport();
final List<DatanodeStorageReport> trimmed = new ArrayList<DatanodeStorageReport>();
// create network topology and classify utilization collections:
@@ -793,8 +864,18 @@ public class Dispatcher {
return trimmed;
}
- public DDatanode newDatanode(DatanodeStorageReport r) {
- return new DDatanode(r, maxConcurrentMovesPerNode);
+ public DDatanode newDatanode(DatanodeInfo datanode) {
+ return new DDatanode(datanode, maxConcurrentMovesPerNode);
+ }
+
+ public void executePendingMove(final PendingMove p) {
+ // move the block
+ moveExecutor.execute(new Runnable() {
+ @Override
+ public void run() {
+ p.dispatch();
+ }
+ });
}
public boolean dispatchAndCheckContinue() throws InterruptedException {
@@ -834,7 +915,7 @@ public class Dispatcher {
}
// wait for all block moving to be done
- waitForMoveCompletion();
+ waitForMoveCompletion(targets);
return bytesMoved.get() - bytesLastMoved;
}
@@ -842,23 +923,25 @@ public class Dispatcher {
/** The sleeping period before checking if block move is completed again */
static private long blockMoveWaitTime = 30000L;
- /** set the sleeping period for block move completion check */
- static void setBlockMoveWaitTime(long time) {
- blockMoveWaitTime = time;
- }
-
- /** Wait for all block move confirmations. */
- private void waitForMoveCompletion() {
+ /**
+ * Wait for all block move confirmations.
+ * @return true if there is failed move execution
+ */
+ public static boolean waitForMoveCompletion(
+ Iterable<? extends StorageGroup> targets) {
+ boolean hasFailure = false;
for(;;) {
boolean empty = true;
for (StorageGroup t : targets) {
if (!t.getDDatanode().isPendingQEmpty()) {
empty = false;
break;
+ } else {
+ hasFailure |= t.getDDatanode().hasFailure;
}
}
if (empty) {
- return; //all pending queues are empty
+ return hasFailure; // all pending queues are empty
}
try {
Thread.sleep(blockMoveWaitTime);
@@ -869,14 +952,14 @@ public class Dispatcher {
/**
* Decide if the block is a good candidate to be moved from source to target.
- * A block is a good candidate if
+ * A block is a good candidate if
* 1. the block is not in the process of being moved/has not been moved;
* 2. the block does not have a replica on the target;
* 3. doing the move does not reduce the number of racks that the block has
*/
- private boolean isGoodBlockCandidate(Source source, StorageGroup target,
- DBlock block) {
- if (source.storageType != target.storageType) {
+ private boolean isGoodBlockCandidate(StorageGroup source, StorageGroup target,
+ StorageType targetStorageType, DBlock block) {
+ if (target.storageType != targetStorageType) {
return false;
}
// check if the block is moved or not
@@ -887,7 +970,7 @@ public class Dispatcher {
return false;
}
if (cluster.isNodeGroupAware()
- && isOnSameNodeGroupWithReplicas(target, block, source)) {
+ && isOnSameNodeGroupWithReplicas(source, target, block)) {
return false;
}
if (reduceNumOfRacks(source, target, block)) {
@@ -900,7 +983,7 @@ public class Dispatcher {
* Determine whether moving the given block replica from source to target
* would reduce the number of racks of the block replicas.
*/
- private boolean reduceNumOfRacks(Source source, StorageGroup target,
+ private boolean reduceNumOfRacks(StorageGroup source, StorageGroup target,
DBlock block) {
final DatanodeInfo sourceDn = source.getDatanodeInfo();
if (cluster.isOnSameRack(sourceDn, target.getDatanodeInfo())) {
@@ -933,12 +1016,12 @@ public class Dispatcher {
* Check if there are any replica (other than source) on the same node group
* with target. If true, then target is not a good candidate for placing
* specific replica as we don't want 2 replicas under the same nodegroup.
- *
+ *
* @return true if there are any replica (other than source) on the same node
* group with target
*/
- private boolean isOnSameNodeGroupWithReplicas(
- StorageGroup target, DBlock block, Source source) {
+ private boolean isOnSameNodeGroupWithReplicas(StorageGroup source,
+ StorageGroup target, DBlock block) {
final DatanodeInfo targetDn = target.getDatanodeInfo();
for (StorageGroup g : block.getLocations()) {
if (g != source && cluster.isOnSameNodeGroup(g.getDatanodeInfo(), targetDn)) {
@@ -958,9 +1041,22 @@ public class Dispatcher {
movedBlocks.cleanup();
}
+ /** set the sleeping period for block move completion check */
+ @VisibleForTesting
+ public static void setBlockMoveWaitTime(long time) {
+ blockMoveWaitTime = time;
+ }
+
+ @VisibleForTesting
+ public static void setDelayAfterErrors(long time) {
+ delayAfterErrors = time;
+ }
+
/** shutdown thread pools */
- void shutdownNow() {
- dispatchExecutor.shutdownNow();
+ public void shutdownNow() {
+ if (dispatchExecutor != null) {
+ dispatchExecutor.shutdownNow();
+ }
moveExecutor.shutdownNow();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
index 54febc6..f8d0071 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
@@ -31,6 +31,11 @@ public interface Matcher {
public boolean match(NetworkTopology cluster, Node left, Node right) {
return cluster.isOnSameNodeGroup(left, right);
}
+
+ @Override
+ public String toString() {
+ return "SAME_NODE_GROUP";
+ }
};
/** Match datanodes in the same rack. */
@@ -39,6 +44,11 @@ public interface Matcher {
public boolean match(NetworkTopology cluster, Node left, Node right) {
return cluster.isOnSameRack(left, right);
}
+
+ @Override
+ public String toString() {
+ return "SAME_RACK";
+ }
};
/** Match any datanode with any other datanode. */
@@ -47,5 +57,10 @@ public interface Matcher {
public boolean match(NetworkTopology cluster, Node left, Node right) {
return left != right;
}
+
+ @Override
+ public String toString() {
+ return "ANY_OTHER";
+ }
};
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/022474c8/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/MovedBlocks.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/MovedBlocks.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/MovedBlocks.java
index 557bfd3..18b9cd8 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/MovedBlocks.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/MovedBlocks.java
@@ -40,7 +40,7 @@ public class MovedBlocks<L> {
public static class Locations<L> {
private final Block block; // the block
/** The locations of the replicas of the block. */
- private final List<L> locations = new ArrayList<L>(3);
+ protected final List<L> locations = new ArrayList<L>(3);
public Locations(Block block) {
this.block = block;