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 xy...@apache.org on 2017/08/08 20:36:39 UTC
[4/8] hadoop git commit: HDFS-12259. Ozone: OzoneClient: Refactor
move ozone client from hadoop-hdfs to hadoop-hdfs-client. Contributed by
Nandakumar.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/TestArchive.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/TestArchive.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/TestArchive.java
new file mode 100644
index 0000000..6fff47e
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/TestArchive.java
@@ -0,0 +1,106 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.scm;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.hadoop.fs.FileUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Iterator;
+import java.util.Random;
+import java.util.zip.Adler32;
+import java.util.zip.Checksum;
+
+/**
+ * Test archive creation and unpacking.
+ */
+public class TestArchive {
+ private static final int DIR_COUNT = 10;
+ private static final int SUB_DIR_COUNT = 3;
+ private static final int FILE_COUNT = 10;
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder();
+
+ @Rule
+ public TemporaryFolder outputFolder = new TemporaryFolder();
+
+ Checksum crc = new Adler32();
+
+ @Before
+ public void setUp() throws Exception {
+ Random r = new Random();
+ final int megaByte = 1024 * 1024;
+
+ for (int x = 0; x < DIR_COUNT; x++) {
+ File subdir = folder.newFolder(String.format("dir%d", x));
+ for (int y = 0; y < SUB_DIR_COUNT; y++) {
+ File targetDir = new File(subdir.getPath().concat(File.separator)
+ .concat(String.format("subdir%d%d", x, y)));
+ if(!targetDir.mkdirs()) {
+ throw new IOException("Failed to create subdirectory. " +
+ targetDir.toString());
+ }
+ for (int z = 0; z < FILE_COUNT; z++) {
+ Path temp = Paths.get(targetDir.getPath().concat(File.separator)
+ .concat(String.format("File%d.txt", z)));
+ byte[] buf = RandomStringUtils.randomAlphanumeric(r.nextInt(megaByte))
+ .getBytes("UTF-8");
+ Files.write(temp, buf);
+ crc.update(buf, 0, buf.length);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testArchive() throws Exception {
+ Checksum readCrc = new Adler32();
+ File archiveFile = new File(outputFolder.getRoot() + File.separator
+ + "test.container.zip");
+ long zipCheckSum = FileUtil.zip(folder.getRoot(), archiveFile);
+ Assert.assertTrue(zipCheckSum > 0);
+ File decomp = new File(outputFolder.getRoot() + File.separator +
+ "decompress");
+ if (!decomp.exists() && !decomp.mkdirs()) {
+ throw new IOException("Unable to create the destination directory. " +
+ decomp.getPath());
+ }
+
+ FileUtil.unZip(archiveFile, decomp);
+ String[] patterns = {"txt"};
+ Iterator<File> iter = FileUtils.iterateFiles(decomp, patterns, true);
+ int count = 0;
+ while (iter.hasNext()) {
+ count++;
+ byte[] buf = Files.readAllBytes(iter.next().toPath());
+ readCrc.update(buf, 0, buf.length);
+ }
+ Assert.assertEquals(DIR_COUNT * SUB_DIR_COUNT * FILE_COUNT, count);
+ Assert.assertEquals(crc.getValue(), readCrc.getValue());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/package-info.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/package-info.java
new file mode 100644
index 0000000..9c480d6
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/scm/package-info.java
@@ -0,0 +1,21 @@
+/**
+ * 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.scm;
+/**
+ Test cases for SCM client classes.
+ */
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java
deleted file mode 100644
index 35b8961..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * 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.cblock;
-
-import static java.lang.Thread.NORM_PRIORITY;
-
-/**
- * This class contains constants for configuration keys used in CBlock.
- */
-public final class CBlockConfigKeys {
- public static final String DFS_CBLOCK_SERVICERPC_ADDRESS_KEY =
- "dfs.cblock.servicerpc-address";
- public static final int DFS_CBLOCK_SERVICERPC_PORT_DEFAULT =
- 9810;
- public static final String DFS_CBLOCK_SERVICERPC_HOSTNAME_DEFAULT =
- "0.0.0.0";
-
- public static final String DFS_CBLOCK_JSCSIRPC_ADDRESS_KEY =
- "dfs.cblock.jscsi-address";
-
- //The port on CBlockManager node for jSCSI to ask
- public static final String DFS_CBLOCK_JSCSI_PORT_KEY =
- "dfs.cblock.jscsi.port";
- public static final int DFS_CBLOCK_JSCSI_PORT_DEFAULT =
- 9811;
-
- public static final String DFS_CBLOCK_SERVICERPC_BIND_HOST_KEY =
- "dfs.cblock.service.rpc-bind-host";
- public static final String DFS_CBLOCK_JSCSIRPC_BIND_HOST_KEY =
- "dfs.cblock.jscsi.rpc-bind-host";
-
- // default block size is 4KB
- public static final int DFS_CBLOCK_SERVICE_BLOCK_SIZE_DEFAULT =
- 4096;
-
- public static final String DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_KEY =
- "dfs.storage.service.handler.count";
- public static final int DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_DEFAULT = 10;
-
- public static final String DFS_CBLOCK_SERVICE_LEVELDB_PATH_KEY =
- "dfs.cblock.service.leveldb.path";
- //TODO : find a better place
- public static final String DFS_CBLOCK_SERVICE_LEVELDB_PATH_DEFAULT =
- "/tmp/cblock_levelDB.dat";
-
-
- public static final String DFS_CBLOCK_DISK_CACHE_PATH_KEY =
- "dfs.cblock.disk.cache.path";
- public static final String DFS_CBLOCK_DISK_CACHE_PATH_DEFAULT =
- "/tmp/cblockCacheDB";
- /**
- * Setting this flag to true makes the block layer compute a sha256 hash of
- * the data and log that information along with block ID. This is very
- * useful for doing trace based simulation of various workloads. Since it is
- * computing a hash for each block this could be expensive, hence default
- * is false.
- */
- public static final String DFS_CBLOCK_TRACE_IO = "dfs.cblock.trace.io";
- public static final boolean DFS_CBLOCK_TRACE_IO_DEFAULT = false;
-
- public static final String DFS_CBLOCK_ENABLE_SHORT_CIRCUIT_IO =
- "dfs.cblock.short.circuit.io";
- public static final boolean DFS_CBLOCK_ENABLE_SHORT_CIRCUIT_IO_DEFAULT =
- false;
-
- /**
- * Cache size in 1000s of entries. 256 indicates 256 * 1024.
- */
- public static final String DFS_CBLOCK_CACHE_QUEUE_SIZE_KB =
- "dfs.cblock.cache.cache.size.in.kb";
- public static final int DFS_CBLOCK_CACHE_QUEUE_SIZE_KB_DEFAULT = 256;
-
- /**
- * Minimum Number of threads that cache pool will use for background I/O.
- */
- public static final String DFS_CBLOCK_CACHE_CORE_MIN_POOL_SIZE =
- "dfs.cblock.cache.core.min.pool.size";
- public static final int DFS_CBLOCK_CACHE_CORE_MIN_POOL_SIZE_DEFAULT = 16;
-
- /**
- * Maximum Number of threads that cache pool will use for background I/O.
- */
-
- public static final String DFS_CBLOCK_CACHE_MAX_POOL_SIZE =
- "dfs.cblock.cache.max.pool.size";
- public static final int DFS_CBLOCK_CACHE_MAX_POOL_SIZE_DEFAULT = 256;
-
- /**
- * Number of seconds to keep the Thread alive when it is idle.
- */
- public static final String DFS_CBLOCK_CACHE_KEEP_ALIVE_SECONDS =
- "dfs.cblock.cache.keep.alive.seconds";
- public static final long DFS_CBLOCK_CACHE_KEEP_ALIVE_SECONDS_DEFAULT = 60;
-
- /**
- * Priority of cache flusher thread, affecting the relative performance of
- * write and read.
- */
- public static final String DFS_CBLOCK_CACHE_THREAD_PRIORITY =
- "dfs.cblock.cache.thread.priority";
- public static final int DFS_CBLOCK_CACHE_THREAD_PRIORITY_DEFAULT =
- NORM_PRIORITY;
-
- /**
- * Block Buffer size in terms of blockID entries, 512 means 512 blockIDs.
- */
- public static final String DFS_CBLOCK_CACHE_BLOCK_BUFFER_SIZE =
- "dfs.cblock.cache.block.buffer.size";
- public static final int DFS_CBLOCK_CACHE_BLOCK_BUFFER_SIZE_DEFAULT = 512;
-
- public static final String DFS_CBLOCK_BLOCK_BUFFER_FLUSH_INTERVAL_SECONDS =
- "dfs.cblock.block.buffer.flush.interval.seconds";
- public static final int
- DFS_CBLOCK_BLOCK_BUFFER_FLUSH_INTERVAL_SECONDS_DEFAULT = 60;
-
- // jscsi server settings
- public static final String DFS_CBLOCK_JSCSI_SERVER_ADDRESS_KEY =
- "dfs.cblock.jscsi.server.address";
- public static final String DFS_CBLOCK_JSCSI_SERVER_ADDRESS_DEFAULT =
- "127.0.0.1";
- public static final String DFS_CBLOCK_JSCSI_CBLOCK_SERVER_ADDRESS_KEY =
- "dfs.cblock.jscsi.cblock.server.address";
- public static final String DFS_CBLOCK_JSCSI_CBLOCK_SERVER_ADDRESS_DEFAULT =
- "127.0.0.1";
-
- // to what address cblock server should talk to scm?
- public static final String DFS_CBLOCK_SCM_IPADDRESS_KEY =
- "dfs.cblock.scm.ipaddress";
- public static final String DFS_CBLOCK_SCM_IPADDRESS_DEFAULT =
- "127.0.0.1";
- public static final String DFS_CBLOCK_SCM_PORT_KEY =
- "dfs.cblock.scm.port";
- public static final int DFS_CBLOCK_SCM_PORT_DEFAULT = 9860;
-
- public static final String DFS_CBLOCK_CONTAINER_SIZE_GB_KEY =
- "dfs.cblock.container.size";
- public static final int DFS_CBLOCK_CONTAINER_SIZE_GB_DEFAULT =
- 5;
-
- // LevelDB cache file uses an off-heap cache in LevelDB of 256 MB.
- public static final String DFS_CBLOCK_CACHE_LEVELDB_CACHE_SIZE_MB_KEY =
- "dfs.cblock.cache.leveldb.cache.size.mb";
- public static final int DFS_CBLOCK_CACHE_LEVELDB_CACHE_SIZE_MB_DEFAULT = 256;
-
- /**
- * Cache does an best case attempt to write a block to a container.
- * At some point of time, we will need to handle the case where we did try
- * 64K times and is till not able to write to the container.
- *
- * TODO: We will need cBlock Server to allow us to do a remapping of the
- * block location in case of failures, at that point we should reduce the
- * retry count to a more normal number. This is approximately 18 hours of
- * retry.
- */
- public static final String DFS_CBLOCK_CACHE_MAX_RETRY_KEY =
- "dfs.cblock.cache.max.retry";
- public static final int DFS_CBLOCK_CACHE_MAX_RETRY_DEFAULT =
- 64 * 1024;
-
- private CBlockConfigKeys() {
-
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
index 144c3bb..d7349cf 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
@@ -34,7 +34,7 @@ import org.apache.hadoop.cblock.protocolPB.CBlockServiceProtocolPB;
import org.apache.hadoop.cblock.protocolPB
.CBlockServiceProtocolServerSideTranslatorPB;
import org.apache.hadoop.ipc.Client;
-import org.apache.hadoop.ozone.OzoneClientUtils;
+import org.apache.hadoop.ozone.client.OzoneClientUtils;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.scm.XceiverClientManager;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java
index f70e8a4..90a16ce 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java
@@ -22,7 +22,7 @@ import org.apache.hadoop.cblock.protocolPB.CBlockServiceProtocolPB;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.ozone.OzoneClientUtils;
+import org.apache.hadoop.ozone.client.OzoneClientUtils;
import org.apache.hadoop.ozone.OzoneConfiguration;
import org.apache.hadoop.security.UserGroupInformation;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/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 5b76179..e6dff98 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
@@ -31,12 +31,13 @@ import com.sun.jersey.api.container.ContainerFactory;
import com.sun.jersey.api.core.ApplicationAdapter;
import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.ksm.protocolPB
+import org.apache.hadoop.ozone.ksm.protocolPB
.KeySpaceManagerProtocolClientSideTranslatorPB;
-import org.apache.hadoop.ksm.protocolPB.KeySpaceManagerProtocolPB;
-import org.apache.hadoop.ozone.OzoneClientUtils;
+import org.apache.hadoop.ozone.ksm.protocolPB.KeySpaceManagerProtocolPB;
+import org.apache.hadoop.ozone.client.OzoneClientUtils;
import org.apache.hadoop.ozone.OzoneConsts;
-import org.apache.hadoop.scm.protocolPB.ScmBlockLocationProtocolClientSideTranslatorPB;
+import org.apache.hadoop.scm.protocolPB
+ .ScmBlockLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.scm.protocolPB.ScmBlockLocationProtocolPB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,7 +48,8 @@ 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.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
+import org.apache.hadoop.scm.protocolPB
+ .StorageContainerLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.scm.protocolPB.StorageContainerLocationProtocolPB;
import org.apache.hadoop.ozone.web.handlers.ServiceFilter;
import org.apache.hadoop.ozone.web.interfaces.StorageHandler;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/URLDispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/URLDispatcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/URLDispatcher.java
index bfdecce..5114298 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/URLDispatcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/web/URLDispatcher.java
@@ -25,7 +25,7 @@ import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.HttpRequest;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.server.datanode.web.webhdfs.WebHdfsHandler;
-import org.apache.hadoop.ozone.web.headers.Header;
+import org.apache.hadoop.ozone.client.rest.headers.Header;
import org.apache.hadoop.ozone.web.netty.ObjectStoreJerseyContainer;
import org.apache.hadoop.ozone.web.netty.RequestDispatchObjectStoreChannelHandler;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/GetConf.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/GetConf.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/GetConf.java
index 725b20f..d2aaf73 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/GetConf.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/GetConf.java
@@ -34,7 +34,7 @@ import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.DFSUtil.ConfiguredNNAddress;
-import org.apache.hadoop.ozone.OzoneClientUtils;
+import org.apache.hadoop.ozone.client.OzoneClientUtils;
import org.apache.hadoop.ozone.OzoneConfiguration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.StringUtils;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneBucket.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneBucket.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneBucket.java
deleted file mode 100644
index 51eefdc..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneBucket.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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;
-
-
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.ksm.helpers.KsmBucketInfo;
-import org.apache.hadoop.ozone.OzoneConsts.Versioning;
-
-import java.util.List;
-
-/**
- * A class that encapsulates OzoneBucket.
- */
-public class OzoneBucket {
-
- /**
- * Name of the volume in which the bucket belongs to.
- */
- private final String volumeName;
- /**
- * Name of the bucket.
- */
- private final String bucketName;
- /**
- * Bucket ACLs.
- */
- private final List<OzoneAcl> acls;
-
- /**
- * Type of storage to be used for this bucket.
- * [RAM_DISK, SSD, DISK, ARCHIVE]
- */
- private final StorageType storageType;
-
- /**
- * Bucket Version flag.
- */
- private final Versioning versioning;
-
-
- /**
- * Constructs OzoneBucket from KsmBucketInfo.
- *
- * @param ksmBucketInfo
- */
- public OzoneBucket(KsmBucketInfo ksmBucketInfo) {
- this.volumeName = ksmBucketInfo.getVolumeName();
- this.bucketName = ksmBucketInfo.getBucketName();
- this.acls = ksmBucketInfo.getAcls();
- this.storageType = ksmBucketInfo.getStorageType();
- this.versioning = ksmBucketInfo.getIsVersionEnabled() ?
- Versioning.ENABLED : Versioning.DISABLED;
- }
-
- /**
- * Returns Volume Name.
- *
- * @return volumeName
- */
- public String getVolumeName() {
- return volumeName;
- }
-
- /**
- * Returns Bucket Name.
- *
- * @return bucketName
- */
- public String getBucketName() {
- return bucketName;
- }
-
- /**
- * Returns ACL's associated with the Bucket.
- *
- * @return acls
- */
- public List<OzoneAcl> getAcls() {
- return acls;
- }
-
- /**
- * Returns StorageType of the Bucket.
- *
- * @return storageType
- */
- public StorageType getStorageType() {
- return storageType;
- }
-
- /**
- * Returns Versioning associated with the Bucket.
- *
- * @return versioning
- */
- public Versioning getVersioning() {
- return versioning;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClient.java
deleted file mode 100644
index dd52a57..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClient.java
+++ /dev/null
@@ -1,414 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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;
-
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.ozone.OzoneConsts.Versioning;
-import org.apache.hadoop.ozone.io.OzoneInputStream;
-import org.apache.hadoop.ozone.io.OzoneOutputStream;
-
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.List;
-
-/**
- * OzoneClient can connect to a Ozone Cluster and
- * perform basic operations.
- */
-public interface OzoneClient {
-
- /**
- * Creates a new Volume.
- *
- * @param volumeName Name of the Volume
- *
- * @throws IOException
- */
- void createVolume(String volumeName)
- throws IOException;
-
- /**
- * Creates a new Volume, with owner set.
- *
- * @param volumeName Name of the Volume
- * @param owner Owner to be set for Volume
- *
- * @throws IOException
- */
- void createVolume(String volumeName, String owner)
- throws IOException;
-
- /**
- * Creates a new Volume, with owner and quota set.
- *
- * @param volumeName Name of the Volume
- * @param owner Owner to be set for Volume
- * @param acls ACLs to be added to the Volume
- *
- * @throws IOException
- */
- void createVolume(String volumeName, String owner,
- OzoneAcl... acls)
- throws IOException;
-
- /**
- * Creates a new Volume, with owner and quota set.
- *
- * @param volumeName Name of the Volume
- * @param owner Owner to be set for Volume
- * @param quota Volume Quota
- *
- * @throws IOException
- */
- void createVolume(String volumeName, String owner,
- long quota)
- throws IOException;
-
- /**
- * Creates a new Volume, with owner and quota set.
- *
- * @param volumeName Name of the Volume
- * @param owner Owner to be set for Volume
- * @param quota Volume Quota
- * @param acls ACLs to be added to the Volume
- *
- * @throws IOException
- */
- void createVolume(String volumeName, String owner,
- long quota, OzoneAcl... acls)
- throws IOException;
-
- /**
- * Sets the owner of the volume.
- *
- * @param volumeName Name of the Volume
- * @param owner to be set for the Volume
- *
- * @throws IOException
- */
- void setVolumeOwner(String volumeName, String owner) throws IOException;
-
- /**
- * Set Volume Quota.
- *
- * @param volumeName Name of the Volume
- * @param quota Quota to be set for the Volume
- *
- * @throws IOException
- */
- void setVolumeQuota(String volumeName, long quota)
- throws IOException;
-
- /**
- * Returns {@link OzoneVolume}.
- *
- * @param volumeName Name of the Volume
- *
- * @return KsmVolumeArgs
- *
- * @throws OzoneVolume
- * */
- OzoneVolume getVolumeDetails(String volumeName)
- throws IOException;
-
- /**
- * Checks if a Volume exists and the user with a role specified has access
- * to the Volume.
- *
- * @param volumeName Name of the Volume
- * @param acl requested acls which needs to be checked for access
- *
- * @return Boolean - True if the user with a role can access the volume.
- * This is possible for owners of the volume and admin users
- *
- * @throws IOException
- */
- boolean checkVolumeAccess(String volumeName, OzoneAcl acl)
- throws IOException;
-
- /**
- * Deletes an Empty Volume.
- *
- * @param volumeName Name of the Volume
- *
- * @throws IOException
- */
- void deleteVolume(String volumeName) throws IOException;
-
- /**
- * Returns the List of Volumes owned by current user.
- *
- * @param volumePrefix Volume prefix to match
- *
- * @return KsmVolumeArgs Iterator
- *
- * @throws IOException
- */
- Iterator<OzoneVolume> listVolumes(String volumePrefix)
- throws IOException;
-
- /**
- * Returns the List of Volumes owned by the specific user.
- *
- * @param volumePrefix Volume prefix to match
- * @param user User Name
- *
- * @return KsmVolumeArgs Iterator
- *
- * @throws IOException
- */
- Iterator<OzoneVolume> listVolumes(String volumePrefix, String user)
- throws IOException;
-
- /**
- * Creates a new Bucket in the Volume.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @throws IOException
- */
- void createBucket(String volumeName, String bucketName)
- throws IOException;
-
- /**
- * Creates a new Bucket in the Volume, with versioning set.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- * @param versioning Bucket versioning
- *
- * @throws IOException
- */
- void createBucket(String volumeName, String bucketName,
- Versioning versioning)
- throws IOException;
-
- /**
- * Creates a new Bucket in the Volume, with storage type set.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- * @param storageType StorageType for the Bucket
- *
- * @throws IOException
- */
- void createBucket(String volumeName, String bucketName,
- StorageType storageType)
- throws IOException;
-
- /**
- * Creates a new Bucket in the Volume, with ACLs set.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- * @param acls OzoneAcls for the Bucket
- *
- * @throws IOException
- */
- void createBucket(String volumeName, String bucketName,
- OzoneAcl... acls)
- throws IOException;
-
-
- /**
- * Creates a new Bucket in the Volume, with versioning
- * storage type and ACLs set.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- * @param storageType StorageType for the Bucket
- *
- * @throws IOException
- */
- void createBucket(String volumeName, String bucketName,
- OzoneConsts.Versioning versioning,
- StorageType storageType, OzoneAcl... acls)
- throws IOException;
-
- /**
- * Adds or Removes ACLs from a Bucket.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @throws IOException
- */
- void addBucketAcls(String volumeName, String bucketName,
- List<OzoneAcl> addAcls)
- throws IOException;
-
- /**
- * Adds or Removes ACLs from a Bucket.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @throws IOException
- */
- void removeBucketAcls(String volumeName, String bucketName,
- List<OzoneAcl> removeAcls)
- throws IOException;
-
-
- /**
- * Enables or disables Bucket Versioning.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @throws IOException
- */
- void setBucketVersioning(String volumeName, String bucketName,
- Versioning versioning)
- throws IOException;
-
- /**
- * Sets the Storage Class of a Bucket.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @throws IOException
- */
- void setBucketStorageType(String volumeName, String bucketName,
- StorageType storageType)
- throws IOException;
-
- /**
- * Deletes a bucket if it is empty.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @throws IOException
- */
- void deleteBucket(String volumeName, String bucketName)
- throws IOException;
-
- /**
- * true if the bucket exists and user has read access
- * to the bucket else throws Exception.
- *
- * @param volumeName Name of the Volume
- *
- * @throws IOException
- */
- void checkBucketAccess(String volumeName, String bucketName)
- throws IOException;
-
- /**
- * Returns {@link OzoneBucket}.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @return OzoneBucket
- *
- * @throws IOException
- */
- OzoneBucket getBucketDetails(String volumeName, String bucketName)
- throws IOException;
-
- /**
- * Returns the List of Buckets in the Volume.
- *
- * @param volumeName Name of the Volume
- * @param bucketPrefix Bucket prefix to match
- *
- * @return KsmVolumeArgs Iterator
- *
- * @throws IOException
- */
- Iterator<OzoneBucket> listBuckets(String volumeName, String bucketPrefix)
- throws IOException;
-
- /**
- * Writes a key in an existing bucket.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- * @param size Size of the data
- *
- * @return OutputStream
- *
- */
- OzoneOutputStream createKey(String volumeName, String bucketName,
- String keyName, long size)
- throws IOException;
-
- /**
- * Reads a key from an existing bucket.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @return LengthInputStream
- *
- * @throws IOException
- */
- OzoneInputStream getKey(String volumeName, String bucketName, String keyName)
- throws IOException;
-
-
- /**
- * Deletes an existing key.
- *
- * @param volumeName Name of the Volume
- *
- * @throws IOException
- */
- void deleteKey(String volumeName, String bucketName, String keyName)
- throws IOException;
-
-
- /**
- * Returns list of {@link OzoneKey} in Volume/Bucket.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- *
- * @return OzoneKey
- *
- * @throws IOException
- */
- List<OzoneKey> listKeys(String volumeName, String bucketName,
- String keyPrefix)
- throws IOException;
-
-
- /**
- * Get OzoneKey.
- *
- * @param volumeName Name of the Volume
- * @param bucketName Name of the Bucket
- * @param keyName Key name
- *
- * @return OzoneKey
- *
- * @throws IOException
- */
- OzoneKey getKeyDetails(String volumeName, String bucketName,
- String keyName)
- throws IOException;
-
- /**
- * Close and release the resources.
- */
- void close() throws IOException;
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientFactory.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientFactory.java
deleted file mode 100644
index 7866f58..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientFactory.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/**
- * 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;
-
-import org.apache.hadoop.conf.Configuration;
-
-import java.io.IOException;
-
-/**
- * Factory class to create different types of OzoneClients.
- */
-public final class OzoneClientFactory {
-
- /**
- * Private constructor, class is not meant to be initialized.
- */
- private OzoneClientFactory(){}
-
- private static Configuration configuration;
-
- /**
- * Returns an OzoneClient which will use RPC protocol to perform
- * client operations.
- *
- * @return OzoneClient
- * @throws IOException
- */
- public static OzoneClient getRpcClient() throws IOException {
- return new OzoneClientImpl(getConfiguration());
- }
-
- /**
- * Sets the configuration, which will be used while creating OzoneClient.
- *
- * @param conf
- */
- public static void setConfiguration(Configuration conf) {
- configuration = conf;
- }
-
- /**
- * Returns the configuration if it's already set, else creates a new
- * {@link OzoneConfiguration} and returns it.
- *
- * @return Configuration
- */
- private static synchronized Configuration getConfiguration() {
- if(configuration == null) {
- setConfiguration(new OzoneConfiguration());
- }
- return configuration;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientImpl.java
deleted file mode 100644
index feb4586..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientImpl.java
+++ /dev/null
@@ -1,570 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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;
-
-import com.google.common.base.Preconditions;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdfs.server.datanode.fsdataset.LengthInputStream;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.ipc.Client;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.ksm.helpers.KsmBucketArgs;
-import org.apache.hadoop.ksm.helpers.KsmBucketInfo;
-import org.apache.hadoop.ksm.helpers.KsmKeyArgs;
-import org.apache.hadoop.ksm.helpers.KsmKeyInfo;
-import org.apache.hadoop.ksm.helpers.KsmVolumeArgs;
-import org.apache.hadoop.ksm.protocolPB
- .KeySpaceManagerProtocolClientSideTranslatorPB;
-import org.apache.hadoop.ksm.protocolPB
- .KeySpaceManagerProtocolPB;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.ozone.web.storage.ChunkGroupInputStream;
-import org.apache.hadoop.ozone.io.OzoneInputStream;
-import org.apache.hadoop.ozone.io.OzoneOutputStream;
-import org.apache.hadoop.ozone.ksm.KSMConfigKeys;
-import org.apache.hadoop.ozone.OzoneConsts.Versioning;
-import org.apache.hadoop.ozone.protocolPB.KSMPBHelper;
-import org.apache.hadoop.ozone.web.storage.ChunkGroupOutputStream;
-import org.apache.hadoop.scm.ScmConfigKeys;
-import org.apache.hadoop.scm.XceiverClientManager;
-import org.apache.hadoop.scm.protocolPB
- .StorageContainerLocationProtocolClientSideTranslatorPB;
-import org.apache.hadoop.scm.protocolPB
- .StorageContainerLocationProtocolPB;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Collectors;
-
-/**
- * Ozone Client Implementation, it connects to KSM, SCM and DataNode
- * to execute client calls. This uses RPC protocol for communication
- * with the servers.
- */
-public class OzoneClientImpl implements OzoneClient, Closeable {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(OzoneClient.class);
-
- private final StorageContainerLocationProtocolClientSideTranslatorPB
- storageContainerLocationClient;
- private final KeySpaceManagerProtocolClientSideTranslatorPB
- keySpaceManagerClient;
- private final XceiverClientManager xceiverClientManager;
- private final int chunkSize;
-
-
- private final UserGroupInformation ugi;
- private final OzoneAcl.OzoneACLRights userRights;
- private final OzoneAcl.OzoneACLRights groupRights;
-
- /**
- * Creates OzoneClientImpl instance with new OzoneConfiguration.
- *
- * @throws IOException
- */
- public OzoneClientImpl() throws IOException {
- this(new OzoneConfiguration());
- }
-
- /**
- * Creates OzoneClientImpl instance with the given configuration.
- *
- * @param conf
- *
- * @throws IOException
- */
- public OzoneClientImpl(Configuration conf) throws IOException {
- Preconditions.checkNotNull(conf);
- this.ugi = UserGroupInformation.getCurrentUser();
- this.userRights = conf.getEnum(KSMConfigKeys.OZONE_KSM_USER_RIGHTS,
- KSMConfigKeys.OZONE_KSM_USER_RIGHTS_DEFAULT);
- this.groupRights = conf.getEnum(KSMConfigKeys.OZONE_KSM_GROUP_RIGHTS,
- KSMConfigKeys.OZONE_KSM_GROUP_RIGHTS_DEFAULT);
-
- long scmVersion =
- RPC.getProtocolVersion(StorageContainerLocationProtocolPB.class);
- InetSocketAddress scmAddress =
- OzoneClientUtils.getScmAddressForClients(conf);
- RPC.setProtocolEngine(conf, StorageContainerLocationProtocolPB.class,
- ProtobufRpcEngine.class);
- this.storageContainerLocationClient =
- new StorageContainerLocationProtocolClientSideTranslatorPB(
- RPC.getProxy(StorageContainerLocationProtocolPB.class, scmVersion,
- scmAddress, UserGroupInformation.getCurrentUser(), conf,
- NetUtils.getDefaultSocketFactory(conf),
- Client.getRpcTimeout(conf)));
-
- long ksmVersion =
- RPC.getProtocolVersion(KeySpaceManagerProtocolPB.class);
- InetSocketAddress ksmAddress = OzoneClientUtils.getKsmAddress(conf);
- RPC.setProtocolEngine(conf, KeySpaceManagerProtocolPB.class,
- ProtobufRpcEngine.class);
- this.keySpaceManagerClient =
- new KeySpaceManagerProtocolClientSideTranslatorPB(
- RPC.getProxy(KeySpaceManagerProtocolPB.class, ksmVersion,
- ksmAddress, UserGroupInformation.getCurrentUser(), conf,
- NetUtils.getDefaultSocketFactory(conf),
- Client.getRpcTimeout(conf)));
-
- this.xceiverClientManager = new XceiverClientManager(conf);
-
- int configuredChunkSize = conf.getInt(
- ScmConfigKeys.OZONE_SCM_CHUNK_SIZE_KEY,
- ScmConfigKeys.OZONE_SCM_CHUNK_SIZE_DEFAULT);
- if(configuredChunkSize > ScmConfigKeys.OZONE_SCM_CHUNK_MAX_SIZE) {
- LOG.warn("The chunk size ({}) is not allowed to be more than"
- + " the maximum size ({}),"
- + " resetting to the maximum size.",
- configuredChunkSize, ScmConfigKeys.OZONE_SCM_CHUNK_MAX_SIZE);
- chunkSize = ScmConfigKeys.OZONE_SCM_CHUNK_MAX_SIZE;
- } else {
- chunkSize = configuredChunkSize;
- }
- }
-
- @Override
- public void createVolume(String volumeName)
- throws IOException {
- createVolume(volumeName, ugi.getUserName());
- }
-
- @Override
- public void createVolume(String volumeName, String owner)
- throws IOException {
-
- createVolume(volumeName, owner, OzoneConsts.MAX_QUOTA_IN_BYTES,
- (OzoneAcl[])null);
- }
-
- @Override
- public void createVolume(String volumeName, String owner,
- OzoneAcl... acls)
- throws IOException {
- createVolume(volumeName, owner, OzoneConsts.MAX_QUOTA_IN_BYTES, acls);
- }
-
- @Override
- public void createVolume(String volumeName, String owner,
- long quota)
- throws IOException {
- createVolume(volumeName, owner, quota, (OzoneAcl[])null);
- }
-
- @Override
- public void createVolume(String volumeName, String owner,
- long quota, OzoneAcl... acls)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(owner);
- Preconditions.checkNotNull(quota);
- Preconditions.checkState(quota >= 0);
- OzoneAcl userAcl =
- new OzoneAcl(OzoneAcl.OzoneACLType.USER,
- owner, userRights);
- KsmVolumeArgs.Builder builder = KsmVolumeArgs.newBuilder();
- builder.setAdminName(ugi.getUserName())
- .setOwnerName(owner)
- .setVolume(volumeName)
- .setQuotaInBytes(quota)
- .addOzoneAcls(KSMPBHelper.convertOzoneAcl(userAcl));
-
- List<OzoneAcl> listOfAcls = new ArrayList<>();
-
- //Group ACLs of the User
- List<String> userGroups = Arrays.asList(UserGroupInformation
- .createRemoteUser(owner).getGroupNames());
- userGroups.stream().forEach((group) -> listOfAcls.add(
- new OzoneAcl(OzoneAcl.OzoneACLType.GROUP, group, groupRights)));
-
- //ACLs passed as argument
- if(acls != null) {
- listOfAcls.addAll(Arrays.asList(acls));
- }
-
- //Remove duplicates and set
- for (OzoneAcl ozoneAcl :
- listOfAcls.stream().distinct().collect(Collectors.toList())) {
- builder.addOzoneAcls(KSMPBHelper.convertOzoneAcl(ozoneAcl));
- }
-
- LOG.info("Creating Volume: {}, with {} as owner and quota set to {} bytes.",
- volumeName, owner, quota);
- keySpaceManagerClient.createVolume(builder.build());
- }
-
- @Override
- public void setVolumeOwner(String volumeName, String owner)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(owner);
- keySpaceManagerClient.setOwner(volumeName, owner);
- }
-
- @Override
- public void setVolumeQuota(String volumeName, long quota)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(quota);
- Preconditions.checkState(quota >= 0);
- keySpaceManagerClient.setQuota(volumeName, quota);
- }
-
- @Override
- public OzoneVolume getVolumeDetails(String volumeName)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- KsmVolumeArgs volumeArgs =
- keySpaceManagerClient.getVolumeInfo(volumeName);
- return new OzoneVolume(volumeArgs);
- }
-
- @Override
- public boolean checkVolumeAccess(String volumeName, OzoneAcl acl)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- return keySpaceManagerClient.checkVolumeAccess(volumeName,
- KSMPBHelper.convertOzoneAcl(acl));
- }
-
- @Override
- public void deleteVolume(String volumeName)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- keySpaceManagerClient.deleteVolume(volumeName);
- }
-
- @Override
- public Iterator<OzoneVolume> listVolumes(String volumePrefix)
- throws IOException {
- throw new UnsupportedOperationException("Not yet implemented.");
- }
-
- @Override
- public Iterator<OzoneVolume> listVolumes(String volumePrefix,
- String user)
- throws IOException {
- throw new UnsupportedOperationException("Not yet implemented.");
- }
-
- @Override
- public void createBucket(String volumeName, String bucketName)
- throws IOException {
- createBucket(volumeName, bucketName, Versioning.NOT_DEFINED,
- StorageType.DEFAULT, (OzoneAcl[])null);
- }
-
- @Override
- public void createBucket(String volumeName, String bucketName,
- Versioning versioning)
- throws IOException {
- createBucket(volumeName, bucketName, versioning,
- StorageType.DEFAULT, (OzoneAcl[])null);
- }
-
- @Override
- public void createBucket(String volumeName, String bucketName,
- StorageType storageType)
- throws IOException {
- createBucket(volumeName, bucketName, Versioning.NOT_DEFINED,
- storageType, (OzoneAcl[])null);
- }
-
- @Override
- public void createBucket(String volumeName, String bucketName,
- OzoneAcl... acls)
- throws IOException {
- createBucket(volumeName, bucketName, Versioning.NOT_DEFINED,
- StorageType.DEFAULT, acls);
- }
-
- @Override
- public void createBucket(String volumeName, String bucketName,
- Versioning versioning, StorageType storageType,
- OzoneAcl... acls)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(versioning);
- Preconditions.checkNotNull(storageType);
-
- KsmBucketInfo.Builder builder = KsmBucketInfo.newBuilder();
- builder.setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setStorageType(storageType)
- .setIsVersionEnabled(getBucketVersioningProtobuf(
- versioning));
-
- String owner = ugi.getUserName();
- final List<OzoneAcl> listOfAcls = new ArrayList<>();
-
- //User ACL
- OzoneAcl userAcl =
- new OzoneAcl(OzoneAcl.OzoneACLType.USER,
- owner, userRights);
- listOfAcls.add(userAcl);
-
- //Group ACLs of the User
- List<String> userGroups = Arrays.asList(UserGroupInformation
- .createRemoteUser(owner).getGroupNames());
- userGroups.stream().forEach((group) -> listOfAcls.add(
- new OzoneAcl(OzoneAcl.OzoneACLType.GROUP, group, groupRights)));
-
- //ACLs passed as argument
- if(acls != null) {
- Arrays.stream(acls).forEach((acl) -> listOfAcls.add(acl));
- }
-
- //Remove duplicates and set
- builder.setAcls(listOfAcls.stream().distinct()
- .collect(Collectors.toList()));
- LOG.info("Creating Bucket: {}/{}, with Versioning {} and " +
- "Storage Type set to {}", volumeName, bucketName, versioning,
- storageType);
- keySpaceManagerClient.createBucket(builder.build());
- }
-
- /**
- * Converts OzoneConts.Versioning enum to boolean.
- *
- * @param version
- * @return corresponding boolean value
- */
- private boolean getBucketVersioningProtobuf(
- Versioning version) {
- if(version != null) {
- switch(version) {
- case ENABLED:
- return true;
- case NOT_DEFINED:
- case DISABLED:
- default:
- return false;
- }
- }
- return false;
- }
-
- @Override
- public void addBucketAcls(String volumeName, String bucketName,
- List<OzoneAcl> addAcls)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(addAcls);
- KsmBucketArgs.Builder builder = KsmBucketArgs.newBuilder();
- builder.setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setAddAcls(addAcls);
- keySpaceManagerClient.setBucketProperty(builder.build());
- }
-
- @Override
- public void removeBucketAcls(String volumeName, String bucketName,
- List<OzoneAcl> removeAcls)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(removeAcls);
- KsmBucketArgs.Builder builder = KsmBucketArgs.newBuilder();
- builder.setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setRemoveAcls(removeAcls);
- keySpaceManagerClient.setBucketProperty(builder.build());
- }
-
- @Override
- public void setBucketVersioning(String volumeName, String bucketName,
- Versioning versioning)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(versioning);
- KsmBucketArgs.Builder builder = KsmBucketArgs.newBuilder();
- builder.setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setIsVersionEnabled(getBucketVersioningFlag(
- versioning));
- keySpaceManagerClient.setBucketProperty(builder.build());
- }
-
- @Override
- public void setBucketStorageType(String volumeName, String bucketName,
- StorageType storageType)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(storageType);
- KsmBucketArgs.Builder builder = KsmBucketArgs.newBuilder();
- builder.setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setStorageType(storageType);
- keySpaceManagerClient.setBucketProperty(builder.build());
- }
-
- @Override
- public void deleteBucket(String volumeName, String bucketName)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- keySpaceManagerClient.deleteBucket(volumeName, bucketName);
- }
-
- @Override
- public void checkBucketAccess(String volumeName, String bucketName)
- throws IOException {
- throw new UnsupportedOperationException("Not yet implemented.");
- }
-
- @Override
- public OzoneBucket getBucketDetails(String volumeName,
- String bucketName)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- KsmBucketInfo bucketInfo =
- keySpaceManagerClient.getBucketInfo(volumeName, bucketName);
- return new OzoneBucket(bucketInfo);
- }
-
- @Override
- public Iterator<OzoneBucket> listBuckets(String volumeName,
- String bucketPrefix)
- throws IOException {
- throw new UnsupportedOperationException("Not yet implemented.");
- }
-
- @Override
- public OzoneOutputStream createKey(String volumeName, String bucketName,
- String keyName, long size)
- throws IOException {
- String requestId = UUID.randomUUID().toString();
- KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()
- .setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setKeyName(keyName)
- .setDataSize(size)
- .build();
-
- KsmKeyInfo keyInfo = keySpaceManagerClient.allocateKey(keyArgs);
- ChunkGroupOutputStream groupOutputStream =
- ChunkGroupOutputStream.getFromKsmKeyInfo(keyInfo, xceiverClientManager,
- storageContainerLocationClient, chunkSize, requestId);
- return new OzoneOutputStream(groupOutputStream);
- }
-
- @Override
- public OzoneInputStream getKey(String volumeName, String bucketName,
- String keyName)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(keyName);
- String requestId = UUID.randomUUID().toString();
- KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()
- .setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setKeyName(keyName)
- .build();
- KsmKeyInfo keyInfo = keySpaceManagerClient.lookupKey(keyArgs);
- LengthInputStream lengthInputStream =
- ChunkGroupInputStream.getFromKsmKeyInfo(
- keyInfo, xceiverClientManager, storageContainerLocationClient,
- requestId);
- return new OzoneInputStream(
- (ChunkGroupInputStream)lengthInputStream.getWrappedStream());
- }
-
- @Override
- public void deleteKey(String volumeName, String bucketName,
- String keyName)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(keyName);
- KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()
- .setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setKeyName(keyName)
- .build();
- keySpaceManagerClient.deleteKey(keyArgs);
- }
-
- @Override
- public List<OzoneKey> listKeys(String volumeName, String bucketName,
- String keyPrefix)
- throws IOException {
- throw new UnsupportedOperationException("Not yet implemented.");
- }
-
- @Override
- public OzoneKey getKeyDetails(String volumeName, String bucketName,
- String keyName)
- throws IOException {
- Preconditions.checkNotNull(volumeName);
- Preconditions.checkNotNull(bucketName);
- Preconditions.checkNotNull(keyName);
- KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()
- .setVolumeName(volumeName)
- .setBucketName(bucketName)
- .setKeyName(keyName)
- .build();
- KsmKeyInfo keyInfo =
- keySpaceManagerClient.lookupKey(keyArgs);
- return new OzoneKey(keyInfo);
- }
-
- /**
- * Converts Versioning to boolean.
- *
- * @param version
- * @return corresponding boolean value
- */
- private boolean getBucketVersioningFlag(
- Versioning version) {
- if(version != null) {
- switch(version) {
- case ENABLED:
- return true;
- case DISABLED:
- case NOT_DEFINED:
- default:
- return false;
- }
- }
- return false;
- }
-
- @Override
- public void close() throws IOException {
- IOUtils.cleanupWithLogger(LOG, storageContainerLocationClient);
- IOUtils.cleanupWithLogger(LOG, keySpaceManagerClient);
- IOUtils.cleanupWithLogger(LOG, xceiverClientManager);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/43d38114/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientUtils.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientUtils.java
deleted file mode 100644
index d92f49f..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/OzoneClientUtils.java
+++ /dev/null
@@ -1,705 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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;
-
-import com.google.common.base.Optional;
-
-import com.google.common.net.HostAndPort;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.scm.ScmConfigKeys;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.client.methods.HttpRequestBase;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClients;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.net.InetSocketAddress;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-import static org.apache.hadoop.cblock.CBlockConfigKeys
- .DFS_CBLOCK_SERVICERPC_ADDRESS_KEY;
-import static org.apache.hadoop.cblock.CBlockConfigKeys
- .DFS_CBLOCK_SERVICERPC_HOSTNAME_DEFAULT;
-import static org.apache.hadoop.cblock.CBlockConfigKeys
- .DFS_CBLOCK_SERVICERPC_PORT_DEFAULT;
-import static org.apache.hadoop.cblock.CBlockConfigKeys
- .DFS_CBLOCK_JSCSIRPC_ADDRESS_KEY;
-import static org.apache.hadoop.cblock.CBlockConfigKeys
- .DFS_CBLOCK_JSCSI_PORT_DEFAULT;
-
-import static org.apache.hadoop.ozone.ksm.KSMConfigKeys.OZONE_KSM_ADDRESS_KEY;
-import static org.apache.hadoop.ozone.ksm.KSMConfigKeys
- .OZONE_KSM_BIND_HOST_DEFAULT;
-import static org.apache.hadoop.ozone.ksm.KSMConfigKeys.OZONE_KSM_PORT_DEFAULT;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_DEADNODE_INTERVAL_DEFAULT;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_DEADNODE_INTERVAL_MS;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_HEARTBEAT_INTERVAL_SECONDS;
-
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_HEARTBEAT_LOG_WARN_DEFAULT;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_HEARTBEAT_LOG_WARN_INTERVAL_COUNT;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS;
-
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_HEARTBEAT_RPC_TIMEOUT;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_HEARTBEAT_RPC_TIMEOUT_DEFAULT;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_STALENODE_INTERVAL_DEFAULT;
-import static org.apache.hadoop.scm.ScmConfigKeys
- .OZONE_SCM_STALENODE_INTERVAL_MS;
-
-/**
- * Utility methods for Ozone and Container Clients.
- *
- * The methods to retrieve SCM service endpoints assume there is a single
- * SCM service instance. This will change when we switch to replicated service
- * instances for redundancy.
- */
-@InterfaceAudience.Public
-@InterfaceStability.Unstable
-public final class OzoneClientUtils {
- private static final Logger LOG = LoggerFactory.getLogger(
- OzoneClientUtils.class);
- private static final int NO_PORT = -1;
-
- /**
- * The service ID of the solitary Ozone SCM service.
- */
- public static final String OZONE_SCM_SERVICE_ID = "OzoneScmService";
- public static final String OZONE_SCM_SERVICE_INSTANCE_ID =
- "OzoneScmServiceInstance";
-
- private OzoneClientUtils() {
- // Never constructed
- }
-
- /**
- * Retrieve the socket addresses of all storage container managers.
- *
- * @param conf
- * @return A collection of SCM addresses
- * @throws IllegalArgumentException If the configuration is invalid
- */
- public static Collection<InetSocketAddress> getSCMAddresses(
- Configuration conf) throws IllegalArgumentException {
- Collection<InetSocketAddress> addresses =
- new HashSet<InetSocketAddress>();
- Collection<String> names =
- conf.getTrimmedStringCollection(ScmConfigKeys.OZONE_SCM_NAMES);
- if (names == null || names.isEmpty()) {
- throw new IllegalArgumentException(ScmConfigKeys.OZONE_SCM_NAMES
- + " need to be a set of valid DNS names or IP addresses."
- + " Null or empty address list found.");
- }
-
- final com.google.common.base.Optional<Integer>
- defaultPort = com.google.common.base.Optional.of(ScmConfigKeys
- .OZONE_SCM_DEFAULT_PORT);
- for (String address : names) {
- com.google.common.base.Optional<String> hostname =
- OzoneClientUtils.getHostName(address);
- if (!hostname.isPresent()) {
- throw new IllegalArgumentException("Invalid hostname for SCM: "
- + hostname);
- }
- com.google.common.base.Optional<Integer> port =
- OzoneClientUtils.getHostPort(address);
- InetSocketAddress addr = NetUtils.createSocketAddr(hostname.get(),
- port.or(defaultPort.get()));
- addresses.add(addr);
- }
- return addresses;
- }
-
- /**
- * Retrieve the socket address that should be used by clients to connect
- * to the SCM.
- *
- * @param conf
- * @return Target InetSocketAddress for the SCM client endpoint.
- */
- public static InetSocketAddress getScmAddressForClients(Configuration conf) {
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY);
-
- if (!host.isPresent()) {
- throw new IllegalArgumentException(
- ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY +
- " must be defined. See" +
- " https://wiki.apache.org/hadoop/Ozone#Configuration for details" +
- " on configuring Ozone.");
- }
-
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(host.get() + ":" +
- port.or(ScmConfigKeys.OZONE_SCM_CLIENT_PORT_DEFAULT));
- }
-
- /**
- * Retrieve the socket address that should be used by clients to connect
- * to the SCM for block service. If
- * {@link ScmConfigKeys#OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY} is not defined
- * then {@link ScmConfigKeys#OZONE_SCM_CLIENT_ADDRESS_KEY} is used.
- *
- * @param conf
- * @return Target InetSocketAddress for the SCM block client endpoint.
- * @throws IllegalArgumentException if configuration is not defined.
- */
- public static InetSocketAddress getScmAddressForBlockClients(
- Configuration conf) {
- Optional<String> host = getHostNameFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY);
-
- if (!host.isPresent()) {
- host = getHostNameFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY);
- if (!host.isPresent()) {
- throw new IllegalArgumentException(
- ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY +
- " must be defined. See" +
- " https://wiki.apache.org/hadoop/Ozone#Configuration for details" +
- " on configuring Ozone.");
- }
- }
-
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(host.get() + ":" +
- port.or(ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_PORT_DEFAULT));
- }
-
- /**
- * Retrieve the socket address that should be used by DataNodes to connect
- * to the SCM.
- *
- * @param conf
- * @return Target InetSocketAddress for the SCM service endpoint.
- */
- public static InetSocketAddress getScmAddressForDataNodes(
- Configuration conf) {
- // We try the following settings in decreasing priority to retrieve the
- // target host.
- // - OZONE_SCM_DATANODE_ADDRESS_KEY
- // - OZONE_SCM_CLIENT_ADDRESS_KEY
- //
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY,
- ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY);
-
- if (!host.isPresent()) {
- throw new IllegalArgumentException(
- ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY +
- " must be defined. See" +
- " https://wiki.apache.org/hadoop/Ozone#Configuration for details" +
- " on configuring Ozone.");
- }
-
- // If no port number is specified then we'll just try the defaultBindPort.
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY);
-
- InetSocketAddress addr = NetUtils.createSocketAddr(host.get() + ":" +
- port.or(ScmConfigKeys.OZONE_SCM_DATANODE_PORT_DEFAULT));
-
- return addr;
- }
-
- /**
- * Retrieve the socket address that should be used by clients to connect
- * to the SCM.
- *
- * @param conf
- * @return Target InetSocketAddress for the SCM client endpoint.
- */
- public static InetSocketAddress getScmClientBindAddress(
- Configuration conf) {
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_CLIENT_BIND_HOST_KEY);
-
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(
- host.or(ScmConfigKeys.OZONE_SCM_CLIENT_BIND_HOST_DEFAULT) + ":" +
- port.or(ScmConfigKeys.OZONE_SCM_CLIENT_PORT_DEFAULT));
- }
-
- /**
- * Retrieve the socket address that should be used by clients to connect
- * to the SCM Block service.
- *
- * @param conf
- * @return Target InetSocketAddress for the SCM block client endpoint.
- */
- public static InetSocketAddress getScmBlockClientBindAddress(
- Configuration conf) {
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_BIND_HOST_KEY);
-
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(
- host.or(ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_BIND_HOST_DEFAULT) +
- ":" + port.or(ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_PORT_DEFAULT));
- }
-
- /**
- * Retrieve the socket address that should be used by DataNodes to connect
- * to the SCM.
- *
- * @param conf
- * @return Target InetSocketAddress for the SCM service endpoint.
- */
- public static InetSocketAddress getScmDataNodeBindAddress(
- Configuration conf) {
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_DATANODE_BIND_HOST_KEY);
-
- // If no port number is specified then we'll just try the defaultBindPort.
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(
- host.or(ScmConfigKeys.OZONE_SCM_DATANODE_BIND_HOST_DEFAULT) + ":" +
- port.or(ScmConfigKeys.OZONE_SCM_DATANODE_PORT_DEFAULT));
- }
-
-
- /**
- * Retrieve the socket address that is used by KSM.
- * @param conf
- * @return Target InetSocketAddress for the SCM service endpoint.
- */
- public static InetSocketAddress getKsmAddress(
- Configuration conf) {
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- OZONE_KSM_ADDRESS_KEY);
-
- // If no port number is specified then we'll just try the defaultBindPort.
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- OZONE_KSM_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(
- host.or(OZONE_KSM_BIND_HOST_DEFAULT) + ":" +
- port.or(OZONE_KSM_PORT_DEFAULT));
- }
-
- /**
- * Retrieve the socket address that is used by CBlock Service.
- * @param conf
- * @return Target InetSocketAddress for the CBlock Service endpoint.
- */
- public static InetSocketAddress getCblockServiceRpcAddr(
- Configuration conf) {
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- DFS_CBLOCK_SERVICERPC_ADDRESS_KEY);
-
- // If no port number is specified then we'll just try the defaultBindPort.
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- DFS_CBLOCK_SERVICERPC_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(
- host.or(DFS_CBLOCK_SERVICERPC_HOSTNAME_DEFAULT) + ":" +
- port.or(DFS_CBLOCK_SERVICERPC_PORT_DEFAULT));
- }
-
- /**
- * Retrieve the socket address that is used by CBlock Server.
- * @param conf
- * @return Target InetSocketAddress for the CBlock Server endpoint.
- */
- public static InetSocketAddress getCblockServerRpcAddr(
- Configuration conf) {
- final Optional<String> host = getHostNameFromConfigKeys(conf,
- DFS_CBLOCK_JSCSIRPC_ADDRESS_KEY);
-
- // If no port number is specified then we'll just try the defaultBindPort.
- final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
- DFS_CBLOCK_JSCSIRPC_ADDRESS_KEY);
-
- return NetUtils.createSocketAddr(
- host.or(DFS_CBLOCK_SERVICERPC_HOSTNAME_DEFAULT) + ":" +
- port.or(DFS_CBLOCK_JSCSI_PORT_DEFAULT));
- }
-
- /**
- * Retrieve the hostname, trying the supplied config keys in order.
- * Each config value may be absent, or if present in the format
- * host:port (the :port part is optional).
- *
- * @param conf - Conf
- * @param keys a list of configuration key names.
- *
- * @return first hostname component found from the given keys, or absent.
- * @throws IllegalArgumentException if any values are not in the 'host'
- * or host:port format.
- */
- public static Optional<String> getHostNameFromConfigKeys(Configuration conf,
- String... keys) {
- for (final String key : keys) {
- final String value = conf.getTrimmed(key);
- final Optional<String> hostName = getHostName(value);
- if (hostName.isPresent()) {
- return hostName;
- }
- }
- return Optional.absent();
- }
-
- /**
- * Gets the hostname or Indicates that it is absent.
- * @param value host or host:port
- * @return hostname
- */
- public static Optional<String> getHostName(String value) {
- if ((value == null) || value.isEmpty()) {
- return Optional.absent();
- }
- return Optional.of(HostAndPort.fromString(value).getHostText());
- }
-
- /**
- * Gets the port if there is one, throws otherwise.
- * @param value String in host:port format.
- * @return Port
- */
- public static Optional<Integer> getHostPort(String value) {
- if((value == null) || value.isEmpty()) {
- return Optional.absent();
- }
- int port = HostAndPort.fromString(value).getPortOrDefault(NO_PORT);
- if (port == NO_PORT) {
- return Optional.absent();
- } else {
- return Optional.of(port);
- }
- }
-
- /**
- * Retrieve the port number, trying the supplied config keys in order.
- * Each config value may be absent, or if present in the format
- * host:port (the :port part is optional).
- *
- * @param conf Conf
- * @param keys a list of configuration key names.
- *
- * @return first port number component found from the given keys, or absent.
- * @throws IllegalArgumentException if any values are not in the 'host'
- * or host:port format.
- */
- public static Optional<Integer> getPortNumberFromConfigKeys(
- Configuration conf, String... keys) {
- for (final String key : keys) {
- final String value = conf.getTrimmed(key);
- final Optional<Integer> hostPort = getHostPort(value);
- if (hostPort.isPresent()) {
- return hostPort;
- }
- }
- return Optional.absent();
- }
-
- /**
- * Return the list of service addresses for the Ozone SCM. This method is used
- * by the DataNodes to determine the service instances to connect to.
- *
- * @param conf
- * @return list of SCM service addresses.
- */
- public static Map<String, ? extends Map<String, InetSocketAddress>>
- getScmServiceRpcAddresses(Configuration conf) {
- final Map<String, InetSocketAddress> serviceInstances = new HashMap<>();
- serviceInstances.put(OZONE_SCM_SERVICE_INSTANCE_ID,
- getScmAddressForDataNodes(conf));
-
- final Map<String, Map<String, InetSocketAddress>> services =
- new HashMap<>();
- services.put(OZONE_SCM_SERVICE_ID, serviceInstances);
- return services;
- }
-
- /**
- * Checks that a given value is with a range.
- *
- * For example, sanitizeUserArgs(17, 3, 5, 10)
- * ensures that 17 is greater/equal than 3 * 5 and less/equal to 3 * 10.
- *
- * @param valueTocheck - value to check
- * @param baseValue - the base value that is being used.
- * @param minFactor - range min - a 2 here makes us ensure that value
- * valueTocheck is at least twice the baseValue.
- * @param maxFactor - range max
- * @return long
- */
- private static long sanitizeUserArgs(long valueTocheck, long baseValue,
- long minFactor, long maxFactor)
- throws IllegalArgumentException {
- if ((valueTocheck >= (baseValue * minFactor)) &&
- (valueTocheck <= (baseValue * maxFactor))) {
- return valueTocheck;
- }
- String errMsg = String.format("%d is not within min = %d or max = " +
- "%d", valueTocheck, baseValue * minFactor, baseValue * maxFactor);
- throw new IllegalArgumentException(errMsg);
- }
-
- /**
- * Returns the interval in which the heartbeat processor thread runs.
- *
- * @param conf - Configuration
- * @return long in Milliseconds.
- */
- public static long getScmheartbeatCheckerInterval(Configuration conf) {
- return conf.getLong(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS,
- ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS_DEFAULT);
- }
-
- /**
- * Heartbeat Interval - Defines the heartbeat frequency from a datanode to
- * SCM.
- *
- * @param conf - Ozone Config
- * @return - HB interval in seconds.
- */
- public static long getScmHeartbeatInterval(Configuration conf) {
- return conf.getTimeDuration(
- OZONE_SCM_HEARTBEAT_INTERVAL_SECONDS,
- ScmConfigKeys.OZONE_SCM_HEARBEAT_INTERVAL_SECONDS_DEFAULT,
- TimeUnit.SECONDS);
- }
-
- /**
- * Get the Stale Node interval, which is used by SCM to flag a datanode as
- * stale, if the heartbeat from that node has been missing for this duration.
- *
- * @param conf - Configuration.
- * @return - Long, Milliseconds to wait before flagging a node as stale.
- */
- public static long getStaleNodeInterval(Configuration conf) {
-
- long staleNodeIntevalMs = conf.getLong(OZONE_SCM_STALENODE_INTERVAL_MS,
- OZONE_SCM_STALENODE_INTERVAL_DEFAULT);
-
- long heartbeatThreadFrequencyMs = getScmheartbeatCheckerInterval(conf);
-
- long heartbeatIntervalMs = getScmHeartbeatInterval(conf) * 1000;
-
-
- // Make sure that StaleNodeInterval is configured way above the frequency
- // at which we run the heartbeat thread.
- //
- // Here we check that staleNodeInterval is at least five times more than the
- // frequency at which the accounting thread is going to run.
- try {
- sanitizeUserArgs(staleNodeIntevalMs, heartbeatThreadFrequencyMs, 5, 1000);
- } catch (IllegalArgumentException ex) {
- LOG.error("Stale Node Interval MS is cannot be honored due to " +
- "mis-configured {}. ex: {}",
- OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS, ex);
- throw ex;
- }
-
- // Make sure that stale node value is greater than configured value that
- // datanodes are going to send HBs.
- try {
- sanitizeUserArgs(staleNodeIntevalMs, heartbeatIntervalMs, 3, 1000);
- } catch (IllegalArgumentException ex) {
- LOG.error("Stale Node Interval MS is cannot be honored due to " +
- "mis-configured {}. ex: {}",
- OZONE_SCM_HEARTBEAT_INTERVAL_SECONDS, ex);
- throw ex;
- }
- return staleNodeIntevalMs;
- }
-
- /**
- * Gets the interval for dead node flagging. This has to be a value that is
- * greater than stale node value, and by transitive relation we also know
- * that this value is greater than heartbeat interval and heartbeatProcess
- * Interval.
- *
- * @param conf - Configuration.
- * @return - the interval for dead node flagging.
- */
- public static long getDeadNodeInterval(Configuration conf) {
- long staleNodeIntervalMs = getStaleNodeInterval(conf);
- long deadNodeIntervalMs = conf.getLong(
- OZONE_SCM_DEADNODE_INTERVAL_MS, OZONE_SCM_DEADNODE_INTERVAL_DEFAULT);
-
- try {
- // Make sure that dead nodes Ms is at least twice the time for staleNodes
- // with a max of 1000 times the staleNodes.
- sanitizeUserArgs(deadNodeIntervalMs, staleNodeIntervalMs, 2, 1000);
- } catch (IllegalArgumentException ex) {
- LOG.error("Dead Node Interval MS is cannot be honored due to " +
- "mis-configured {}. ex: {}",
- OZONE_SCM_STALENODE_INTERVAL_MS, ex);
- throw ex;
- }
- return deadNodeIntervalMs;
- }
-
- /**
- * Returns the maximum number of heartbeat to process per loop of the process
- * thread.
- * @param conf Configuration
- * @return - int -- Number of HBs to process
- */
- public static int getMaxHBToProcessPerLoop(Configuration conf) {
- return conf.getInt(ScmConfigKeys.OZONE_SCM_MAX_HB_COUNT_TO_PROCESS,
- ScmConfigKeys.OZONE_SCM_MAX_HB_COUNT_TO_PROCESS_DEFAULT);
- }
-
- /**
- * Timeout value for the RPC from Datanode to SCM, primarily used for
- * Heartbeats and container reports.
- *
- * @param conf - Ozone Config
- * @return - Rpc timeout in Milliseconds.
- */
- public static long getScmRpcTimeOutInMilliseconds(Configuration conf) {
- return conf.getTimeDuration(OZONE_SCM_HEARTBEAT_RPC_TIMEOUT,
- OZONE_SCM_HEARTBEAT_RPC_TIMEOUT_DEFAULT, TimeUnit.MILLISECONDS);
- }
-
- /**
- * Log Warn interval.
- *
- * @param conf - Ozone Config
- * @return - Log warn interval.
- */
- public static int getLogWarnInterval(Configuration conf) {
- return conf.getInt(OZONE_SCM_HEARTBEAT_LOG_WARN_INTERVAL_COUNT,
- OZONE_SCM_HEARTBEAT_LOG_WARN_DEFAULT);
- }
-
- /**
- * returns the Container port.
- * @param conf - Conf
- * @return port number.
- */
- public static int getContainerPort(Configuration conf) {
- return conf.getInt(ScmConfigKeys.DFS_CONTAINER_IPC_PORT, ScmConfigKeys
- .DFS_CONTAINER_IPC_PORT_DEFAULT);
- }
-
- /**
- * After starting an RPC server, updates configuration with the actual
- * listening address of that server. The listening address may be different
- * from the configured address if, for example, the configured address uses
- * port 0 to request use of an ephemeral port.
- *
- * @param conf configuration to update
- * @param rpcAddressKey configuration key for RPC server address
- * @param addr configured address
- * @param rpcServer started RPC server.
- */
- public static InetSocketAddress updateRPCListenAddress(
- OzoneConfiguration conf, String rpcAddressKey,
- InetSocketAddress addr, RPC.Server rpcServer) {
- return updateListenAddress(conf, rpcAddressKey, addr,
- rpcServer.getListenerAddress());
- }
-
- /**
- * After starting an server, updates configuration with the actual
- * listening address of that server. The listening address may be different
- * from the configured address if, for example, the configured address uses
- * port 0 to request use of an ephemeral port.
- *
- * @param conf configuration to update
- * @param addressKey configuration key for RPC server address
- * @param addr configured address
- * @param listenAddr the real listening address.
- */
- public static InetSocketAddress updateListenAddress(OzoneConfiguration conf,
- String addressKey, InetSocketAddress addr, InetSocketAddress listenAddr) {
- InetSocketAddress updatedAddr = new InetSocketAddress(addr.getHostString(),
- listenAddr.getPort());
- conf.set(addressKey,
- addr.getHostString() + ":" + listenAddr.getPort());
- return updatedAddr;
- }
-
- /**
- * Releases a http connection if the request is not null.
- * @param request
- */
- public static void releaseConnection(HttpRequestBase request) {
- if (request != null) {
- request.releaseConnection();
- }
- }
-
- /**
- * @return a default instance of {@link CloseableHttpClient}.
- */
- public static CloseableHttpClient newHttpClient() {
- return OzoneClientUtils.newHttpClient(null);
- }
-
- /**
- * Returns a {@link CloseableHttpClient} configured by given configuration.
- * If conf is null, returns a default instance.
- *
- * @param conf configuration
- * @return a {@link CloseableHttpClient} instance.
- */
- public static CloseableHttpClient newHttpClient(Configuration conf) {
- int socketTimeout = OzoneConfigKeys
- .OZONE_CLIENT_SOCKET_TIMEOUT_MS_DEFAULT;
- int connectionTimeout = OzoneConfigKeys
- .OZONE_CLIENT_CONNECTION_TIMEOUT_MS_DEFAULT;
- if (conf != null) {
- socketTimeout = conf.getInt(
- OzoneConfigKeys.OZONE_CLIENT_SOCKET_TIMEOUT_MS,
- OzoneConfigKeys.OZONE_CLIENT_SOCKET_TIMEOUT_MS_DEFAULT);
- connectionTimeout = conf.getInt(
- OzoneConfigKeys.OZONE_CLIENT_CONNECTION_TIMEOUT_MS,
- OzoneConfigKeys.OZONE_CLIENT_CONNECTION_TIMEOUT_MS_DEFAULT);
- }
-
- CloseableHttpClient client = HttpClients.custom()
- .setDefaultRequestConfig(
- RequestConfig.custom()
- .setSocketTimeout(socketTimeout)
- .setConnectTimeout(connectionTimeout)
- .build())
- .build();
- return client;
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org