You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ae...@apache.org on 2016/04/07 23:41:20 UTC
[2/2] hadoop git commit: HDFS-10268. Ozone: end-to-end integration
for create/get volumes, buckets and keys. Contributed by Chris Nauroth.
HDFS-10268. Ozone: end-to-end integration for create/get volumes, buckets and keys. Contributed by Chris Nauroth.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/fedb22d9
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/fedb22d9
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/fedb22d9
Branch: refs/heads/HDFS-7240
Commit: fedb22d9b642da94f6cd3fb79239924708ec34eb
Parents: b3044db
Author: Anu Engineer <ae...@apache.org>
Authored: Thu Apr 7 14:38:54 2016 -0700
Committer: Anu Engineer <ae...@apache.org>
Committed: Thu Apr 7 14:38:54 2016 -0700
----------------------------------------------------------------------
.../hadoop/hdfs/server/datanode/DataNode.java | 25 +-
.../server/datanode/ObjectStoreHandler.java | 60 ++++-
.../org/apache/hadoop/ozone/OzoneConsts.java | 2 +
.../container/common/helpers/ChunkUtils.java | 3 +-
.../ozone/container/common/impl/Dispatcher.java | 6 +-
.../common/transport/client/XceiverClient.java | 14 +-
.../transport/client/XceiverClientHandler.java | 2 +-
.../transport/client/XceiverClientManager.java | 83 ++++++
.../common/transport/client/package-info.java | 24 ++
.../common/transport/server/package-info.java | 24 ++
.../ozone/storage/StorageContainerManager.java | 110 +++++++-
.../hadoop/ozone/web/client/OzoneBucket.java | 9 +-
.../ozone/web/exceptions/OzoneException.java | 2 +-
.../hadoop/ozone/web/request/OzoneQuota.java | 10 +
.../ozone/web/storage/ChunkInputStream.java | 193 ++++++++++++++
.../ozone/web/storage/ChunkOutputStream.java | 193 ++++++++++++++
.../web/storage/ContainerProtocolCalls.java | 198 ++++++++++++++
.../web/storage/DistributedStorageHandler.java | 266 +++++++++++++++----
.../web/storage/OzoneContainerTranslation.java | 261 ++++++++++++++++++
.../apache/hadoop/ozone/MiniOzoneCluster.java | 71 +++--
.../ozone/web/TestOzoneRestWithMiniCluster.java | 253 ++++++++++++++++++
21 files changed, 1695 insertions(+), 114 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index a5d5015..ff71653 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -1846,16 +1846,6 @@ public class DataNode extends ReconfigurableBase
public void shutdown() {
stopMetricsLogger();
- if(this.ozoneEnabled) {
- if(ozoneServer != null) {
- try {
- ozoneServer.stop();
- } catch (Exception e) {
- LOG.error("Error is ozone shutdown. ex {}", e.toString());
- }
- }
- }
-
if (plugins != null) {
for (ServicePlugin p : plugins) {
try {
@@ -1914,6 +1904,21 @@ public class DataNode extends ReconfigurableBase
}
}
+ // Stop the object store handler
+ if (this.objectStoreHandler != null) {
+ this.objectStoreHandler.close();
+ }
+
+ if(this.ozoneEnabled) {
+ if(ozoneServer != null) {
+ try {
+ ozoneServer.stop();
+ } catch (Exception e) {
+ LOG.error("Error is ozone shutdown. ex {}", e.toString());
+ }
+ }
+ }
+
if (pauseMonitor != null) {
pauseMonitor.stop();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ObjectStoreHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ObjectStoreHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ObjectStoreHandler.java
index 6413ac0..b8c6a13 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ObjectStoreHandler.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ObjectStoreHandler.java
@@ -17,36 +17,58 @@
*/
package org.apache.hadoop.hdfs.server.datanode;
-import static com.sun.jersey.api.core.ResourceConfig.PROPERTY_CONTAINER_REQUEST_FILTERS;
-import static com.sun.jersey.api.core.ResourceConfig.FEATURE_TRACE;
-import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_HANDLER_TYPE_DEFAULT;
-import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_HANDLER_TYPE_KEY;
import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_OBJECTSTORE_TRACE_ENABLED_KEY;
import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_OBJECTSTORE_TRACE_ENABLED_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_HANDLER_TYPE_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_HANDLER_TYPE_KEY;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_RPC_ADDRESS_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_RPC_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_RPC_BIND_HOST_KEY;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_STORAGE_RPC_DEFAULT_PORT;
+import static com.sun.jersey.api.core.ResourceConfig.PROPERTY_CONTAINER_REQUEST_FILTERS;
+import static com.sun.jersey.api.core.ResourceConfig.FEATURE_TRACE;
+import java.io.Closeable;
import java.io.IOException;
+import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import com.sun.jersey.api.container.ContainerFactory;
import com.sun.jersey.api.core.ApplicationAdapter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.Client;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.ozone.OzoneConfiguration;
+import org.apache.hadoop.ozone.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
+import org.apache.hadoop.ozone.protocolPB.StorageContainerLocationProtocolPB;
import org.apache.hadoop.ozone.web.handlers.ServiceFilter;
import org.apache.hadoop.ozone.web.interfaces.StorageHandler;
import org.apache.hadoop.ozone.web.ObjectStoreApplication;
import org.apache.hadoop.ozone.web.netty.ObjectStoreJerseyContainer;
import org.apache.hadoop.ozone.web.storage.DistributedStorageHandler;
import org.apache.hadoop.ozone.web.localstorage.LocalStorageHandler;
+import org.apache.hadoop.security.UserGroupInformation;
/**
* Implements object store handling within the DataNode process. This class is
* responsible for initializing and maintaining the RPC clients and servers and
* the web application required for the object store implementation.
*/
-public final class ObjectStoreHandler {
+public final class ObjectStoreHandler implements Closeable {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ObjectStoreJerseyContainer.class);
private final ObjectStoreJerseyContainer objectStoreJerseyContainer;
+ private final StorageContainerLocationProtocolClientSideTranslatorPB
+ storageContainerLocationClient;
/**
* Creates a new ObjectStoreHandler.
@@ -57,14 +79,32 @@ public final class ObjectStoreHandler {
public ObjectStoreHandler(Configuration conf) throws IOException {
String shType = conf.getTrimmed(DFS_STORAGE_HANDLER_TYPE_KEY,
DFS_STORAGE_HANDLER_TYPE_DEFAULT);
+ LOG.info("ObjectStoreHandler initializing with {}: {}",
+ DFS_STORAGE_HANDLER_TYPE_KEY, shType);
boolean ozoneTrace = conf.getBoolean(DFS_OBJECTSTORE_TRACE_ENABLED_KEY,
DFS_OBJECTSTORE_TRACE_ENABLED_DEFAULT);
final StorageHandler storageHandler;
+
+ // Initialize Jersey container for object store web application.
if ("distributed".equalsIgnoreCase(shType)) {
- storageHandler = new DistributedStorageHandler();
+ RPC.setProtocolEngine(conf, StorageContainerLocationProtocolPB.class,
+ ProtobufRpcEngine.class);
+ long version =
+ RPC.getProtocolVersion(StorageContainerLocationProtocolPB.class);
+ InetSocketAddress address = conf.getSocketAddr(
+ DFS_STORAGE_RPC_BIND_HOST_KEY, DFS_STORAGE_RPC_ADDRESS_KEY,
+ DFS_STORAGE_RPC_ADDRESS_DEFAULT, DFS_STORAGE_RPC_DEFAULT_PORT);
+ this.storageContainerLocationClient =
+ new StorageContainerLocationProtocolClientSideTranslatorPB(
+ RPC.getProxy(StorageContainerLocationProtocolPB.class, version,
+ address, UserGroupInformation.getCurrentUser(), conf,
+ NetUtils.getDefaultSocketFactory(conf), Client.getTimeout(conf)));
+ storageHandler = new DistributedStorageHandler(new OzoneConfiguration(),
+ this.storageContainerLocationClient);
} else {
if ("local".equalsIgnoreCase(shType)) {
storageHandler = new LocalStorageHandler(conf);
+ this.storageContainerLocationClient = null;
} else {
throw new IllegalArgumentException(
String.format("Unrecognized value for %s: %s",
@@ -91,4 +131,12 @@ public final class ObjectStoreHandler {
public ObjectStoreJerseyContainer getObjectStoreJerseyContainer() {
return this.objectStoreJerseyContainer;
}
+
+ @Override
+ public void close() {
+ LOG.info("Closing ObjectStoreHandler.");
+ if (this.storageContainerLocationClient != null) {
+ this.storageContainerLocationClient.close();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
index bebbb78..1ffaa2f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
@@ -68,6 +68,8 @@ public final class OzoneConsts {
public static final String FILE_HASH = "SHA-256";
public final static String CHUNK_OVERWRITE = "OverWriteRequested";
+ public static final int CHUNK_SIZE = 1 * 1024 * 1024; // 1 MB
+
/**
* Supports Bucket Versioning.
*/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java
index 15e4524..b4c8aa6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java
@@ -169,7 +169,8 @@ public final class ChunkUtils {
StandardOpenOption.SPARSE,
StandardOpenOption.SYNC);
lock = file.lock().get();
- if (!chunkInfo.getChecksum().isEmpty()) {
+ if (chunkInfo.getChecksum() != null &&
+ !chunkInfo.getChecksum().isEmpty()) {
verifyChecksum(chunkInfo, data, log);
}
int size = file.write(ByteBuffer.wrap(data), chunkInfo.getOffset()).get();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/impl/Dispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/impl/Dispatcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/impl/Dispatcher.java
index 66ff1ba..bad1d23 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/impl/Dispatcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/impl/Dispatcher.java
@@ -127,7 +127,7 @@ public class Dispatcher implements ContainerDispatcher {
msg.getCreateContainer().getContainerData().getName(),
msg.getCmdType().name(),
msg.getTraceID(),
- ex.toString());
+ ex.toString(), ex);
// TODO : Replace with finer error codes.
return ContainerUtils.getContainerResponse(msg,
@@ -169,7 +169,7 @@ public class Dispatcher implements ContainerDispatcher {
msg.getCreateContainer().getContainerData().getName(),
msg.getCmdType().name(),
msg.getTraceID(),
- ex.toString());
+ ex.toString(), ex);
// TODO : Replace with finer error codes.
return ContainerUtils.getContainerResponse(msg,
@@ -210,7 +210,7 @@ public class Dispatcher implements ContainerDispatcher {
msg.getCreateContainer().getContainerData().getName(),
msg.getCmdType().name(),
msg.getTraceID(),
- ex.toString());
+ ex.toString(), ex);
// TODO : Replace with finer error codes.
return ContainerUtils.getContainerResponse(msg,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClient.java
index 3b9ba8d..e6d914a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClient.java
@@ -34,12 +34,13 @@ import org.apache.hadoop.ozone.container.common.helpers.Pipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.Closeable;
import java.io.IOException;
/**
* A Client for the storageContainer protocol.
*/
-public class XceiverClient {
+public class XceiverClient implements Closeable {
static final Logger LOG = LoggerFactory.getLogger(XceiverClient.class);
private final Pipeline pipeline;
private final Configuration config;
@@ -92,6 +93,7 @@ public class XceiverClient {
/**
* Close the client.
*/
+ @Override
public void close() {
if(group != null) {
group.shutdownGracefully();
@@ -103,6 +105,16 @@ public class XceiverClient {
}
/**
+ * Returns the pipeline of machines that host the container used by this
+ * client.
+ *
+ * @return pipeline of machines that host the container
+ */
+ public Pipeline getPipeline() {
+ return pipeline;
+ }
+
+ /**
* Sends a given command to server and gets the reply back.
* @param request Request
* @return Response to the command
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientHandler.java
index a219e4e..c9a3ad3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientHandler.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientHandler.java
@@ -94,7 +94,7 @@ public class XceiverClientHandler extends
ContainerProtos.ContainerCommandResponseProto response;
channel.writeAndFlush(request);
boolean interrupted = false;
- for (; ; ) {
+ for (;;) {
try {
response = responses.take();
break;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientManager.java
new file mode 100644
index 0000000..8123ae9
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/XceiverClientManager.java
@@ -0,0 +1,83 @@
+/*
+ * 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.container.common.transport.client;
+
+import java.io.IOException;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.hadoop.ozone.OzoneConfiguration;
+import org.apache.hadoop.ozone.container.common.helpers.Pipeline;
+
+/**
+ * XceiverClientManager is responsible for the lifecycle of XceiverClient
+ * instances. Callers use this class to acquire an XceiverClient instance
+ * connected to the desired container pipeline. When done, the caller also uses
+ * this class to release the previously acquired XceiverClient instance.
+ *
+ * This class may evolve to implement efficient lifecycle management policies by
+ * caching container location information and pooling connected client instances
+ * for reuse without needing to reestablish a socket connection. The current
+ * implementation simply allocates and closes a new instance every time.
+ */
+public class XceiverClientManager {
+
+ private final OzoneConfiguration conf;
+
+ /**
+ * Creates a new XceiverClientManager.
+ *
+ * @param conf configuration
+ */
+ public XceiverClientManager(OzoneConfiguration conf) {
+ Preconditions.checkNotNull(conf);
+ this.conf = conf;
+ }
+
+ /**
+ * Acquires a XceiverClient connected to a container capable of storing the
+ * specified key.
+ *
+ * @param pipeline the container pipeline for the client connection
+ * @return XceiverClient connected to a container
+ * @throws IOException if an XceiverClient cannot be acquired
+ */
+ public XceiverClient acquireClient(Pipeline pipeline) throws IOException {
+ Preconditions.checkNotNull(pipeline);
+ Preconditions.checkArgument(pipeline.getMachines() != null);
+ Preconditions.checkArgument(!pipeline.getMachines().isEmpty());
+ XceiverClient xceiverClient = new XceiverClient(pipeline, conf);
+ try {
+ xceiverClient.connect();
+ } catch (Exception e) {
+ throw new IOException("Exception connecting XceiverClient.", e);
+ }
+ return xceiverClient;
+ }
+
+ /**
+ * Releases an XceiverClient after use.
+ *
+ * @param xceiverClient client to release
+ */
+ public void releaseClient(XceiverClient xceiverClient) {
+ Preconditions.checkNotNull(xceiverClient);
+ xceiverClient.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/package-info.java
new file mode 100644
index 0000000..d3c0278
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/client/package-info.java
@@ -0,0 +1,24 @@
+/**
+ * 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.container.common.transport.client;
+
+/**
+ * This package contains classes for the client of the storage container
+ * protocol.
+ */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/package-info.java
new file mode 100644
index 0000000..59c96f1
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/package-info.java
@@ -0,0 +1,24 @@
+/**
+ * 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.container.common.transport.server;
+
+/**
+ * This package contains classes for the server of the storage container
+ * protocol.
+ */
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/storage/StorageContainerManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/storage/StorageContainerManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/storage/StorageContainerManager.java
index 90e200a..16863c9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/storage/StorageContainerManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/storage/StorageContainerManager.java
@@ -48,10 +48,15 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ha.HAServiceProtocol;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.DFSUtilClient;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ContainerCommandRequestProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ContainerCommandResponseProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ContainerData;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.CreateContainerRequestProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Result;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Type;
import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
@@ -81,6 +86,10 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.ProtobufRpcEngine;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.ozone.OzoneConfiguration;
+import org.apache.hadoop.ozone.container.common.helpers.Pipeline;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClient;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClientManager;
import org.apache.hadoop.ozone.protocol.LocatedContainer;
import org.apache.hadoop.ozone.protocol.StorageContainerLocationProtocol;
import org.apache.hadoop.ozone.protocol.proto.StorageContainerLocationProtocolProtos;
@@ -94,11 +103,14 @@ import org.apache.hadoop.util.StringUtils;
*
* The current implementation is a stub suitable to begin end-to-end testing of
* Ozone service interactions. DataNodes report to StorageContainerManager
- * using the existing heartbeat messages. StorageContainerManager tells clients
- * container locations by reporting that all registered nodes are a viable
- * location. This will evolve from a stub to a full-fledged implementation
- * capable of partitioning the keyspace across multiple containers, with
- * appropriate distribution across nodes.
+ * using the existing heartbeat messages. StorageContainerManager lazily
+ * initializes a single storage container to be served by those DataNodes.
+ * All subsequent requests for container locations will reply with that single
+ * pipeline, using all registered nodes.
+ *
+ * This will evolve from a stub to a full-fledged implementation capable of
+ * partitioning the keyspace across multiple containers, with appropriate
+ * distribution across nodes.
*/
@InterfaceAudience.Private
public class StorageContainerManager
@@ -109,6 +121,8 @@ public class StorageContainerManager
private final StorageContainerNameService ns;
private final BlockManager blockManager;
+ private final XceiverClientManager xceiverClientManager;
+ private Pipeline singlePipeline;
/** The RPC server that listens to requests from DataNodes. */
private final RPC.Server serviceRpcServer;
@@ -128,11 +142,12 @@ public class StorageContainerManager
*
* @param conf configuration
*/
- public StorageContainerManager(Configuration conf)
+ public StorageContainerManager(OzoneConfiguration conf)
throws IOException {
ns = new StorageContainerNameService();
boolean haEnabled = false;
blockManager = new BlockManager(ns, haEnabled, conf);
+ xceiverClientManager = new XceiverClientManager(conf);
RPC.setProtocolEngine(conf, DatanodeProtocolPB.class,
ProtobufRpcEngine.class);
@@ -193,20 +208,20 @@ public class StorageContainerManager
public Set<LocatedContainer> getStorageContainerLocations(Set<String> keys)
throws IOException {
LOG.trace("getStorageContainerLocations keys = {}", keys);
+ Pipeline pipeline = initSingleContainerPipeline();
List<DatanodeDescriptor> liveNodes = new ArrayList<DatanodeDescriptor>();
blockManager.getDatanodeManager().fetchDatanodes(liveNodes, null, false);
if (liveNodes.isEmpty()) {
throw new IOException("Storage container locations not found.");
}
- String containerName = UUID.randomUUID().toString();
Set<DatanodeInfo> locations =
Sets.<DatanodeInfo>newLinkedHashSet(liveNodes);
DatanodeInfo leader = liveNodes.get(0);
Set<LocatedContainer> locatedContainers =
Sets.newLinkedHashSetWithExpectedSize(keys.size());
for (String key: keys) {
- locatedContainers.add(new LocatedContainer(key, key, containerName,
- locations, leader));
+ locatedContainers.add(new LocatedContainer(key, key,
+ pipeline.getContainerName(), locations, leader));
}
LOG.trace("getStorageContainerLocations keys = {}, locatedContainers = {}",
keys, locatedContainers);
@@ -416,6 +431,56 @@ public class StorageContainerManager
}
/**
+ * Lazily initializes a single container pipeline using all registered
+ * DataNodes via a synchronous call to the container protocol. This single
+ * container pipeline will be reused for container requests for the lifetime
+ * of this StorageContainerManager.
+ *
+ * @throws IOException if there is an I/O error
+ */
+ private synchronized Pipeline initSingleContainerPipeline()
+ throws IOException {
+ if (singlePipeline == null) {
+ List<DatanodeDescriptor> liveNodes = new ArrayList<DatanodeDescriptor>();
+ blockManager.getDatanodeManager().fetchDatanodes(liveNodes, null, false);
+ if (liveNodes.isEmpty()) {
+ throw new IOException("Storage container locations not found.");
+ }
+ Pipeline newPipeline = newPipelineFromNodes(liveNodes,
+ UUID.randomUUID().toString());
+ XceiverClient xceiverClient =
+ xceiverClientManager.acquireClient(newPipeline);
+ try {
+ ContainerData containerData = ContainerData
+ .newBuilder()
+ .setName(newPipeline.getContainerName())
+ .build();
+ CreateContainerRequestProto createContainerRequest =
+ CreateContainerRequestProto.newBuilder()
+ .setPipeline(newPipeline.getProtobufMessage())
+ .setContainerData(containerData)
+ .build();
+ ContainerCommandRequestProto request = ContainerCommandRequestProto
+ .newBuilder()
+ .setCmdType(Type.CreateContainer)
+ .setCreateContainer(createContainerRequest)
+ .build();
+ ContainerCommandResponseProto response = xceiverClient.sendCommand(
+ request);
+ Result result = response.getResult();
+ if (result != Result.SUCCESS) {
+ throw new IOException(
+ "Failed to initialize container due to result code: " + result);
+ }
+ singlePipeline = newPipeline;
+ } finally {
+ xceiverClientManager.releaseClient(xceiverClient);
+ }
+ }
+ return singlePipeline;
+ }
+
+ /**
* Builds a message for logging startup information about an RPC server.
*
* @param description RPC server description
@@ -430,6 +495,25 @@ public class StorageContainerManager
}
/**
+ * Translates a list of nodes, ordered such that the first is the leader, into
+ * a corresponding {@link Pipeline} object.
+ *
+ * @param nodes list of nodes
+ * @param containerName container name
+ * @return pipeline corresponding to nodes
+ */
+ private static Pipeline newPipelineFromNodes(List<DatanodeDescriptor> nodes,
+ String containerName) {
+ String leaderId = nodes.get(0).getDatanodeUuid();
+ Pipeline pipeline = new Pipeline(leaderId);
+ for (DatanodeDescriptor node : nodes) {
+ pipeline.addMember(node);
+ }
+ pipeline.setContainerName(containerName);
+ return pipeline;
+ }
+
+ /**
* Starts an RPC server, if configured.
*
* @param conf configuration
@@ -443,7 +527,7 @@ public class StorageContainerManager
* @return RPC server, or null if addr is null
* @throws IOException if there is an I/O error while creating RPC server
*/
- private static RPC.Server startRpcServer(Configuration conf,
+ private static RPC.Server startRpcServer(OzoneConfiguration conf,
InetSocketAddress addr, Class<?> protocol, BlockingService instance,
String bindHostKey, String handlerCountKey, int handlerCountDefault)
throws IOException {
@@ -480,7 +564,7 @@ public class StorageContainerManager
* @param rpcServer started RPC server. If null, then the server was not
* started, and this method is a no-op.
*/
- private static InetSocketAddress updateListenAddress(Configuration conf,
+ private static InetSocketAddress updateListenAddress(OzoneConfiguration conf,
String rpcAddressKey, InetSocketAddress addr, RPC.Server rpcServer) {
if (rpcServer == null) {
return null;
@@ -502,7 +586,7 @@ public class StorageContainerManager
StringUtils.startupShutdownMessage(
StorageContainerManager.class, argv, LOG);
StorageContainerManager scm = new StorageContainerManager(
- new Configuration());
+ new OzoneConfiguration());
scm.start();
scm.join();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java
index 4df303a..3441bf9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/client/OzoneBucket.java
@@ -174,9 +174,12 @@ public class OzoneBucket {
InputStream is = new ByteArrayInputStream(data.getBytes(ENCODING));
putRequest.setEntity(new InputStreamEntity(is, data.length()));
- putRequest.setHeader(Header.CONTENT_MD5, DigestUtils.md5Hex(is));
- putRequest
- .setHeader(HttpHeaders.CONTENT_LENGTH, Long.toString(data.length()));
+ is.mark(data.length());
+ try {
+ putRequest.setHeader(Header.CONTENT_MD5, DigestUtils.md5Hex(is));
+ } finally {
+ is.reset();
+ }
executePutKey(putRequest, httpClient);
} catch (IOException | URISyntaxException ex) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/exceptions/OzoneException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/exceptions/OzoneException.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/exceptions/OzoneException.java
index 2bd4a69..c2e64da 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/exceptions/OzoneException.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/exceptions/OzoneException.java
@@ -92,7 +92,7 @@ public class OzoneException extends Exception {
*/
public OzoneException(long httpCode, String shortMessage, String message) {
this.shortMessage = shortMessage;
- this.resource = message;
+ this.message = message;
this.httpCode = httpCode;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/request/OzoneQuota.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/request/OzoneQuota.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/request/OzoneQuota.java
index ca6ddeb..501b239 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/request/OzoneQuota.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/request/OzoneQuota.java
@@ -78,6 +78,16 @@ public class OzoneQuota {
}
/**
+ * Formats a quota as a string.
+ *
+ * @param quota the quota to format
+ * @return string representation of quota
+ */
+ public static String formatQuota(OzoneQuota quota) {
+ return String.valueOf(quota.size) + quota.unit;
+ }
+
+ /**
* Parses a user provided string and returns the
* Quota Object.
*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkInputStream.java
new file mode 100644
index 0000000..166e71c
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkInputStream.java
@@ -0,0 +1,193 @@
+/*
+ * 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.web.storage;
+
+import static org.apache.hadoop.ozone.web.storage.ContainerProtocolCalls.*;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.protobuf.ByteString;
+
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ReadChunkResponseProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClient;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClientManager;
+import org.apache.hadoop.ozone.web.exceptions.OzoneException;
+import org.apache.hadoop.ozone.web.handlers.UserArgs;
+
+/**
+ * An {@link InputStream} used by the REST service in combination with the
+ * {@link DistributedStorageHandler} to read the value of a key from a sequence
+ * of container chunks. All bytes of the key value are stored in container
+ * chunks. Each chunk may contain multiple underlying {@link ByteBuffer}
+ * instances. This class encapsulates all state management for iterating
+ * through the sequence of chunks and the sequence of buffers within each chunk.
+ */
+class ChunkInputStream extends InputStream {
+
+ private static final int EOF = -1;
+
+ private final String key;
+ private final UserArgs args;
+ private XceiverClientManager xceiverClientManager;
+ private XceiverClient xceiverClient;
+ private List<ChunkInfo> chunks;
+ private int chunkOffset;
+ private List<ByteBuffer> buffers;
+ private int bufferOffset;
+
+ /**
+ * Creates a new ChunkInputStream.
+ *
+ * @param key chunk key
+ * @param xceiverClientManager client manager that controls client
+ * @param xceiverClient client to perform container calls
+ * @param chunks list of chunks to read
+ * @param args container protocol call args
+ */
+ public ChunkInputStream(String key, XceiverClientManager xceiverClientManager,
+ XceiverClient xceiverClient, List<ChunkInfo> chunks, UserArgs args) {
+ this.key = key;
+ this.args = args;
+ this.xceiverClientManager = xceiverClientManager;
+ this.xceiverClient = xceiverClient;
+ this.chunks = chunks;
+ this.chunkOffset = 0;
+ this.buffers = null;
+ this.bufferOffset = 0;
+ }
+
+ @Override
+ public synchronized int read()
+ throws IOException {
+ checkOpen();
+ int available = prepareRead(1);
+ return available == EOF ? EOF : buffers.get(bufferOffset).get();
+ }
+
+ @Override
+ public synchronized int read(byte[] b, int off, int len) throws IOException {
+ // According to the JavaDocs for InputStream, it is recommended that
+ // subclasses provide an override of bulk read if possible for performance
+ // reasons. In addition to performance, we need to do it for correctness
+ // reasons. The Ozone REST service uses PipedInputStream and
+ // PipedOutputStream to relay HTTP response data between a Jersey thread and
+ // a Netty thread. It turns out that PipedInputStream/PipedOutputStream
+ // have a subtle dependency (bug?) on the wrapped stream providing separate
+ // implementations of single-byte read and bulk read. Without this, get key
+ // responses might close the connection before writing all of the bytes
+ // advertised in the Content-Length.
+ if (b == null) {
+ throw new NullPointerException();
+ }
+ if (off < 0 || len < 0 || len > b.length - off) {
+ throw new IndexOutOfBoundsException();
+ }
+ if (len == 0) {
+ return 0;
+ }
+ checkOpen();
+ int available = prepareRead(len);
+ if (available == EOF) {
+ return EOF;
+ }
+ buffers.get(bufferOffset).get(b, off, available);
+ return available;
+ }
+
+ @Override
+ public synchronized void close() {
+ if (xceiverClientManager != null && xceiverClient != null) {
+ xceiverClientManager.releaseClient(xceiverClient);
+ xceiverClientManager = null;
+ xceiverClient = null;
+ }
+ }
+
+ /**
+ * Checks if the stream is open. If not, throws an exception.
+ *
+ * @throws IOException if stream is closed
+ */
+ private synchronized void checkOpen() throws IOException {
+ if (xceiverClient == null) {
+ throw new IOException("ChunkInputStream has been closed.");
+ }
+ }
+
+ /**
+ * Prepares to read by advancing through chunks and buffers as needed until it
+ * finds data to return or encounters EOF.
+ *
+ * @param len desired length of data to read
+ * @return length of data available to read, possibly less than desired length
+ */
+ private synchronized int prepareRead(int len) throws IOException {
+ for (;;) {
+ if (chunks == null || chunks.isEmpty()) {
+ // This must be an empty key.
+ return EOF;
+ } else if (buffers == null) {
+ // The first read triggers fetching the first chunk.
+ readChunkFromContainer(0);
+ } else if (!buffers.isEmpty() &&
+ buffers.get(bufferOffset).hasRemaining()) {
+ // Data is available from the current buffer.
+ ByteBuffer bb = buffers.get(bufferOffset);
+ return len > bb.remaining() ? bb.remaining() : len;
+ } else if (!buffers.isEmpty() &&
+ !buffers.get(bufferOffset).hasRemaining() &&
+ bufferOffset < buffers.size() - 1) {
+ // There are additional buffers available.
+ ++bufferOffset;
+ } else if (chunkOffset < chunks.size() - 1) {
+ // There are additional chunks available.
+ readChunkFromContainer(chunkOffset + 1);
+ } else {
+ // All available input has been consumed.
+ return EOF;
+ }
+ }
+ }
+
+ /**
+ * Attempts to read the chunk at the specified offset in the chunk list. If
+ * successful, then the data of the read chunk is saved so that its bytes can
+ * be returned from subsequent read calls.
+ *
+ * @param readChunkOffset offset in the chunk list of which chunk to read
+ * @throws IOException if there is an I/O error while performing the call
+ */
+ private synchronized void readChunkFromContainer(int readChunkOffset)
+ throws IOException {
+ final ReadChunkResponseProto readChunkResponse;
+ try {
+ readChunkResponse = readChunk(xceiverClient, chunks.get(readChunkOffset),
+ key, args);
+ } catch (OzoneException e) {
+ throw new IOException("Unexpected OzoneException", e);
+ }
+ chunkOffset = readChunkOffset;
+ ByteString byteString = readChunkResponse.getData();
+ buffers = byteString.asReadOnlyByteBufferList();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkOutputStream.java
new file mode 100644
index 0000000..d4e639f
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ChunkOutputStream.java
@@ -0,0 +1,193 @@
+/*
+ * 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.web.storage;
+
+import static org.apache.hadoop.ozone.OzoneConsts.CHUNK_SIZE;
+import static org.apache.hadoop.ozone.web.storage.ContainerProtocolCalls.*;
+import static org.apache.hadoop.ozone.web.storage.OzoneContainerTranslation.*;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+
+import com.google.protobuf.ByteString;
+
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyData;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClient;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClientManager;
+import org.apache.hadoop.ozone.web.exceptions.OzoneException;
+import org.apache.hadoop.ozone.web.handlers.UserArgs;
+import org.apache.hadoop.ozone.web.response.KeyInfo;
+
+/**
+ * An {@link OutputStream} used by the REST service in combination with the
+ * {@link DistributedStorageHandler} to write the value of a key to a sequence
+ * of container chunks. Writes are buffered locally and periodically written to
+ * the container as a new chunk. In order to preserve the semantics that
+ * replacement of a pre-existing key is atomic, each instance of the stream has
+ * an internal unique identifier. This unique identifier and a monotonically
+ * increasing chunk index form a composite key that is used as the chunk name.
+ * After all data is written, a putKey call creates or updates the corresponding
+ * container key, and this call includes the full list of chunks that make up
+ * the key data. The list of chunks is updated all at once. Therefore, a
+ * concurrent reader never can see an intermediate state in which different
+ * chunks of data from different versions of the key data are interleaved.
+ * This class encapsulates all state management for buffering and writing
+ * through to the container.
+ */
+class ChunkOutputStream extends OutputStream {
+
+ private final String containerKey;
+ private final KeyInfo key;
+ private final UserArgs args;
+ private final KeyData.Builder containerKeyData;
+ private XceiverClientManager xceiverClientManager;
+ private XceiverClient xceiverClient;
+ private ByteBuffer buffer;
+ private final String streamId;
+ private int chunkIndex;
+
+ /**
+ * Creates a new ChunkOutputStream.
+ *
+ * @param containerKey container key
+ * @param key chunk key
+ * @param xceiverClientManager client manager that controls client
+ * @param xceiverClient client to perform container calls
+ * @param args container protocol call args
+ */
+ public ChunkOutputStream(String containerKey, KeyInfo key,
+ XceiverClientManager xceiverClientManager, XceiverClient xceiverClient,
+ UserArgs args) {
+ this.containerKey = containerKey;
+ this.key = key;
+ this.args = args;
+ this.containerKeyData = fromKeyToContainerKeyDataBuilder(
+ xceiverClient.getPipeline().getContainerName(), containerKey, key);
+ this.xceiverClientManager = xceiverClientManager;
+ this.xceiverClient = xceiverClient;
+ this.buffer = ByteBuffer.allocate(CHUNK_SIZE);
+ this.streamId = UUID.randomUUID().toString();
+ this.chunkIndex = 0;
+ }
+
+ @Override
+ public synchronized void write(int b) throws IOException {
+ checkOpen();
+ int rollbackPosition = buffer.position();
+ int rollbackLimit = buffer.limit();
+ buffer.put((byte)b);
+ if (buffer.position() == CHUNK_SIZE) {
+ flushBufferToChunk(rollbackPosition, rollbackLimit);
+ }
+ }
+
+ @Override
+ public synchronized void flush() throws IOException {
+ checkOpen();
+ if (buffer.position() > 0) {
+ int rollbackPosition = buffer.position();
+ int rollbackLimit = buffer.limit();
+ flushBufferToChunk(rollbackPosition, rollbackLimit);
+ }
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ if (xceiverClientManager != null && xceiverClient != null &&
+ buffer != null) {
+ try {
+ if (buffer.position() > 0) {
+ writeChunkToContainer();
+ }
+ putKey(xceiverClient, containerKeyData.build(), args);
+ } catch (OzoneException e) {
+ throw new IOException("Unexpected OzoneException", e);
+ } finally {
+ xceiverClientManager.releaseClient(xceiverClient);
+ xceiverClientManager = null;
+ xceiverClient = null;
+ buffer = null;
+ }
+ }
+
+ }
+
+ /**
+ * Checks if the stream is open. If not, throws an exception.
+ *
+ * @throws IOException if stream is closed
+ */
+ private synchronized void checkOpen() throws IOException {
+ if (xceiverClient == null) {
+ throw new IOException("ChunkOutputStream has been closed.");
+ }
+ }
+
+ /**
+ * Attempts to flush buffered writes by writing a new chunk to the container.
+ * If successful, then clears the buffer to prepare to receive writes for a
+ * new chunk.
+ *
+ * @param rollbackPosition position to restore in buffer if write fails
+ * @param rollbackLimit limit to restore in buffer if write fails
+ * @throws IOException if there is an I/O error while performing the call
+ */
+ private synchronized void flushBufferToChunk(int rollbackPosition,
+ int rollbackLimit) throws IOException {
+ boolean success = false;
+ try {
+ writeChunkToContainer();
+ success = true;
+ } finally {
+ if (success) {
+ buffer.clear();
+ } else {
+ buffer.position(rollbackPosition);
+ buffer.limit(rollbackLimit);
+ }
+ }
+ }
+
+ /**
+ * Writes buffered data as a new chunk to the container and saves chunk
+ * information to be used later in putKey call.
+ *
+ * @throws IOException if there is an I/O error while performing the call
+ */
+ private synchronized void writeChunkToContainer() throws IOException {
+ buffer.flip();
+ ByteString data = ByteString.copyFrom(buffer);
+ ChunkInfo chunk = ChunkInfo
+ .newBuilder()
+ .setChunkName(
+ key.getKeyName() + "_stream_" + streamId + "_chunk_" + ++chunkIndex)
+ .setOffset(0)
+ .setLen(data.size())
+ .build();
+ try {
+ writeChunk(xceiverClient, chunk, key.getKeyName(), data, args);
+ } catch (OzoneException e) {
+ throw new IOException("Unexpected OzoneException", e);
+ }
+ containerKeyData.addChunks(chunk);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ContainerProtocolCalls.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ContainerProtocolCalls.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ContainerProtocolCalls.java
new file mode 100644
index 0000000..4cb3ab9
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/ContainerProtocolCalls.java
@@ -0,0 +1,198 @@
+/*
+ * 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.web.storage;
+
+import static java.net.HttpURLConnection.HTTP_BAD_REQUEST;
+import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR;
+
+import java.io.IOException;
+
+import com.google.protobuf.ByteString;
+
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ContainerCommandRequestProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ContainerCommandResponseProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.GetKeyRequestProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.GetKeyResponseProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyData;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.PutKeyRequestProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ReadChunkRequestProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ReadChunkResponseProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.Type;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.WriteChunkRequestProto;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClient;
+import org.apache.hadoop.ozone.web.exceptions.ErrorTable;
+import org.apache.hadoop.ozone.web.exceptions.OzoneException;
+import org.apache.hadoop.ozone.web.handlers.UserArgs;
+
+/**
+ * Implementation of all container protocol calls performed by
+ * {@link DistributedStorageHandler}.
+ */
+final class ContainerProtocolCalls {
+
+ /**
+ * Calls the container protocol to get a container key.
+ *
+ * @param xceiverClient client to perform call
+ * @param containerKeyData key data to identify container
+ * @param args container protocol call args
+ * @returns container protocol get key response
+ * @throws IOException if there is an I/O error while performing the call
+ * @throws OzoneException if the container protocol call failed
+ */
+ public static GetKeyResponseProto getKey(XceiverClient xceiverClient,
+ KeyData containerKeyData, UserArgs args) throws IOException,
+ OzoneException {
+ GetKeyRequestProto.Builder readKeyRequest = GetKeyRequestProto
+ .newBuilder()
+ .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
+ .setKeyData(containerKeyData);
+ ContainerCommandRequestProto request = ContainerCommandRequestProto
+ .newBuilder()
+ .setCmdType(Type.GetKey)
+ .setTraceID(args.getRequestID())
+ .setGetKey(readKeyRequest)
+ .build();
+ ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
+ validateContainerResponse(response, args);
+ return response.getGetKey();
+ }
+
+ /**
+ * Calls the container protocol to put a container key.
+ *
+ * @param xceiverClient client to perform call
+ * @param containerKeyData key data to identify container
+ * @param args container protocol call args
+ * @throws IOException if there is an I/O error while performing the call
+ * @throws OzoneException if the container protocol call failed
+ */
+ public static void putKey(XceiverClient xceiverClient,
+ KeyData containerKeyData, UserArgs args) throws IOException,
+ OzoneException {
+ PutKeyRequestProto.Builder createKeyRequest = PutKeyRequestProto
+ .newBuilder()
+ .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
+ .setKeyData(containerKeyData);
+ ContainerCommandRequestProto request = ContainerCommandRequestProto
+ .newBuilder()
+ .setCmdType(Type.PutKey)
+ .setTraceID(args.getRequestID())
+ .setPutKey(createKeyRequest)
+ .build();
+ ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
+ validateContainerResponse(response, args);
+ }
+
+ /**
+ * Calls the container protocol to read a chunk.
+ *
+ * @param xceiverClient client to perform call
+ * @param chunk information about chunk to read
+ * @param key the key name
+ * @param args container protocol call args
+ * @returns container protocol read chunk response
+ * @throws IOException if there is an I/O error while performing the call
+ * @throws OzoneException if the container protocol call failed
+ */
+ public static ReadChunkResponseProto readChunk(XceiverClient xceiverClient,
+ ChunkInfo chunk, String key, UserArgs args)
+ throws IOException, OzoneException {
+ ReadChunkRequestProto.Builder readChunkRequest = ReadChunkRequestProto
+ .newBuilder()
+ .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
+ .setKeyName(key)
+ .setChunkData(chunk);
+ ContainerCommandRequestProto request = ContainerCommandRequestProto
+ .newBuilder()
+ .setCmdType(Type.ReadChunk)
+ .setTraceID(args.getRequestID())
+ .setReadChunk(readChunkRequest)
+ .build();
+ ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
+ validateContainerResponse(response, args);
+ return response.getReadChunk();
+ }
+
+ /**
+ * Calls the container protocol to write a chunk.
+ *
+ * @param xceiverClient client to perform call
+ * @param chunk information about chunk to write
+ * @param key the key name
+ * @param data the data of the chunk to write
+ * @param args container protocol call args
+ * @throws IOException if there is an I/O error while performing the call
+ * @throws OzoneException if the container protocol call failed
+ */
+ public static void writeChunk(XceiverClient xceiverClient, ChunkInfo chunk,
+ String key, ByteString data, UserArgs args)
+ throws IOException, OzoneException {
+ WriteChunkRequestProto.Builder writeChunkRequest = WriteChunkRequestProto
+ .newBuilder()
+ .setPipeline(xceiverClient.getPipeline().getProtobufMessage())
+ .setKeyName(key)
+ .setChunkData(chunk)
+ .setData(data);
+ ContainerCommandRequestProto request = ContainerCommandRequestProto
+ .newBuilder()
+ .setCmdType(Type.WriteChunk)
+ .setTraceID(args.getRequestID())
+ .setWriteChunk(writeChunkRequest)
+ .build();
+ ContainerCommandResponseProto response = xceiverClient.sendCommand(request);
+ validateContainerResponse(response, args);
+ }
+
+ /**
+ * Validates a response from a container protocol call. Any non-successful
+ * return code is mapped to a corresponding exception and thrown.
+ *
+ * @param response container protocol call response
+ * @param args container protocol call args
+ * @throws OzoneException if the container protocol call failed
+ */
+ private static void validateContainerResponse(
+ ContainerCommandResponseProto response, UserArgs args)
+ throws OzoneException {
+ switch (response.getResult()) {
+ case SUCCESS:
+ break;
+ case MALFORMED_REQUEST:
+ throw ErrorTable.newError(new OzoneException(HTTP_BAD_REQUEST,
+ "badRequest", "Bad container request."), args);
+ case UNSUPPORTED_REQUEST:
+ throw ErrorTable.newError(new OzoneException(HTTP_INTERNAL_ERROR,
+ "internalServerError", "Unsupported container request."), args);
+ case CONTAINER_INTERNAL_ERROR:
+ throw ErrorTable.newError(new OzoneException(HTTP_INTERNAL_ERROR,
+ "internalServerError", "Container internal error."), args);
+ default:
+ throw ErrorTable.newError(new OzoneException(HTTP_INTERNAL_ERROR,
+ "internalServerError", "Unrecognized container response."), args);
+ }
+ }
+
+ /**
+ * There is no need to instantiate this class.
+ */
+ private ContainerProtocolCalls() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java
index 9cb8430..8d4868c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java
@@ -18,7 +18,32 @@
package org.apache.hadoop.ozone.web.storage;
+import static org.apache.hadoop.ozone.web.storage.ContainerProtocolCalls.*;
+import static org.apache.hadoop.ozone.web.storage.OzoneContainerTranslation.*;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.TimeZone;
+
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.ChunkInfo;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.GetKeyResponseProto;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyData;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.LengthInputStream;
+import org.apache.hadoop.ozone.OzoneConfiguration;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.container.common.helpers.Pipeline;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClient;
+import org.apache.hadoop.ozone.container.common.transport.client.XceiverClientManager;
+import org.apache.hadoop.ozone.protocol.LocatedContainer;
+import org.apache.hadoop.ozone.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.ozone.web.exceptions.OzoneException;
import org.apache.hadoop.ozone.web.handlers.BucketArgs;
import org.apache.hadoop.ozone.web.handlers.KeyArgs;
@@ -27,13 +52,13 @@ import org.apache.hadoop.ozone.web.handlers.UserArgs;
import org.apache.hadoop.ozone.web.handlers.VolumeArgs;
import org.apache.hadoop.ozone.web.interfaces.StorageHandler;
import org.apache.hadoop.ozone.web.response.BucketInfo;
+import org.apache.hadoop.ozone.web.response.KeyInfo;
import org.apache.hadoop.ozone.web.response.ListBuckets;
import org.apache.hadoop.ozone.web.response.ListKeys;
import org.apache.hadoop.ozone.web.response.ListVolumes;
import org.apache.hadoop.ozone.web.response.VolumeInfo;
-
-import java.io.IOException;
-import java.io.OutputStream;
+import org.apache.hadoop.ozone.web.response.VolumeOwner;
+import org.apache.hadoop.util.StringUtils;
/**
* A {@link StorageHandler} implementation that distributes object storage
@@ -41,156 +66,283 @@ import java.io.OutputStream;
*/
public final class DistributedStorageHandler implements StorageHandler {
- @Override
- public void createVolume(VolumeArgs args) throws
- IOException, OzoneException {
+ private final StorageContainerLocationProtocolClientSideTranslatorPB
+ storageContainerLocation;
+ private final XceiverClientManager xceiverClientManager;
+
+ /**
+ * Creates a new DistributedStorageHandler.
+ *
+ * @param conf configuration
+ * @param storageContainerLocation StorageContainerLocationProtocol proxy
+ */
+ public DistributedStorageHandler(OzoneConfiguration conf,
+ StorageContainerLocationProtocolClientSideTranslatorPB
+ storageContainerLocation) {
+ this.storageContainerLocation = storageContainerLocation;
+ this.xceiverClientManager = new XceiverClientManager(conf);
+ }
+ @Override
+ public void createVolume(VolumeArgs args) throws IOException, OzoneException {
+ String containerKey = buildContainerKey(args.getVolumeName());
+ XceiverClient xceiverClient = acquireXceiverClient(containerKey);
+ try {
+ VolumeInfo volume = new VolumeInfo();
+ volume.setVolumeName(args.getVolumeName());
+ volume.setQuota(args.getQuota());
+ volume.setOwner(new VolumeOwner(args.getUserName()));
+ volume.setCreatedOn(dateToString(new Date()));
+ volume.setCreatedBy(args.getAdminName());
+ KeyData containerKeyData = fromVolumeToContainerKeyData(
+ xceiverClient.getPipeline().getContainerName(), containerKey, volume);
+ putKey(xceiverClient, containerKeyData, args);
+ } finally {
+ xceiverClientManager.releaseClient(xceiverClient);
+ }
}
@Override
public void setVolumeOwner(VolumeArgs args) throws
IOException, OzoneException {
-
+ throw new UnsupportedOperationException("setVolumeOwner not implemented");
}
@Override
public void setVolumeQuota(VolumeArgs args, boolean remove)
throws IOException, OzoneException {
-
+ throw new UnsupportedOperationException("setVolumeQuota not implemented");
}
@Override
public boolean checkVolumeAccess(VolumeArgs args)
throws IOException, OzoneException {
- return false;
+ throw new UnsupportedOperationException("checkVolumeAccessnot implemented");
}
@Override
public ListVolumes listVolumes(UserArgs args)
throws IOException, OzoneException {
- return null;
+ throw new UnsupportedOperationException("listVolumes not implemented");
}
@Override
public void deleteVolume(VolumeArgs args)
throws IOException, OzoneException {
-
+ throw new UnsupportedOperationException("deleteVolume not implemented");
}
@Override
public VolumeInfo getVolumeInfo(VolumeArgs args)
throws IOException, OzoneException {
- return null;
+ String containerKey = buildContainerKey(args.getVolumeName());
+ XceiverClient xceiverClient = acquireXceiverClient(containerKey);
+ try {
+ KeyData containerKeyData = containerKeyDataForRead(
+ xceiverClient.getPipeline().getContainerName(), containerKey);
+ GetKeyResponseProto response = getKey(xceiverClient, containerKeyData,
+ args);
+ return fromContainerKeyValueListToVolume(
+ response.getKeyData().getMetadataList());
+ } finally {
+ xceiverClientManager.releaseClient(xceiverClient);
+ }
}
@Override
- public void createBucket(BucketArgs args)
+ public void createBucket(final BucketArgs args)
throws IOException, OzoneException {
-
+ String containerKey = buildContainerKey(args.getVolumeName(),
+ args.getBucketName());
+ XceiverClient xceiverClient = acquireXceiverClient(containerKey);
+ try {
+ BucketInfo bucket = new BucketInfo();
+ bucket.setVolumeName(args.getVolumeName());
+ bucket.setBucketName(args.getBucketName());
+ bucket.setAcls(args.getAddAcls());
+ bucket.setVersioning(args.getVersioning());
+ bucket.setStorageType(args.getStorageType());
+ KeyData containerKeyData = fromBucketToContainerKeyData(
+ xceiverClient.getPipeline().getContainerName(), containerKey, bucket);
+ putKey(xceiverClient, containerKeyData, args);
+ } finally {
+ xceiverClientManager.releaseClient(xceiverClient);
+ }
}
@Override
public void setBucketAcls(BucketArgs args)
throws IOException, OzoneException {
-
+ throw new UnsupportedOperationException("setBucketAcls not implemented");
}
@Override
public void setBucketVersioning(BucketArgs args)
throws IOException, OzoneException {
-
+ throw new UnsupportedOperationException(
+ "setBucketVersioning not implemented");
}
@Override
public void setBucketStorageClass(BucketArgs args)
throws IOException, OzoneException {
-
+ throw new UnsupportedOperationException(
+ "setBucketStorageClass not implemented");
}
@Override
public void deleteBucket(BucketArgs args)
throws IOException, OzoneException {
-
+ throw new UnsupportedOperationException("deleteBucket not implemented");
}
@Override
public void checkBucketAccess(BucketArgs args)
throws IOException, OzoneException {
-
+ throw new UnsupportedOperationException(
+ "checkBucketAccess not implemented");
}
@Override
public ListBuckets listBuckets(VolumeArgs args)
throws IOException, OzoneException {
- return null;
+ throw new UnsupportedOperationException("listBuckets not implemented");
}
@Override
public BucketInfo getBucketInfo(BucketArgs args)
throws IOException, OzoneException {
- return null;
+ String containerKey = buildContainerKey(args.getVolumeName(),
+ args.getBucketName());
+ XceiverClient xceiverClient = acquireXceiverClient(containerKey);
+ try {
+ KeyData containerKeyData = containerKeyDataForRead(
+ xceiverClient.getPipeline().getContainerName(), containerKey);
+ GetKeyResponseProto response = getKey(xceiverClient, containerKeyData,
+ args);
+ return fromContainerKeyValueListToBucket(
+ response.getKeyData().getMetadataList());
+ } finally {
+ xceiverClientManager.releaseClient(xceiverClient);
+ }
}
- /**
- * Writes a key in an existing bucket.
- *
- * @param args KeyArgs
- * @return InputStream
- * @throws OzoneException
- */
@Override
public OutputStream newKeyWriter(KeyArgs args) throws IOException,
OzoneException {
- return null;
+ String containerKey = buildContainerKey(args.getVolumeName(),
+ args.getBucketName(), args.getKeyName());
+ KeyInfo key = new KeyInfo();
+ key.setKeyName(args.getKeyName());
+ key.setCreatedOn(dateToString(new Date()));
+ XceiverClient xceiverClient = acquireXceiverClient(containerKey);
+ return new ChunkOutputStream(containerKey, key, xceiverClientManager,
+ xceiverClient, args);
}
- /**
- * Tells the file system that the object has been written out completely and
- * it can do any house keeping operation that needs to be done.
- *
- * @param args Key Args
- * @param stream
- * @throws IOException
- */
@Override
public void commitKey(KeyArgs args, OutputStream stream) throws
IOException, OzoneException {
+ stream.close();
+ }
+
+ @Override
+ public LengthInputStream newKeyReader(KeyArgs args) throws IOException,
+ OzoneException {
+ String containerKey = buildContainerKey(args.getVolumeName(),
+ args.getBucketName(), args.getKeyName());
+ XceiverClient xceiverClient = acquireXceiverClient(containerKey);
+ boolean success = false;
+ try {
+ KeyData containerKeyData = containerKeyDataForRead(
+ xceiverClient.getPipeline().getContainerName(), containerKey);
+ GetKeyResponseProto response = getKey(xceiverClient, containerKeyData,
+ args);
+ long length = 0;
+ List<ChunkInfo> chunks = response.getKeyData().getChunksList();
+ for (ChunkInfo chunk : chunks) {
+ length += chunk.getLen();
+ }
+ success = true;
+ return new LengthInputStream(new ChunkInputStream(
+ containerKey, xceiverClientManager, xceiverClient, chunks, args),
+ length);
+ } finally {
+ if (!success) {
+ xceiverClientManager.releaseClient(xceiverClient);
+ }
+ }
+ }
+
+ @Override
+ public void deleteKey(KeyArgs args) throws IOException, OzoneException {
+ throw new UnsupportedOperationException("deleteKey not implemented");
+ }
+ @Override
+ public ListKeys listKeys(ListArgs args) throws IOException, OzoneException {
+ throw new UnsupportedOperationException("listKeys not implemented");
}
/**
- * Reads a key from an existing bucket.
+ * Acquires an {@link XceiverClient} connected to a {@link Pipeline} of nodes
+ * capable of serving container protocol operations. The container is
+ * selected based on the specified container key.
*
- * @param args KeyArgs
- * @return LengthInputStream
- * @throws IOException
+ * @param containerKey container key
+ * @return XceiverClient connected to a container
+ * @throws IOException if an XceiverClient cannot be acquired
*/
- @Override
- public LengthInputStream newKeyReader(KeyArgs args) throws IOException,
- OzoneException {
- return null;
+ private XceiverClient acquireXceiverClient(String containerKey)
+ throws IOException {
+ Set<LocatedContainer> locatedContainers =
+ storageContainerLocation.getStorageContainerLocations(
+ new HashSet<>(Arrays.asList(containerKey)));
+ Pipeline pipeline = newPipelineFromLocatedContainer(
+ locatedContainers.iterator().next());
+ return xceiverClientManager.acquireClient(pipeline);
}
/**
- * Deletes an existing key.
+ * Creates a container key from any number of components by combining all
+ * components with a delimiter.
*
- * @param args KeyArgs
- * @throws OzoneException
+ * @param parts container key components
+ * @return container key
*/
- @Override
- public void deleteKey(KeyArgs args) throws IOException, OzoneException {
+ private static String buildContainerKey(String... parts) {
+ return '/' + StringUtils.join('/', parts);
+ }
+ /**
+ * Formats a date in the expected string format.
+ *
+ * @param date the date to format
+ * @return formatted string representation of date
+ */
+ private static String dateToString(Date date) {
+ SimpleDateFormat sdf =
+ new SimpleDateFormat(OzoneConsts.OZONE_DATE_FORMAT, Locale.US);
+ sdf.setTimeZone(TimeZone.getTimeZone(OzoneConsts.OZONE_TIME_ZONE));
+ return sdf.format(date);
}
/**
- * Returns a list of Key.
+ * Translates a set of container locations, ordered such that the first is the
+ * leader, into a corresponding {@link Pipeline} object.
*
- * @param args KeyArgs
- * @return BucketList
- * @throws IOException
+ * @param locatedContainer container location
+ * @return pipeline corresponding to container locations
*/
- @Override
- public ListKeys listKeys(ListArgs args) throws IOException, OzoneException {
- return null;
+ private static Pipeline newPipelineFromLocatedContainer(
+ LocatedContainer locatedContainer) {
+ Set<DatanodeInfo> locations = locatedContainer.getLocations();
+ String leaderId = locations.iterator().next().getDatanodeUuid();
+ Pipeline pipeline = new Pipeline(leaderId);
+ for (DatanodeInfo location : locations) {
+ pipeline.addMember(location);
+ }
+ pipeline.setContainerName(locatedContainer.getContainerName());
+ return pipeline;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fedb22d9/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/OzoneContainerTranslation.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/OzoneContainerTranslation.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/OzoneContainerTranslation.java
new file mode 100644
index 0000000..9333fe6
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/OzoneContainerTranslation.java
@@ -0,0 +1,261 @@
+/*
+ * 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.web.storage;
+
+import java.util.List;
+
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyData;
+import org.apache.hadoop.hdfs.ozone.protocol.proto.ContainerProtos.KeyValue;
+import org.apache.hadoop.ozone.OzoneConsts.Versioning;
+import org.apache.hadoop.ozone.web.request.OzoneQuota;
+import org.apache.hadoop.ozone.web.response.BucketInfo;
+import org.apache.hadoop.ozone.web.response.KeyInfo;
+import org.apache.hadoop.ozone.web.response.VolumeInfo;
+import org.apache.hadoop.ozone.web.response.VolumeOwner;
+import org.apache.hadoop.util.StringUtils;
+
+/**
+ * This class contains methods that define the translation between the Ozone
+ * domain model and the storage container domain model.
+ */
+final class OzoneContainerTranslation {
+
+ private static final String ACLS = "ACLS";
+ private static final String BUCKET = "BUCKET";
+ private static final String BUCKET_NAME = "BUCKET_NAME";
+ private static final String CREATED_BY = "CREATED_BY";
+ private static final String CREATED_ON = "CREATED_ON";
+ private static final String KEY = "KEY";
+ private static final String OWNER = "OWNER";
+ private static final String QUOTA = "QUOTA";
+ private static final String STORAGE_TYPE = "STORAGE_TYPE";
+ private static final String TYPE = "TYPE";
+ private static final String VERSIONING = "VERSIONING";
+ private static final String VOLUME = "VOLUME";
+ private static final String VOLUME_NAME = "VOLUME_NAME";
+
+ /**
+ * Creates key data intended for reading a container key.
+ *
+ * @param containerName container name
+ * @param containerKey container key
+ * @return KeyData intended for reading the container key
+ */
+ public static KeyData containerKeyDataForRead(String containerName,
+ String containerKey) {
+ return KeyData
+ .newBuilder()
+ .setContainerName(containerName)
+ .setName(containerKey)
+ .build();
+ }
+
+ /**
+ * Translates a bucket to its container representation.
+ *
+ * @param containerName container name
+ * @param containerKey container key
+ * @param bucket the bucket to translate
+ * @return KeyData representation of bucket
+ */
+ public static KeyData fromBucketToContainerKeyData(
+ String containerName, String containerKey, BucketInfo bucket) {
+ KeyData.Builder containerKeyData = KeyData
+ .newBuilder()
+ .setContainerName(containerName)
+ .setName(containerKey)
+ .addMetadata(newKeyValue(TYPE, BUCKET))
+ .addMetadata(newKeyValue(VOLUME_NAME, bucket.getVolumeName()))
+ .addMetadata(newKeyValue(BUCKET_NAME, bucket.getBucketName()));
+
+ if (bucket.getAcls() != null) {
+ containerKeyData.addMetadata(newKeyValue(ACLS,
+ StringUtils.join(',', bucket.getAcls())));
+ }
+
+ if (bucket.getVersioning() != null &&
+ bucket.getVersioning() != Versioning.NOT_DEFINED) {
+ containerKeyData.addMetadata(newKeyValue(VERSIONING,
+ bucket.getVersioning().name()));
+ }
+
+ if (bucket.getStorageType() != StorageType.RAM_DISK) {
+ containerKeyData.addMetadata(newKeyValue(STORAGE_TYPE,
+ bucket.getStorageType().name()));
+ }
+
+ return containerKeyData.build();
+ }
+
+ /**
+ * Translates a bucket from its container representation.
+ *
+ * @param metadata container metadata representing the bucket
+ * @return bucket translated from container representation
+ */
+ public static BucketInfo fromContainerKeyValueListToBucket(
+ List<KeyValue> metadata) {
+ BucketInfo bucket = new BucketInfo();
+ for (KeyValue keyValue : metadata) {
+ switch (keyValue.getKey()) {
+ case VOLUME_NAME:
+ bucket.setVolumeName(keyValue.getValue());
+ break;
+ case BUCKET_NAME:
+ bucket.setBucketName(keyValue.getValue());
+ break;
+ case VERSIONING:
+ bucket.setVersioning(
+ Enum.valueOf(Versioning.class, keyValue.getValue()));
+ break;
+ case STORAGE_TYPE:
+ bucket.setStorageType(
+ Enum.valueOf(StorageType.class, keyValue.getValue()));
+ break;
+ default:
+ break;
+ }
+ }
+ return bucket;
+ }
+
+ /**
+ * Translates a volume from its container representation.
+ *
+ * @param metadata container metadata representing the volume
+ * @return volume translated from container representation
+ */
+ public static VolumeInfo fromContainerKeyValueListToVolume(
+ List<KeyValue> metadata) {
+ VolumeInfo volume = new VolumeInfo();
+ for (KeyValue keyValue : metadata) {
+ switch (keyValue.getKey()) {
+ case VOLUME_NAME:
+ volume.setVolumeName(keyValue.getValue());
+ break;
+ case CREATED_BY:
+ volume.setCreatedBy(keyValue.getValue());
+ break;
+ case CREATED_ON:
+ volume.setCreatedOn(keyValue.getValue());
+ break;
+ case OWNER:
+ volume.setOwner(new VolumeOwner(keyValue.getValue()));
+ break;
+ case QUOTA:
+ volume.setQuota(OzoneQuota.parseQuota(keyValue.getValue()));
+ break;
+ default:
+ break;
+ }
+ }
+ return volume;
+ }
+
+ /**
+ * Translates a key to its container representation.
+ *
+ * @param containerName container name
+ * @param containerKey container key
+ * @param keyInfo key information received from call
+ * @return KeyData intended for reading the container key
+ */
+ public static KeyData fromKeyToContainerKeyData(String containerName,
+ String containerKey, KeyInfo key) {
+ return KeyData
+ .newBuilder()
+ .setContainerName(containerName)
+ .setName(containerKey)
+ .addMetadata(newKeyValue(TYPE, KEY))
+ .build();
+ }
+
+ /**
+ * Translates a key to its container representation. The return value is a
+ * builder that can be manipulated further before building the result.
+ *
+ * @param containerName container name
+ * @param containerKey container key
+ * @param keyInfo key information received from call
+ * @return KeyData builder
+ */
+ public static KeyData.Builder fromKeyToContainerKeyDataBuilder(
+ String containerName, String containerKey, KeyInfo key) {
+ return KeyData
+ .newBuilder()
+ .setContainerName(containerName)
+ .setName(containerKey)
+ .addMetadata(newKeyValue(TYPE, KEY));
+ }
+
+ /**
+ * Translates a volume to its container representation.
+ *
+ * @param containerName container name
+ * @param containerKey container key
+ * @param volume the volume to translate
+ * @return KeyData representation of volume
+ */
+ public static KeyData fromVolumeToContainerKeyData(
+ String containerName, String containerKey, VolumeInfo volume) {
+ KeyData.Builder containerKeyData = KeyData
+ .newBuilder()
+ .setContainerName(containerName)
+ .setName(containerKey)
+ .addMetadata(newKeyValue(TYPE, VOLUME))
+ .addMetadata(newKeyValue(VOLUME_NAME, volume.getVolumeName()))
+ .addMetadata(newKeyValue(CREATED_ON, volume.getCreatedOn()));
+
+ if (volume.getQuota() != null && volume.getQuota().sizeInBytes() != -1L) {
+ containerKeyData.addMetadata(newKeyValue(QUOTA,
+ OzoneQuota.formatQuota(volume.getQuota())));
+ }
+
+ if (volume.getOwner() != null && volume.getOwner().getName() != null &&
+ !volume.getOwner().getName().isEmpty()) {
+ containerKeyData.addMetadata(newKeyValue(OWNER,
+ volume.getOwner().getName()));
+ }
+
+ if (volume.getCreatedBy() != null && !volume.getCreatedBy().isEmpty()) {
+ containerKeyData.addMetadata(
+ newKeyValue(CREATED_BY, volume.getCreatedBy()));
+ }
+
+ return containerKeyData.build();
+ }
+
+ /**
+ * Translates a key-value pair to its container representation.
+ *
+ * @param key the key
+ * @param value the value
+ * @return container representation of key-value pair
+ */
+ private static KeyValue newKeyValue(String key, Object value) {
+ return KeyValue.newBuilder().setKey(key).setValue(value.toString()).build();
+ }
+
+ /**
+ * There is no need to instantiate this class.
+ */
+ private OzoneContainerTranslation() {
+ }
+}