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 2018/03/15 16:33:05 UTC
[50/51] [partial] hadoop git commit: HDFS-13258. Ozone: restructure
Hdsl/Ozone code to separated maven subprojects. Contributed by Elek Marton,
Mukul Kumar Singh, Xiaoyu Yao, Ajay Kumar, Anu Engineer, Lokesh Jain,
Nanda Kumar.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/pom.xml b/hadoop-cblock/server/pom.xml
new file mode 100644
index 0000000..17af85d
--- /dev/null
+++ b/hadoop-cblock/server/pom.xml
@@ -0,0 +1,169 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed 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. See accompanying LICENSE file.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-cblock</artifactId>
+ <version>3.2.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>hadoop-cblock-server</artifactId>
+ <version>3.2.0-SNAPSHOT</version>
+ <description>Apache Hadoop CBlock Server</description>
+ <name>Apache Hadoop CBlock Server</name>
+ <packaging>jar</packaging>
+
+ <properties>
+ <hadoop.component>cblock</hadoop.component>
+ <is.hadoop.component>true</is.hadoop.component>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdsl-server-framework</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-ozone-common</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdsl-common</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdsl-client</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-ozone-integration-test</artifactId>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.jscsi</groupId>
+ <artifactId>target</artifactId>
+ <version>2.6.0</version>
+ <optional>true</optional>
+ <exclusions>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>io.kubernetes</groupId>
+ <artifactId>client-java</artifactId>
+ <version>1.0.0-beta1</version>
+ <exclusions>
+ <exclusion>
+ <groupId>io.swagger</groupId>
+ <artifactId>swagger-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.github.stefanbirkner</groupId>
+ <artifactId>system-rules</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ </dependencies>
+
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <exclude>src/test/resources/dynamicprovisioner/expected1-pv.json</exclude>
+ <exclude>src/test/resources/dynamicprovisioner/input1-pvc.json</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-maven-plugins</artifactId>
+ <executions>
+ <execution>
+ <id>compile-protoc</id>
+ <goals>
+ <goal>protoc</goal>
+ </goals>
+ <configuration>
+ <protocVersion>${protobuf.version}</protocVersion>
+ <protocCommand>${protoc.path}</protocCommand>
+ <imports>
+ <param>
+ ${basedir}/../../hadoop-common-project/hadoop-common/src/main/proto
+ </param>
+ <param>
+ ${basedir}/../../hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/
+ </param>
+ <param>
+ ${basedir}/../../hadoop-hdfs-project/hadoop-hdfs/src/main/proto/
+ </param>
+ <param>
+ ${basedir}/../../hadoop-hdsl/common/src/main/proto/
+ </param>
+ <param>${basedir}/src/main/proto</param>
+ </imports>
+ <source>
+ <directory>${basedir}/src/main/proto</directory>
+ <includes>
+ <include>CBlockClientServerProtocol.proto</include>
+ <include>CBlockServiceProtocol.proto</include>
+ </includes>
+ </source>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>findbugs-maven-plugin</artifactId>
+ <configuration>
+ <excludeFilterFile>${basedir}/dev-support/findbugsExcludeFile.xml</excludeFilterFile>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java
new file mode 100644
index 0000000..ec5d7d2
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockConfigKeys.java
@@ -0,0 +1,222 @@
+/*
+ * 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.cblock.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.queue.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 =
+ "dfs.cblock.cache.keep.alive";
+ public static final String DFS_CBLOCK_CACHE_KEEP_ALIVE_DEFAULT = "60s";
+
+ /**
+ * 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 =
+ "dfs.cblock.block.buffer.flush.interval";
+ public static final String DFS_CBLOCK_BLOCK_BUFFER_FLUSH_INTERVAL_DEFAULT =
+ "60s";
+
+ // 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 =
+ "0.0.0.0";
+ 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.gb";
+ 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;
+
+ /**
+ * Cblock CLI configs.
+ */
+ public static final String DFS_CBLOCK_MANAGER_POOL_SIZE =
+ "dfs.cblock.manager.pool.size";
+ public static final int DFS_CBLOCK_MANAGER_POOL_SIZE_DEFAULT = 16;
+
+ /**
+ * currently the largest supported volume is about 8TB, which might take
+ * > 20 seconds to finish creating containers. thus set timeout to 30 sec.
+ */
+ public static final String DFS_CBLOCK_RPC_TIMEOUT =
+ "dfs.cblock.rpc.timeout";
+ public static final String DFS_CBLOCK_RPC_TIMEOUT_DEFAULT = "300s";
+
+ public static final String DFS_CBLOCK_ISCSI_ADVERTISED_IP =
+ "dfs.cblock.iscsi.advertised.ip";
+
+ public static final String DFS_CBLOCK_ISCSI_ADVERTISED_PORT =
+ "dfs.cblock.iscsi.advertised.port";
+
+ public static final int DFS_CBLOCK_ISCSI_ADVERTISED_PORT_DEFAULT = 3260;
+
+
+ public static final String
+ DFS_CBLOCK_KUBERNETES_DYNAMIC_PROVISIONER_ENABLED
+ = "dfs.cblock.kubernetes.dynamic-provisioner.enabled";
+
+ public static final boolean
+ DFS_CBLOCK_KUBERNETES_DYNAMIC_PROVISIONER_ENABLED_DEFAULT = false;
+
+ public static final String
+ DFS_CBLOCK_KUBERNETES_CBLOCK_USER =
+ "dfs.cblock.kubernetes.cblock-user";
+
+ public static final String
+ DFS_CBLOCK_KUBERNETES_CBLOCK_USER_DEFAULT =
+ "iqn.2001-04.org.apache.hadoop";
+
+ public static final String
+ DFS_CBLOCK_KUBERNETES_CONFIG_FILE_KEY =
+ "dfs.cblock.kubernetes.configfile";
+
+ private CBlockConfigKeys() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockManager.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
new file mode 100644
index 0000000..788d01e
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
@@ -0,0 +1,426 @@
+/*
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.BlockingService;
+import org.apache.hadoop.cblock.kubernetes.DynamicProvisioner;
+import org.apache.hadoop.cblock.meta.VolumeDescriptor;
+import org.apache.hadoop.cblock.meta.VolumeInfo;
+import org.apache.hadoop.cblock.proto.CBlockClientProtocol;
+import org.apache.hadoop.cblock.proto.CBlockServiceProtocol;
+import org.apache.hadoop.cblock.proto.MountVolumeResponse;
+import org.apache.hadoop.cblock.protocol.proto
+ .CBlockClientServerProtocolProtos;
+import org.apache.hadoop.cblock.protocol.proto.CBlockServiceProtocolProtos;
+import org.apache.hadoop.cblock.protocolPB.CBlockClientServerProtocolPB;
+import org.apache.hadoop.cblock.protocolPB
+ .CBlockClientServerProtocolServerSideTranslatorPB;
+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.OzoneConfigKeys;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.scm.XceiverClientManager;
+import org.apache.hadoop.scm.client.ContainerOperationClient;
+import org.apache.hadoop.scm.client.ScmClient;
+import org.apache.hadoop.cblock.storage.StorageManager;
+import org.apache.hadoop.cblock.util.KeyUtil;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.hdsl.conf.OzoneConfiguration;
+import org.apache.hadoop.scm.protocolPB
+ .StorageContainerLocationProtocolClientSideTranslatorPB;
+import org.apache.hadoop.scm.protocolPB.StorageContainerLocationProtocolPB;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.utils.LevelDBStore;
+
+import static org.apache.hadoop.cblock.CblockUtils.getCblockServerRpcAddr;
+import static org.apache.hadoop.cblock.CblockUtils.getCblockServiceRpcAddr;
+import static org.apache.hadoop.ozone.web.util.ServerUtils
+ .updateRPCListenAddress;
+import org.iq80.leveldb.DBIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_CONTAINER_SIZE_GB_DEFAULT;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_CONTAINER_SIZE_GB_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_JSCSIRPC_ADDRESS_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_JSCSIRPC_BIND_HOST_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SCM_IPADDRESS_DEFAULT;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SCM_IPADDRESS_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SCM_PORT_DEFAULT;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SCM_PORT_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SERVICERPC_ADDRESS_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SERVICERPC_BIND_HOST_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_DEFAULT;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SERVICE_LEVELDB_PATH_DEFAULT;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_SERVICE_LEVELDB_PATH_KEY;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_KUBERNETES_DYNAMIC_PROVISIONER_ENABLED;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_KUBERNETES_DYNAMIC_PROVISIONER_ENABLED_DEFAULT;
+
+
+/**
+ * The main entry point of CBlock operations, ALL the CBlock operations
+ * will go through this class. But NOTE that:
+ *
+ * volume operations (create/
+ * delete/info) are:
+ * client -> CBlockManager -> StorageManager -> CBlock client
+ *
+ * IO operations (put/get block) are;
+ * client -> CBlock client -> container
+ *
+ */
+public class CBlockManager implements CBlockServiceProtocol,
+ CBlockClientProtocol {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(CBlockManager.class);
+
+ private final RPC.Server cblockService;
+ private final RPC.Server cblockServer;
+
+ private final StorageManager storageManager;
+
+ private final LevelDBStore levelDBStore;
+ private final String dbPath;
+
+ private final DynamicProvisioner kubernetesDynamicProvisioner;
+
+ private Charset encoding = Charset.forName("UTF-8");
+
+ public CBlockManager(OzoneConfiguration conf,
+ ScmClient storageClient) throws IOException {
+ // Fix the cBlockManagerId generattion code here. Should support
+ // cBlockManager --init command which will generate a cBlockManagerId and
+ // persist it locally.
+ storageManager =
+ new StorageManager(storageClient, conf, "CBLOCK");
+
+ dbPath = conf.getTrimmed(DFS_CBLOCK_SERVICE_LEVELDB_PATH_KEY,
+ DFS_CBLOCK_SERVICE_LEVELDB_PATH_DEFAULT);
+ levelDBStore = new LevelDBStore(new File(dbPath), true);
+ LOG.info("Try to load exising volume information");
+ readFromPersistentStore();
+
+ RPC.setProtocolEngine(conf, CBlockServiceProtocolPB.class,
+ ProtobufRpcEngine.class);
+ RPC.setProtocolEngine(conf, CBlockClientServerProtocolPB.class,
+ ProtobufRpcEngine.class);
+ // start service for client command-to-cblock server service
+ InetSocketAddress serviceRpcAddr =
+ getCblockServiceRpcAddr(conf);
+ BlockingService cblockProto =
+ CBlockServiceProtocolProtos
+ .CBlockServiceProtocolService
+ .newReflectiveBlockingService(
+ new CBlockServiceProtocolServerSideTranslatorPB(this)
+ );
+ cblockService = startRpcServer(conf, CBlockServiceProtocolPB.class,
+ cblockProto, serviceRpcAddr,
+ DFS_CBLOCK_SERVICERPC_BIND_HOST_KEY,
+ DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_KEY,
+ DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_DEFAULT);
+ InetSocketAddress cblockServiceRpcAddress =
+ updateRPCListenAddress(conf,
+ DFS_CBLOCK_SERVICERPC_ADDRESS_KEY, serviceRpcAddr, cblockService);
+ LOG.info("CBlock manager listening for client commands on: {}",
+ cblockServiceRpcAddress);
+ // now start service for cblock client-to-cblock server communication
+
+ InetSocketAddress serverRpcAddr =
+ getCblockServerRpcAddr(conf);
+ BlockingService serverProto =
+ CBlockClientServerProtocolProtos
+ .CBlockClientServerProtocolService
+ .newReflectiveBlockingService(
+ new CBlockClientServerProtocolServerSideTranslatorPB(this)
+ );
+ cblockServer = startRpcServer(
+ conf, CBlockClientServerProtocolPB.class,
+ serverProto, serverRpcAddr,
+ DFS_CBLOCK_JSCSIRPC_BIND_HOST_KEY,
+ DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_KEY,
+ DFS_CBLOCK_SERVICERPC_HANDLER_COUNT_DEFAULT);
+ InetSocketAddress cblockServerRpcAddress =
+ updateRPCListenAddress(conf,
+ DFS_CBLOCK_JSCSIRPC_ADDRESS_KEY, serverRpcAddr, cblockServer);
+ LOG.info("CBlock server listening for client commands on: {}",
+ cblockServerRpcAddress);
+
+ if (conf.getBoolean(DFS_CBLOCK_KUBERNETES_DYNAMIC_PROVISIONER_ENABLED,
+ DFS_CBLOCK_KUBERNETES_DYNAMIC_PROVISIONER_ENABLED_DEFAULT)) {
+
+ kubernetesDynamicProvisioner =
+ new DynamicProvisioner(conf, storageManager);
+ kubernetesDynamicProvisioner.init();
+
+ } else {
+ kubernetesDynamicProvisioner = null;
+ }
+ }
+
+ public void start() {
+ cblockService.start();
+ cblockServer.start();
+ if (kubernetesDynamicProvisioner != null) {
+ kubernetesDynamicProvisioner.start();
+ }
+ LOG.info("CBlock manager started!");
+ }
+
+ public void stop() {
+ cblockService.stop();
+ cblockServer.stop();
+ if (kubernetesDynamicProvisioner != null) {
+ kubernetesDynamicProvisioner.stop();
+ }
+ }
+
+ public void join() {
+ try {
+ cblockService.join();
+ cblockServer.join();
+ } catch (InterruptedException e) {
+ LOG.error("Interrupted during join");
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Starts an RPC server, if configured.
+ *
+ * @param conf configuration
+ * @param protocol RPC protocol provided by RPC server
+ * @param instance RPC protocol implementation instance
+ * @param addr configured address of RPC server
+ * @param bindHostKey configuration key for setting explicit bind host. If
+ * the property is not configured, then the bind host is taken from addr.
+ * @param handlerCountKey configuration key for RPC server handler count
+ * @param handlerCountDefault default RPC server handler count if unconfigured
+ * @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(OzoneConfiguration conf,
+ Class<?> protocol, BlockingService instance,
+ InetSocketAddress addr, String bindHostKey,
+ String handlerCountKey, int handlerCountDefault) throws IOException {
+ if (addr == null) {
+ return null;
+ }
+ String bindHost = conf.getTrimmed(bindHostKey);
+ if (bindHost == null || bindHost.isEmpty()) {
+ bindHost = addr.getHostName();
+ }
+ int numHandlers = conf.getInt(handlerCountKey, handlerCountDefault);
+ RPC.Server rpcServer = new RPC.Builder(conf)
+ .setProtocol(protocol)
+ .setInstance(instance)
+ .setBindAddress(bindHost)
+ .setPort(addr.getPort())
+ .setNumHandlers(numHandlers)
+ .setVerbose(false)
+ .setSecretManager(null)
+ .build();
+ return rpcServer;
+ }
+
+ @Override
+ public synchronized MountVolumeResponse mountVolume(
+ String userName, String volumeName) throws IOException {
+ return storageManager.isVolumeValid(userName, volumeName);
+ }
+
+ @Override
+ public List<VolumeInfo> listVolumes() throws IOException {
+ return listVolume(null);
+ }
+
+ @Override
+ public synchronized void createVolume(String userName, String volumeName,
+ long volumeSize, int blockSize) throws IOException {
+ LOG.info("Create volume received: userName: {} volumeName: {} " +
+ "volumeSize: {} blockSize: {}", userName, volumeName,
+ volumeSize, blockSize);
+ // It is important to create in-memory representation of the
+ // volume first, then writes to persistent storage (levelDB)
+ // such that it is guaranteed that when there is an entry in
+ // levelDB, the volume is allocated. (more like a UNDO log fashion)
+ // TODO: what if creation failed? we allocated containers but lost
+ // the reference to the volume and all it's containers. How to release
+ // the containers?
+ storageManager.createVolume(userName, volumeName, volumeSize, blockSize);
+ VolumeDescriptor volume = storageManager.getVolume(userName, volumeName);
+ if (volume == null) {
+ throw new IOException("Volume creation failed!");
+ }
+ String volumeKey = KeyUtil.getVolumeKey(userName, volumeName);
+ writeToPersistentStore(volumeKey.getBytes(encoding),
+ volume.toProtobuf().toByteArray());
+ }
+
+ @Override
+ public synchronized void deleteVolume(String userName,
+ String volumeName, boolean force) throws IOException {
+ LOG.info("Delete volume received: volume: {} {} ", volumeName, force);
+ storageManager.deleteVolume(userName, volumeName, force);
+ // being here means volume is successfully deleted now
+ String volumeKey = KeyUtil.getVolumeKey(userName, volumeName);
+ removeFromPersistentStore(volumeKey.getBytes(encoding));
+ }
+
+ // No need to synchronize on the following three methods, since write and
+ // remove's caller are synchronized. read's caller is the constructor and
+ // no other method call can happen at that time.
+ @VisibleForTesting
+ public void writeToPersistentStore(byte[] key, byte[] value) {
+ levelDBStore.put(key, value);
+ }
+
+ @VisibleForTesting
+ public void removeFromPersistentStore(byte[] key) {
+ levelDBStore.delete(key);
+ }
+
+ public void readFromPersistentStore() throws IOException {
+ try (DBIterator iter = levelDBStore.getIterator()) {
+ iter.seekToFirst();
+ while (iter.hasNext()) {
+ Map.Entry<byte[], byte[]> entry = iter.next();
+ String volumeKey = new String(entry.getKey(), encoding);
+ try {
+ VolumeDescriptor volumeDescriptor =
+ VolumeDescriptor.fromProtobuf(entry.getValue());
+ storageManager.addVolume(volumeDescriptor);
+ } catch (IOException e) {
+ LOG.error("Loading volume " + volumeKey + " error " + e);
+ }
+ }
+ }
+ }
+
+ @Override
+ public synchronized VolumeInfo infoVolume(String userName, String volumeName
+ ) throws IOException {
+ LOG.info("Info volume received: volume: {}", volumeName);
+ return storageManager.infoVolume(userName, volumeName);
+ }
+
+ @VisibleForTesting
+ public synchronized List<VolumeDescriptor> getAllVolumes() {
+ return storageManager.getAllVolume(null);
+ }
+
+ public synchronized List<VolumeDescriptor> getAllVolumes(String userName) {
+ return storageManager.getAllVolume(userName);
+ }
+
+ public synchronized void close() {
+ try {
+ levelDBStore.close();
+ } catch (IOException e) {
+ LOG.error("Error when closing levelDB " + e);
+ }
+ }
+
+ public synchronized void clean() {
+ try {
+ levelDBStore.close();
+ levelDBStore.destroy();
+ } catch (IOException e) {
+ LOG.error("Error when deleting levelDB " + e);
+ }
+ }
+
+ @Override
+ public synchronized List<VolumeInfo> listVolume(String userName)
+ throws IOException {
+ ArrayList<VolumeInfo> response = new ArrayList<>();
+ List<VolumeDescriptor> allVolumes =
+ storageManager.getAllVolume(userName);
+ for (VolumeDescriptor volume : allVolumes) {
+ VolumeInfo info =
+ new VolumeInfo(volume.getUserName(), volume.getVolumeName(),
+ volume.getVolumeSize(), volume.getBlockSize());
+ response.add(info);
+ }
+ return response;
+ }
+
+ public static void main(String[] args) throws Exception {
+ long version = RPC.getProtocolVersion(
+ StorageContainerLocationProtocolPB.class);
+ CblockUtils.activateConfigs();
+ OzoneConfiguration ozoneConf = new OzoneConfiguration();
+ String scmAddress = ozoneConf.get(DFS_CBLOCK_SCM_IPADDRESS_KEY,
+ DFS_CBLOCK_SCM_IPADDRESS_DEFAULT);
+ int scmPort = ozoneConf.getInt(DFS_CBLOCK_SCM_PORT_KEY,
+ DFS_CBLOCK_SCM_PORT_DEFAULT);
+ int containerSizeGB = ozoneConf.getInt(DFS_CBLOCK_CONTAINER_SIZE_GB_KEY,
+ DFS_CBLOCK_CONTAINER_SIZE_GB_DEFAULT);
+ ContainerOperationClient.setContainerSizeB(containerSizeGB* OzoneConsts.GB);
+ InetSocketAddress address = new InetSocketAddress(scmAddress, scmPort);
+
+ ozoneConf.set(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT,
+ OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT_DEFAULT);
+ LOG.info(
+ "Creating StorageContainerLocationProtocol RPC client with address {}",
+ address);
+ RPC.setProtocolEngine(ozoneConf, StorageContainerLocationProtocolPB.class,
+ ProtobufRpcEngine.class);
+ StorageContainerLocationProtocolClientSideTranslatorPB client =
+ new StorageContainerLocationProtocolClientSideTranslatorPB(
+ RPC.getProxy(StorageContainerLocationProtocolPB.class, version,
+ address, UserGroupInformation.getCurrentUser(), ozoneConf,
+ NetUtils.getDefaultSocketFactory(ozoneConf),
+ Client.getRpcTimeout(ozoneConf)));
+ ScmClient storageClient = new ContainerOperationClient(
+ client, new XceiverClientManager(ozoneConf));
+ CBlockManager cbm = new CBlockManager(ozoneConf, storageClient);
+ cbm.start();
+ cbm.join();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CblockUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CblockUtils.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CblockUtils.java
new file mode 100644
index 0000000..99ffde0
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/CblockUtils.java
@@ -0,0 +1,129 @@
+/**
+ * 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.cblock;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.net.NetUtils;
+
+import com.google.common.base.Optional;
+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.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.hdsl.HdslUtils.getHostNameFromConfigKeys;
+import static org.apache.hadoop.hdsl.HdslUtils.getPortNumberFromConfigKeys;
+
+/**
+ * Generic stateless utility functions for CBlock components.
+ */
+public class CblockUtils {
+
+ private CblockUtils() {
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * Parse size with size prefix string and return in bytes.
+ *
+ */
+ public static long parseSize(String volumeSizeArgs) throws IOException {
+ long multiplier = 1;
+
+ Pattern p = Pattern.compile("([0-9]+)([a-zA-Z]+)");
+ Matcher m = p.matcher(volumeSizeArgs);
+
+ if (!m.find()) {
+ throw new IOException("Invalid volume size args " + volumeSizeArgs);
+ }
+
+ int size = Integer.parseInt(m.group(1));
+ String s = m.group(2);
+
+ if (s.equalsIgnoreCase("MB") ||
+ s.equalsIgnoreCase("Mi")) {
+ multiplier = 1024L * 1024;
+ } else if (s.equalsIgnoreCase("GB") ||
+ s.equalsIgnoreCase("Gi")) {
+ multiplier = 1024L * 1024 * 1024;
+ } else if (s.equalsIgnoreCase("TB") ||
+ s.equalsIgnoreCase("Ti")) {
+ multiplier = 1024L * 1024 * 1024 * 1024;
+ } else {
+ throw new IOException("Invalid volume size args " + volumeSizeArgs);
+ }
+ return size * multiplier;
+ }
+
+ public static void activateConfigs(){
+ Configuration.addDefaultResource("hdfs-default.xml");
+ Configuration.addDefaultResource("hdfs-site.xml");
+ Configuration.addDefaultResource("ozone-default.xml");
+ Configuration.addDefaultResource("ozone-site.xml");
+ Configuration.addDefaultResource("cblock-default.xml");
+ Configuration.addDefaultResource("cblock-site.xml");
+
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockServiceProtocolClientSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockServiceProtocolClientSideTranslatorPB.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockServiceProtocolClientSideTranslatorPB.java
new file mode 100644
index 0000000..ec38c9c
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockServiceProtocolClientSideTranslatorPB.java
@@ -0,0 +1,135 @@
+/*
+ * 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.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.cblock.meta.VolumeInfo;
+import org.apache.hadoop.cblock.proto.CBlockServiceProtocol;
+import org.apache.hadoop.cblock.protocol.proto.CBlockServiceProtocolProtos;
+import org.apache.hadoop.cblock.protocolPB.CBlockServiceProtocolPB;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.ipc.ProtobufHelper;
+import org.apache.hadoop.ipc.ProtocolTranslator;
+import org.apache.hadoop.ipc.RPC;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The client side implement of CBlockServiceProtocol.
+ */
+@InterfaceAudience.Private
+public final class CBlockServiceProtocolClientSideTranslatorPB
+ implements CBlockServiceProtocol, ProtocolTranslator, Closeable {
+
+ private final CBlockServiceProtocolPB rpcProxy;
+
+ public CBlockServiceProtocolClientSideTranslatorPB(
+ CBlockServiceProtocolPB rpcProxy) {
+ this.rpcProxy = rpcProxy;
+ }
+
+ @Override
+ public void close() throws IOException {
+ RPC.stopProxy(rpcProxy);
+ }
+
+ @Override
+ public void createVolume(String userName, String volumeName,
+ long volumeSize, int blockSize) throws IOException {
+ CBlockServiceProtocolProtos.CreateVolumeRequestProto.Builder req =
+ CBlockServiceProtocolProtos.CreateVolumeRequestProto.newBuilder();
+ req.setUserName(userName);
+ req.setVolumeName(volumeName);
+ req.setVolumeSize(volumeSize);
+ req.setBlockSize(blockSize);
+ try {
+ rpcProxy.createVolume(null, req.build());
+ } catch (ServiceException e) {
+ throw ProtobufHelper.getRemoteException(e);
+ }
+ }
+
+ @Override
+ public void deleteVolume(String userName, String volumeName, boolean force)
+ throws IOException {
+ CBlockServiceProtocolProtos.DeleteVolumeRequestProto.Builder req =
+ CBlockServiceProtocolProtos.DeleteVolumeRequestProto.newBuilder();
+ req.setUserName(userName);
+ req.setVolumeName(volumeName);
+ req.setForce(force);
+ try {
+ rpcProxy.deleteVolume(null, req.build());
+ } catch (ServiceException e) {
+ throw ProtobufHelper.getRemoteException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingProxyObject() {
+ return rpcProxy;
+ }
+
+ @Override
+ public VolumeInfo infoVolume(String userName, String volumeName)
+ throws IOException {
+ CBlockServiceProtocolProtos.InfoVolumeRequestProto.Builder req =
+ CBlockServiceProtocolProtos.InfoVolumeRequestProto.newBuilder();
+ req.setUserName(userName);
+ req.setVolumeName(volumeName);
+ try {
+ CBlockServiceProtocolProtos.InfoVolumeResponseProto resp =
+ rpcProxy.infoVolume(null, req.build());
+ return new VolumeInfo(resp.getVolumeInfo().getUserName(),
+ resp.getVolumeInfo().getVolumeName(),
+ resp.getVolumeInfo().getVolumeSize(),
+ resp.getVolumeInfo().getBlockSize(),
+ resp.getVolumeInfo().getUsage());
+ } catch (ServiceException e) {
+ throw ProtobufHelper.getRemoteException(e);
+ }
+ }
+
+ @Override
+ public List<VolumeInfo> listVolume(String userName) throws IOException {
+ CBlockServiceProtocolProtos.ListVolumeRequestProto.Builder req =
+ CBlockServiceProtocolProtos.ListVolumeRequestProto.newBuilder();
+ if (userName != null) {
+ req.setUserName(userName);
+ }
+ try {
+ CBlockServiceProtocolProtos.ListVolumeResponseProto resp =
+ rpcProxy.listVolume(null, req.build());
+ List<VolumeInfo> respList = new ArrayList<>();
+ for (CBlockServiceProtocolProtos.VolumeInfoProto entry :
+ resp.getVolumeEntryList()) {
+ VolumeInfo volumeInfo = new VolumeInfo(
+ entry.getUserName(), entry.getVolumeName(), entry.getVolumeSize(),
+ entry.getBlockSize());
+ respList.add(volumeInfo);
+ }
+ return respList;
+ } catch (ServiceException e) {
+ throw ProtobufHelper.getRemoteException(e);
+ } catch (Exception e) {
+ throw new IOException("got" + e.getCause() + " " + e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.java
new file mode 100644
index 0000000..4a3878a
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/CBlockVolumeClient.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.cblock.client;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.cblock.CBlockConfigKeys;
+import org.apache.hadoop.cblock.meta.VolumeInfo;
+import org.apache.hadoop.cblock.protocolPB.CBlockServiceProtocolPB;
+import org.apache.hadoop.hdsl.conf.OzoneConfiguration;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import static org.apache.hadoop.cblock.CblockUtils.getCblockServiceRpcAddr;
+
+/**
+ * Implementation of client used by CBlock command line tool.
+ */
+public class CBlockVolumeClient {
+ private final CBlockServiceProtocolClientSideTranslatorPB cblockClient;
+
+ public CBlockVolumeClient(OzoneConfiguration conf) throws IOException {
+ this(conf, null);
+ }
+
+ public CBlockVolumeClient(OzoneConfiguration conf,
+ InetSocketAddress serverAddress) throws IOException {
+ InetSocketAddress address = serverAddress != null ? serverAddress :
+ getCblockServiceRpcAddr(conf);
+ long version = RPC.getProtocolVersion(CBlockServiceProtocolPB.class);
+ int rpcTimeout = Math.toIntExact(
+ conf.getTimeDuration(CBlockConfigKeys.DFS_CBLOCK_RPC_TIMEOUT,
+ CBlockConfigKeys.DFS_CBLOCK_RPC_TIMEOUT_DEFAULT,
+ TimeUnit.MILLISECONDS));
+ cblockClient = new CBlockServiceProtocolClientSideTranslatorPB(
+ RPC.getProtocolProxy(CBlockServiceProtocolPB.class, version,
+ address, UserGroupInformation.getCurrentUser(), conf,
+ NetUtils.getDefaultSocketFactory(conf), rpcTimeout, RetryPolicies
+ .retryUpToMaximumCountWithFixedSleep(
+ 300, 1, TimeUnit.SECONDS)).getProxy());
+ }
+
+ public void createVolume(String userName, String volumeName,
+ long volumeSize, int blockSize) throws IOException {
+ cblockClient.createVolume(userName, volumeName,
+ volumeSize, blockSize);
+ }
+
+ public void deleteVolume(String userName, String volumeName, boolean force)
+ throws IOException {
+ cblockClient.deleteVolume(userName, volumeName, force);
+ }
+
+ public VolumeInfo infoVolume(String userName, String volumeName)
+ throws IOException {
+ return cblockClient.infoVolume(userName, volumeName);
+ }
+
+ public List<VolumeInfo> listVolume(String userName)
+ throws IOException {
+ return cblockClient.listVolume(userName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/package-info.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/package-info.java
new file mode 100644
index 0000000..761b71e
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/client/package-info.java
@@ -0,0 +1,18 @@
+/**
+ * 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.client;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/CBlockException.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/CBlockException.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/CBlockException.java
new file mode 100644
index 0000000..8f6b82b
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/CBlockException.java
@@ -0,0 +1,29 @@
+/*
+ * 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.exception;
+
+import java.io.IOException;
+
+/**
+ * The exception class used in CBlock.
+ */
+public class CBlockException extends IOException {
+ public CBlockException(String message) {
+ super(message);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/package-info.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/package-info.java
new file mode 100644
index 0000000..268b8cb
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/exception/package-info.java
@@ -0,0 +1,18 @@
+/**
+ * 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.exception;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/BlockWriterTask.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/BlockWriterTask.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/BlockWriterTask.java
new file mode 100644
index 0000000..04fe3a4
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/BlockWriterTask.java
@@ -0,0 +1,175 @@
+/**
+ * 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.cblock.jscsiHelper;
+
+import com.google.common.base.Preconditions;
+import com.google.common.primitives.Longs;
+import org.apache.hadoop.cblock.jscsiHelper.cache.LogicalBlock;
+import org.apache.hadoop.cblock.jscsiHelper.cache.impl.AsyncBlockWriter;
+import org.apache.hadoop.scm.XceiverClientSpi;
+import org.apache.hadoop.scm.container.common.helpers.Pipeline;
+import org.apache.hadoop.scm.storage.ContainerProtocolCalls;
+import org.apache.hadoop.util.Time;
+import org.apache.hadoop.utils.LevelDBStore;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Paths;
+
+/**
+ * The blockWriter task.
+ */
+public class BlockWriterTask implements Runnable {
+ private final LogicalBlock block;
+ private int tryCount;
+ private final ContainerCacheFlusher flusher;
+ private final String dbPath;
+ private final String fileName;
+ private final int maxRetryCount;
+
+ /**
+ * Constructs a BlockWriterTask.
+ *
+ * @param block - Block Information.
+ * @param flusher - ContainerCacheFlusher.
+ */
+ public BlockWriterTask(LogicalBlock block, ContainerCacheFlusher flusher,
+ String dbPath, int tryCount, String fileName, int maxRetryCount) {
+ this.block = block;
+ this.flusher = flusher;
+ this.dbPath = dbPath;
+ this.tryCount = tryCount;
+ this.fileName = fileName;
+ this.maxRetryCount = maxRetryCount;
+ }
+
+ /**
+ * When an object implementing interface <code>Runnable</code> is used
+ * to create a thread, starting the thread causes the object's
+ * <code>run</code> method to be called in that separately executing
+ * thread.
+ * <p>
+ * The general contract of the method <code>run</code> is that it may
+ * take any action whatsoever.
+ *
+ * @see Thread#run()
+ */
+ @Override
+ public void run() {
+ String containerName = null;
+ XceiverClientSpi client = null;
+ LevelDBStore levelDBStore = null;
+ String traceID = flusher.getTraceID(new File(dbPath), block.getBlockID());
+ flusher.getLOG().debug(
+ "Writing block to remote. block ID: {}", block.getBlockID());
+ try {
+ incTryCount();
+ Pipeline pipeline = flusher.getPipeline(this.dbPath, block.getBlockID());
+ client = flusher.getXceiverClientManager().acquireClient(pipeline);
+ containerName = pipeline.getContainerName();
+ byte[] keybuf = Longs.toByteArray(block.getBlockID());
+ byte[] data;
+ long startTime = Time.monotonicNow();
+ levelDBStore = flusher.getCacheDB(this.dbPath);
+ data = levelDBStore.get(keybuf);
+ Preconditions.checkNotNull(data);
+ long endTime = Time.monotonicNow();
+ Preconditions.checkState(data.length > 0, "Block data is zero length");
+ startTime = Time.monotonicNow();
+ ContainerProtocolCalls.writeSmallFile(client, containerName,
+ Long.toString(block.getBlockID()), data, traceID);
+ endTime = Time.monotonicNow();
+ flusher.getTargetMetrics().updateContainerWriteLatency(
+ endTime - startTime);
+ flusher.getLOG().debug("Time taken for Write Small File : {} ms",
+ endTime - startTime);
+
+ flusher.incrementRemoteIO();
+
+ } catch (Exception ex) {
+ flusher.getLOG().error("Writing of block:{} failed, We have attempted " +
+ "to write this block {} times to the container {}.Trace ID:{}",
+ block.getBlockID(), this.getTryCount(), containerName, traceID, ex);
+ writeRetryBlock(block);
+ if (ex instanceof IOException) {
+ flusher.getTargetMetrics().incNumWriteIOExceptionRetryBlocks();
+ } else {
+ flusher.getTargetMetrics().incNumWriteGenericExceptionRetryBlocks();
+ }
+ if (this.getTryCount() >= maxRetryCount) {
+ flusher.getTargetMetrics().incNumWriteMaxRetryBlocks();
+ }
+ } finally {
+ flusher.incFinishCount(fileName);
+ if (levelDBStore != null) {
+ flusher.releaseCacheDB(dbPath);
+ }
+ if(client != null) {
+ flusher.getXceiverClientManager().releaseClient(client);
+ }
+ }
+ }
+
+
+ private void writeRetryBlock(LogicalBlock currentBlock) {
+ boolean append = false;
+ String retryFileName =
+ String.format("%s.%d.%s.%s", AsyncBlockWriter.RETRY_LOG_PREFIX,
+ currentBlock.getBlockID(), Time.monotonicNow(), tryCount);
+ File logDir = new File(this.dbPath);
+ if (!logDir.exists() && !logDir.mkdirs()) {
+ flusher.getLOG().error(
+ "Unable to create the log directory, Critical error cannot continue");
+ return;
+ }
+ String log = Paths.get(this.dbPath, retryFileName).toString();
+ ByteBuffer buffer = ByteBuffer.allocate(Long.SIZE / Byte.SIZE);
+ buffer.putLong(currentBlock.getBlockID());
+ buffer.flip();
+ try {
+ FileChannel channel = new FileOutputStream(log, append).getChannel();
+ channel.write(buffer);
+ channel.close();
+ flusher.processDirtyBlocks(this.dbPath, retryFileName);
+ } catch (IOException e) {
+ flusher.getTargetMetrics().incNumFailedRetryLogFileWrites();
+ flusher.getLOG().error("Unable to write the retry block. Block ID: {}",
+ currentBlock.getBlockID(), e);
+ }
+ }
+
+ /**
+ * Increments the try count. This is done each time we try this block
+ * write to the container.
+ */
+ private void incTryCount() {
+ tryCount++;
+ }
+
+ /**
+ * Get the retry count.
+ *
+ * @return int
+ */
+ public int getTryCount() {
+ return tryCount;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockClientProtocolClientSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockClientProtocolClientSideTranslatorPB.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockClientProtocolClientSideTranslatorPB.java
new file mode 100644
index 0000000..84b68e3
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockClientProtocolClientSideTranslatorPB.java
@@ -0,0 +1,147 @@
+/*
+ * 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.jscsiHelper;
+
+import com.google.common.primitives.Longs;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.cblock.exception.CBlockException;
+import org.apache.hadoop.cblock.meta.VolumeInfo;
+import org.apache.hadoop.cblock.proto.CBlockClientProtocol;
+import org.apache.hadoop.cblock.proto.MountVolumeResponse;
+import org.apache.hadoop.cblock.protocol.proto
+ .CBlockClientServerProtocolProtos.ContainerIDProto;
+import org.apache.hadoop.cblock.protocol.proto
+ .CBlockClientServerProtocolProtos.ListVolumesRequestProto;
+import org.apache.hadoop.cblock.protocol.proto
+ .CBlockClientServerProtocolProtos.ListVolumesResponseProto;
+import org.apache.hadoop.cblock.protocol.proto
+ .CBlockClientServerProtocolProtos.MountVolumeRequestProto;
+import org.apache.hadoop.cblock.protocol.proto
+ .CBlockClientServerProtocolProtos.MountVolumeResponseProto;
+import org.apache.hadoop.cblock.protocol.proto.CBlockServiceProtocolProtos
+ .VolumeInfoProto;
+import org.apache.hadoop.cblock.protocolPB.CBlockClientServerProtocolPB;
+import org.apache.hadoop.ipc.ProtobufHelper;
+import org.apache.hadoop.ipc.ProtocolTranslator;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.scm.container.common.helpers.Pipeline;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * The client side of CBlockClientProtocol.
+ *
+ * CBlockClientProtocol is the protocol used between cblock client side
+ * and cblock manager (cblock client side is just the node where jscsi daemon
+ * process runs. a machines talks to jscsi daemon for mounting a volume).
+ *
+ * Right now, the only communication carried by this protocol is for client side
+ * to request mounting a volume.
+ */
+public class CBlockClientProtocolClientSideTranslatorPB
+ implements CBlockClientProtocol, ProtocolTranslator, Closeable {
+
+ private final CBlockClientServerProtocolPB rpcProxy;
+
+ public CBlockClientProtocolClientSideTranslatorPB(
+ CBlockClientServerProtocolPB rpcProxy) {
+ this.rpcProxy = rpcProxy;
+ }
+
+
+ @Override
+ public void close() throws IOException {
+ RPC.stopProxy(rpcProxy);
+ }
+
+ @Override
+ public Object getUnderlyingProxyObject() {
+ return rpcProxy;
+ }
+
+ @Override
+ public MountVolumeResponse mountVolume(
+ String userName, String volumeName) throws IOException {
+ MountVolumeRequestProto.Builder
+ request
+ = MountVolumeRequestProto
+ .newBuilder();
+ request.setUserName(userName);
+ request.setVolumeName(volumeName);
+ try {
+ MountVolumeResponseProto resp
+ = rpcProxy.mountVolume(null, request.build());
+ if (!resp.getIsValid()) {
+ throw new CBlockException(
+ "Not a valid volume:" + userName + ":" + volumeName);
+ }
+ List<Pipeline> containerIDs = new ArrayList<>();
+ HashMap<String, Pipeline> containerPipelines = new HashMap<>();
+ if (resp.getAllContainerIDsList().size() == 0) {
+ throw new CBlockException("Mount volume request returned no container");
+ }
+ for (ContainerIDProto containerID :
+ resp.getAllContainerIDsList()) {
+ if (containerID.hasPipeline()) {
+ // it should always have a pipeline only except for tests.
+ Pipeline p = Pipeline.getFromProtoBuf(containerID.getPipeline());
+ p.setData(Longs.toByteArray(containerID.getIndex()));
+ containerIDs.add(p);
+ containerPipelines.put(containerID.getContainerID(), p);
+ } else {
+ throw new CBlockException("ContainerID does not have pipeline!");
+ }
+ }
+ return new MountVolumeResponse(
+ resp.getIsValid(),
+ resp.getUserName(),
+ resp.getVolumeName(),
+ resp.getVolumeSize(),
+ resp.getBlockSize(),
+ containerIDs,
+ containerPipelines);
+ } catch (ServiceException e) {
+ throw ProtobufHelper.getRemoteException(e);
+ }
+ }
+
+ @Override
+ public List<VolumeInfo> listVolumes() throws IOException {
+ try {
+ List<VolumeInfo> result = new ArrayList<>();
+ ListVolumesResponseProto
+ listVolumesResponseProto = this.rpcProxy.listVolumes(null,
+ ListVolumesRequestProto.newBuilder()
+ .build());
+ for (VolumeInfoProto volumeInfoProto :
+ listVolumesResponseProto
+ .getVolumeEntryList()) {
+ result.add(new VolumeInfo(volumeInfoProto.getUserName(),
+ volumeInfoProto.getVolumeName(), volumeInfoProto.getVolumeSize(),
+ volumeInfoProto.getBlockSize()));
+ }
+ return result;
+ } catch (ServiceException e) {
+ throw ProtobufHelper.getRemoteException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockIStorageImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockIStorageImpl.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockIStorageImpl.java
new file mode 100644
index 0000000..2f35668
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockIStorageImpl.java
@@ -0,0 +1,440 @@
+/**
+ * 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.cblock.jscsiHelper;
+
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.cblock.jscsiHelper.cache.CacheModule;
+import org.apache.hadoop.cblock.jscsiHelper.cache.LogicalBlock;
+import org.apache.hadoop.cblock.jscsiHelper.cache.impl.CBlockLocalCache;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.scm.XceiverClientManager;
+import org.apache.hadoop.scm.container.common.helpers.Pipeline;
+import org.jscsi.target.storage.IStorageModule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.hadoop.cblock.CBlockConfigKeys.DFS_CBLOCK_TRACE_IO;
+import static org.apache.hadoop.cblock.CBlockConfigKeys
+ .DFS_CBLOCK_TRACE_IO_DEFAULT;
+
+/**
+ * The SCSI Target class for CBlockSCSIServer.
+ */
+final public class CBlockIStorageImpl implements IStorageModule {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(CBlockIStorageImpl.class);
+ private static final Logger TRACER =
+ LoggerFactory.getLogger("TraceIO");
+
+ private CacheModule cache;
+ private final long volumeSize;
+ private final int blockSize;
+ private final String userName;
+ private final String volumeName;
+ private final boolean traceEnabled;
+ private final Configuration conf;
+ private final ContainerCacheFlusher flusher;
+ private List<Pipeline> fullContainerList;
+
+ /**
+ * private: constructs a SCSI Target.
+ *
+ * @param config - config
+ * @param userName - Username
+ * @param volumeName - Name of the volume
+ * @param volumeSize - Size of the volume
+ * @param blockSize - Size of the block
+ * @param fullContainerList - Ordered list of containers that make up this
+ * volume.
+ * @param flusher - flusher which is used to flush data from
+ * level db cache to containers
+ * @throws IOException - Throws IOException.
+ */
+ private CBlockIStorageImpl(Configuration config, String userName,
+ String volumeName, long volumeSize, int blockSize,
+ List<Pipeline> fullContainerList, ContainerCacheFlusher flusher) {
+ this.conf = config;
+ this.userName = userName;
+ this.volumeName = volumeName;
+ this.volumeSize = volumeSize;
+ this.blockSize = blockSize;
+ this.fullContainerList = new ArrayList<>(fullContainerList);
+ this.flusher = flusher;
+ this.traceEnabled = conf.getBoolean(DFS_CBLOCK_TRACE_IO,
+ DFS_CBLOCK_TRACE_IO_DEFAULT);
+ }
+
+ /**
+ * private: initialize the cache.
+ *
+ * @param xceiverClientManager - client manager that is used for creating new
+ * connections to containers.
+ * @param metrics - target metrics to maintain metrics for target server
+ * @throws IOException - Throws IOException.
+ */
+ private void initCache(XceiverClientManager xceiverClientManager,
+ CBlockTargetMetrics metrics) throws IOException {
+ this.cache = CBlockLocalCache.newBuilder()
+ .setConfiguration(conf)
+ .setVolumeName(this.volumeName)
+ .setUserName(this.userName)
+ .setPipelines(this.fullContainerList)
+ .setClientManager(xceiverClientManager)
+ .setBlockSize(blockSize)
+ .setVolumeSize(volumeSize)
+ .setFlusher(flusher)
+ .setCBlockTargetMetrics(metrics)
+ .build();
+ this.cache.start();
+ }
+
+ /**
+ * Gets a new builder for CBlockStorageImpl.
+ *
+ * @return builder
+ */
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /**
+ * Get Cache.
+ *
+ * @return - Cache
+ */
+ public CacheModule getCache() {
+ return cache;
+ }
+
+ /**
+ * Returns block size of this volume.
+ *
+ * @return int size of block for this volume.
+ */
+ @Override
+ public int getBlockSize() {
+ return blockSize;
+ }
+
+ /**
+ * Checks the index boundary of a block address.
+ *
+ * @param logicalBlockAddress the index of the first block of data to be read
+ * or written
+ * @param transferLengthInBlocks the total number of consecutive blocks about
+ * to be read or written
+ * @return 0 == Success, 1 indicates the LBA address is out of bounds and 2
+ * indicates that LBA + transfer size is out of bounds.
+ */
+ @Override
+ public int checkBounds(long logicalBlockAddress, int transferLengthInBlocks) {
+ long sizeInBlocks = volumeSize / blockSize;
+ int res = 0;
+ if (logicalBlockAddress < 0 || logicalBlockAddress >= sizeInBlocks) {
+ res = 1;
+ }
+
+ if (transferLengthInBlocks < 0 ||
+ logicalBlockAddress + transferLengthInBlocks > sizeInBlocks) {
+ if (res == 0) {
+ res = 2;
+ }
+ }
+ return res;
+ }
+
+ /**
+ * Number of blocks that make up this volume.
+ *
+ * @return long - count of blocks.
+ */
+ @Override
+ public long getSizeInBlocks() {
+ return volumeSize / blockSize;
+ }
+
+ /**
+ * Reads the number of bytes that can be read into the bytes buffer from the
+ * location indicated.
+ *
+ * @param bytes the array into which the data will be copied will be filled
+ * with data from storage
+ * @param storageIndex the position of the first byte to be copied
+ * @throws IOException
+ */
+ @Override
+ public void read(byte[] bytes, long storageIndex) throws IOException {
+ int startingIdxInBlock = (int) storageIndex % blockSize;
+ int idxInBytes = 0;
+ if (this.traceEnabled) {
+ TRACER.info("Task=ReadStart,length={},location={}",
+ bytes.length, storageIndex);
+ }
+ while (idxInBytes < bytes.length - 1) {
+ long blockId = (storageIndex + idxInBytes) / blockSize;
+ byte[] dataBytes;
+
+ try {
+ LogicalBlock block = this.cache.get(blockId);
+ dataBytes = block.getData().array();
+
+ if (this.traceEnabled) {
+ TRACER.info("Task=ReadBlock,BlockID={},length={},SHA={}",
+ blockId,
+ dataBytes.length,
+ dataBytes.length > 0 ? DigestUtils.sha256Hex(dataBytes) : null);
+ }
+ } catch (IOException e) {
+ // For an non-existing block cache.get will return a block with zero
+ // bytes filled. So any error here is a real error.
+ LOGGER.error("getting errors when reading data:" + e);
+ throw e;
+ }
+
+ int length = blockSize - startingIdxInBlock;
+ if (length > bytes.length - idxInBytes) {
+ length = bytes.length - idxInBytes;
+ }
+ if (dataBytes.length >= length) {
+ System.arraycopy(dataBytes, startingIdxInBlock, bytes, idxInBytes,
+ length);
+ }
+ startingIdxInBlock = 0;
+ idxInBytes += length;
+ }
+ if (this.traceEnabled) {
+ TRACER.info("Task=ReadEnd,length={},location={},SHA={}",
+ bytes.length, storageIndex, DigestUtils.sha256Hex(bytes));
+ }
+ }
+
+ @Override
+ public void write(byte[] bytes, long storageIndex) throws IOException {
+ int startingIdxInBlock = (int) storageIndex % blockSize;
+ int idxInBytes = 0;
+ if (this.traceEnabled) {
+ TRACER.info("Task=WriteStart,length={},location={},SHA={}",
+ bytes.length, storageIndex,
+ bytes.length > 0 ? DigestUtils.sha256Hex(bytes) : null);
+ }
+
+ ByteBuffer dataByte = ByteBuffer.allocate(blockSize);
+ while (idxInBytes < bytes.length - 1) {
+ long blockId = (storageIndex + idxInBytes) / blockSize;
+ int length = blockSize - startingIdxInBlock;
+ if (length > bytes.length - idxInBytes) {
+ length = bytes.length - idxInBytes;
+ }
+ System.arraycopy(bytes, idxInBytes, dataByte.array(), startingIdxInBlock,
+ length);
+ this.cache.put(blockId, dataByte.array());
+
+ if (this.traceEnabled) {
+ TRACER.info("Task=WriteBlock,BlockID={},length={},SHA={}",
+ blockId, dataByte.array().length,
+ dataByte.array().length > 0 ?
+ DigestUtils.sha256Hex(dataByte.array()) : null);
+ }
+ dataByte.clear();
+ startingIdxInBlock = 0;
+ idxInBytes += length;
+ }
+
+ if (this.traceEnabled) {
+ TRACER.info("Task=WriteEnd,length={},location={} ",
+ bytes.length, storageIndex);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ cache.close();
+ } catch (IllegalStateException ise) {
+ LOGGER.error("Can not close the storage {}", ise);
+ throw ise;
+ }
+ }
+
+ /**
+ * Builder class for CBlocklocalCache.
+ */
+ public static class Builder {
+ private String userName;
+ private String volumeName;
+ private long volumeSize;
+ private int blockSize;
+ private List<Pipeline> containerList;
+ private Configuration conf;
+ private XceiverClientManager clientManager;
+ private ContainerCacheFlusher flusher;
+ private CBlockTargetMetrics metrics;
+
+ /**
+ * Constructs a builder.
+ */
+ Builder() {
+
+ }
+
+ public Builder setFlusher(ContainerCacheFlusher cacheFlusher) {
+ this.flusher = cacheFlusher;
+ return this;
+ }
+
+ /**
+ * set config.
+ *
+ * @param config - config
+ * @return Builder
+ */
+ public Builder setConf(Configuration config) {
+ this.conf = config;
+ return this;
+ }
+
+ /**
+ * set user name.
+ *
+ * @param cblockUserName - user name
+ * @return Builder
+ */
+ public Builder setUserName(String cblockUserName) {
+ this.userName = cblockUserName;
+ return this;
+ }
+
+ /**
+ * set volume name.
+ *
+ * @param cblockVolumeName -- volume name
+ * @return Builder
+ */
+ public Builder setVolumeName(String cblockVolumeName) {
+ this.volumeName = cblockVolumeName;
+ return this;
+ }
+
+ /**
+ * set volume size.
+ *
+ * @param cblockVolumeSize -- set volume size.
+ * @return Builder
+ */
+ public Builder setVolumeSize(long cblockVolumeSize) {
+ this.volumeSize = cblockVolumeSize;
+ return this;
+ }
+
+ /**
+ * set block size.
+ *
+ * @param cblockBlockSize -- block size
+ * @return Builder
+ */
+ public Builder setBlockSize(int cblockBlockSize) {
+ this.blockSize = cblockBlockSize;
+ return this;
+ }
+
+ /**
+ * Set contianer list.
+ *
+ * @param cblockContainerList - set the pipeline list
+ * @return Builder
+ */
+ public Builder setContainerList(List<Pipeline> cblockContainerList) {
+ this.containerList = cblockContainerList;
+ return this;
+ }
+
+ /**
+ * Set client manager.
+ *
+ * @param xceiverClientManager -- sets the client manager.
+ * @return Builder
+ */
+ public Builder setClientManager(XceiverClientManager xceiverClientManager) {
+ this.clientManager = xceiverClientManager;
+ return this;
+ }
+
+ /**
+ * Set Cblock Target Metrics.
+ *
+ * @param targetMetrics -- sets the cblock target metrics
+ * @return Builder
+ */
+ public Builder setCBlockTargetMetrics(CBlockTargetMetrics targetMetrics) {
+ this.metrics = targetMetrics;
+ return this;
+ }
+
+ /**
+ * Builds the CBlockStorageImpl.
+ *
+ * @return builds the CBlock Scsi Target.
+ */
+ public CBlockIStorageImpl build() throws IOException {
+ if (StringUtils.isBlank(userName)) {
+ throw new IllegalArgumentException("User name cannot be null or empty" +
+ ".");
+ }
+ if (StringUtils.isBlank(volumeName)) {
+ throw new IllegalArgumentException("Volume name cannot be null or " +
+ "empty");
+ }
+
+ if (volumeSize < 1) {
+ throw new IllegalArgumentException("Volume size cannot be negative or" +
+ " zero.");
+ }
+
+ if (blockSize < 1) {
+ throw new IllegalArgumentException("Block size cannot be negative or " +
+ "zero.");
+ }
+
+ if (containerList == null || containerList.size() == 0) {
+ throw new IllegalArgumentException("Container list cannot be null or " +
+ "empty");
+ }
+ if (clientManager == null) {
+ throw new IllegalArgumentException("Client manager cannot be null");
+ }
+ if (conf == null) {
+ throw new IllegalArgumentException("Configuration cannot be null");
+ }
+
+ if (flusher == null) {
+ throw new IllegalArgumentException("Flusher Cannot be null.");
+ }
+ CBlockIStorageImpl impl = new CBlockIStorageImpl(this.conf, this.userName,
+ this.volumeName, this.volumeSize, this.blockSize, this.containerList,
+ this.flusher);
+ impl.initCache(this.clientManager, this.metrics);
+ return impl;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ce23d9ad/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockManagerHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockManagerHandler.java b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockManagerHandler.java
new file mode 100644
index 0000000..6367c61
--- /dev/null
+++ b/hadoop-cblock/server/src/main/java/org/apache/hadoop/cblock/jscsiHelper/CBlockManagerHandler.java
@@ -0,0 +1,50 @@
+/**
+ * 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.cblock.jscsiHelper;
+
+import org.apache.hadoop.cblock.meta.VolumeInfo;
+import org.apache.hadoop.cblock.proto.MountVolumeResponse;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * This class is the handler of CBlockManager used by target server
+ * to communicate with CBlockManager.
+ *
+ * More specifically, this class will expose local methods to target
+ * server, and make RPC calls to CBlockManager accordingly
+ */
+public class CBlockManagerHandler {
+
+ private final CBlockClientProtocolClientSideTranslatorPB handler;
+
+ public CBlockManagerHandler(
+ CBlockClientProtocolClientSideTranslatorPB handler) {
+ this.handler = handler;
+ }
+
+ public MountVolumeResponse mountVolume(
+ String userName, String volumeName) throws IOException {
+ return handler.mountVolume(userName, volumeName);
+ }
+
+ public List<VolumeInfo> listVolumes() throws IOException {
+ return handler.listVolumes();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org