HDDS-805. Block token: Client api changes for block token. Contributed by Ajay Kumar.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/38861ebd Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/38861ebd Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/38861ebd Branch: refs/heads/HDDS-4 Commit: 38861ebd2574e3c99d8fc8a4f69e0955aadcf179 Parents: f6f75dd Author: Ajay Kumar <[email protected]> Authored: Wed Dec 19 16:34:51 2018 -0800 Committer: Ajay Kumar <[email protected]> Committed: Wed Dec 19 16:34:51 2018 -0800 ---------------------------------------------------------------------- .../hadoop/hdds/scm/XceiverClientGrpc.java | 61 ++-- .../hadoop/hdds/scm/XceiverClientManager.java | 16 +- .../hadoop/hdds/scm/XceiverClientRatis.java | 6 + .../scm/client/ContainerOperationClient.java | 10 +- .../org/apache/hadoop/hdds/HddsConfigKeys.java | 4 + .../org/apache/hadoop/hdds/client/BlockID.java | 8 +- .../hadoop/hdds/client/ContainerBlockID.java | 10 +- .../hadoop/hdds/scm/XceiverClientSpi.java | 6 + .../scm/storage/ContainerProtocolCalls.java | 161 +++++++-- .../hdds/security/token/BlockTokenVerifier.java | 12 +- .../security/token/OzoneBlockTokenSelector.java | 22 +- .../hdds/security/x509/SecurityConfig.java | 22 +- .../main/proto/DatanodeContainerProtocol.proto | 1 + .../common/src/main/resources/ozone-default.xml | 14 +- .../ozone/client/io/ChunkGroupInputStream.java | 7 +- .../ozone/client/io/ChunkGroupOutputStream.java | 199 ++--------- .../ozone/client/io/ChunkOutputStreamEntry.java | 352 +++++++++++++++++++ .../ozone/om/helpers/OmKeyLocationInfo.java | 44 ++- .../security/OzoneBlockTokenSecretManager.java | 11 +- .../OzoneDelegationTokenSecretManager.java | 4 +- .../ozone/security/OzoneSecretManager.java | 7 +- .../src/main/proto/OzoneManagerProtocol.proto | 1 + .../TestOzoneBlockTokenSecretManager.java | 3 +- .../apache/hadoop/ozone/MiniOzoneCluster.java | 14 + .../hadoop/ozone/MiniOzoneClusterImpl.java | 1 + .../TestContainerStateMachineIdempotency.java | 9 +- .../ozone/client/CertificateClientTestImpl.java | 148 ++++++++ .../ozone/client/rpc/TestOzoneRpcClient.java | 34 +- .../client/rpc/TestSecureOzoneRpcClient.java | 238 +++++++++++++ .../container/ozoneimpl/TestOzoneContainer.java | 13 +- .../ozoneimpl/TestSecureOzoneContainer.java | 2 +- .../ozone/scm/TestContainerSmallFile.java | 8 +- .../TestGetCommittedBlockLengthAndPutKey.java | 8 +- .../ozone/scm/TestXceiverClientManager.java | 6 +- .../ozone/web/OzoneHddsDatanodeService.java | 4 +- .../apache/hadoop/ozone/om/KeyManagerImpl.java | 80 ++++- .../apache/hadoop/ozone/om/OzoneManager.java | 133 +++++-- .../hadoop/ozone/om/TestKeyDeletingService.java | 6 +- .../hadoop/ozone/om/TestKeyManagerImpl.java | 119 ++++++- 39 files changed, 1451 insertions(+), 353 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java index 61d1a95..94798e0 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java @@ -32,15 +32,10 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.scm.client.HddsClientUtils; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.security.exception.SCMSecurityException; -import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; -import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSelector; import org.apache.hadoop.hdds.security.x509.SecurityConfig; -import org.apache.hadoop.io.Text; import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.hadoop.ozone.OzoneConsts; -import org.apache.hadoop.security.SecurityUtil; import org.apache.hadoop.security.UserGroupInformation; -import org.apache.hadoop.security.token.Token; import org.apache.hadoop.util.Time; import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; import org.apache.ratis.thirdparty.io.grpc.Status; @@ -53,7 +48,6 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; -import java.net.InetSocketAddress; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -76,6 +70,7 @@ public class XceiverClientGrpc extends XceiverClientSpi { private Map<UUID, ManagedChannel> channels; private final Semaphore semaphore; private boolean closed = false; + private SecurityConfig secConfig; /** * Constructs a client that can communicate with the Container framework on @@ -90,6 +85,7 @@ public class XceiverClientGrpc extends XceiverClientSpi { Preconditions.checkNotNull(config); this.pipeline = pipeline; this.config = config; + this.secConfig = new SecurityConfig(config); this.semaphore = new Semaphore(HddsClientUtils.getMaxOutstandingRequests(config)); this.metrics = XceiverClientManager.getXceiverClientMetrics(); @@ -97,17 +93,30 @@ public class XceiverClientGrpc extends XceiverClientSpi { this.asyncStubs = new HashMap<>(); } + /** + * To be used when grpc token is not enabled. + * */ @Override public void connect() throws Exception { - // leader by default is the 1st datanode in the datanode list of pipleline DatanodeDetails dn = this.pipeline.getFirstNode(); // just make a connection to the 1st datanode at the beginning - connectToDatanode(dn); + connectToDatanode(dn, null); } + /** + * Passed encoded token to GRPC header when security is enabled. + * */ + @Override + public void connect(String encodedToken) throws Exception { + // leader by default is the 1st datanode in the datanode list of pipleline + DatanodeDetails dn = this.pipeline.getFirstNode(); + // just make a connection to the 1st datanode at the beginning + connectToDatanode(dn, encodedToken); + } - private void connectToDatanode(DatanodeDetails dn) throws IOException { + private void connectToDatanode(DatanodeDetails dn, String encodedToken) + throws IOException { // read port from the data node, on failure use default configured // port. int port = dn.getPort(DatanodeDetails.Port.Name.STANDALONE).getValue(); @@ -119,19 +128,6 @@ public class XceiverClientGrpc extends XceiverClientSpi { // Add credential context to the client call String userName = UserGroupInformation.getCurrentUser() .getShortUserName(); - - // Add block token if block token (mutual auth) is required but the client - // does not have a mTLS (private key and ca signed certificate) - String encodedToken = null; - SecurityConfig secConfig = new SecurityConfig(config); - if (secConfig.isGrpcBlockTokenEnabled()) { - InetSocketAddress addr = new InetSocketAddress(dn.getIpAddress(), port); - encodedToken = getEncodedBlockToken(addr); - if (encodedToken == null) { - throw new SCMSecurityException("No Block token available to access " + - "service at : " + addr.toString()); - } - } LOG.debug("Connecting to server Port : " + dn.getIpAddress()); NettyChannelBuilder channelBuilder = NettyChannelBuilder.forAddress(dn .getIpAddress(), port).usePlaintext() @@ -166,20 +162,6 @@ public class XceiverClientGrpc extends XceiverClientSpi { channels.put(dn.getUuid(), channel); } - private String getEncodedBlockToken(InetSocketAddress addr) - throws IOException{ - UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); - OzoneBlockTokenSelector tokenSelector = new OzoneBlockTokenSelector(); - Text service = SecurityUtil.buildTokenService(addr); - Token<OzoneBlockTokenIdentifier> token = tokenSelector.selectToken( - service, ugi.getTokens()); - if (token != null) { - token.setService(service); - return token.encodeToUrlString(); - } - return null; - } - /** * Returns if the xceiver client connects to all servers in the pipeline. * @@ -301,8 +283,9 @@ public class XceiverClientGrpc extends XceiverClientSpi { ManagedChannel channel = channels.get(dnId); // If the channel doesn't exist for this specific datanode or the channel // is closed, just reconnect + String token = request.getEncodedToken(); if (!isConnected(channel)) { - reconnect(dn); + reconnect(dn, token); } final CompletableFuture<ContainerCommandResponseProto> replyFuture = @@ -347,11 +330,11 @@ public class XceiverClientGrpc extends XceiverClientSpi { return new XceiverClientAsyncReply(replyFuture); } - private void reconnect(DatanodeDetails dn) + private void reconnect(DatanodeDetails dn, String encodedToken) throws IOException { ManagedChannel channel; try { - connectToDatanode(dn); + connectToDatanode(dn, encodedToken); channel = channels.get(dn.getUuid()); } catch (Exception e) { LOG.error("Error while connecting: ", e); http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java index b2735bc..e9ca93d 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java @@ -27,6 +27,8 @@ import com.google.common.cache.RemovalNotification; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.ozone.OzoneSecurityUtil; +import org.apache.hadoop.security.UserGroupInformation; import java.io.Closeable; import java.io.IOException; @@ -62,6 +64,7 @@ public class XceiverClientManager implements Closeable { private final boolean useRatis; private static XceiverClientMetrics metrics; + private boolean isSecurityEnabled; /** * Creates a new XceiverClientManager. * @@ -78,6 +81,7 @@ public class XceiverClientManager implements Closeable { ScmConfigKeys.DFS_CONTAINER_RATIS_ENABLED_KEY, ScmConfigKeys.DFS_CONTAINER_RATIS_ENABLED_DEFAULT); this.conf = conf; + this.isSecurityEnabled = OzoneSecurityUtil.isSecurityEnabled(conf); this.clientCache = CacheBuilder.newBuilder() .expireAfterAccess(staleThresholdMs, TimeUnit.MILLISECONDS) .maximumSize(maxSize) @@ -141,14 +145,19 @@ public class XceiverClientManager implements Closeable { throws IOException { HddsProtos.ReplicationType type = pipeline.getType(); try { - return clientCache.get(pipeline.getId().getId().toString() + type, - new Callable<XceiverClientSpi>() { - @Override + String key = pipeline.getId().getId().toString() + type; + // Append user short name to key to prevent a different user + // from using same instance of xceiverClient. + key = isSecurityEnabled ? + key + UserGroupInformation.getCurrentUser().getShortUserName() : key; + return clientCache.get(key, new Callable<XceiverClientSpi>() { + @Override public XceiverClientSpi call() throws Exception { XceiverClientSpi client = null; switch (type) { case RATIS: client = XceiverClientRatis.newXceiverClientRatis(pipeline, conf); + client.connect(); break; case STAND_ALONE: client = new XceiverClientGrpc(pipeline, conf); @@ -157,7 +166,6 @@ public class XceiverClientManager implements Closeable { default: throw new IOException("not implemented" + pipeline.getType()); } - client.connect(); return client; } }); http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java index 28d3e7a..01e76af 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java @@ -120,6 +120,12 @@ public final class XceiverClientRatis extends XceiverClientSpi { } @Override + public void connect(String encodedToken) throws Exception { + throw new UnsupportedOperationException("Block tokens are not " + + "implemented for Ratis clients."); + } + + @Override public void close() { final RaftClient c = client.getAndSet(null); if (c != null) { http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/client/ContainerOperationClient.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/client/ContainerOperationClient.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/client/ContainerOperationClient.java index 85b5d29..1689e07 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/client/ContainerOperationClient.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/client/ContainerOperationClient.java @@ -115,7 +115,7 @@ public class ContainerOperationClient implements ScmClient { public void createContainer(XceiverClientSpi client, long containerId) throws IOException { String traceID = UUID.randomUUID().toString(); - ContainerProtocolCalls.createContainer(client, containerId, traceID); + ContainerProtocolCalls.createContainer(client, containerId, traceID, null); // Let us log this info after we let SCM know that we have completed the // creation state. @@ -260,7 +260,7 @@ public class ContainerOperationClient implements ScmClient { client = xceiverClientManager.acquireClient(pipeline); String traceID = UUID.randomUUID().toString(); ContainerProtocolCalls - .deleteContainer(client, containerId, force, traceID); + .deleteContainer(client, containerId, force, traceID, null); storageContainerLocationClient .deleteContainer(containerId); if (LOG.isDebugEnabled()) { @@ -310,7 +310,8 @@ public class ContainerOperationClient implements ScmClient { client = xceiverClientManager.acquireClient(pipeline); String traceID = UUID.randomUUID().toString(); ReadContainerResponseProto response = - ContainerProtocolCalls.readContainer(client, containerID, traceID); + ContainerProtocolCalls.readContainer(client, containerID, traceID, + null); if (LOG.isDebugEnabled()) { LOG.debug("Read container {}, machines: {} ", containerID, pipeline.getNodes()); @@ -401,7 +402,8 @@ public class ContainerOperationClient implements ScmClient { ObjectStageChangeRequestProto.Op.close, ObjectStageChangeRequestProto.Stage.begin); - ContainerProtocolCalls.closeContainer(client, containerId, traceID); + ContainerProtocolCalls.closeContainer(client, containerId, traceID, + null); // Notify SCM to close the container storageContainerLocationClient.notifyObjectStageChange( ObjectStageChangeRequestProto.Type.container, http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java index ba84f44..895f627 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java @@ -111,6 +111,10 @@ public final class HddsConfigKeys { public static final String HDDS_PUBLIC_KEY_FILE_NAME = "hdds.public.key.file" + ".name"; public static final String HDDS_PUBLIC_KEY_FILE_NAME_DEFAULT = "public.pem"; + + public static final String HDDS_BLOCK_TOKEN_EXPIRY_TIME = + "hdds.block.token.expiry.time"; + public static final String HDDS_BLOCK_TOKEN_EXPIRY_TIME_DEFAULT = "1d"; /** * Maximum duration of certificates issued by SCM including Self-Signed Roots. * The formats accepted are based on the ISO-8601 duration format PnDTnHnMn.nS http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java index a863437..a49f8ae 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java @@ -16,7 +16,6 @@ */ package org.apache.hadoop.hdds.client; -import org.apache.commons.lang3.builder.ToStringBuilder; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; @@ -75,10 +74,9 @@ public class BlockID { @Override public String toString() { - return new ToStringBuilder(this) - .append("containerID", containerBlockID.getContainerID()) - .append("localID", containerBlockID.getLocalID()) - .append("blockCommitSequenceId", blockCommitSequenceId) + return new StringBuffer().append(getContainerBlockID().toString()) + .append(" bcId: ") + .append(blockCommitSequenceId) .toString(); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/ContainerBlockID.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/ContainerBlockID.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/ContainerBlockID.java index 82084f2..1e30cc3 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/ContainerBlockID.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/ContainerBlockID.java @@ -16,7 +16,6 @@ */ package org.apache.hadoop.hdds.client; -import org.apache.commons.lang3.builder.ToStringBuilder; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import java.util.Objects; @@ -43,10 +42,11 @@ public class ContainerBlockID { @Override public String toString() { - return new ToStringBuilder(this). - append("containerID", containerID). - append("localID", localID). - toString(); + return new StringBuffer() + .append("conID: ") + .append(containerID) + .append(" locID: ") + .append(localID).toString(); } public HddsProtos.ContainerBlockID getProtobuf() { http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java index e9896dc..b30d4af 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java @@ -78,6 +78,12 @@ public abstract class XceiverClientSpi implements Closeable { */ public abstract void connect() throws Exception; + /** + * Connects to the leader in the pipeline using encoded token. To be used + * in a secure cluster. + */ + public abstract void connect(String encodedToken) throws Exception; + @Override public abstract void close(); http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java index 8af3973..114b6e6 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java @@ -21,8 +21,13 @@ package org.apache.hadoop.hdds.scm.storage; import org.apache.hadoop.hdds.scm.XceiverClientAsyncReply; import org.apache.hadoop.hdds.scm.container.common.helpers .BlockNotCommittedException; +import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; +import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSelector; +import org.apache.hadoop.io.Text; import org.apache.hadoop.ozone.common.Checksum; import org.apache.hadoop.ozone.common.ChecksumData; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.security.token.Token; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.apache.hadoop.hdds.scm.XceiverClientSpi; import org.apache.hadoop.hdds.scm.container.common.helpers @@ -97,14 +102,19 @@ public final class ContainerProtocolCalls { .setBlockID(datanodeBlockID); String id = xceiverClient.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = ContainerCommandRequestProto + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.GetBlock) .setContainerID(datanodeBlockID.getContainerID()) .setTraceID(traceID) .setDatanodeUuid(id) - .setGetBlock(readBlockRequest) - .build(); + .setGetBlock(readBlockRequest); + String encodedToken = getEncodedBlockToken(getService(datanodeBlockID)); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + + ContainerCommandRequestProto request = builder.build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); validateContainerResponse(response); @@ -129,13 +139,19 @@ public final class ContainerProtocolCalls { ContainerProtos.GetCommittedBlockLengthRequestProto.newBuilder(). setBlockID(blockID.getDatanodeBlockIDProtobuf()); String id = xceiverClient.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto.newBuilder() .setCmdType(Type.GetCommittedBlockLength) .setContainerID(blockID.getContainerID()) .setTraceID(traceID) .setDatanodeUuid(id) - .setGetCommittedBlockLength(getBlockLengthRequestBuilder).build(); + .setGetCommittedBlockLength(getBlockLengthRequestBuilder); + String encodedToken = getEncodedBlockToken(new Text(blockID. + getContainerBlockID().toString())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); validateContainerResponse(response); return response.getGetCommittedBlockLength(); @@ -156,11 +172,17 @@ public final class ContainerProtocolCalls { PutBlockRequestProto.Builder createBlockRequest = PutBlockRequestProto.newBuilder().setBlockData(containerBlockData); String id = xceiverClient.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto.newBuilder().setCmdType(Type.PutBlock) .setContainerID(containerBlockData.getBlockID().getContainerID()) .setTraceID(traceID).setDatanodeUuid(id) - .setPutBlock(createBlockRequest).build(); + .setPutBlock(createBlockRequest); + String encodedToken = + getEncodedBlockToken(getService(containerBlockData.getBlockID())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); validateContainerResponse(response); return response.getPutBlock(); @@ -184,11 +206,17 @@ public final class ContainerProtocolCalls { PutBlockRequestProto.Builder createBlockRequest = PutBlockRequestProto.newBuilder().setBlockData(containerBlockData); String id = xceiverClient.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto.newBuilder().setCmdType(Type.PutBlock) .setContainerID(containerBlockData.getBlockID().getContainerID()) .setTraceID(traceID).setDatanodeUuid(id) - .setPutBlock(createBlockRequest).build(); + .setPutBlock(createBlockRequest); + String encodedToken = + getEncodedBlockToken(getService(containerBlockData.getBlockID())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); xceiverClient.sendCommand(request); return xceiverClient.sendCommandAsync(request); } @@ -204,20 +232,25 @@ public final class ContainerProtocolCalls { * @throws IOException if there is an I/O error while performing the call */ public static ReadChunkResponseProto readChunk(XceiverClientSpi xceiverClient, - ChunkInfo chunk, BlockID blockID, String traceID) throws IOException { + ChunkInfo chunk, BlockID blockID, String traceID) throws IOException { ReadChunkRequestProto.Builder readChunkRequest = ReadChunkRequestProto .newBuilder() .setBlockID(blockID.getDatanodeBlockIDProtobuf()) .setChunkData(chunk); String id = xceiverClient.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = ContainerCommandRequestProto + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.ReadChunk) .setContainerID(blockID.getContainerID()) .setTraceID(traceID) .setDatanodeUuid(id) - .setReadChunk(readChunkRequest) - .build(); + .setReadChunk(readChunkRequest); + String encodedToken = getEncodedBlockToken(new Text(blockID. + getContainerBlockID().toString())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); validateContainerResponse(response); return response.getReadChunk(); @@ -242,14 +275,19 @@ public final class ContainerProtocolCalls { .setChunkData(chunk) .setData(data); String id = xceiverClient.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = ContainerCommandRequestProto + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.WriteChunk) .setContainerID(blockID.getContainerID()) .setTraceID(traceID) .setDatanodeUuid(id) - .setWriteChunk(writeChunkRequest) - .build(); + .setWriteChunk(writeChunkRequest); + String encodedToken = getEncodedBlockToken(new Text(blockID. + getContainerBlockID().toString())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); validateContainerResponse(response); } @@ -273,10 +311,16 @@ public final class ContainerProtocolCalls { .setBlockID(blockID.getDatanodeBlockIDProtobuf()) .setChunkData(chunk).setData(data); String id = xceiverClient.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto.newBuilder().setCmdType(Type.WriteChunk) .setContainerID(blockID.getContainerID()).setTraceID(traceID) - .setDatanodeUuid(id).setWriteChunk(writeChunkRequest).build(); + .setDatanodeUuid(id).setWriteChunk(writeChunkRequest); + String encodedToken = getEncodedBlockToken(new Text(blockID. + getContainerBlockID().toString())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); return xceiverClient.sendCommandAsync(request); } @@ -294,8 +338,8 @@ public final class ContainerProtocolCalls { * @throws IOException */ public static PutSmallFileResponseProto writeSmallFile( - XceiverClientSpi client, BlockID blockID, byte[] data, String traceID) - throws IOException { + XceiverClientSpi client, BlockID blockID, byte[] data, + String traceID) throws IOException { BlockData containerBlockData = BlockData.newBuilder().setBlockID(blockID.getDatanodeBlockIDProtobuf()) @@ -324,14 +368,19 @@ public final class ContainerProtocolCalls { .build(); String id = client.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto.newBuilder() .setCmdType(Type.PutSmallFile) .setContainerID(blockID.getContainerID()) .setTraceID(traceID) .setDatanodeUuid(id) - .setPutSmallFile(putSmallFileRequest) - .build(); + .setPutSmallFile(putSmallFileRequest); + String encodedToken = getEncodedBlockToken(new Text(blockID. + getContainerBlockID().toString())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); ContainerCommandResponseProto response = client.sendCommand(request); validateContainerResponse(response); return response.getPutSmallFile(); @@ -342,10 +391,11 @@ public final class ContainerProtocolCalls { * @param client - client * @param containerID - ID of container * @param traceID - traceID + * @param encodedToken - encodedToken if security is enabled * @throws IOException */ public static void createContainer(XceiverClientSpi client, long containerID, - String traceID) throws IOException { + String traceID, String encodedToken) throws IOException { ContainerProtos.CreateContainerRequestProto.Builder createRequest = ContainerProtos.CreateContainerRequestProto .newBuilder(); @@ -355,6 +405,9 @@ public final class ContainerProtocolCalls { String id = client.getPipeline().getFirstNode().getUuidString(); ContainerCommandRequestProto.Builder request = ContainerCommandRequestProto.newBuilder(); + if (encodedToken != null) { + request.setEncodedToken(encodedToken); + } request.setCmdType(ContainerProtos.Type.CreateContainer); request.setContainerID(containerID); request.setCreateContainer(createRequest.build()); @@ -371,10 +424,11 @@ public final class ContainerProtocolCalls { * @param client * @param force whether or not to forcibly delete the container. * @param traceID + * @param encodedToken - encodedToken if security is enabled * @throws IOException */ public static void deleteContainer(XceiverClientSpi client, long containerID, - boolean force, String traceID) throws IOException { + boolean force, String traceID, String encodedToken) throws IOException { ContainerProtos.DeleteContainerRequestProto.Builder deleteRequest = ContainerProtos.DeleteContainerRequestProto.newBuilder(); deleteRequest.setForceDelete(force); @@ -387,6 +441,9 @@ public final class ContainerProtocolCalls { request.setDeleteContainer(deleteRequest); request.setTraceID(traceID); request.setDatanodeUuid(id); + if(encodedToken != null) { + request.setEncodedToken(encodedToken); + } ContainerCommandResponseProto response = client.sendCommand(request.build()); validateContainerResponse(response); @@ -398,10 +455,12 @@ public final class ContainerProtocolCalls { * @param client * @param containerID * @param traceID + * @param encodedToken - encodedToken if security is enabled * @throws IOException */ public static void closeContainer(XceiverClientSpi client, - long containerID, String traceID) throws IOException { + long containerID, String traceID, String encodedToken) + throws IOException { String id = client.getPipeline().getFirstNode().getUuidString(); ContainerCommandRequestProto.Builder request = @@ -411,6 +470,9 @@ public final class ContainerProtocolCalls { request.setCloseContainer(CloseContainerRequestProto.getDefaultInstance()); request.setTraceID(traceID); request.setDatanodeUuid(id); + if(encodedToken != null) { + request.setEncodedToken(encodedToken); + } ContainerCommandResponseProto response = client.sendCommand(request.build()); validateContainerResponse(response); @@ -419,13 +481,14 @@ public final class ContainerProtocolCalls { /** * readContainer call that gets meta data from an existing container. * - * @param client - client - * @param traceID - trace ID + * @param client - client + * @param traceID - trace ID + * @param encodedToken - encodedToken if security is enabled * @throws IOException */ public static ReadContainerResponseProto readContainer( XceiverClientSpi client, long containerID, - String traceID) throws IOException { + String traceID, String encodedToken) throws IOException { String id = client.getPipeline().getFirstNode().getUuidString(); ContainerCommandRequestProto.Builder request = @@ -435,6 +498,9 @@ public final class ContainerProtocolCalls { request.setReadContainer(ReadContainerRequestProto.getDefaultInstance()); request.setDatanodeUuid(id); request.setTraceID(traceID); + if(encodedToken != null) { + request.setEncodedToken(encodedToken); + } ContainerCommandResponseProto response = client.sendCommand(request.build()); validateContainerResponse(response); @@ -462,14 +528,19 @@ public final class ContainerProtocolCalls { .build(); String id = client.getPipeline().getFirstNode().getUuidString(); - ContainerCommandRequestProto request = ContainerCommandRequestProto + ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.GetSmallFile) .setContainerID(blockID.getContainerID()) .setTraceID(traceID) .setDatanodeUuid(id) - .setGetSmallFile(getSmallFileRequest) - .build(); + .setGetSmallFile(getSmallFileRequest); + String encodedToken = getEncodedBlockToken(new Text(blockID. + getContainerBlockID().toString())); + if (encodedToken != null) { + builder.setEncodedToken(encodedToken); + } + ContainerCommandRequestProto request = builder.build(); ContainerCommandResponseProto response = client.sendCommand(request); validateContainerResponse(response); @@ -495,4 +566,30 @@ public final class ContainerProtocolCalls { throw new StorageContainerException( response.getMessage(), response.getResult()); } + + /** + * Returns a url encoded block token. Service param should match the service + * field of token. + * @param service + * + * */ + private static String getEncodedBlockToken(Text service) + throws IOException { + UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); + Token<OzoneBlockTokenIdentifier> token = + OzoneBlockTokenSelector.selectBlockToken(service, ugi.getTokens()); + if (token != null) { + return token.encodeToUrlString(); + } + return null; + } + + private static Text getService(DatanodeBlockID blockId) { + return new Text(new StringBuffer() + .append("conID: ") + .append(blockId.getContainerID()) + .append(" locID: ") + .append(blockId.getLocalID()) + .toString()); + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/BlockTokenVerifier.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/BlockTokenVerifier.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/BlockTokenVerifier.java index 8c0d19e..9867879 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/BlockTokenVerifier.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/BlockTokenVerifier.java @@ -39,6 +39,7 @@ public class BlockTokenVerifier implements TokenVerifier { private final CertificateClient caClient; private final SecurityConfig conf; + private static boolean testStub = false; public BlockTokenVerifier(SecurityConfig conf, CertificateClient caClient) { this.conf = conf; @@ -53,7 +54,7 @@ public class BlockTokenVerifier implements TokenVerifier { public UserGroupInformation verify(String user, String tokenStr) throws SCMSecurityException { if (conf.isGrpcBlockTokenEnabled()) { - if (Strings.isNullOrEmpty(tokenStr)) { + if (Strings.isNullOrEmpty(tokenStr) || isTestStub()) { throw new BlockTokenException("Fail to find any token (empty or " + "null."); } @@ -110,4 +111,13 @@ public class BlockTokenVerifier implements TokenVerifier { return UserGroupInformation.createRemoteUser(user); } } + + public static boolean isTestStub() { + return testStub; + } + + // For testing purpose only. + public static void setTestStub(boolean isTestStub) { + BlockTokenVerifier.testStub = isTestStub; + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/OzoneBlockTokenSelector.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/OzoneBlockTokenSelector.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/OzoneBlockTokenSelector.java index 1943cce..83797c3 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/OzoneBlockTokenSelector.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/token/OzoneBlockTokenSelector.java @@ -45,7 +45,27 @@ public class OzoneBlockTokenSelector implements return null; } for (Token<? extends TokenIdentifier> token : tokens) { - if (OzoneBlockTokenIdentifier.KIND_NAME.equals(token.getKind())) { + if (OzoneBlockTokenIdentifier.KIND_NAME.equals(token.getKind()) + && token.getService().equals(service)) { + LOG.trace("Getting token for service:{}", service); + return (Token<OzoneBlockTokenIdentifier>) token; + } + } + return null; + } + + /** + * Static method to avoid instantiation. + * */ + @SuppressWarnings("unchecked") + public static Token<OzoneBlockTokenIdentifier> selectBlockToken(Text service, + Collection<Token<? extends TokenIdentifier>> tokens) { + if (service == null) { + return null; + } + for (Token<? extends TokenIdentifier> token : tokens) { + if (OzoneBlockTokenIdentifier.KIND_NAME.equals(token.getKind()) + && token.getService().equals(service)) { LOG.trace("Getting token for service:{}", service); return (Token<OzoneBlockTokenIdentifier>) token; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/SecurityConfig.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/SecurityConfig.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/SecurityConfig.java index c0ae070..cb7d87d 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/SecurityConfig.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/SecurityConfig.java @@ -72,6 +72,9 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_X509_MAX_DURATION_DEFAU import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_X509_SIGNATURE_ALGO; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_X509_SIGNATURE_ALGO_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.OZONE_METADATA_DIRS; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_DATANODE_DIR_KEY; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SECURITY_ENABLED_DEFAULT; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SECURITY_ENABLED_KEY; /** * A class that deals with all Security related configs in HDDS. @@ -101,6 +104,7 @@ public class SecurityConfig { private String trustStoreFileName; private String serverCertChainFileName; private String clientCertChainFileName; + private final boolean isSecurityEnabled; /** * Constructs a SecurityConfig. @@ -120,8 +124,8 @@ public class SecurityConfig { // HDDS metadata dir and if that is not set, we will use Ozone directory. // TODO: We might want to fix this later. this.metadatDir = this.configuration.get(HDDS_METADATA_DIR_NAME, - configuration.get(OZONE_METADATA_DIRS)); - + configuration.get(OZONE_METADATA_DIRS, + configuration.get(HDDS_DATANODE_DIR_KEY))); Preconditions.checkNotNull(this.metadatDir, "Metadata directory can't be" + " null. Please check configs."); this.keyDir = this.configuration.get(HDDS_KEY_DIR_NAME, @@ -164,6 +168,10 @@ public class SecurityConfig { HDDS_GRPC_TLS_TEST_CERT, HDDS_GRPC_TLS_TEST_CERT_DEFAULT); } + this.isSecurityEnabled = this.configuration.getBoolean( + OZONE_SECURITY_ENABLED_KEY, + OZONE_SECURITY_ENABLED_DEFAULT); + // First Startup -- if the provider is null, check for the provider. if (SecurityConfig.provider == null) { synchronized (SecurityConfig.class) { @@ -178,6 +186,16 @@ public class SecurityConfig { } /** + * Returns true if security is enabled for OzoneCluster. This is determined + * by value of OZONE_SECURITY_ENABLED_KEY. + * + * @return true if security is enabled for OzoneCluster. + */ + public boolean isSecurityEnabled() { + return isSecurityEnabled; + } + + /** * Returns the Standard Certificate file name. * * @return String - Name of the Certificate File. http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto b/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto index 661d910..3e4f64d 100644 --- a/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto +++ b/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto @@ -194,6 +194,7 @@ message ContainerCommandRequestProto { optional PutSmallFileRequestProto putSmallFile = 20; optional GetSmallFileRequestProto getSmallFile = 21; optional GetCommittedBlockLengthRequestProto getCommittedBlockLength = 22; + optional string encodedToken = 23; } message ContainerCommandResponseProto { http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-hdds/common/src/main/resources/ozone-default.xml ---------------------------------------------------------------------- diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index b2c91ac..ae2b7d2 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -1116,7 +1116,7 @@ <property> <name>ozone.tags.system</name> - <value>OZONE,MANAGEMENT,SECURITY,PERFORMANCE,DEBUG,CLIENT,SERVER,OM,SCM,CRITICAL,RATIS,CONTAINER,REQUIRED,REST,STORAGE,PIPELINE,STANDALONE,S3GATEWAY</value> + <value>OZONE,MANAGEMENT,SECURITY,PERFORMANCE,DEBUG,CLIENT,SERVER,OM,SCM,CRITICAL,RATIS,CONTAINER,REQUIRED,REST,STORAGE,PIPELINE,STANDALONE,S3GATEWAY,TOKEN</value> </property> @@ -1474,6 +1474,18 @@ </description> </property> <property> + <name>hdds.block.token.expiry.time</name> + <value>1d</value> + <tag>OZONE, HDDS, SECURITY, TOKEN</tag> + <description> + Default value for expiry time of block token. This + setting supports multiple time unit suffixes as described in + dfs.heartbeat.interval. If no suffix is specified, then milliseconds is + assumed. + </description> + </property> + + <property> <name>hdds.metadata.dir</name> <value/> <tag>X509, SECURITY</tag> http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java index 5632756..cf94aa2 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupInputStream.java @@ -32,6 +32,7 @@ import org.apache.hadoop.hdds.scm.XceiverClientSpi; import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB; import org.apache.hadoop.hdds.scm.storage.ChunkInputStream; import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls; +import org.apache.hadoop.security.UserGroupInformation; import org.apache.ratis.util.Preconditions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -136,7 +137,7 @@ public class ChunkGroupInputStream extends InputStream implements Seekable { off += numBytesRead; len -= numBytesRead; if (current.getRemaining() <= 0 && - ((currentStreamIndex + 1) < streamEntries.size())) { + ((currentStreamIndex + 1) < streamEntries.size())) { currentStreamIndex += 1; } } @@ -296,6 +297,10 @@ public class ChunkGroupInputStream extends InputStream implements Seekable { groupInputStream.streamOffset[i] = length; ContainerProtos.DatanodeBlockID datanodeBlockID = blockID .getDatanodeBlockIDProtobuf(); + if (omKeyLocationInfo.getToken() != null) { + UserGroupInformation.getCurrentUser(). + addToken(omKeyLocationInfo.getToken()); + } ContainerProtos.GetBlockResponseProto response = ContainerProtocolCalls .getBlock(xceiverClient, datanodeBlockID, requestId); List<ContainerProtos.ChunkInfo> chunks = http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java index 4d76395..03a70a4 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java @@ -21,7 +21,6 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import org.apache.hadoop.fs.FSExceptionMessages; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result; -import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerNotOpenException; import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline; import org.apache.hadoop.ozone.common.Checksum; @@ -39,7 +38,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers .StorageContainerException; import org.apache.hadoop.hdds.scm.protocolPB .StorageContainerLocationProtocolClientSideTranslatorPB; -import org.apache.hadoop.hdds.scm.storage.ChunkOutputStream; +import org.apache.hadoop.security.UserGroupInformation; import org.apache.ratis.protocol.RaftRetryFailureException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -133,12 +132,13 @@ public class ChunkGroupOutputStream extends OutputStream { List<OmKeyLocationInfo> locationInfoList = new ArrayList<>(); for (ChunkOutputStreamEntry streamEntry : streamEntries) { OmKeyLocationInfo info = - new OmKeyLocationInfo.Builder().setBlockID(streamEntry.blockID) - .setLength(streamEntry.currentPosition).setOffset(0) + new OmKeyLocationInfo.Builder().setBlockID(streamEntry.getBlockID()) + .setLength(streamEntry.getCurrentPosition()).setOffset(0) + .setToken(streamEntry.getToken()) .build(); - LOG.debug("block written " + streamEntry.blockID + ", length " - + streamEntry.currentPosition + " bcsID " + streamEntry.blockID - .getBlockCommitSequenceId()); + LOG.debug("block written " + streamEntry.getBlockID() + ", length " + + streamEntry.getCurrentPosition() + " bcsID " + + streamEntry.getBlockID().getBlockCommitSequenceId()); locationInfoList.add(info); } return locationInfoList; @@ -217,12 +217,26 @@ public class ChunkGroupOutputStream extends OutputStream { throws IOException { ContainerWithPipeline containerWithPipeline = scmClient .getContainerWithPipeline(subKeyInfo.getContainerID()); + UserGroupInformation.getCurrentUser().addToken(subKeyInfo.getToken()); XceiverClientSpi xceiverClient = - xceiverClientManager.acquireClient(containerWithPipeline.getPipeline()); - streamEntries.add(new ChunkOutputStreamEntry(subKeyInfo.getBlockID(), - keyArgs.getKeyName(), xceiverClientManager, xceiverClient, requestID, - chunkSize, subKeyInfo.getLength(), streamBufferFlushSize, - streamBufferMaxSize, watchTimeout, buffer, checksum)); + xceiverClientManager.acquireClient(containerWithPipeline.getPipeline() + ); + ChunkOutputStreamEntry.Builder builder = + new ChunkOutputStreamEntry.Builder() + .setBlockID(subKeyInfo.getBlockID()) + .setKey(keyArgs.getKeyName()) + .setXceiverClientManager(xceiverClientManager) + .setXceiverClient(xceiverClient) + .setRequestId(requestID) + .setChunkSize(chunkSize) + .setLength(subKeyInfo.getLength()) + .setStreamBufferFlushSize(streamBufferFlushSize) + .setStreamBufferMaxSize(streamBufferMaxSize) + .setWatchTimeout(watchTimeout) + .setBuffer(buffer) + .setChecksum(checksum) + .setToken(subKeyInfo.getToken()); + streamEntries.add(builder.build()); } @VisibleForTesting @@ -329,7 +343,7 @@ public class ChunkGroupOutputStream extends OutputStream { ListIterator<ChunkOutputStreamEntry> streamEntryIterator = streamEntries.listIterator(currentStreamIndex); while (streamEntryIterator.hasNext()) { - if (streamEntryIterator.next().blockID.getContainerID() + if (streamEntryIterator.next().getBlockID().getContainerID() == containerID) { streamEntryIterator.remove(); } @@ -348,7 +362,7 @@ public class ChunkGroupOutputStream extends OutputStream { ListIterator<ChunkOutputStreamEntry> streamEntryIterator = streamEntries.listIterator(currentStreamIndex); while (streamEntryIterator.hasNext()) { - if (streamEntryIterator.next().currentPosition == 0) { + if (streamEntryIterator.next().getCurrentPosition() == 0) { streamEntryIterator.remove(); } } @@ -379,7 +393,7 @@ public class ChunkGroupOutputStream extends OutputStream { if (buffer.position() > 0) { //set the correct length for the current stream - streamEntry.currentPosition = lastSuccessfulFlushIndex; + streamEntry.setCurrentPosition(lastSuccessfulFlushIndex); // If the data is still cached in the underlying stream, we need to // allocate new block and write this data in the datanode. currentStreamIndex += 1; @@ -395,7 +409,7 @@ public class ChunkGroupOutputStream extends OutputStream { } // discard subsequent pre allocated blocks from the streamEntries list // from the closed container - discardPreallocatedBlocks(streamEntry.blockID.getContainerID()); + discardPreallocatedBlocks(streamEntry.getBlockID().getContainerID()); } private boolean checkIfContainerIsClosed(IOException ioe) { @@ -433,8 +447,8 @@ public class ChunkGroupOutputStream extends OutputStream { } private long getKeyLength() { - return streamEntries.parallelStream().mapToLong(e -> e.currentPosition) - .sum(); + return streamEntries.parallelStream().mapToLong( + e -> e.getCurrentPosition()).sum(); } /** @@ -616,155 +630,6 @@ public class ChunkGroupOutputStream extends OutputStream { } } - private static class ChunkOutputStreamEntry extends OutputStream { - private OutputStream outputStream; - private BlockID blockID; - private final String key; - private final XceiverClientManager xceiverClientManager; - private final XceiverClientSpi xceiverClient; - private final Checksum checksum; - private final String requestId; - private final int chunkSize; - // total number of bytes that should be written to this stream - private final long length; - // the current position of this stream 0 <= currentPosition < length - private long currentPosition; - - private final long streamBufferFlushSize; - private final long streamBufferMaxSize; - private final long watchTimeout; - private ByteBuffer buffer; - - ChunkOutputStreamEntry(BlockID blockID, String key, - XceiverClientManager xceiverClientManager, - XceiverClientSpi xceiverClient, String requestId, int chunkSize, - long length, long streamBufferFlushSize, long streamBufferMaxSize, - long watchTimeout, ByteBuffer buffer, Checksum checksum) { - this.outputStream = null; - this.blockID = blockID; - this.key = key; - this.xceiverClientManager = xceiverClientManager; - this.xceiverClient = xceiverClient; - this.requestId = requestId; - this.chunkSize = chunkSize; - - this.length = length; - this.currentPosition = 0; - this.streamBufferFlushSize = streamBufferFlushSize; - this.streamBufferMaxSize = streamBufferMaxSize; - this.watchTimeout = watchTimeout; - this.buffer = buffer; - this.checksum = checksum; - } - - /** - * For testing purpose, taking a some random created stream instance. - * @param outputStream a existing writable output stream - * @param length the length of data to write to the stream - */ - ChunkOutputStreamEntry(OutputStream outputStream, long length, - Checksum checksum) { - this.outputStream = outputStream; - this.blockID = null; - this.key = null; - this.xceiverClientManager = null; - this.xceiverClient = null; - this.requestId = null; - this.chunkSize = -1; - - this.length = length; - this.currentPosition = 0; - streamBufferFlushSize = 0; - streamBufferMaxSize = 0; - buffer = null; - watchTimeout = 0; - this.checksum = checksum; - } - - long getLength() { - return length; - } - - long getRemaining() { - return length - currentPosition; - } - - private void checkStream() { - if (this.outputStream == null) { - this.outputStream = - new ChunkOutputStream(blockID, key, xceiverClientManager, - xceiverClient, requestId, chunkSize, streamBufferFlushSize, - streamBufferMaxSize, watchTimeout, buffer, checksum); - } - } - - @Override - public void write(int b) throws IOException { - checkStream(); - outputStream.write(b); - this.currentPosition += 1; - } - - @Override - public void write(byte[] b, int off, int len) throws IOException { - checkStream(); - outputStream.write(b, off, len); - this.currentPosition += len; - } - - @Override - public void flush() throws IOException { - if (this.outputStream != null) { - this.outputStream.flush(); - } - } - - @Override - public void close() throws IOException { - if (this.outputStream != null) { - this.outputStream.close(); - // after closing the chunkOutPutStream, blockId would have been - // reconstructed with updated bcsId - if (this.outputStream instanceof ChunkOutputStream) { - this.blockID = ((ChunkOutputStream) outputStream).getBlockID(); - } - } - } - - int getLastSuccessfulFlushIndex() throws IOException { - if (this.outputStream instanceof ChunkOutputStream) { - ChunkOutputStream out = (ChunkOutputStream) this.outputStream; - blockID = out.getBlockID(); - return out.getLastSuccessfulFlushIndex(); - } else if (outputStream == null) { - // For a pre allocated block for which no write has been initiated, - // the OutputStream will be null here. - // In such cases, the default blockCommitSequenceId will be 0 - return 0; - } - throw new IOException("Invalid Output Stream for Key: " + key); - } - - void cleanup() { - checkStream(); - if (this.outputStream instanceof ChunkOutputStream) { - ChunkOutputStream out = (ChunkOutputStream) this.outputStream; - out.cleanup(); - } - } - - void writeOnRetry(int len) throws IOException { - checkStream(); - if (this.outputStream instanceof ChunkOutputStream) { - ChunkOutputStream out = (ChunkOutputStream) this.outputStream; - out.writeOnRetry(len); - this.currentPosition += len; - } else { - throw new IOException("Invalid Output Stream for Key: " + key); - } - } - } - /** * Verify that the output stream is open. Non blocking; this gives * the last state of the volatile {@link #closed} field. http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkOutputStreamEntry.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkOutputStreamEntry.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkOutputStreamEntry.java new file mode 100644 index 0000000..55cfb0b --- /dev/null +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkOutputStreamEntry.java @@ -0,0 +1,352 @@ +/* + * 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.io; + +import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.scm.XceiverClientManager; +import org.apache.hadoop.hdds.scm.XceiverClientSpi; +import org.apache.hadoop.hdds.scm.storage.ChunkOutputStream; +import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; +import org.apache.hadoop.ozone.common.Checksum; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.security.token.Token; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; + +/** + * Helper class used inside {@link ChunkGroupOutputStream}. + * */ +public class ChunkOutputStreamEntry extends OutputStream { + + private OutputStream outputStream; + private BlockID blockID; + private final String key; + private final XceiverClientManager xceiverClientManager; + private final XceiverClientSpi xceiverClient; + private final Checksum checksum; + private final String requestId; + private final int chunkSize; + // total number of bytes that should be written to this stream + private final long length; + // the current position of this stream 0 <= currentPosition < length + private long currentPosition; + private Token<OzoneBlockTokenIdentifier> token; + + private final long streamBufferFlushSize; + private final long streamBufferMaxSize; + private final long watchTimeout; + private ByteBuffer buffer; + + private ChunkOutputStreamEntry(BlockID blockID, String key, + XceiverClientManager xceiverClientManager, + XceiverClientSpi xceiverClient, String requestId, int chunkSize, + long length, long streamBufferFlushSize, long streamBufferMaxSize, + long watchTimeout, ByteBuffer buffer, Checksum checksum, + Token<OzoneBlockTokenIdentifier> token) { + this.outputStream = null; + this.blockID = blockID; + this.key = key; + this.xceiverClientManager = xceiverClientManager; + this.xceiverClient = xceiverClient; + this.requestId = requestId; + this.chunkSize = chunkSize; + this.token = token; + this.length = length; + this.currentPosition = 0; + this.streamBufferFlushSize = streamBufferFlushSize; + this.streamBufferMaxSize = streamBufferMaxSize; + this.watchTimeout = watchTimeout; + this.buffer = buffer; + this.checksum = checksum; + } + + /** + * For testing purpose, taking a some random created stream instance. + * + * @param outputStream a existing writable output stream + * @param length the length of data to write to the stream + */ + ChunkOutputStreamEntry(OutputStream outputStream, long length, + Checksum checksum) { + this.outputStream = outputStream; + this.blockID = null; + this.key = null; + this.xceiverClientManager = null; + this.xceiverClient = null; + this.requestId = null; + this.chunkSize = -1; + this.token = null; + this.length = length; + this.currentPosition = 0; + streamBufferFlushSize = 0; + streamBufferMaxSize = 0; + buffer = null; + watchTimeout = 0; + this.checksum = checksum; + } + + long getLength() { + return length; + } + + Token<OzoneBlockTokenIdentifier> getToken() { + return token; + } + + long getRemaining() { + return length - currentPosition; + } + + private void checkStream() throws IOException { + if (this.outputStream == null) { + if (getToken() != null) { + UserGroupInformation.getCurrentUser().addToken(getToken()); + } + this.outputStream = + new ChunkOutputStream(blockID, key, xceiverClientManager, + xceiverClient, requestId, chunkSize, streamBufferFlushSize, + streamBufferMaxSize, watchTimeout, buffer, checksum); + } + } + + @Override + public void write(int b) throws IOException { + checkStream(); + outputStream.write(b); + this.currentPosition += 1; + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + checkStream(); + outputStream.write(b, off, len); + this.currentPosition += len; + } + + @Override + public void flush() throws IOException { + if (this.outputStream != null) { + this.outputStream.flush(); + } + } + + @Override + public void close() throws IOException { + if (this.outputStream != null) { + this.outputStream.close(); + // after closing the chunkOutPutStream, blockId would have been + // reconstructed with updated bcsId + if (this.outputStream instanceof ChunkOutputStream) { + this.blockID = ((ChunkOutputStream) outputStream).getBlockID(); + } + } + } + + int getLastSuccessfulFlushIndex() throws IOException { + if (this.outputStream instanceof ChunkOutputStream) { + ChunkOutputStream out = (ChunkOutputStream) this.outputStream; + blockID = out.getBlockID(); + return out.getLastSuccessfulFlushIndex(); + } else { + if (outputStream == null) { + // For a pre allocated block for which no write has been initiated, + // the OutputStream will be null here. + // In such cases, the default blockCommitSequenceId will be 0 + return 0; + } + } + throw new IOException("Invalid Output Stream for Key: " + key); + } + + void cleanup() { + if (this.outputStream == null) { + this.outputStream = + new ChunkOutputStream(blockID, key, xceiverClientManager, + xceiverClient, requestId, chunkSize, streamBufferFlushSize, + streamBufferMaxSize, watchTimeout, buffer, checksum); + } + if (this.outputStream instanceof ChunkOutputStream) { + ChunkOutputStream out = (ChunkOutputStream) this.outputStream; + out.cleanup(); + } + } + + void writeOnRetry(int len) throws IOException { + checkStream(); + if (this.outputStream instanceof ChunkOutputStream) { + ChunkOutputStream out = (ChunkOutputStream) this.outputStream; + out.writeOnRetry(len); + this.currentPosition += len; + } else { + throw new IOException("Invalid Output Stream for Key: " + key); + } + } + + /** + * Builder class for ChunkGroupOutputStreamEntry. + * */ + public static class Builder { + + private BlockID blockID; + private String key; + private XceiverClientManager xceiverClientManager; + private XceiverClientSpi xceiverClient; + private String requestId; + private int chunkSize; + private long length; + private long streamBufferFlushSize; + private long streamBufferMaxSize; + private long watchTimeout; + private ByteBuffer buffer; + private Token<OzoneBlockTokenIdentifier> token; + private Checksum checksum; + + public Builder setChecksum(Checksum cs) { + this.checksum = cs; + return this; + } + + public Builder setBlockID(BlockID bID) { + this.blockID = bID; + return this; + } + + public Builder setKey(String keys) { + this.key = keys; + return this; + } + + public Builder setXceiverClientManager(XceiverClientManager + xClientManager) { + this.xceiverClientManager = xClientManager; + return this; + } + + public Builder setXceiverClient(XceiverClientSpi client) { + this.xceiverClient = client; + return this; + } + + public Builder setRequestId(String request) { + this.requestId = request; + return this; + } + + public Builder setChunkSize(int cSize) { + this.chunkSize = cSize; + return this; + } + + public Builder setLength(long len) { + this.length = len; + return this; + } + + public Builder setStreamBufferFlushSize(long bufferFlushSize) { + this.streamBufferFlushSize = bufferFlushSize; + return this; + } + + public Builder setStreamBufferMaxSize(long bufferMaxSize) { + this.streamBufferMaxSize = bufferMaxSize; + return this; + } + + public Builder setWatchTimeout(long timeout) { + this.watchTimeout = timeout; + return this; + } + + public Builder setBuffer(ByteBuffer buff) { + this.buffer = buff; + return this; + } + + public Builder setToken(Token<OzoneBlockTokenIdentifier> bToken) { + this.token = bToken; + return this; + } + + public ChunkOutputStreamEntry build() { + return new ChunkOutputStreamEntry(blockID, key, + xceiverClientManager, xceiverClient, requestId, chunkSize, + length, streamBufferFlushSize, streamBufferMaxSize, watchTimeout, + buffer, checksum, token); + } + } + + public OutputStream getOutputStream() { + return outputStream; + } + + public BlockID getBlockID() { + return blockID; + } + + public String getKey() { + return key; + } + + public XceiverClientManager getXceiverClientManager() { + return xceiverClientManager; + } + + public XceiverClientSpi getXceiverClient() { + return xceiverClient; + } + + public Checksum getChecksum() { + return checksum; + } + + public String getRequestId() { + return requestId; + } + + public int getChunkSize() { + return chunkSize; + } + + public long getCurrentPosition() { + return currentPosition; + } + + public long getStreamBufferFlushSize() { + return streamBufferFlushSize; + } + + public long getStreamBufferMaxSize() { + return streamBufferMaxSize; + } + + public long getWatchTimeout() { + return watchTimeout; + } + + public ByteBuffer getBuffer() { + return buffer; + } + + public void setCurrentPosition(int curPosition) { + this.currentPosition = curPosition; + } +} + + http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java index cf61f3c..e4eb0f5 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java @@ -17,7 +17,9 @@ package org.apache.hadoop.ozone.om.helpers; import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation; +import org.apache.hadoop.security.token.Token; /** * One key can be too huge to fit in one container. In which case it gets split @@ -28,6 +30,8 @@ public final class OmKeyLocationInfo { // the id of this subkey in all the subkeys. private long length; private final long offset; + // Block token, required for client authentication when security is enabled. + private Token<OzoneBlockTokenIdentifier> token; // the version number indicating when this block was added private long createVersion; @@ -37,6 +41,14 @@ public final class OmKeyLocationInfo { this.offset = offset; } + private OmKeyLocationInfo(BlockID blockID, long length, long offset, + Token<OzoneBlockTokenIdentifier> token) { + this.blockID = blockID; + this.length = length; + this.offset = offset; + this.token = token; + } + public void setCreateVersion(long version) { createVersion = version; } @@ -73,6 +85,13 @@ public final class OmKeyLocationInfo { return blockID.getBlockCommitSequenceId(); } + public Token<OzoneBlockTokenIdentifier> getToken() { + return token; + } + + public void setToken(Token<OzoneBlockTokenIdentifier> token) { + this.token = token; + } /** * Builder of OmKeyLocationInfo. */ @@ -80,6 +99,7 @@ public final class OmKeyLocationInfo { private BlockID blockID; private long length; private long offset; + private Token<OzoneBlockTokenIdentifier> token; public Builder setBlockID(BlockID blockId) { this.blockID = blockId; @@ -96,18 +116,30 @@ public final class OmKeyLocationInfo { return this; } + public Builder setToken(Token<OzoneBlockTokenIdentifier> bToken) { + this.token = bToken; + return this; + } + public OmKeyLocationInfo build() { - return new OmKeyLocationInfo(blockID, length, offset); + if (token == null) { + return new OmKeyLocationInfo(blockID, length, offset); + } else { + return new OmKeyLocationInfo(blockID, length, offset, token); + } } } public KeyLocation getProtobuf() { - return KeyLocation.newBuilder() + KeyLocation.Builder builder = KeyLocation.newBuilder() .setBlockID(blockID.getProtobuf()) .setLength(length) .setOffset(offset) - .setCreateVersion(createVersion) - .build(); + .setCreateVersion(createVersion); + if (this.token != null) { + builder.setToken(this.token.toTokenProto()); + } + return builder.build(); } public static OmKeyLocationInfo getFromProtobuf(KeyLocation keyLocation) { @@ -115,6 +147,9 @@ public final class OmKeyLocationInfo { BlockID.getFromProtobuf(keyLocation.getBlockID()), keyLocation.getLength(), keyLocation.getOffset()); + if(keyLocation.hasToken()) { + info.token = new Token<>(keyLocation.getToken()); + } info.setCreateVersion(keyLocation.getCreateVersion()); return info; } @@ -125,6 +160,7 @@ public final class OmKeyLocationInfo { ", localID=" + blockID.getLocalID() + "}" + ", length=" + length + ", offset=" + offset + + ", token=" + token + ", createVersion=" + createVersion + '}'; } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneBlockTokenSecretManager.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneBlockTokenSecretManager.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneBlockTokenSecretManager.java index 3b833cb..611d251 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneBlockTokenSecretManager.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneBlockTokenSecretManager.java @@ -19,9 +19,9 @@ package org.apache.hadoop.ozone.security; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.BlockTokenSecretProto.AccessModeProto; +import org.apache.hadoop.hdds.security.x509.SecurityConfig; import org.apache.hadoop.io.Text; import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.security.token.Token; @@ -55,7 +55,7 @@ public class OzoneBlockTokenSecretManager extends * @param blockTokenExpirytime token expiry time for expired tokens in * milliseconds */ - public OzoneBlockTokenSecretManager(OzoneConfiguration conf, + public OzoneBlockTokenSecretManager(SecurityConfig conf, long blockTokenExpirytime, String omCertSerialId) { super(conf, blockTokenExpirytime, blockTokenExpirytime, SERVICE, LOG); this.omCertSerialId = omCertSerialId; @@ -74,7 +74,8 @@ public class OzoneBlockTokenSecretManager extends } /** - * Generate an block token for specified user, blockId. + * Generate an block token for specified user, blockId. Service field for + * token is set to blockId. * * @param user * @param blockId @@ -92,8 +93,10 @@ public class OzoneBlockTokenSecretManager extends LOG.trace("Issued delegation token -> expiryTime:{},tokenId:{}", expiryTime, tokenId); } + // Pass blockId as service. return new Token<>(tokenIdentifier.getBytes(), - createPassword(tokenIdentifier), tokenIdentifier.getKind(), SERVICE); + createPassword(tokenIdentifier), tokenIdentifier.getKind(), + new Text(blockId)); } /** http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneDelegationTokenSecretManager.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneDelegationTokenSecretManager.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneDelegationTokenSecretManager.java index 1b9414b..b13c5e7 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneDelegationTokenSecretManager.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneDelegationTokenSecretManager.java @@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.security; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.security.x509.SecurityConfig; import org.apache.hadoop.io.Text; import org.apache.hadoop.ozone.security.OzoneSecretStore.OzoneManagerSecretState; import org.apache.hadoop.ozone.security.OzoneTokenIdentifier.TokenInfo; @@ -75,7 +76,8 @@ public class OzoneDelegationTokenSecretManager<T extends OzoneTokenIdentifier> public OzoneDelegationTokenSecretManager(OzoneConfiguration conf, long tokenMaxLifetime, long tokenRenewInterval, long dtRemoverScanInterval, Text service) throws IOException { - super(conf, tokenMaxLifetime, tokenRenewInterval, service, LOG); + super(new SecurityConfig(conf), tokenMaxLifetime, tokenRenewInterval, + service, LOG); currentTokens = new ConcurrentHashMap(); this.tokenRemoverScanInterval = dtRemoverScanInterval; this.store = new OzoneSecretStore(conf); http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneSecretManager.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneSecretManager.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneSecretManager.java index 5f909bb..12be4c9 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneSecretManager.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/OzoneSecretManager.java @@ -20,7 +20,6 @@ package org.apache.hadoop.ozone.security; import com.google.common.base.Preconditions; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.security.x509.SecurityConfig; import org.apache.hadoop.io.Text; import org.apache.hadoop.security.AccessControlException; @@ -66,16 +65,16 @@ public abstract class OzoneSecretManager<T extends TokenIdentifier> /** * Create a secret manager. * - * @param conf configuration. + * @param secureConf configuration. * @param tokenMaxLifetime the maximum lifetime of the delegation tokens in * milliseconds * @param tokenRenewInterval how often the tokens must be renewed in * milliseconds * @param service name of service */ - public OzoneSecretManager(OzoneConfiguration conf, long tokenMaxLifetime, + public OzoneSecretManager(SecurityConfig secureConf, long tokenMaxLifetime, long tokenRenewInterval, Text service, Logger logger) { - this.securityConfig = new SecurityConfig(conf); + this.securityConfig = secureConf; this.tokenMaxLifetime = tokenMaxLifetime; this.tokenRenewInterval = tokenRenewInterval; currentKeyId = new AtomicInteger(); http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto ---------------------------------------------------------------------- diff --git a/hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto b/hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto index 23a6222..39d709d 100644 --- a/hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto +++ b/hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto @@ -252,6 +252,7 @@ message KeyLocation { required uint64 length = 4; // indicated at which version this block gets created. optional uint64 createVersion = 5; + optional hadoop.common.TokenProto token = 6; } message KeyLocationList { http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/security/TestOzoneBlockTokenSecretManager.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/security/TestOzoneBlockTokenSecretManager.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/security/TestOzoneBlockTokenSecretManager.java index 4692266..fca67e1 100644 --- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/security/TestOzoneBlockTokenSecretManager.java +++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/security/TestOzoneBlockTokenSecretManager.java @@ -22,6 +22,7 @@ import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.BlockTokenSecretProto.AccessModeProto; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; +import org.apache.hadoop.hdds.security.x509.SecurityConfig; import org.apache.hadoop.security.ssl.KeyStoreTestUtil; import org.apache.hadoop.security.token.Token; import org.apache.hadoop.test.GenericTestUtils; @@ -64,7 +65,7 @@ public class TestOzoneBlockTokenSecretManager { x509Certificate = KeyStoreTestUtil .generateCertificate("CN=OzoneMaster", keyPair, 30, "SHA256withRSA"); omCertSerialId = x509Certificate.getSerialNumber().toString(); - secretManager = new OzoneBlockTokenSecretManager(conf, + secretManager = new OzoneBlockTokenSecretManager(new SecurityConfig(conf), expiryTime, omCertSerialId); secretManager.start(keyPair); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java index cbed243..bef39cd 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java @@ -21,6 +21,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.scm.server.StorageContainerManager; +import org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient; import org.apache.hadoop.ozone.client.OzoneClient; import org.apache.hadoop.ozone.om.OzoneManager; import org.apache.hadoop.hdds.scm.protocolPB @@ -240,6 +241,7 @@ public interface MiniOzoneCluster { protected int numOfScmHandlers = 20; protected int numOfDatanodes = 1; protected boolean startDataNodes = true; + protected CertificateClient certClient; protected Builder(OzoneConfiguration conf) { this.conf = conf; @@ -260,6 +262,18 @@ public interface MiniOzoneCluster { return this; } + /** + * Sets the certificate client. + * + * @param client + * + * @return MiniOzoneCluster.Builder + */ + public Builder setCertificateClient(CertificateClient client) { + this.certClient = client; + return this; + } + public Builder setStartDataNodes(boolean startDataNodes) { this.startDataNodes = startDataNodes; return this; http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java index d7f74b9..bea9566 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java @@ -377,6 +377,7 @@ public final class MiniOzoneClusterImpl implements MiniOzoneCluster { scm = createSCM(); scm.start(); om = createOM(); + om.setCertClient(certClient); } catch (AuthenticationException ex) { throw new IOException("Unable to build MiniOzoneCluster. ", ex); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/38861ebd/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java index 78a8511..98ff8dc 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java @@ -89,7 +89,8 @@ public class TestContainerStateMachineIdempotency { XceiverClientSpi client = xceiverClientManager.acquireClient(pipeline); try { //create the container - ContainerProtocolCalls.createContainer(client, containerID, traceID); + ContainerProtocolCalls.createContainer(client, containerID, traceID, + null); // call create Container again BlockID blockID = ContainerTestHelper.getTestBlockID(containerID); byte[] data = @@ -111,8 +112,10 @@ public class TestContainerStateMachineIdempotency { client.sendCommand(putKeyRequest); // close container call - ContainerProtocolCalls.closeContainer(client, containerID, traceID); - ContainerProtocolCalls.closeContainer(client, containerID, traceID); + ContainerProtocolCalls.closeContainer(client, containerID, traceID, + null); + ContainerProtocolCalls.closeContainer(client, containerID, traceID, + null); } catch (IOException ioe) { Assert.fail("Container operation failed" + ioe); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
