|
@@ -16,7 +16,7 @@
|
|
|
* limitations under the License.
|
|
|
*/
|
|
|
|
|
|
-package org.apache.hadoop.hdfs;
|
|
|
+package org.apache.hadoop.hdfs.protocol;
|
|
|
|
|
|
import java.util.Arrays;
|
|
|
import java.util.EnumSet;
|
|
@@ -24,12 +24,10 @@ 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;
|
|
|
+import org.apache.hadoop.hdfs.StorageType;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
|
|
|
/**
|
|
|
* A block storage policy describes how to select the storage types
|
|
@@ -37,67 +35,8 @@ import org.apache.hadoop.fs.XAttr.NameSpace;
|
|
|
*/
|
|
|
@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;
|
|
|
- }
|
|
|
- }
|
|
|
+ public static final Logger LOG = LoggerFactory.getLogger(BlockStoragePolicy
|
|
|
+ .class);
|
|
|
|
|
|
/** A 4-bit policy ID */
|
|
|
private final byte id;
|
|
@@ -160,7 +99,7 @@ public class BlockStoragePolicy {
|
|
|
/**
|
|
|
* 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 unavailable storage types. It uses fallback storage in case that
|
|
|
* the desired storage type is unavailable.
|
|
|
*
|
|
|
* @param replication the replication number.
|
|
@@ -195,7 +134,7 @@ public class BlockStoragePolicy {
|
|
|
// 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
|
|
|
+ LOG.warn("Failed to place enough replicas: expected size is " + expectedSize
|
|
|
+ " but only " + storageTypes.size() + " storage types can be selected "
|
|
|
+ "(replication=" + replication
|
|
|
+ ", selected=" + storageTypes
|
|
@@ -207,7 +146,8 @@ public class BlockStoragePolicy {
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * Compute the list difference t = t - c.
|
|
|
+ * Compute the difference between two lists t and c so that after the diff
|
|
|
+ * computation we have: t = t - c;
|
|
|
* Further, if e is not null, set e = e + c - t;
|
|
|
*/
|
|
|
private static void diff(List<StorageType> t, Iterable<StorageType> c,
|
|
@@ -242,7 +182,7 @@ public class BlockStoragePolicy {
|
|
|
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);
|
|
@@ -280,140 +220,25 @@ public class BlockStoragePolicy {
|
|
|
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;
|
|
|
+ public StorageType[] getStorageTypes() {
|
|
|
+ return this.storageTypes;
|
|
|
}
|
|
|
|
|
|
- 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);
|
|
|
- }
|
|
|
+ public StorageType[] getCreationFallbacks() {
|
|
|
+ return this.creationFallbacks;
|
|
|
}
|
|
|
|
|
|
- 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);
|
|
|
+ public StorageType[] getReplicationFallbacks() {
|
|
|
+ return this.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) ";
|
|
|
+ private static StorageType getFallback(EnumSet<StorageType> unavailables,
|
|
|
+ StorageType[] fallbacks) {
|
|
|
+ for(StorageType fb : fallbacks) {
|
|
|
+ if (!unavailables.contains(fb)) {
|
|
|
+ return fb;
|
|
|
}
|
|
|
- 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) + "\".");
|
|
|
+ return null;
|
|
|
}
|
|
|
-}
|
|
|
+}
|