This is an automated email from the ASF dual-hosted git repository.
devmadhuu pushed a commit to branch HDDS-11233
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-11233 by this push:
new 1ebaf1383cc HDDS-16665. Create key should support StoragePolicy
(#11377)
1ebaf1383cc is described below
commit 1ebaf1383cc4252311f9c676816c29536db0a7f8
Author: Devesh Kumar Singh <[email protected]>
AuthorDate: Wed Oct 14 09:36:22 2026 +0530
HDDS-16665. Create key should support StoragePolicy (#11377)
---
.../hadoop/hdds/scm/storage/BlockOutputStream.java | 17 +-
.../hdds/scm/storage/ECBlockOutputStream.java | 5 +-
.../hdds/scm/storage/RatisBlockOutputStream.java | 7 +-
.../storage/TestBlockOutputStreamCorrectness.java | 4 +-
.../hadoop/hdds/client/OzoneStoragePolicy.java | 2 +
.../hadoop/hdds/client/StorageTypeUtils.java | 2 +-
.../apache/hadoop/ozone/client/OzoneBucket.java | 25 ++
.../org/apache/hadoop/ozone/client/OzoneKey.java | 39 +-
.../hadoop/ozone/client/OzoneKeyDetails.java | 19 +-
.../ozone/client/io/BlockOutputStreamEntry.java | 13 +-
.../client/io/BlockOutputStreamEntryPool.java | 20 +
.../ozone/client/protocol/ClientProtocol.java | 25 ++
.../apache/hadoop/ozone/client/rpc/RpcClient.java | 17 +-
.../apache/hadoop/ozone/om/helpers/OmKeyArgs.java | 21 +
.../apache/hadoop/ozone/om/helpers/OmKeyInfo.java | 25 ++
...OzoneManagerProtocolClientSideTranslatorPB.java | 10 +
.../ozone/client/rpc/TestOzoneStoragePolicy.java | 446 +++++++++++++++++++++
.../src/main/proto/OmClientProtocol.proto | 9 +
.../ozone/om/request/file/OMFileCreateRequest.java | 9 +-
.../om/request/key/OMAllocateBlockRequest.java | 5 +-
.../ozone/om/request/key/OMKeyCreateRequest.java | 11 +-
.../om/request/key/OMKeyCreateRequestWithFSO.java | 12 +
.../hadoop/ozone/om/request/key/OMKeyRequest.java | 44 +-
.../om/request/key/TestOMAllocateBlockRequest.java | 10 +-
24 files changed, 770 insertions(+), 27 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
index 98855d28ac0..3e0624e4050 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
@@ -100,6 +100,9 @@ public class BlockOutputStream extends OutputStream {
private AtomicReference<BlockID> blockID;
// planned block full size
private long blockSize;
+ // Storage type the block should be written to, from the key's storage
policy.
+ // Null leaves the datanode free to pick any volume.
+ private final StorageType storageType;
private AtomicBoolean eofSent = new AtomicBoolean(false);
private final AtomicReference<ChunkInfo> previousChunkInfo
= new AtomicReference<>();
@@ -179,19 +182,23 @@ public BlockOutputStream(
OzoneClientConfig config,
Token<? extends TokenIdentifier> token,
ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs,
- Supplier<ExecutorService> blockOutputStreamResourceProvider
+ Supplier<ExecutorService> blockOutputStreamResourceProvider,
+ StorageType storageType
) throws IOException {
this.xceiverClientFactory = xceiverClientManager;
this.config = config;
this.blockID = new AtomicReference<>(blockID);
this.blockSize = blockSize;
+ this.storageType = storageType;
replicationIndex = pipeline.getReplicaIndex(pipeline.getClosestNode());
KeyValue keyValue =
KeyValue.newBuilder().setKey("TYPE").setValue("KEY").build();
+ // Send the storage type SCM chose for this block so the datanode places
it on a volume of
+ // that type. Null leaves the datanode free to pick any volume.
this.containerBlockData = BlockData.newBuilder()
.setBlockID(blockID.getDatanodeBlockIDProtobufBuilder(replicationIndex,
- StorageTypeUtils.getID(StorageType.DISK)))
+ storageType == null ? null : StorageTypeUtils.getID(storageType)))
.addMetadata(keyValue);
this.pipeline = pipeline;
// tell DataNode I will send incremental chunk list
@@ -959,12 +966,10 @@ private CompletableFuture<PutBlockResult>
writeChunkToContainer(
byteBufferList = null;
}
- // TODO: Pass the requested storage type from allocation/storage policy
- // once that context is wired through BlockOutputStream. Null preserves
- // the current any-volume behavior.
asyncReply = writeChunkAsync(xceiverClient, chunkInfo,
blockID.get(), data, tokenString, replicationIndex, blockData, close,
- null, containerAutoCreate());
+ storageType == null ? null :
StorageTypeUtils.getStorageTypeProto(storageType),
+ containerAutoCreate());
CompletableFuture<ContainerCommandResponseProto>
respFuture = asyncReply.getResponse();
validateFuture = respFuture.thenApplyAsync(e -> {
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
index 5f33e02f4d5..7b55a4be312 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
@@ -101,8 +101,11 @@ public ECBlockOutputStream(
Supplier<ExecutorService> executorServiceSupplier,
boolean containerAutoCreate
) throws IOException {
+ // EC keys do not carry a per-block storage type yet; null leaves the
datanode
+ // free to pick any volume. EC storage policy support is tracked
separately.
super(blockID, -1, xceiverClientManager,
- pipeline, bufferPool, config, token, clientMetrics, streamBufferArgs,
executorServiceSupplier);
+ pipeline, bufferPool, config, token, clientMetrics, streamBufferArgs,
executorServiceSupplier,
+ null);
// In EC stream, there will be only one node in pipeline.
this.datanodeDetails = pipeline.getClosestNode();
this.containerAutoCreate = containerAutoCreate;
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/RatisBlockOutputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/RatisBlockOutputStream.java
index 684707908fb..d2b0eff67d1 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/RatisBlockOutputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/RatisBlockOutputStream.java
@@ -26,6 +26,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.function.Supplier;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.fs.Syncable;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.scm.ContainerClientMetrics;
@@ -78,10 +79,12 @@ public RatisBlockOutputStream(
OzoneClientConfig config,
Token<? extends TokenIdentifier> token,
ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs,
- Supplier<ExecutorService> blockOutputStreamResourceProvider
+ Supplier<ExecutorService> blockOutputStreamResourceProvider,
+ StorageType storageType
) throws IOException {
super(blockID, blockSize, xceiverClientManager, pipeline,
- bufferPool, config, token, clientMetrics, streamBufferArgs,
blockOutputStreamResourceProvider);
+ bufferPool, config, token, clientMetrics, streamBufferArgs,
blockOutputStreamResourceProvider,
+ storageType);
this.commitWatcher = new CommitWatcher(bufferPool, getXceiverClient());
}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockOutputStreamCorrectness.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockOutputStreamCorrectness.java
index 81deaaf36bb..460c6519668 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockOutputStreamCorrectness.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockOutputStreamCorrectness.java
@@ -32,6 +32,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -221,7 +222,8 @@ private BlockOutputStream
createBlockOutputStream(BufferPool bufferPool)
null,
ContainerClientMetrics.acquire(),
streamBufferArgs,
- () -> newFixedThreadPool(10));
+ () -> newFixedThreadPool(10),
+ StorageType.DISK);
}
private ECBlockOutputStream createECBlockOutputStream(OzoneClientConfig
clientConfig,
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneStoragePolicy.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneStoragePolicy.java
index cdfbf2953db..b27297fa6ea 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneStoragePolicy.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneStoragePolicy.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.client;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StoragePolicyProto;
+import org.apache.ratis.util.Preconditions;
/**
* Enum defining different storage policies using StorageTier.
@@ -117,6 +118,7 @@ public static OzoneStoragePolicy getDefaultPolicy() {
}
public static void setDefaultPolicy(OzoneStoragePolicy storagePolicy) {
+ Preconditions.assertNotNull(storagePolicy, "storagePolicy == null");
defaultPolicy = storagePolicy;
}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
index 0b56ccd55a2..55cf6ec74d5 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
@@ -108,7 +108,7 @@ public static int getIDFromProtobuf(@Nonnull
HddsProtos.StorageTypeProto proto)
* Returns integer representation of FileSystem StorageType.
* @return storageType int ID value
*/
- public static int getID(StorageType storageType) throws
IllegalArgumentException {
+ public static int getID(@Nonnull StorageType storageType) throws
IllegalArgumentException {
return getStorageTypeProto(storageType).getNumber();
}
}
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java
index 930d3d9db6c..e539247b486 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java
@@ -578,6 +578,31 @@ public OzoneOutputStream createKey(String key, long size,
.createKey(volumeName, name, key, size, replicationConfig,
keyMetadata, tags, derivedKeyPiggyBacking);
}
+ /**
+ * Creates a new key in this bucket and places it according to the given
storage
+ * policy.
+ *
+ * @param key Name of the key to be created.
+ * @param size Size of the data the key will hold.
+ * @param replicationConfig Replication configuration.
+ * @param keyMetadata Custom key metadata.
+ * @param tags Tags used for S3 object tags.
+ * @param keyStoragePolicy Storage policy for the key, or null to inherit
this
+ * bucket's policy.
+ * @return OzoneOutputStream to which the data has to be written.
+ * @throws IOException
+ */
+ @SuppressWarnings("checkstyle:parameternumber")
+ public OzoneOutputStream createKey(String key, long size,
+ ReplicationConfig replicationConfig,
+ Map<String, String> keyMetadata,
+ Map<String, String> tags,
+ StoragePolicy keyStoragePolicy)
+ throws IOException {
+ return proxy
+ .createKey(volumeName, name, key, size, replicationConfig,
keyMetadata, tags, keyStoragePolicy);
+ }
+
/**
* This API allows to atomically update an existing key. The key read before
invoking this API
* should remain unchanged for this key to be written. This is controlled by
the generation
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKey.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKey.java
index 7cd0bc42c59..2a5df6a6c42 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKey.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKey.java
@@ -23,6 +23,7 @@
import java.util.Map;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.client.ReplicationType;
+import org.apache.hadoop.hdds.client.StoragePolicy;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
/**
@@ -70,11 +71,26 @@ public class OzoneKey {
*/
private final boolean isFile;
+ /**
+ * Storage policy the key was created with. Null for keys written before
storage
+ * policy support.
+ */
+ private final StoragePolicy storagePolicy;
+
@SuppressWarnings("parameternumber")
public OzoneKey(String volumeName, String bucketName,
String keyName, long size, long creationTime,
long modificationTime, ReplicationConfig replicationConfig,
boolean isFile, String owner) {
+ this(volumeName, bucketName, keyName, size, creationTime, modificationTime,
+ replicationConfig, isFile, owner, null);
+ }
+
+ @SuppressWarnings("parameternumber")
+ public OzoneKey(String volumeName, String bucketName,
+ String keyName, long size, long creationTime,
+ long modificationTime, ReplicationConfig replicationConfig,
+ boolean isFile, String owner, StoragePolicy storagePolicy) {
this.volumeName = volumeName;
this.bucketName = bucketName;
this.name = keyName;
@@ -84,6 +100,7 @@ public OzoneKey(String volumeName, String bucketName,
this.replicationConfig = replicationConfig;
this.isFile = isFile;
this.owner = owner;
+ this.storagePolicy = storagePolicy;
}
@SuppressWarnings("parameternumber")
@@ -92,8 +109,18 @@ public OzoneKey(String volumeName, String bucketName,
long modificationTime, ReplicationConfig replicationConfig,
Map<String, String> metadata, boolean isFile, String owner,
Map<String, String> tags) {
+ this(volumeName, bucketName, keyName, size, creationTime, modificationTime,
+ replicationConfig, metadata, isFile, owner, tags, null);
+ }
+
+ @SuppressWarnings("parameternumber")
+ public OzoneKey(String volumeName, String bucketName,
+ String keyName, long size, long creationTime,
+ long modificationTime, ReplicationConfig replicationConfig,
+ Map<String, String> metadata, boolean isFile, String owner,
+ Map<String, String> tags, StoragePolicy storagePolicy) {
this(volumeName, bucketName, keyName, size, creationTime,
- modificationTime, replicationConfig, isFile, owner);
+ modificationTime, replicationConfig, isFile, owner, storagePolicy);
this.metadata.putAll(metadata);
this.tags.putAll(tags);
}
@@ -213,6 +240,14 @@ public boolean isFile() {
return isFile;
}
+ /**
+ * Returns the storage policy the key was created with, or null when the key
+ * predates storage policy support.
+ */
+ public StoragePolicy getStoragePolicy() {
+ return storagePolicy;
+ }
+
/**
* Constructs OzoneKey from OmKeyInfo.
*
@@ -222,7 +257,7 @@ public static OzoneKey fromKeyInfo(OmKeyInfo keyInfo) {
keyInfo.getKeyName(), keyInfo.getDataSize(), keyInfo.getCreationTime(),
keyInfo.getModificationTime(), keyInfo.getReplicationConfig(),
keyInfo.getMetadata(), keyInfo.isFile(), keyInfo.getOwnerName(),
- keyInfo.getTags());
+ keyInfo.getTags(), keyInfo.getStoragePolicy());
}
}
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKeyDetails.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKeyDetails.java
index 7514f2a7cd7..52976b795f7 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKeyDetails.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneKeyDetails.java
@@ -23,6 +23,7 @@
import java.util.Map;
import org.apache.hadoop.fs.FileEncryptionInfo;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StoragePolicy;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.client.io.OzoneInputStream;
import org.apache.ratis.util.function.CheckedSupplier;
@@ -59,8 +60,24 @@ public OzoneKeyDetails(String volumeName, String bucketName,
String keyName,
FileEncryptionInfo feInfo,
CheckedSupplier<OzoneInputStream, IOException> contentSupplier,
boolean isFile, String owner, Map<String, String> tags, Long generation)
{
+ this(volumeName, bucketName, keyName, size, creationTime, modificationTime,
+ ozoneKeyLocations, replicationConfig, metadata, feInfo,
contentSupplier,
+ isFile, owner, tags, generation, null);
+ }
+
+ @SuppressWarnings("parameternumber")
+ public OzoneKeyDetails(String volumeName, String bucketName, String keyName,
+ long size, long creationTime, long modificationTime,
+ List<OzoneKeyLocation> ozoneKeyLocations,
+ ReplicationConfig replicationConfig,
+ Map<String, String> metadata,
+ FileEncryptionInfo feInfo,
+ CheckedSupplier<OzoneInputStream, IOException> contentSupplier,
+ boolean isFile, String owner, Map<String, String> tags, Long generation,
+ StoragePolicy storagePolicy) {
super(volumeName, bucketName, keyName, size, creationTime,
- modificationTime, replicationConfig, metadata, isFile, owner, tags);
+ modificationTime, replicationConfig, metadata, isFile, owner, tags,
+ storagePolicy);
this.ozoneKeyLocations = ozoneKeyLocations;
this.feInfo = feInfo;
this.contentSupplier = contentSupplier;
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntry.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntry.java
index 5fc0e92fb72..da62af663d1 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntry.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntry.java
@@ -26,6 +26,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.function.Supplier;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.fs.Syncable;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -70,6 +71,9 @@ public class BlockOutputStreamEntry extends OutputStream {
private final ContainerClientMetrics clientMetrics;
private final StreamBufferArgs streamBufferArgs;
private final Supplier<ExecutorService> executorServiceSupplier;
+ // Storage type the block should be written to, derived from the key's
storage
+ // tier. Null lets the datanode pick any volume.
+ private final StorageType storageType;
/**
* An indicator that this BlockOutputStream is created to handoff writes
from another faulty BlockOutputStream.
@@ -97,6 +101,7 @@ public class BlockOutputStreamEntry extends OutputStream {
this.streamBufferArgs = b.streamBufferArgs;
this.executorServiceSupplier = b.executorServiceSupplier;
this.isHandlingRetry = b.forRetry;
+ this.storageType = b.storageType;
}
@Override
@@ -155,7 +160,7 @@ void waitForAllPendingFlushes() throws IOException {
void createOutputStream() throws IOException {
outputStream = new RatisBlockOutputStream(blockID, length,
xceiverClientManager,
pipeline, bufferPool, config, token, clientMetrics, streamBufferArgs,
- executorServiceSupplier);
+ executorServiceSupplier, storageType);
}
ContainerClientMetrics getClientMetrics() {
@@ -415,6 +420,7 @@ public static class Builder {
private StreamBufferArgs streamBufferArgs;
private Supplier<ExecutorService> executorServiceSupplier;
private boolean forRetry;
+ private StorageType storageType;
public Pipeline getPipeline() {
return pipeline;
@@ -485,6 +491,11 @@ public Builder setForRetry(boolean forRetry) {
return this;
}
+ public Builder setStorageType(StorageType storageType) {
+ this.storageType = storageType;
+ return this;
+ }
+
public BlockOutputStreamEntry build() {
return new BlockOutputStreamEntry(this);
}
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
index 0abfc001ee0..24ae5f57179 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
@@ -30,7 +30,9 @@
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.function.Supplier;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ContainerBlockID;
+import org.apache.hadoop.hdds.client.StorageTier;
import org.apache.hadoop.hdds.scm.ByteStringConversion;
import org.apache.hadoop.hdds.scm.ContainerClientMetrics;
import org.apache.hadoop.hdds.scm.OzoneClientConfig;
@@ -102,6 +104,10 @@ public BlockOutputStreamEntryPool(KeyOutputStream.Builder
b) {
.setBucketName(info.getBucketName()).setKeyName(info.getKeyName())
.setReplicationConfig(b.getReplicationConfig())
.setDataSize(info.getDataSize())
+ // Carry the key's storage policy into later allocateBlock calls, so
blocks allocated
+ // after the pre-allocated ones run out land on the same tier. Null
when the key did not
+ // ask for a policy, which leaves OM to resolve it from the bucket as
it did on create.
+ .setStoragePolicy(info.getStoragePolicy())
.setIsMultipartKey(b.isMultipartKey())
.setMultipartUploadID(b.getMultipartUploadID())
.setMultipartUploadPartNumber(b.getMultipartNumber());
@@ -169,9 +175,23 @@ BlockOutputStreamEntry createStreamEntry(OmKeyLocationInfo
subKeyInfo, boolean f
.setStreamBufferArgs(streamBufferArgs)
.setExecutorServiceSupplier(executorServiceSupplier)
.setForRetry(forRetry)
+ .setStorageType(getStorageType(subKeyInfo))
.build();
}
+ /**
+ * Resolves the storage type the block should land on from the storage tier
SCM
+ * assigned to it. Returns null when the tier is absent or spans more than
one
+ * storage type, which leaves the datanode free to pick any volume.
+ */
+ private static StorageType getStorageType(OmKeyLocationInfo subKeyInfo) {
+ StorageTier storageTier = subKeyInfo.getStorageTier();
+ if (storageTier == null || !storageTier.isUniform()) {
+ return null;
+ }
+ return storageTier.getUniformStorageType();
+ }
+
private synchronized void addKeyLocationInfo(OmKeyLocationInfo subKeyInfo,
boolean forRetry) {
Objects.requireNonNull(subKeyInfo.getPipeline(), "subKeyInfo.getPipeline()
== null");
streamEntries.add(createStreamEntry(subKeyInfo, forRetry));
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
index 3ee60408bd3..4bb1a9a31ae 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.ozone.client.protocol;
import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
import java.io.IOException;
import java.net.URI;
import java.util.List;
@@ -479,6 +480,30 @@ OzoneOutputStream createKey(String volumeName, String
bucketName,
Map<String, String> metadata, Map<String, String> tags)
throws IOException;
+ /**
+ * Writes a key in an existing bucket, placing it according to the given
storage
+ * policy.
+ * @param volumeName Name of the Volume
+ * @param bucketName Name of the Bucket
+ * @param keyName Name of the Key
+ * @param size Size of the data
+ * @param metadata Custom key value metadata
+ * @param tags Tags used for S3 object tags
+ * @param storagePolicy storage policy for the key, or null to inherit the
+ * bucket's policy
+ * @return {@link OzoneOutputStream}
+ *
+ */
+ @SuppressWarnings("checkstyle:parameternumber")
+ default OzoneOutputStream createKey(String volumeName, String bucketName,
+ String keyName, long size, ReplicationConfig replicationConfig,
+ Map<String, String> metadata, Map<String, String> tags,
+ @Nullable StoragePolicy storagePolicy)
+ throws IOException {
+ return createKey(volumeName, bucketName, keyName, size, replicationConfig,
+ metadata, tags);
+ }
+
/**
* Writes a key in an existing bucket.
* @param volumeName Name of the Volume
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
index 220476d9e6d..36edc0dd0d1 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
@@ -1449,6 +1449,21 @@ public OzoneOutputStream createKey(
return openOutputStream(builder.build(), size);
}
+ @Override
+ @SuppressWarnings("checkstyle:parameternumber")
+ public OzoneOutputStream createKey(
+ String volumeName, String bucketName, String keyName, long size,
+ ReplicationConfig replicationConfig,
+ Map<String, String> metadata, Map<String, String> tags,
+ StoragePolicy storagePolicy) throws IOException {
+ String ownerName = getRealUserInfo().getShortUserName();
+ OmKeyArgs.Builder builder = createWriteKeyArgsBuilder(volumeName,
+ bucketName, keyName, size, replicationConfig, metadata, tags);
+ builder.setOwnerName(ownerName);
+ builder.setStoragePolicy(storagePolicy);
+ return openOutputStream(builder.build(), size);
+ }
+
@Override
public OzoneOutputStream rewriteKey(String volumeName, String bucketName,
String keyName,
long size, long existingKeyGeneration, ReplicationConfig
replicationConfig,
@@ -1940,7 +1955,7 @@ private OzoneKeyDetails getOzoneKeyDetails(OmKeyInfo
keyInfo) {
keyInfo.getFileEncryptionInfo(),
() -> getInputStreamWithRetryFunction(keyInfo), keyInfo.isFile(),
keyInfo.getOwnerName(), keyInfo.getTags(),
- keyInfo.getGeneration()
+ keyInfo.getGeneration(), keyInfo.getStoragePolicy()
);
}
diff --git
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java
index 03ff4bd8f61..c99fc5f69c0 100644
---
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java
+++
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java
@@ -21,11 +21,14 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StoragePolicy;
import org.apache.hadoop.ozone.OzoneAcl;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.audit.Auditable;
@@ -66,6 +69,8 @@ public final class OmKeyArgs extends WithMetadata implements
Auditable {
private final String listPrefix;
private final String expectedETag;
private final boolean derivedKeyPiggyBacking;
+ // When null, OM resolves the policy from the bucket, then the cluster
default.
+ private final StoragePolicy storagePolicy;
private OmKeyArgs(Builder b) {
super(b);
@@ -90,6 +95,7 @@ private OmKeyArgs(Builder b) {
this.listPrefix = b.listPrefix;
this.expectedETag = b.expectedETag;
this.derivedKeyPiggyBacking = b.derivedKeyPiggyBacking;
+ this.storagePolicy = b.storagePolicy;
}
public boolean getIsMultipartKey() {
@@ -188,6 +194,11 @@ public boolean isDerivedKeyPiggyBacking() {
return derivedKeyPiggyBacking;
}
+ @Nullable
+ public StoragePolicy getStoragePolicy() {
+ return storagePolicy;
+ }
+
@Override
public Map<String, String> toAuditMap() {
Map<String, String> auditMap = new LinkedHashMap<>();
@@ -236,6 +247,9 @@ public KeyArgs toProtobuf() {
if (expectedETag != null) {
builder.setExpectedETag(expectedETag);
}
+ if (storagePolicy != null) {
+ builder.setStoragePolicy(OzoneStoragePolicy.toProto(storagePolicy));
+ }
return builder.build();
}
@@ -264,6 +278,7 @@ public static class Builder extends WithMetadata.Builder {
private String listPrefix = null;
private String expectedETag;
private boolean derivedKeyPiggyBacking;
+ private StoragePolicy storagePolicy;
public Builder() {
this(AclListBuilder.empty());
@@ -311,6 +326,7 @@ public Builder(OmKeyArgs obj) {
this.expectedDataGeneration = obj.expectedDataGeneration;
this.expectedETag = obj.expectedETag;
this.derivedKeyPiggyBacking = obj.derivedKeyPiggyBacking;
+ this.storagePolicy = obj.storagePolicy;
this.tags = MapBuilder.of(obj.tags);
this.acls = AclListBuilder.of(obj.acls);
this.listPrefix = obj.listPrefix;
@@ -462,6 +478,11 @@ public Builder setDerivedKeyPiggyBacking(boolean
derivedKeyPiggyBacking) {
return this;
}
+ public Builder setStoragePolicy(StoragePolicy storagePolicy) {
+ this.storagePolicy = storagePolicy;
+ return this;
+ }
+
public OmKeyArgs build() {
return new OmKeyArgs(this);
}
diff --git
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
index c2dc49a2d1e..adba2860ef0 100644
---
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
+++
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
@@ -33,8 +33,10 @@
import org.apache.hadoop.fs.FileEncryptionInfo;
import org.apache.hadoop.hdds.client.ContainerBlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StoragePolicy;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.utils.db.Codec;
import org.apache.hadoop.hdds.utils.db.CopyObject;
@@ -125,6 +127,10 @@ public final class OmKeyInfo extends WithParentObjectId
// been modified.
private final Long expectedDataGeneration;
+ // Storage policy the key was created with. Null for keys written before
+ // storage policy support; those fall back to the bucket policy on read.
+ private final StoragePolicy storagePolicy;
+
private OmKeyInfo(Builder b) {
super(b);
this.volumeName = b.volumeName;
@@ -143,6 +149,7 @@ private OmKeyInfo(Builder b) {
this.ownerName = b.ownerName;
this.tags = b.tags.build();
this.expectedDataGeneration = b.expectedDataGeneration;
+ this.storagePolicy = b.storagePolicy;
}
/**
@@ -221,6 +228,11 @@ public Long getExpectedDataGeneration() {
return expectedDataGeneration;
}
+ @Nullable
+ public StoragePolicy getStoragePolicy() {
+ return storagePolicy;
+ }
+
public String getOwnerName() {
return ownerName;
}
@@ -526,6 +538,7 @@ public static class Builder extends
WithParentObjectId.Builder<OmKeyInfo> {
private boolean isFile;
private final MapBuilder<String, String> tags;
private Long expectedDataGeneration = null;
+ private StoragePolicy storagePolicy;
public Builder() {
this.acls = AclListBuilder.empty();
@@ -548,6 +561,7 @@ public Builder(OmKeyInfo obj) {
this.fileChecksum = obj.fileChecksum;
this.isFile = obj.isFile;
this.expectedDataGeneration = obj.expectedDataGeneration;
+ this.storagePolicy = obj.storagePolicy;
this.tags = MapBuilder.of(obj.tags);
obj.keyLocationVersions.forEach(keyLocationVersion ->
this.omKeyLocationInfoGroups.add(
@@ -719,6 +733,11 @@ public Builder setExpectedDataGeneration(Long
existingGeneration) {
return this;
}
+ public Builder setStoragePolicy(StoragePolicy storagePolicy) {
+ this.storagePolicy = storagePolicy;
+ return this;
+ }
+
@Override
protected void validate() {
super.validate();
@@ -870,6 +889,9 @@ private KeyInfo getProtobuf(boolean ignorePipeline, String
fullKeyName,
if (ownerName != null) {
kb.setOwnerName(ownerName);
}
+ if (storagePolicy != null) {
+ kb.setStoragePolicy(OzoneStoragePolicy.toProto(storagePolicy));
+ }
return kb.build();
}
@@ -924,6 +946,9 @@ public static Builder builderFromProtobuf(KeyInfo keyInfo) {
if (keyInfo.hasOwnerName()) {
builder.setOwnerName(keyInfo.getOwnerName());
}
+ if (keyInfo.hasStoragePolicy()) {
+
builder.setStoragePolicy(OzoneStoragePolicy.fromProto(keyInfo.getStoragePolicy()));
+ }
return builder;
}
diff --git
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
index caa6acc5ce2..8b686be238d 100644
---
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
+++
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
@@ -46,6 +46,7 @@
import org.apache.hadoop.fs.SafeModeAction;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import
org.apache.hadoop.hdds.protocol.proto.HddsProtos.TransferLeadershipRequestProto;
import
org.apache.hadoop.hdds.protocol.proto.HddsProtos.UpgradeFinalizationStatus;
@@ -789,6 +790,9 @@ public OpenKeySession openKey(OmKeyArgs args) throws
IOException {
if (args.getExpectedETag() != null) {
keyArgs.setExpectedETag(args.getExpectedETag());
}
+ if (args.getStoragePolicy() != null) {
+
keyArgs.setStoragePolicy(OzoneStoragePolicy.toProto(args.getStoragePolicy()));
+ }
req.setKeyArgs(keyArgs.build());
req.setDerivedKeyPiggyBacking(args.isDerivedKeyPiggyBacking());
@@ -837,6 +841,12 @@ public OmKeyLocationInfo allocateBlock(OmKeyArgs args,
long clientId,
keyArgs.setType(args.getReplicationConfig().getReplicationType());
}
+ // Send the key's storage policy so OM allocates on the same tier as the
key's earlier
+ // blocks. Absent when the key inherited its policy, which OM then
resolves from the bucket.
+ if (args.getStoragePolicy() != null) {
+
keyArgs.setStoragePolicy(OzoneStoragePolicy.toProto(args.getStoragePolicy()));
+ }
+
req.setKeyArgs(keyArgs);
req.setClientID(clientId);
req.setExcludeList(excludeList.getProtoBuf());
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneStoragePolicy.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneStoragePolicy.java
new file mode 100644
index 00000000000..4e4981acd31
--- /dev/null
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneStoragePolicy.java
@@ -0,0 +1,446 @@
+/*
+ * 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.ozone.client.rpc;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static
org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_RATIS_PIPELINE_LIMIT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.client.DefaultReplicationConfig;
+import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.conf.StorageUnit;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.XceiverClientGrpc;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
+import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls;
+import org.apache.hadoop.hdds.utils.IOUtils;
+import org.apache.hadoop.ozone.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.client.BucketArgs;
+import org.apache.hadoop.ozone.client.ObjectStore;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneClientFactory;
+import org.apache.hadoop.ozone.client.OzoneKeyDetails;
+import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.container.common.impl.ContainerData;
+import org.apache.hadoop.ozone.container.common.interfaces.Container;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that a key honours its storage policy end to end: the policy is
+ * persisted on the key, SCM allocates the container on the policy's creation
+ * tier, and the block physically lands on a datanode volume of that tier.
+ *
+ * Each test builds its own cluster because the datanode volume topology
differs
+ * per case.
+ */
+public class TestOzoneStoragePolicy {
+
+ private static final ReplicationConfig RATIS_THREE =
+ RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE);
+
+ private MiniOzoneCluster cluster;
+ private OzoneClient ozClient;
+ private ObjectStore store;
+ private OzoneManager ozoneManager;
+ private ContainerManager containerManager;
+ private PipelineManager pipelineManager;
+ private OzoneConfiguration conf;
+
+ /**
+ * Starts a cluster whose datanodes each expose one volume per storage type,
so
+ * every storage policy has a matching volume to land on.
+ */
+ private void startCluster(List<List<StorageType>> storageTypeList)
+ throws Exception {
+ conf = new OzoneConfiguration();
+ conf.setInt(OZONE_SCM_RATIS_PIPELINE_LIMIT, 10);
+ conf.setStorageSize(OZONE_SCM_BLOCK_SIZE, 4, StorageUnit.MB);
+ cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(storageTypeList.size())
+ .setNumDataVolumes(storageTypeList.get(0).size())
+ .setDatanodeStorageType(storageTypeList)
+ .build();
+ cluster.waitForClusterToBeReady();
+ cluster.waitTobeOutOfSafeMode();
+
+ ozClient = OzoneClientFactory.getRpcClient(conf);
+ store = ozClient.getObjectStore();
+ ozoneManager = cluster.getOzoneManager();
+ containerManager =
cluster.getStorageContainerManager().getContainerManager();
+ pipelineManager =
cluster.getStorageContainerManager().getPipelineManager();
+ }
+
+ private static List<List<StorageType>> allTierTopology(int datanodes) {
+ List<StorageType> volumes =
+ Arrays.asList(StorageType.DISK, StorageType.SSD, StorageType.ARCHIVE);
+ return Collections.nCopies(datanodes, volumes);
+ }
+
+ @AfterEach
+ public void shutdownCluster() {
+ IOUtils.closeQuietly(ozClient);
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ /**
+ * A key created with an explicit storage policy records that policy and
lands
+ * on the policy's creation tier, in SCM and on the datanodes.
+ */
+ @Test
+ public void testKeyStoragePolicyIsHonoured() throws Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ for (OzoneStoragePolicy storagePolicy : OzoneStoragePolicy.values()) {
+ OzoneBucket bucket = createBucket(null, BucketLayout.OBJECT_STORE);
+ OmKeyInfo keyInfo = createKeyAndLookup(bucket, storagePolicy);
+
+ assertKeyInfo(keyInfo, storagePolicy);
+ assertScmContainerAndPipeline(keyInfo, storagePolicy.getCreationTier());
+ assertDatanodeContainerAndBlock(keyInfo,
storagePolicy.getCreationTier());
+ }
+ }
+
+ /**
+ * A key created without a storage policy inherits the bucket's policy, and
an
+ * explicit key level policy overrides the bucket's.
+ */
+ @Test
+ public void testKeyInheritsAndOverridesBucketStoragePolicy() throws
Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ OzoneStoragePolicy bucketPolicy = OzoneStoragePolicy.COLD;
+ OzoneStoragePolicy keyPolicy = OzoneStoragePolicy.HOT;
+
+ OzoneBucket bucket = createBucket(bucketPolicy, BucketLayout.OBJECT_STORE);
+ assertEquals(bucketPolicy, bucket.getStoragePolicy());
+
+ // No key level policy: the bucket's policy applies.
+ OmKeyInfo inherited = createKeyAndLookup(bucket, null);
+ assertKeyInfo(inherited, bucketPolicy);
+ assertScmContainerAndPipeline(inherited, bucketPolicy.getCreationTier());
+ assertDatanodeContainerAndBlock(inherited, bucketPolicy.getCreationTier());
+
+ // Explicit key level policy wins over the bucket's.
+ OmKeyInfo overridden = createKeyAndLookup(bucket, keyPolicy);
+ assertKeyInfo(overridden, keyPolicy);
+ assertScmContainerAndPipeline(overridden, keyPolicy.getCreationTier());
+ assertDatanodeContainerAndBlock(overridden, keyPolicy.getCreationTier());
+ }
+
+ /**
+ * With neither a key nor a bucket policy, the key falls back to the cluster
+ * default rather than being left without one.
+ */
+ @Test
+ public void testKeyWithoutAnyPolicyUsesClusterDefault() throws Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ OzoneBucket bucket = createBucket(null, BucketLayout.OBJECT_STORE);
+ OmKeyInfo keyInfo = createKeyAndLookup(bucket, null);
+
+ assertKeyInfo(keyInfo, OzoneStoragePolicy.getDefaultPolicy());
+ assertScmContainerAndPipeline(keyInfo,
+ OzoneStoragePolicy.getDefaultPolicy().getCreationTier());
+ }
+
+ /**
+ * The storage policy survives the round trip to the client on read.
+ */
+ @Test
+ public void testStoragePolicyVisibleToClient() throws Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ OzoneBucket bucket = createBucket(null, BucketLayout.OBJECT_STORE);
+ String keyName = UUID.randomUUID().toString();
+ writeKey(bucket, keyName, OzoneStoragePolicy.COLD);
+
+ OzoneKeyDetails details = bucket.getKey(keyName);
+ assertEquals(OzoneStoragePolicy.COLD, details.getStoragePolicy());
+ }
+
+ /**
+ * A key written through an FSO bucket carries its storage policy too, since
FSO
+ * creates keys through a separate OM request path.
+ */
+ @Test
+ public void testStoragePolicyOnFsoBucket() throws Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ OzoneBucket bucket =
+ createBucket(null, BucketLayout.FILE_SYSTEM_OPTIMIZED);
+ OmKeyInfo keyInfo = createKeyAndLookup(bucket, OzoneStoragePolicy.HOT);
+
+ assertKeyInfo(keyInfo, OzoneStoragePolicy.HOT);
+ assertScmContainerAndPipeline(keyInfo,
+ OzoneStoragePolicy.HOT.getCreationTier());
+ assertDatanodeContainerAndBlock(keyInfo,
+ OzoneStoragePolicy.HOT.getCreationTier());
+ }
+
+ /**
+ * An empty key skips block allocation but must still record its policy,
+ * otherwise a later rewrite would lose it.
+ */
+ @Test
+ public void testEmptyKeyRecordsStoragePolicy() throws Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ OzoneBucket bucket = createBucket(null, BucketLayout.OBJECT_STORE);
+ String keyName = UUID.randomUUID().toString();
+ try (OzoneOutputStream out = bucket.createKey(keyName, 0, RATIS_THREE,
+ Collections.emptyMap(), Collections.emptyMap(),
+ OzoneStoragePolicy.COLD)) {
+ // Intentionally write nothing.
+ assertNotNull(out);
+ }
+
+ OmKeyInfo keyInfo = lookupKey(bucket, keyName);
+ assertEquals(OzoneStoragePolicy.COLD, keyInfo.getStoragePolicy());
+ assertTrue(keyInfo.getLatestVersionLocations()
+ .getBlocksLatestVersionOnly().isEmpty());
+ }
+
+ /**
+ * Blocks allocated after the pre-allocated ones run out must stay on the
key's tier.
+ *
+ * <p>Creating the key with size 0 pre-allocates nothing, so every block
comes from a separate
+ * allocateBlock call. The key asks for COLD while its bucket is WARM, so if
the policy is not
+ * carried into those calls, OM falls back to the bucket and the later
blocks land on DISK
+ * instead of ARCHIVE.
+ */
+ @Test
+ public void testStoragePolicyHonouredOnSubsequentBlockAllocation()
+ throws Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ OzoneBucket bucket =
+ createBucket(OzoneStoragePolicy.WARM, BucketLayout.OBJECT_STORE);
+ String keyName = UUID.randomUUID().toString();
+
+ // Size 0 means OM pre-allocates no blocks, so the first write already
needs an
+ // allocateBlock round trip. Write enough to span more than one block.
+ byte[] data = new byte[(int) (2 * ozoneManager.getScmBlockSize() + 1024)];
+ Arrays.fill(data, (byte) 'p');
+ try (OzoneOutputStream out = bucket.createKey(keyName, 0, RATIS_THREE,
+ Collections.emptyMap(), Collections.emptyMap(),
+ OzoneStoragePolicy.COLD)) {
+ out.write(data);
+ }
+
+ OmKeyInfo keyInfo = lookupKey(bucket, keyName);
+ assertEquals(OzoneStoragePolicy.COLD, keyInfo.getStoragePolicy());
+
+ List<OmKeyLocationInfo> blocks =
+ keyInfo.getLatestVersionLocations().getBlocksLatestVersionOnly();
+ assertThat(blocks.size())
+ .withFailMessage("Expected more than one block so a later allocation
is exercised, "
+ + "but the key has %d", blocks.size())
+ .isGreaterThan(1);
+
+ // Every block, not just the first, must sit on the tier the key asked for.
+ assertAllBlocksOnTier(keyInfo, OzoneStoragePolicy.COLD.getCreationTier());
+ }
+
+ /**
+ * Overwriting a key with a different storage policy must update the
recorded policy, not just
+ * the tier of the new blocks.
+ *
+ * <p>The overwrite takes the existing-key branch of {@code
prepareFileInfo}, which starts from
+ * a copy of the old key. If that branch does not apply the new policy, the
blocks land on the new
+ * tier while {@code getStoragePolicy()} keeps reporting the old one.
+ */
+ @Test
+ public void testOverwriteWithDifferentStoragePolicyUpdatesKey()
+ throws Exception {
+ startCluster(allTierTopology(RATIS_THREE.getRequiredNodes()));
+
+ OzoneBucket bucket = createBucket(null, BucketLayout.OBJECT_STORE);
+ String keyName = UUID.randomUUID().toString();
+
+ writeKey(bucket, keyName, OzoneStoragePolicy.HOT);
+ OmKeyInfo original = lookupKey(bucket, keyName);
+ assertEquals(OzoneStoragePolicy.HOT, original.getStoragePolicy());
+ assertAllBlocksOnTier(original, OzoneStoragePolicy.HOT.getCreationTier());
+
+ // Overwrite the same key name with a different policy.
+ writeKey(bucket, keyName, OzoneStoragePolicy.COLD);
+ OmKeyInfo overwritten = lookupKey(bucket, keyName);
+
+ // The recorded policy and the tier the data actually sits on must agree.
+ assertEquals(OzoneStoragePolicy.COLD, overwritten.getStoragePolicy());
+ assertAllBlocksOnTier(overwritten,
OzoneStoragePolicy.COLD.getCreationTier());
+ }
+
+ /**
+ * Checks every block of a key landed on a container of the expected tier.
The
+ * {@code assertDatanodeContainerAndBlock} helper only inspects the first
block, which cannot
+ * catch a tier that changes partway through a key.
+ */
+ private void assertAllBlocksOnTier(OmKeyInfo keyInfo, StorageTier
expectedTier)
+ throws IOException {
+ StorageType expectedStorageType = expectedTier.getUniformStorageType();
+ List<OmKeyLocationInfo> blocks =
+ keyInfo.getLatestVersionLocations().getBlocksLatestVersionOnly();
+ for (int i = 0; i < blocks.size(); i++) {
+ OmKeyLocationInfo block = blocks.get(i);
+ ContainerInfo containerInfo = containerManager.getContainer(
+ ContainerID.valueOf(block.getContainerID()));
+ assertEquals(expectedStorageType,
+ containerInfo.getStorageTier().getUniformStorageType(),
+ "block " + i + " of " + blocks.size() + " landed on the wrong tier");
+ }
+ }
+
+ private OzoneBucket createBucket(OzoneStoragePolicy storagePolicy,
+ BucketLayout bucketLayout) throws IOException {
+ String volumeName = UUID.randomUUID().toString();
+ String bucketName = UUID.randomUUID().toString();
+ store.createVolume(volumeName);
+ BucketArgs.Builder builder = BucketArgs.newBuilder()
+ .setBucketLayout(bucketLayout)
+ .setDefaultReplicationConfig(new
DefaultReplicationConfig(RATIS_THREE));
+ if (storagePolicy != null) {
+ builder.setStoragePolicy(storagePolicy);
+ }
+ store.getVolume(volumeName).createBucket(bucketName, builder.build());
+ return store.getVolume(volumeName).getBucket(bucketName);
+ }
+
+ private void writeKey(OzoneBucket bucket, String keyName,
+ OzoneStoragePolicy storagePolicy) throws IOException {
+ byte[] data = UUID.randomUUID().toString().getBytes(UTF_8);
+ try (OzoneOutputStream out = bucket.createKey(keyName, data.length,
+ RATIS_THREE, Collections.emptyMap(), Collections.emptyMap(),
+ storagePolicy)) {
+ out.write(data);
+ }
+ }
+
+ private OmKeyInfo createKeyAndLookup(OzoneBucket bucket,
+ OzoneStoragePolicy storagePolicy) throws IOException {
+ String keyName = UUID.randomUUID().toString();
+ writeKey(bucket, keyName, storagePolicy);
+ return lookupKey(bucket, keyName);
+ }
+
+ private OmKeyInfo lookupKey(OzoneBucket bucket, String keyName)
+ throws IOException {
+ OmKeyArgs keyArgs = new OmKeyArgs.Builder()
+ .setVolumeName(bucket.getVolumeName())
+ .setBucketName(bucket.getName())
+ .setKeyName(keyName)
+ .build();
+ return ozoneManager.lookupKey(keyArgs);
+ }
+
+ private void assertKeyInfo(OmKeyInfo keyInfo,
+ OzoneStoragePolicy expectedPolicy) {
+ assertEquals(1, keyInfo.getKeyLocationVersions().size());
+ assertNotNull(keyInfo.getLatestVersionLocations()
+ .getBlocksLatestVersionOnly());
+ assertEquals(expectedPolicy, keyInfo.getStoragePolicy());
+ assertEquals(RATIS_THREE, keyInfo.getReplicationConfig());
+ }
+
+ private void assertScmContainerAndPipeline(OmKeyInfo keyInfo,
+ StorageTier expectedTier) throws IOException {
+ List<OmKeyLocationInfo> blocks =
+ keyInfo.getLatestVersionLocations().getBlocksLatestVersionOnly();
+ if (blocks.isEmpty()) {
+ return;
+ }
+ ContainerInfo containerInfo = containerManager.getContainer(
+ ContainerID.valueOf(blocks.get(0).getContainerID()));
+ Pipeline pipeline =
+ pipelineManager.getPipeline(containerInfo.getPipelineID());
+
+ assertEquals(RATIS_THREE, pipeline.getReplicationConfig());
+ assertEquals(expectedTier, containerInfo.getStorageTier());
+ assertEquals(expectedTier, pipeline.getSupportedStorageTier());
+ assertEquals(RATIS_THREE.getRequiredNodes(), pipeline.getNodeSet().size());
+ }
+
+ /**
+ * Checks the block actually landed on a volume of the expected tier, and
that
+ * the storage type travelled to the datanode on the block id.
+ */
+ private void assertDatanodeContainerAndBlock(OmKeyInfo keyInfo,
+ StorageTier expectedTier) throws IOException {
+ StorageType expectedStorageType = expectedTier.getUniformStorageType();
+ OmKeyLocationInfo block =
+
keyInfo.getLatestVersionLocations().getBlocksLatestVersionOnly().get(0);
+ ContainerInfo containerInfo = containerManager.getContainer(
+ ContainerID.valueOf(block.getContainerID()));
+ Pipeline pipeline =
+ pipelineManager.getPipeline(containerInfo.getPipelineID());
+
+ int replicasChecked = 0;
+ try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf)) {
+ for (HddsDatanodeService datanode : cluster.getHddsDatanodes()) {
+ Container<?> container = datanode.getDatanodeStateMachine()
+ .getContainer().getContainerSet()
+ .getContainer(block.getContainerID());
+ if (container == null) {
+ continue;
+ }
+ ContainerData containerData = container.getContainerData();
+ assertEquals(expectedStorageType, containerData.getStorageType());
+ assertEquals(expectedStorageType,
+ containerData.getVolume().getStorageType());
+
assertTrue(pipeline.getNodeSet().contains(datanode.getDatanodeDetails()));
+
+ ContainerProtos.BlockData blockData = ContainerProtocolCalls.getBlock(
+ client, block.getBlockID(), null,
client.getPipeline()).getBlockData();
+ assertEquals(expectedStorageType, StorageTypeUtils
+ .getStorageTypeFromID(blockData.getBlockID().getStorageTypeID()));
+ replicasChecked++;
+ }
+ }
+ assertEquals(RATIS_THREE.getRequiredNodes(), replicasChecked);
+ }
+}
diff --git
a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
index d3d31205571..6ea52d9204e 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
@@ -1164,6 +1164,11 @@ message KeyArgs {
// the given ETag for the operation to succeed. This is used for
// S3 conditional writes with the If-Match header.
optional string expectedETag = 24;
+
+ // storagePolicy, when set, overrides the bucket's storage policy for this
+ // key. When unset, OM falls back to the bucket policy, then to the cluster
+ // default, so older clients keep their current behaviour.
+ optional hadoop.hdds.StoragePolicyProto storagePolicy = 25;
}
message KeyLocation {
@@ -1259,6 +1264,10 @@ message KeyInfo {
// This allows a key to be created an committed atomically if the original
has not
// been modified.
optional uint64 expectedDataGeneration = 22;
+
+ // Storage policy this key was created with. Absent for keys written before
+ // storage policy support, which are read back using the bucket policy.
+ optional hadoop.hdds.StoragePolicyProto storagePolicy = 23;
}
// KeyInfoProtoLight is a lightweight subset of KeyInfo message containing
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java
index 5b91f66cf3f..52c526a131f 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java
@@ -31,7 +31,9 @@
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StoragePolicy;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
import org.apache.hadoop.hdds.utils.UniqueId;
@@ -127,12 +129,15 @@ public OMRequest preExecute(OzoneManager ozoneManager)
throws IOException {
// File system client does not know the final file size in advance but use
0 as
// the placeholder for the data size. Therefore, we should at least
allocate a
// single block and we cannot simply skip the allocate block call
+ final StoragePolicy storagePolicy = getStoragePolicy(bucketInfo, keyArgs);
final List< OmKeyLocationInfo > omKeyLocationInfoList = allocateBlock(
- repConfig, new ExcludeList(), requestedSize,
keyArgs.getSortDatanodes(), userInfo, ozoneManager);
+ repConfig, new ExcludeList(), requestedSize,
keyArgs.getSortDatanodes(), userInfo, ozoneManager,
+ storagePolicy, getAllowFallbackStoragePolicy(bucketInfo));
KeyArgs.Builder newKeyArgs = keyArgs.toBuilder()
.setModificationTime(Time.now()).setType(type).setFactor(factor)
- .setDataSize(requestedSize);
+ .setDataSize(requestedSize)
+ .setStoragePolicy(OzoneStoragePolicy.toProto(storagePolicy));
newKeyArgs.addAllKeyLocations(omKeyLocationInfoList.stream()
.map(info -> info.getProtobuf(getOmRequest().getVersion()))
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java
index 6f25c9da497..a829c4262a8 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java
@@ -108,8 +108,11 @@ public OMRequest preExecute(OzoneManager ozoneManager)
throws IOException {
// To allocate atleast one block passing requested size and scmBlockSize
// as same value. When allocating block requested size is same as
// scmBlockSize.
+ final OmBucketInfo bucketInfo = ozoneManager
+ .getBucketInfo(keyArgs.getVolumeName(), keyArgs.getBucketName());
final List<OmKeyLocationInfo> omKeyLocationInfoList =
allocateBlock(repConfig, excludeList,
- ozoneManager.getScmBlockSize(), keyArgs.getSortDatanodes(), userInfo,
ozoneManager);
+ ozoneManager.getScmBlockSize(), keyArgs.getSortDatanodes(), userInfo,
ozoneManager,
+ getStoragePolicy(bucketInfo, keyArgs),
getAllowFallbackStoragePolicy(bucketInfo));
// Set modification time and normalize key if required.
KeyArgs.Builder newKeyArgs =
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java
index 528085a7021..45ce32a8f0f 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java
@@ -31,7 +31,9 @@
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StoragePolicy;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
import org.apache.hadoop.hdds.utils.UniqueId;
@@ -143,6 +145,9 @@ public OMRequest preExecute(OzoneManager ozoneManager)
throws IOException {
// As for a client for the first time this can be executed on any OM,
// till leader is identified.
+ final StoragePolicy storagePolicy = getStoragePolicy(bucketInfo,
keyArgs);
+ final boolean allowFallbackStoragePolicy =
getAllowFallbackStoragePolicy(bucketInfo);
+
final List<OmKeyLocationInfo> omKeyLocationInfoList;
final long effectiveDataSize;
// Skip block allocation if dataSize <= 0. We also consider unspecified
dataSize as
@@ -154,12 +159,14 @@ public OMRequest preExecute(OzoneManager ozoneManager)
throws IOException {
effectiveDataSize = keyArgs.getDataSize();
omKeyLocationInfoList =
captureLatencyNs(perfMetrics.getCreateKeyAllocateBlockLatencyNs(),
() -> allocateBlock(repConfig, new ExcludeList(),
effectiveDataSize,
- keyArgs.getSortDatanodes(), userInfo, ozoneManager));
+ keyArgs.getSortDatanodes(), userInfo, ozoneManager,
+ storagePolicy, allowFallbackStoragePolicy));
}
newKeyArgs = keyArgs.toBuilder().setModificationTime(Time.now())
.setType(type).setFactor(factor)
- .setDataSize(effectiveDataSize);
+ .setDataSize(effectiveDataSize)
+ .setStoragePolicy(OzoneStoragePolicy.toProto(storagePolicy));
newKeyArgs.addAllKeyLocations(omKeyLocationInfoList.stream()
.map(info -> info.getProtobuf(false,
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequestWithFSO.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequestWithFSO.java
index 94caccffc6d..0d3f6ef3189 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequestWithFSO.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequestWithFSO.java
@@ -52,12 +52,17 @@
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Handles CreateKey request layout version1.
*/
public class OMKeyCreateRequestWithFSO extends OMKeyCreateRequest {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(OMKeyCreateRequestWithFSO.class);
+
public OMKeyCreateRequestWithFSO(OMRequest omRequest,
BucketLayout bucketLayout) {
super(omRequest, bucketLayout);
@@ -140,6 +145,13 @@ public OMClientResponse
validateAndUpdateCache(OzoneManager ozoneManager, Execut
OmBucketInfo bucketInfo = omMetadataManager.getBucketTable().get(
omMetadataManager.getBucketKey(volumeName, bucketName));
+ if (!keyArgs.getIsMultipartKey() && !keyArgs.hasStoragePolicy()) {
+ // preExecute always sets a storage policy, so a missing one means
this request
+ // was journaled by an older OM and is being replayed during an
upgrade.
+ LOG.warn("Key {} has no StoragePolicy. This is expected only while
replaying " +
+ "journals written by an OM without storage policy support.",
keyArgs.getKeyName());
+ }
+
// add all missing parents to dir table
missingParentInfos = getAllMissingParentDirInfo(
ozoneManager, keyArgs, bucketInfo, pathInfoFSO, trxnLogIndex);
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java
index 111d80d1222..49f96cc5df4 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java
@@ -58,6 +58,7 @@
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StoragePolicy;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import
org.apache.hadoop.hdds.protocol.proto.HddsProtos.BlockTokenSecretProto.AccessModeProto;
@@ -188,11 +189,14 @@ protected KeyArgs
resolveBucketAndCheckOpenKeyAcls(KeyArgs keyArgs,
}
/** Allocate multiple blocks using one rpc to SCM. */
+ @SuppressWarnings("checkstyle:ParameterNumber")
protected List<OmKeyLocationInfo> allocateBlock(
ReplicationConfig replicationConfig, ExcludeList excludeList,
long requestedSize, boolean shouldSortDatanodes,
- UserInfo userInfo, OzoneManager ozoneManager)
+ UserInfo userInfo, OzoneManager ozoneManager,
+ @Nonnull StoragePolicy storagePolicy, boolean allowFallbackStoragePolicy)
throws IOException {
+ Objects.requireNonNull(storagePolicy, "storagePolicy == null");
final long scmBlockSize = ozoneManager.getScmBlockSize();
final KeyManager keyManager = ozoneManager.getKeyManager();
@@ -232,10 +236,9 @@ protected List<OmKeyLocationInfo> allocateBlock(
String remoteUser = getRemoteUser().getShortUserName();
final List<AllocatedBlock> allocatedBlocks;
try {
- // TODO Use the actually passed `allowFallbackStoragePolicy` instead of
`true`
allocatedBlocks =
ozoneManager.getScmClient().getBlockClient().allocateBlock(
scmBlockSize, numBlocks, replicationConfig,
ozoneManager.getOMServiceId(), excludeList, scmClientMachine,
- OzoneStoragePolicy.getDefaultPolicy(), true);
+ storagePolicy, allowFallbackStoragePolicy);
} catch (IOException ex) {
ozoneManager.getMetrics().incNumBlockAllocateCallFails();
if (ex instanceof SCMException) {
@@ -987,6 +990,33 @@ public static OmBucketInfo
getBucketInfoForUpdate(OMMetadataManager omMetadataMa
return omBucketInfo != null ? omBucketInfo.copyObject() : null;
}
+ /**
+ * Resolves the storage policy for a key: the policy explicitly requested on
the
+ * key wins, otherwise the bucket's policy, otherwise the cluster default.
+ */
+ @Nonnull
+ protected StoragePolicy getStoragePolicy(OmBucketInfo bucketInfo, KeyArgs
keyArgs) {
+ if (keyArgs.hasStoragePolicy()) {
+ return OzoneStoragePolicy.fromProto(keyArgs.getStoragePolicy());
+ }
+ if (bucketInfo != null && bucketInfo.getStoragePolicy() != null) {
+ return bucketInfo.getStoragePolicy();
+ }
+ return OzoneStoragePolicy.getDefaultPolicy();
+ }
+
+ /**
+ * Whether SCM may fall back to the policy's fallback tier when the creation
tier
+ * has no space. Taken from the bucket; defaults to true when the bucket
does not
+ * set it, matching the behaviour before storage policy support.
+ */
+ protected boolean getAllowFallbackStoragePolicy(OmBucketInfo bucketInfo) {
+ if (bucketInfo == null || bucketInfo.getAllowFallbackStoragePolicy() ==
null) {
+ return true;
+ }
+ return bucketInfo.getAllowFallbackStoragePolicy();
+ }
+
/**
* Prepare OmKeyInfo which will be persisted to openKeyTable.
* @return OmKeyInfo
@@ -1060,6 +1090,11 @@ protected OmKeyInfo prepareFileInfo(
if (keyArgs.hasExpectedDataGeneration()) {
builder.setExpectedDataGeneration(keyArgs.getExpectedDataGeneration());
}
+ // The new blocks were allocated with the policy in keyArgs, so record
that policy instead
+ // of the one copied from the overwritten key.
+ if (keyArgs.hasStoragePolicy()) {
+
builder.setStoragePolicy(OzoneStoragePolicy.fromProto(keyArgs.getStoragePolicy()));
+ }
return builder.build();
}
@@ -1110,6 +1145,9 @@ protected OmKeyInfo createFileInfo(
if (keyArgs.hasExpectedDataGeneration()) {
builder.setExpectedDataGeneration(keyArgs.getExpectedDataGeneration());
}
+ if (keyArgs.hasStoragePolicy()) {
+
builder.setStoragePolicy(OzoneStoragePolicy.fromProto(keyArgs.getStoragePolicy()));
+ }
if (omPathInfo instanceof OMFileRequest.OMPathInfoWithFSO) {
// FileTable metadata format
OMFileRequest.OMPathInfoWithFSO omPathInfoFSO
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java
index da85a37e106..fa57c68000d 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java
@@ -43,6 +43,7 @@
import java.util.List;
import java.util.UUID;
import org.apache.hadoop.hdds.client.ContainerBlockID;
+import org.apache.hadoop.hdds.client.OzoneStoragePolicy;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
import org.apache.hadoop.hdds.client.StoragePolicy;
@@ -359,7 +360,8 @@ public void testAllocateBlockSortsSharedPipelineOnce()
throws Exception {
// requestedSize spans two scmBlockSize blocks on the same pipeline.
List<OmKeyLocationInfo> locations =
request.allocateBlock(replicationConfig,
new ExcludeList(), 2 * scmBlockSize, true,
- UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(),
ozoneManager);
+ UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(),
ozoneManager,
+ OzoneStoragePolicy.getDefaultPolicy(), true);
// Sorted once for the shared pipeline...
verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(),
eq("1.2.3.4"), any());
@@ -415,7 +417,8 @@ public void
testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep
getOmAllocateBlockRequest(createAllocateBlockRequest());
List<OmKeyLocationInfo> locations =
request.allocateBlock(replicationConfig,
new ExcludeList(), 2 * scmBlockSize, true,
- UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(),
ozoneManager);
+ UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(),
ozoneManager,
+ OzoneStoragePolicy.getDefaultPolicy(), true);
assertEquals(2, locations.size());
// Each pipeline keeps its own order; the skipped-sort result is not
shared.
@@ -457,7 +460,8 @@ public void
testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception
getOmAllocateBlockRequest(createAllocateBlockRequest());
List<OmKeyLocationInfo> locations =
request.allocateBlock(replicationConfig,
new ExcludeList(), scmBlockSize, true,
- UserInfo.newBuilder().setRemoteAddress("").build(), ozoneManager);
+ UserInfo.newBuilder().setRemoteAddress("").build(), ozoneManager,
+ OzoneStoragePolicy.getDefaultPolicy(), true);
assertEquals("", clientMachine.getValue());
verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(),
any());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
