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 2018/04/24 20:13:58 UTC
[43/83] [abbrv] [partial] hadoop git commit: HDFS-13405. Ozone:
Rename HDSL to HDDS. Contributed by Ajay Kumar, Elek Marton, Mukul Kumar Singh,
Shashikant Banerjee and Anu Engineer.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestRocksDBStoreMBean.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestRocksDBStoreMBean.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestRocksDBStoreMBean.java
new file mode 100644
index 0000000..a7ce60b
--- /dev/null
+++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestRocksDBStoreMBean.java
@@ -0,0 +1,87 @@
+/**
+ * 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.utils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import javax.management.MBeanServer;
+import java.io.File;
+import java.lang.management.ManagementFactory;
+
+/**
+ * Test the JMX interface for the rocksdb metastore implementation.
+ */
+public class TestRocksDBStoreMBean {
+
+ @Test
+ public void testJmxBeans() throws Exception {
+ File testDir =
+ GenericTestUtils.getTestDir(getClass().getSimpleName() + "-withstat");
+
+ Configuration conf = new OzoneConfiguration();
+ conf.set(OzoneConfigKeys.OZONE_METADATA_STORE_IMPL,
+ OzoneConfigKeys.OZONE_METADATA_STORE_IMPL_ROCKSDB);
+
+ RocksDBStore metadataStore =
+ (RocksDBStore) MetadataStoreBuilder.newBuilder().setConf(conf)
+ .setCreateIfMissing(true).setDbFile(testDir).build();
+
+ for (int i = 0; i < 10; i++) {
+ metadataStore.put("key".getBytes(), "value".getBytes());
+ }
+
+ MBeanServer platformMBeanServer =
+ ManagementFactory.getPlatformMBeanServer();
+ Thread.sleep(2000);
+
+ Object keysWritten = platformMBeanServer
+ .getAttribute(metadataStore.getStatMBeanName(), "NUMBER_KEYS_WRITTEN");
+
+ Assert.assertEquals(10L, keysWritten);
+
+ Object dbWriteAverage = platformMBeanServer
+ .getAttribute(metadataStore.getStatMBeanName(), "DB_WRITE_AVERAGE");
+ Assert.assertTrue((double) dbWriteAverage > 0);
+
+ metadataStore.close();
+
+ }
+
+ @Test()
+ public void testDisabledStat() throws Exception {
+ File testDir = GenericTestUtils
+ .getTestDir(getClass().getSimpleName() + "-withoutstat");
+
+ Configuration conf = new OzoneConfiguration();
+ conf.set(OzoneConfigKeys.OZONE_METADATA_STORE_IMPL,
+ OzoneConfigKeys.OZONE_METADATA_STORE_IMPL_ROCKSDB);
+ conf.set(OzoneConfigKeys.OZONE_METADATA_STORE_ROCKSDB_STATISTICS,
+ OzoneConfigKeys.OZONE_METADATA_STORE_ROCKSDB_STATISTICS_OFF);
+
+ RocksDBStore metadataStore =
+ (RocksDBStore) MetadataStoreBuilder.newBuilder().setConf(conf)
+ .setCreateIfMissing(true).setDbFile(testDir).build();
+
+ Assert.assertNull(metadataStore.getStatMBeanName());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/pom.xml b/hadoop-hdds/container-service/pom.xml
new file mode 100644
index 0000000..736272d
--- /dev/null
+++ b/hadoop-hdds/container-service/pom.xml
@@ -0,0 +1,98 @@
+<?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-hdds</artifactId>
+ <version>3.2.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>hadoop-hdds-container-service</artifactId>
+ <version>3.2.0-SNAPSHOT</version>
+ <description>Apache HDDS Container server</description>
+ <name>Apache HDDS Container server</name>
+ <packaging>jar</packaging>
+
+ <properties>
+ <hadoop.component>hdds</hadoop.component>
+ <is.hadoop.component>true</is.hadoop.component>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdds-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdds-server-framework</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>2.2.0</version>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <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-hdds/common/src/main/proto/
+ </param>
+ <param>${basedir}/src/main/proto</param>
+ </imports>
+ <source>
+ <directory>${basedir}/src/main/proto</directory>
+ <includes>
+ <include>StorageContainerDatanodeProtocol.proto</include>
+ </includes>
+ </source>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/HddsServerUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/HddsServerUtil.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/HddsServerUtil.java
new file mode 100644
index 0000000..956aef2
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/HddsServerUtil.java
@@ -0,0 +1,322 @@
+/**
+ * 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.hdds.scm;
+
+import com.google.common.base.Optional;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_DEADNODE_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_DEADNODE_INTERVAL_DEFAULT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_HEARTBEAT_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_HEARTBEAT_LOG_WARN_DEFAULT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_HEARTBEAT_LOG_WARN_INTERVAL_COUNT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_HEARTBEAT_RPC_TIMEOUT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_HEARTBEAT_RPC_TIMEOUT_DEFAULT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_STALENODE_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_SCM_STALENODE_INTERVAL_DEFAULT;
+import static org.apache.hadoop.hdds.HddsUtils.*;
+import static org.apache.hadoop.hdds.server.ServerUtils.sanitizeUserArgs;
+
+/**
+ * Hdds stateless helper functions for server side components.
+ */
+public class HddsServerUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ HddsServerUtil.class);
+
+ /**
+ * 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));
+ }
+
+
+ /**
+ * 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.getTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL,
+ ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * 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,
+ ScmConfigKeys.OZONE_SCM_HEARBEAT_INTERVAL_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 staleNodeIntervalMs =
+ conf.getTimeDuration(OZONE_SCM_STALENODE_INTERVAL,
+ OZONE_SCM_STALENODE_INTERVAL_DEFAULT, TimeUnit.MILLISECONDS);
+
+ 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(staleNodeIntervalMs, heartbeatThreadFrequencyMs,
+ 5, 1000);
+ } catch (IllegalArgumentException ex) {
+ LOG.error("Stale Node Interval is cannot be honored due to " +
+ "mis-configured {}. ex: {}",
+ OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, ex);
+ throw ex;
+ }
+
+ // Make sure that stale node value is greater than configured value that
+ // datanodes are going to send HBs.
+ try {
+ sanitizeUserArgs(staleNodeIntervalMs, 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, ex);
+ throw ex;
+ }
+ return staleNodeIntervalMs;
+ }
+
+ /**
+ * 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.getTimeDuration(OZONE_SCM_DEADNODE_INTERVAL,
+ OZONE_SCM_DEADNODE_INTERVAL_DEFAULT,
+ TimeUnit.MILLISECONDS);
+
+ 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, 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(OzoneConfigKeys.DFS_CONTAINER_IPC_PORT,
+ OzoneConfigKeys.DFS_CONTAINER_IPC_PORT_DEFAULT);
+ }
+
+
+ /**
+ * 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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/VersionInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/VersionInfo.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/VersionInfo.java
new file mode 100644
index 0000000..4e52046
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/VersionInfo.java
@@ -0,0 +1,81 @@
+/**
+ * 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.hdds.scm;
+
+/**
+ * This is a class that tracks versions of SCM.
+ */
+public final class VersionInfo {
+
+ // We will just be normal and use positive counting numbers for versions.
+ private final static VersionInfo[] VERSION_INFOS =
+ {new VersionInfo("First version of SCM", 1)};
+
+
+ public static final String DESCRIPTION_KEY = "Description";
+ private final String description;
+ private final int version;
+
+ /**
+ * Never created outside this class.
+ *
+ * @param description -- description
+ * @param version -- version number
+ */
+ private VersionInfo(String description, int version) {
+ this.description = description;
+ this.version = version;
+ }
+
+ /**
+ * Returns all versions.
+ *
+ * @return Version info array.
+ */
+ public static VersionInfo[] getAllVersions() {
+ return VERSION_INFOS.clone();
+ }
+
+ /**
+ * Returns the latest version.
+ *
+ * @return versionInfo
+ */
+ public static VersionInfo getLatestVersion() {
+ return VERSION_INFOS[VERSION_INFOS.length - 1];
+ }
+
+ /**
+ * Return description.
+ *
+ * @return String
+ */
+ public String getDescription() {
+ return description;
+ }
+
+ /**
+ * Return the version.
+ *
+ * @return int.
+ */
+ public int getVersion() {
+ return version;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
new file mode 100644
index 0000000..7213e7e
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java
@@ -0,0 +1,140 @@
+/*
+ * 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 com.google.common.base.Preconditions;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.datanode.DataNodeServicePlugin;
+import org.apache.hadoop.hdds.HddsUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
+import org.apache.hadoop.ozone.container.common.statemachine
+ .DatanodeStateMachine;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.UUID;
+
+/**
+ * Datanode service plugin to start the HDDS container services.
+ */
+public class HddsDatanodeService implements DataNodeServicePlugin {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ HddsDatanodeService.class);
+
+ private final boolean isOzoneEnabled;
+
+ private Configuration conf;
+ private DatanodeDetails datanodeDetails;
+ private DatanodeStateMachine datanodeStateMachine;
+
+ public HddsDatanodeService() {
+ try {
+ OzoneConfiguration.activate();
+ this.conf = new OzoneConfiguration();
+ this.isOzoneEnabled = HddsUtils.isHddsEnabled(conf);
+ if (isOzoneEnabled) {
+ this.datanodeDetails = getDatanodeDetails(conf);
+ String hostname = DataNode.getHostName(conf);
+ String ip = InetAddress.getByName(hostname).getHostAddress();
+ this.datanodeDetails.setHostName(hostname);
+ this.datanodeDetails.setIpAddress(ip);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Can't start the HDDS datanode plugin", e);
+ }
+ }
+
+ @Override
+ public void start(Object service) {
+ if (isOzoneEnabled) {
+ try {
+ DataNode dataNode = (DataNode) service;
+ datanodeDetails.setInfoPort(dataNode.getInfoPort());
+ datanodeDetails.setInfoSecurePort(dataNode.getInfoSecurePort());
+ datanodeStateMachine = new DatanodeStateMachine(datanodeDetails, conf);
+ datanodeStateMachine.startDaemon();
+ } catch (IOException e) {
+ throw new RuntimeException("Can't start the HDDS datanode plugin", e);
+ }
+ }
+ }
+
+ /**
+ * Returns ContainerNodeIDProto or null in case of Error.
+ *
+ * @return ContainerNodeIDProto
+ */
+ private static DatanodeDetails getDatanodeDetails(Configuration conf)
+ throws IOException {
+ String idFilePath = HddsUtils.getDatanodeIdFilePath(conf);
+ if (idFilePath == null || idFilePath.isEmpty()) {
+ LOG.error("A valid file path is needed for config setting {}",
+ ScmConfigKeys.OZONE_SCM_DATANODE_ID);
+ throw new IllegalArgumentException(ScmConfigKeys.OZONE_SCM_DATANODE_ID +
+ " must be defined. See" +
+ " https://wiki.apache.org/hadoop/Ozone#Configuration" +
+ " for details on configuring Ozone.");
+ }
+
+ Preconditions.checkNotNull(idFilePath);
+ File idFile = new File(idFilePath);
+ if (idFile.exists()) {
+ return ContainerUtils.readDatanodeDetailsFrom(idFile);
+ } else {
+ // There is no datanode.id file, this might be the first time datanode
+ // is started.
+ String datanodeUuid = UUID.randomUUID().toString();
+ return DatanodeDetails.newBuilder().setUuid(datanodeUuid).build();
+ }
+ }
+
+ /**
+ *
+ * Return DatanodeDetails if set, return null otherwise.
+ *
+ * @return DatanodeDetails
+ */
+ public DatanodeDetails getDatanodeDetails() {
+ return datanodeDetails;
+ }
+
+ @InterfaceAudience.Private
+ public DatanodeStateMachine getDatanodeStateMachine() {
+ return datanodeStateMachine;
+ }
+
+ @Override
+ public void stop() {
+ if (datanodeStateMachine != null) {
+ datanodeStateMachine.stopDaemon();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java
new file mode 100644
index 0000000..68bf442
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkUtils.java
@@ -0,0 +1,346 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.container.common.helpers;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.ByteString;
+import org.apache.commons.codec.binary.Hex;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
+import org.apache.hadoop.hdds.scm.container.common.helpers
+ .StorageContainerException;
+import org.apache.hadoop.hdds.protocol.proto.ContainerProtos;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.container.common.impl.ChunkManagerImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousFileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.StandardOpenOption;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .CHECKSUM_MISMATCH;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .CONTAINER_INTERNAL_ERROR;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .CONTAINER_NOT_FOUND;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .INVALID_WRITE_SIZE;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .IO_EXCEPTION;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .OVERWRITE_FLAG_REQUIRED;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .UNABLE_TO_FIND_CHUNK;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .UNABLE_TO_FIND_DATA_DIR;
+
+/**
+ * Set of utility functions used by the chunk Manager.
+ */
+public final class ChunkUtils {
+
+ /* Never constructed. */
+ private ChunkUtils() {
+ }
+
+ /**
+ * Checks if we are getting a request to overwrite an existing range of
+ * chunk.
+ *
+ * @param chunkFile - File
+ * @param chunkInfo - Buffer to write
+ * @return bool
+ */
+ public static boolean isOverWriteRequested(File chunkFile, ChunkInfo
+ chunkInfo) {
+
+ if (!chunkFile.exists()) {
+ return false;
+ }
+
+ long offset = chunkInfo.getOffset();
+ return offset < chunkFile.length();
+ }
+
+ /**
+ * Overwrite is permitted if an only if the user explicitly asks for it. We
+ * permit this iff the key/value pair contains a flag called
+ * [OverWriteRequested, true].
+ *
+ * @param chunkInfo - Chunk info
+ * @return true if the user asks for it.
+ */
+ public static boolean isOverWritePermitted(ChunkInfo chunkInfo) {
+ String overWrite = chunkInfo.getMetadata().get(OzoneConsts.CHUNK_OVERWRITE);
+ return (overWrite != null) &&
+ (!overWrite.isEmpty()) &&
+ (Boolean.valueOf(overWrite));
+ }
+
+ /**
+ * Validates chunk data and returns a file object to Chunk File that we are
+ * expected to write data to.
+ *
+ * @param pipeline - pipeline.
+ * @param data - container data.
+ * @param info - chunk info.
+ * @return File
+ * @throws StorageContainerException
+ */
+ public static File validateChunk(Pipeline pipeline, ContainerData data,
+ ChunkInfo info) throws StorageContainerException {
+
+ Logger log = LoggerFactory.getLogger(ChunkManagerImpl.class);
+
+ File chunkFile = getChunkFile(pipeline, data, info);
+ if (ChunkUtils.isOverWriteRequested(chunkFile, info)) {
+ if (!ChunkUtils.isOverWritePermitted(info)) {
+ log.error("Rejecting write chunk request. Chunk overwrite " +
+ "without explicit request. {}", info.toString());
+ throw new StorageContainerException("Rejecting write chunk request. " +
+ "OverWrite flag required." + info.toString(),
+ OVERWRITE_FLAG_REQUIRED);
+ }
+ }
+ return chunkFile;
+ }
+
+ /**
+ * Validates that Path to chunk file exists.
+ *
+ * @param pipeline - Container Info.
+ * @param data - Container Data
+ * @param info - Chunk info
+ * @return - File.
+ * @throws StorageContainerException
+ */
+ public static File getChunkFile(Pipeline pipeline, ContainerData data,
+ ChunkInfo info) throws StorageContainerException {
+
+ Logger log = LoggerFactory.getLogger(ChunkManagerImpl.class);
+ if (data == null) {
+ log.error("Invalid container Name: {}", pipeline.getContainerName());
+ throw new StorageContainerException("Unable to find the container Name:" +
+ " " +
+ pipeline.getContainerName(), CONTAINER_NOT_FOUND);
+ }
+
+ File dataDir = ContainerUtils.getDataDirectory(data).toFile();
+ if (!dataDir.exists()) {
+ log.error("Unable to find the data directory: {}", dataDir);
+ throw new StorageContainerException("Unable to find the data directory:" +
+ " " + dataDir, UNABLE_TO_FIND_DATA_DIR);
+ }
+
+ return dataDir.toPath().resolve(info.getChunkName()).toFile();
+
+ }
+
+ /**
+ * Writes the data in chunk Info to the specified location in the chunkfile.
+ *
+ * @param chunkFile - File to write data to.
+ * @param chunkInfo - Data stream to write.
+ * @param data - The data buffer.
+ * @throws StorageContainerException
+ */
+ public static void writeData(File chunkFile, ChunkInfo chunkInfo,
+ byte[] data) throws
+ StorageContainerException, ExecutionException, InterruptedException,
+ NoSuchAlgorithmException {
+
+ Logger log = LoggerFactory.getLogger(ChunkManagerImpl.class);
+ if (data.length != chunkInfo.getLen()) {
+ String err = String.format("data array does not match the length " +
+ "specified. DataLen: %d Byte Array: %d",
+ chunkInfo.getLen(), data.length);
+ log.error(err);
+ throw new StorageContainerException(err, INVALID_WRITE_SIZE);
+ }
+
+ AsynchronousFileChannel file = null;
+ FileLock lock = null;
+
+ try {
+ file =
+ AsynchronousFileChannel.open(chunkFile.toPath(),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE,
+ StandardOpenOption.SPARSE,
+ StandardOpenOption.SYNC);
+ lock = file.lock().get();
+ if (chunkInfo.getChecksum() != null &&
+ !chunkInfo.getChecksum().isEmpty()) {
+ verifyChecksum(chunkInfo, data, log);
+ }
+ int size = file.write(ByteBuffer.wrap(data), chunkInfo.getOffset()).get();
+ if (size != data.length) {
+ log.error("Invalid write size found. Size:{} Expected: {} ", size,
+ data.length);
+ throw new StorageContainerException("Invalid write size found. " +
+ "Size: " + size + " Expected: " + data.length, INVALID_WRITE_SIZE);
+ }
+ } catch (IOException e) {
+ throw new StorageContainerException(e, IO_EXCEPTION);
+
+ } finally {
+ if (lock != null) {
+ try {
+ lock.release();
+ } catch (IOException e) {
+ log.error("Unable to release lock ??, Fatal Error.");
+ throw new StorageContainerException(e, CONTAINER_INTERNAL_ERROR);
+
+ }
+ }
+ if (file != null) {
+ try {
+ file.close();
+ } catch (IOException e) {
+ throw new StorageContainerException("Error closing chunk file",
+ e, CONTAINER_INTERNAL_ERROR);
+ }
+ }
+ }
+ }
+
+ /**
+ * Verifies the checksum of a chunk against the data buffer.
+ *
+ * @param chunkInfo - Chunk Info.
+ * @param data - data buffer
+ * @param log - log
+ * @throws NoSuchAlgorithmException
+ * @throws StorageContainerException
+ */
+ private static void verifyChecksum(ChunkInfo chunkInfo, byte[] data, Logger
+ log) throws NoSuchAlgorithmException, StorageContainerException {
+ MessageDigest sha = MessageDigest.getInstance(OzoneConsts.FILE_HASH);
+ sha.update(data);
+ if (!Hex.encodeHexString(sha.digest()).equals(
+ chunkInfo.getChecksum())) {
+ log.error("Checksum mismatch. Provided: {} , computed: {}",
+ chunkInfo.getChecksum(), DigestUtils.sha256Hex(sha.digest()));
+ throw new StorageContainerException("Checksum mismatch. Provided: " +
+ chunkInfo.getChecksum() + " , computed: " +
+ DigestUtils.sha256Hex(sha.digest()), CHECKSUM_MISMATCH);
+ }
+ }
+
+ /**
+ * Reads data from an existing chunk file.
+ *
+ * @param chunkFile - file where data lives.
+ * @param data - chunk definition.
+ * @return ByteBuffer
+ * @throws StorageContainerException
+ * @throws ExecutionException
+ * @throws InterruptedException
+ */
+ public static ByteBuffer readData(File chunkFile, ChunkInfo data) throws
+ StorageContainerException, ExecutionException, InterruptedException,
+ NoSuchAlgorithmException {
+ Logger log = LoggerFactory.getLogger(ChunkManagerImpl.class);
+
+ if (!chunkFile.exists()) {
+ log.error("Unable to find the chunk file. chunk info : {}",
+ data.toString());
+ throw new StorageContainerException("Unable to find the chunk file. " +
+ "chunk info " +
+ data.toString(), UNABLE_TO_FIND_CHUNK);
+ }
+
+ AsynchronousFileChannel file = null;
+ FileLock lock = null;
+ try {
+ file =
+ AsynchronousFileChannel.open(chunkFile.toPath(),
+ StandardOpenOption.READ);
+ lock = file.lock(data.getOffset(), data.getLen(), true).get();
+
+ ByteBuffer buf = ByteBuffer.allocate((int) data.getLen());
+ file.read(buf, data.getOffset()).get();
+
+ if (data.getChecksum() != null && !data.getChecksum().isEmpty()) {
+ verifyChecksum(data, buf.array(), log);
+ }
+
+ return buf;
+ } catch (IOException e) {
+ throw new StorageContainerException(e, IO_EXCEPTION);
+ } finally {
+ if (lock != null) {
+ try {
+ lock.release();
+ } catch (IOException e) {
+ log.error("I/O error is lock release.");
+ }
+ }
+ if (file != null) {
+ IOUtils.closeStream(file);
+ }
+ }
+ }
+
+ /**
+ * Returns a CreateContainer Response. This call is used by create and delete
+ * containers which have null success responses.
+ *
+ * @param msg Request
+ * @return Response.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ getChunkResponse(ContainerProtos.ContainerCommandRequestProto msg) {
+ return ContainerUtils.getContainerResponse(msg);
+ }
+
+ /**
+ * Gets a response to the read chunk calls.
+ *
+ * @param msg - Msg
+ * @param data - Data
+ * @param info - Info
+ * @return Response.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ getReadChunkResponse(ContainerProtos.ContainerCommandRequestProto msg,
+ byte[] data, ChunkInfo info) {
+ Preconditions.checkNotNull(msg);
+
+ ContainerProtos.ReadChunkResponseProto.Builder response =
+ ContainerProtos.ReadChunkResponseProto.newBuilder();
+ response.setChunkData(info.getProtoBufMessage());
+ response.setData(ByteString.copyFrom(data));
+ response.setPipeline(msg.getReadChunk().getPipeline());
+
+ ContainerProtos.ContainerCommandResponseProto.Builder builder =
+ ContainerUtils.getContainerResponse(msg, ContainerProtos.Result
+ .SUCCESS, "");
+ builder.setReadChunk(response);
+ return builder.build();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java
new file mode 100644
index 0000000..c29374c
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java
@@ -0,0 +1,326 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.common.helpers;
+
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.protocol.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.util.Time;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * This class maintains the information about a container in the ozone world.
+ * <p>
+ * A container is a name, along with metadata- which is a set of key value
+ * pair.
+ */
+public class ContainerData {
+
+ private final String containerName;
+ private final Map<String, String> metadata;
+ private String dbPath; // Path to Level DB Store.
+ // Path to Physical file system where container and checksum are stored.
+ private String containerFilePath;
+ private String hash;
+ private AtomicLong bytesUsed;
+ private long maxSize;
+ private Long containerID;
+ private HddsProtos.LifeCycleState state;
+
+ /**
+ * Constructs a ContainerData Object.
+ *
+ * @param containerName - Name
+ */
+ public ContainerData(String containerName, Long containerID,
+ Configuration conf) {
+ this.metadata = new TreeMap<>();
+ this.containerName = containerName;
+ this.maxSize = conf.getLong(ScmConfigKeys.SCM_CONTAINER_CLIENT_MAX_SIZE_KEY,
+ ScmConfigKeys.SCM_CONTAINER_CLIENT_MAX_SIZE_DEFAULT) * OzoneConsts.GB;
+ this.bytesUsed = new AtomicLong(0L);
+ this.containerID = containerID;
+ this.state = HddsProtos.LifeCycleState.OPEN;
+ }
+
+ /**
+ * Constructs a ContainerData object from ProtoBuf classes.
+ *
+ * @param protoData - ProtoBuf Message
+ * @throws IOException
+ */
+ public static ContainerData getFromProtBuf(
+ ContainerProtos.ContainerData protoData, Configuration conf)
+ throws IOException {
+ ContainerData data = new ContainerData(protoData.getName(),
+ protoData.getContainerID(), conf);
+ for (int x = 0; x < protoData.getMetadataCount(); x++) {
+ data.addMetadata(protoData.getMetadata(x).getKey(),
+ protoData.getMetadata(x).getValue());
+ }
+
+ if (protoData.hasContainerPath()) {
+ data.setContainerPath(protoData.getContainerPath());
+ }
+
+ if (protoData.hasDbPath()) {
+ data.setDBPath(protoData.getDbPath());
+ }
+
+ if (protoData.hasState()) {
+ data.setState(protoData.getState());
+ }
+
+ if(protoData.hasHash()) {
+ data.setHash(protoData.getHash());
+ }
+
+ if (protoData.hasBytesUsed()) {
+ data.setBytesUsed(protoData.getBytesUsed());
+ }
+
+ if (protoData.hasSize()) {
+ data.setMaxSize(protoData.getSize());
+ }
+ return data;
+ }
+
+ /**
+ * Returns a ProtoBuf Message from ContainerData.
+ *
+ * @return Protocol Buffer Message
+ */
+ public ContainerProtos.ContainerData getProtoBufMessage() {
+ ContainerProtos.ContainerData.Builder builder = ContainerProtos
+ .ContainerData.newBuilder();
+ builder.setName(this.getContainerName());
+ builder.setContainerID(this.getContainerID());
+
+ if (this.getDBPath() != null) {
+ builder.setDbPath(this.getDBPath());
+ }
+
+ if (this.getHash() != null) {
+ builder.setHash(this.getHash());
+ }
+
+ if (this.getContainerPath() != null) {
+ builder.setContainerPath(this.getContainerPath());
+ }
+
+ builder.setState(this.getState());
+
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ HddsProtos.KeyValue.Builder keyValBuilder =
+ HddsProtos.KeyValue.newBuilder();
+ builder.addMetadata(keyValBuilder.setKey(entry.getKey())
+ .setValue(entry.getValue()).build());
+ }
+
+ if (this.getBytesUsed() >= 0) {
+ builder.setBytesUsed(this.getBytesUsed());
+ }
+
+ if (this.getKeyCount() >= 0) {
+ builder.setKeyCount(this.getKeyCount());
+ }
+
+ if (this.getMaxSize() >= 0) {
+ builder.setSize(this.getMaxSize());
+ }
+
+ return builder.build();
+ }
+
+ /**
+ * Returns the name of the container.
+ *
+ * @return - name
+ */
+ public String getContainerName() {
+ return containerName;
+ }
+
+ /**
+ * Adds metadata.
+ */
+ public void addMetadata(String key, String value) throws IOException {
+ synchronized (this.metadata) {
+ if (this.metadata.containsKey(key)) {
+ throw new IOException("This key already exists. Key " + key);
+ }
+ metadata.put(key, value);
+ }
+ }
+
+ /**
+ * Returns all metadata.
+ */
+ public Map<String, String> getAllMetadata() {
+ synchronized (this.metadata) {
+ return Collections.unmodifiableMap(this.metadata);
+ }
+ }
+
+ /**
+ * Returns value of a key.
+ */
+ public String getValue(String key) {
+ synchronized (this.metadata) {
+ return metadata.get(key);
+ }
+ }
+
+ /**
+ * Deletes a metadata entry from the map.
+ *
+ * @param key - Key
+ */
+ public void deleteKey(String key) {
+ synchronized (this.metadata) {
+ metadata.remove(key);
+ }
+ }
+
+ /**
+ * Returns path.
+ *
+ * @return - path
+ */
+ public String getDBPath() {
+ return dbPath;
+ }
+
+ /**
+ * Sets path.
+ *
+ * @param path - String.
+ */
+ public void setDBPath(String path) {
+ this.dbPath = path;
+ }
+
+ /**
+ * This function serves as the generic key for ContainerCache class. Both
+ * ContainerData and ContainerKeyData overrides this function to appropriately
+ * return the right name that can be used in ContainerCache.
+ *
+ * @return String Name.
+ */
+ public String getName() {
+ return getContainerName();
+ }
+
+ /**
+ * Get container file path.
+ * @return - Physical path where container file and checksum is stored.
+ */
+ public String getContainerPath() {
+ return containerFilePath;
+ }
+
+ /**
+ * Set container Path.
+ * @param containerPath - File path.
+ */
+ public void setContainerPath(String containerPath) {
+ this.containerFilePath = containerPath;
+ }
+
+ /**
+ * Get container ID.
+ * @return - container ID.
+ */
+ public synchronized Long getContainerID() {
+ return containerID;
+ }
+
+ public synchronized void setState(HddsProtos.LifeCycleState state) {
+ this.state = state;
+ }
+
+ public synchronized HddsProtos.LifeCycleState getState() {
+ return this.state;
+ }
+
+ /**
+ * checks if the container is open.
+ * @return - boolean
+ */
+ public synchronized boolean isOpen() {
+ return HddsProtos.LifeCycleState.OPEN == state;
+ }
+
+ /**
+ * Marks this container as closed.
+ */
+ public synchronized void closeContainer() {
+ // TODO: closed or closing here
+ setState(HddsProtos.LifeCycleState.CLOSED);
+
+ // Some thing brain dead for now. name + Time stamp of when we get the close
+ // container message.
+ setHash(DigestUtils.sha256Hex(this.getContainerName() +
+ Long.toString(Time.monotonicNow())));
+ }
+
+ /**
+ * Final hash for this container.
+ * @return - Hash
+ */
+ public String getHash() {
+ return hash;
+ }
+
+ public void setHash(String hash) {
+ this.hash = hash;
+ }
+
+ public void setMaxSize(long maxSize) {
+ this.maxSize = maxSize;
+ }
+
+ public long getMaxSize() {
+ return maxSize;
+ }
+
+ public long getKeyCount() {
+ return metadata.size();
+ }
+
+ public void setBytesUsed(long used) {
+ this.bytesUsed.set(used);
+ }
+
+ public long addBytesUsed(long delta) {
+ return this.bytesUsed.addAndGet(delta);
+ }
+
+ public long getBytesUsed() {
+ return bytesUsed.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerMetrics.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerMetrics.java
new file mode 100644
index 0000000..d4d732b
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerMetrics.java
@@ -0,0 +1,121 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.common.helpers;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdds.protocol.proto.ContainerProtos;
+import org.apache.hadoop.metrics2.MetricsSystem;
+import org.apache.hadoop.metrics2.annotation.Metric;
+import org.apache.hadoop.metrics2.annotation.Metrics;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.metrics2.lib.MetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableCounterLong;
+import org.apache.hadoop.metrics2.lib.MutableQuantiles;
+import org.apache.hadoop.metrics2.lib.MutableRate;
+
+/**
+ *
+ * This class is for maintaining the various Storage Container
+ * DataNode statistics and publishing them through the metrics interfaces.
+ * This also registers the JMX MBean for RPC.
+ * <p>
+ * This class has a number of metrics variables that are publicly accessible;
+ * these variables (objects) have methods to update their values;
+ * for example:
+ * <p> {@link #numOps}.inc()
+ *
+ */
+@InterfaceAudience.Private
+@Metrics(about="Storage Container DataNode Metrics", context="dfs")
+public class ContainerMetrics {
+ @Metric private MutableCounterLong numOps;
+ private MutableCounterLong[] numOpsArray;
+ private MutableCounterLong[] opsBytesArray;
+ private MutableRate[] opsLatency;
+ private MutableQuantiles[][] opsLatQuantiles;
+ private MetricsRegistry registry = null;
+
+ public ContainerMetrics(int[] intervals) {
+ int numEnumEntries = ContainerProtos.Type.values().length;
+ final int len = intervals.length;
+ this.numOpsArray = new MutableCounterLong[numEnumEntries];
+ this.opsBytesArray = new MutableCounterLong[numEnumEntries];
+ this.opsLatency = new MutableRate[numEnumEntries];
+ this.opsLatQuantiles = new MutableQuantiles[numEnumEntries][len];
+ this.registry = new MetricsRegistry("StorageContainerMetrics");
+ for (int i = 0; i < numEnumEntries; i++) {
+ numOpsArray[i] = registry.newCounter(
+ "num" + ContainerProtos.Type.valueOf(i + 1),
+ "number of " + ContainerProtos.Type.valueOf(i + 1) + " ops",
+ (long) 0);
+ opsBytesArray[i] = registry.newCounter(
+ "bytes" + ContainerProtos.Type.valueOf(i + 1),
+ "bytes used by " + ContainerProtos.Type.valueOf(i + 1) + "op",
+ (long) 0);
+ opsLatency[i] = registry.newRate(
+ "latency" + ContainerProtos.Type.valueOf(i + 1),
+ ContainerProtos.Type.valueOf(i + 1) + " op");
+
+ for (int j = 0; j < len; j++) {
+ int interval = intervals[j];
+ String quantileName = ContainerProtos.Type.valueOf(i + 1) + "Nanos"
+ + interval + "s";
+ opsLatQuantiles[i][j] = registry.newQuantiles(quantileName,
+ "latency of Container ops", "ops", "latency", interval);
+ }
+ }
+ }
+
+ public static ContainerMetrics create(Configuration conf) {
+ MetricsSystem ms = DefaultMetricsSystem.instance();
+ // Percentile measurement is off by default, by watching no intervals
+ int[] intervals =
+ conf.getInts(DFSConfigKeys.DFS_METRICS_PERCENTILES_INTERVALS_KEY);
+ return ms.register("StorageContainerMetrics",
+ "Storage Container Node Metrics",
+ new ContainerMetrics(intervals));
+ }
+
+ public void incContainerOpcMetrics(ContainerProtos.Type type){
+ numOps.incr();
+ numOpsArray[type.ordinal()].incr();
+ }
+
+ public long getContainerOpsMetrics(ContainerProtos.Type type){
+ return numOpsArray[type.ordinal()].value();
+ }
+
+ public void incContainerOpsLatencies(ContainerProtos.Type type,
+ long latencyNanos) {
+ opsLatency[type.ordinal()].add(latencyNanos);
+ for (MutableQuantiles q: opsLatQuantiles[type.ordinal()]) {
+ q.add(latencyNanos);
+ }
+ }
+
+ public void incContainerBytesStats(ContainerProtos.Type type, long bytes) {
+ opsBytesArray[type.ordinal()].incr(bytes);
+ }
+
+ public long getContainerBytesMetrics(ContainerProtos.Type type){
+ return opsBytesArray[type.ordinal()].value();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerReport.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerReport.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerReport.java
new file mode 100644
index 0000000..50d2da3
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerReport.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.container.common.helpers;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerInfo;
+
+/**
+ * Container Report iterates the closed containers and sends a container report
+ * to SCM.
+ */
+public class ContainerReport {
+ private static final int UNKNOWN = -1;
+ private final String containerName;
+ private final String finalhash;
+ private long size;
+ private long keyCount;
+ private long bytesUsed;
+ private long readCount;
+ private long writeCount;
+ private long readBytes;
+ private long writeBytes;
+ private long containerID;
+
+ public long getContainerID() {
+ return containerID;
+ }
+
+ public void setContainerID(long containerID) {
+ this.containerID = containerID;
+ }
+
+
+
+
+ /**
+ * Constructs the ContainerReport.
+ *
+ * @param containerName - Container Name.
+ * @param finalhash - Final Hash.
+ */
+ public ContainerReport(String containerName, String finalhash) {
+ this.containerName = containerName;
+ this.finalhash = finalhash;
+ this.size = UNKNOWN;
+ this.keyCount = UNKNOWN;
+ this.bytesUsed = 0L;
+ this.readCount = 0L;
+ this.readBytes = 0L;
+ this.writeCount = 0L;
+ this.writeBytes = 0L;
+ }
+
+ /**
+ * Gets a containerReport from protobuf class.
+ *
+ * @param info - ContainerInfo.
+ * @return - ContainerReport.
+ */
+ public static ContainerReport getFromProtoBuf(ContainerInfo info) {
+ Preconditions.checkNotNull(info);
+ ContainerReport report = new ContainerReport(info.getContainerName(),
+ info.getFinalhash());
+ if (info.hasSize()) {
+ report.setSize(info.getSize());
+ }
+ if (info.hasKeyCount()) {
+ report.setKeyCount(info.getKeyCount());
+ }
+ if (info.hasUsed()) {
+ report.setBytesUsed(info.getUsed());
+ }
+ if (info.hasReadCount()) {
+ report.setReadCount(info.getReadCount());
+ }
+ if (info.hasReadBytes()) {
+ report.setReadBytes(info.getReadBytes());
+ }
+ if (info.hasWriteCount()) {
+ report.setWriteCount(info.getWriteCount());
+ }
+ if (info.hasWriteBytes()) {
+ report.setWriteBytes(info.getWriteBytes());
+ }
+
+ report.setContainerID(info.getContainerID());
+ return report;
+ }
+
+ /**
+ * Gets the container name.
+ *
+ * @return - Name
+ */
+ public String getContainerName() {
+ return containerName;
+ }
+
+ /**
+ * Returns the final signature for this container.
+ *
+ * @return - hash
+ */
+ public String getFinalhash() {
+ return finalhash;
+ }
+
+ /**
+ * Returns a positive number it is a valid number, -1 if not known.
+ *
+ * @return size or -1
+ */
+ public long getSize() {
+ return size;
+ }
+
+ /**
+ * Sets the size of the container on disk.
+ *
+ * @param size - int
+ */
+ public void setSize(long size) {
+ this.size = size;
+ }
+
+ /**
+ * Gets number of keys in the container if known.
+ *
+ * @return - Number of keys or -1 for not known.
+ */
+ public long getKeyCount() {
+ return keyCount;
+ }
+
+ /**
+ * Sets the key count.
+ *
+ * @param keyCount - Key Count
+ */
+ public void setKeyCount(long keyCount) {
+ this.keyCount = keyCount;
+ }
+
+ public long getReadCount() {
+ return readCount;
+ }
+
+ public void setReadCount(long readCount) {
+ this.readCount = readCount;
+ }
+
+ public long getWriteCount() {
+ return writeCount;
+ }
+
+ public void setWriteCount(long writeCount) {
+ this.writeCount = writeCount;
+ }
+
+ public long getReadBytes() {
+ return readBytes;
+ }
+
+ public void setReadBytes(long readBytes) {
+ this.readBytes = readBytes;
+ }
+
+ public long getWriteBytes() {
+ return writeBytes;
+ }
+
+ public void setWriteBytes(long writeBytes) {
+ this.writeBytes = writeBytes;
+ }
+
+ public long getBytesUsed() {
+ return bytesUsed;
+ }
+
+ public void setBytesUsed(long bytesUsed) {
+ this.bytesUsed = bytesUsed;
+ }
+
+ /**
+ * Gets a containerInfo protobuf message from ContainerReports.
+ *
+ * @return ContainerInfo
+ */
+ public ContainerInfo getProtoBufMessage() {
+ return ContainerInfo.newBuilder()
+ .setContainerName(this.getContainerName())
+ .setKeyCount(this.getKeyCount())
+ .setSize(this.getSize())
+ .setUsed(this.getBytesUsed())
+ .setReadCount(this.getReadCount())
+ .setReadBytes(this.getReadBytes())
+ .setWriteCount(this.getWriteCount())
+ .setWriteBytes(this.getWriteBytes())
+ .setFinalhash(this.getFinalhash())
+ .setContainerID(this.getContainerID())
+ .build();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java
new file mode 100644
index 0000000..1818188
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.common.helpers;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
+import org.apache.hadoop.hdds.scm.container.common.helpers
+ .StorageContainerException;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.container.common.impl.ContainerManagerImpl;
+import org.apache.hadoop.utils.MetadataStore;
+import org.apache.hadoop.utils.MetadataStoreBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static org.apache.commons.io.FilenameUtils.removeExtension;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .INVALID_ARGUMENT;
+import static org.apache.hadoop.hdds.protocol.proto.ContainerProtos.Result
+ .UNABLE_TO_FIND_DATA_DIR;
+import static org.apache.hadoop.ozone.OzoneConsts.CONTAINER_EXTENSION;
+import static org.apache.hadoop.ozone.OzoneConsts.CONTAINER_META;
+
+/**
+ * A set of helper functions to create proper responses.
+ */
+public final class ContainerUtils {
+
+ private ContainerUtils() {
+ //never constructed.
+ }
+
+ /**
+ * Returns a CreateContainer Response. This call is used by create and delete
+ * containers which have null success responses.
+ *
+ * @param msg Request
+ * @return Response.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ getContainerResponse(ContainerProtos.ContainerCommandRequestProto msg) {
+ ContainerProtos.ContainerCommandResponseProto.Builder builder =
+ getContainerResponse(msg, ContainerProtos.Result.SUCCESS, "");
+ return builder.build();
+ }
+
+ /**
+ * Returns a ReadContainer Response.
+ *
+ * @param msg Request
+ * @param containerData - data
+ * @return Response.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ getReadContainerResponse(ContainerProtos.ContainerCommandRequestProto msg,
+ ContainerData containerData) {
+ Preconditions.checkNotNull(containerData);
+
+ ContainerProtos.ReadContainerResponseProto.Builder response =
+ ContainerProtos.ReadContainerResponseProto.newBuilder();
+ response.setContainerData(containerData.getProtoBufMessage());
+
+ ContainerProtos.ContainerCommandResponseProto.Builder builder =
+ getContainerResponse(msg, ContainerProtos.Result.SUCCESS, "");
+ builder.setReadContainer(response);
+ return builder.build();
+ }
+
+ /**
+ * We found a command type but no associated payload for the command. Hence
+ * return malformed Command as response.
+ *
+ * @param msg - Protobuf message.
+ * @param result - result
+ * @param message - Error message.
+ * @return ContainerCommandResponseProto - MALFORMED_REQUEST.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto.Builder
+ getContainerResponse(ContainerProtos.ContainerCommandRequestProto msg,
+ ContainerProtos.Result result, String message) {
+ return
+ ContainerProtos.ContainerCommandResponseProto.newBuilder()
+ .setCmdType(msg.getCmdType())
+ .setTraceID(msg.getTraceID())
+ .setResult(result)
+ .setMessage(message);
+ }
+
+ /**
+ * Logs the error and returns a response to the caller.
+ *
+ * @param log - Logger
+ * @param ex - Exception
+ * @param msg - Request Object
+ * @return Response
+ */
+ public static ContainerProtos.ContainerCommandResponseProto logAndReturnError(
+ Logger log, StorageContainerException ex,
+ ContainerProtos.ContainerCommandRequestProto msg) {
+ log.info("Operation: {} : Trace ID: {} : Message: {} : Result: {}",
+ msg.getCmdType().name(), msg.getTraceID(),
+ ex.getMessage(), ex.getResult().getValueDescriptor().getName());
+ return getContainerResponse(msg, ex.getResult(), ex.getMessage()).build();
+ }
+
+ /**
+ * Logs the error and returns a response to the caller.
+ *
+ * @param log - Logger
+ * @param ex - Exception
+ * @param msg - Request Object
+ * @return Response
+ */
+ public static ContainerProtos.ContainerCommandResponseProto logAndReturnError(
+ Logger log, RuntimeException ex,
+ ContainerProtos.ContainerCommandRequestProto msg) {
+ log.info("Operation: {} : Trace ID: {} : Message: {} ",
+ msg.getCmdType().name(), msg.getTraceID(), ex.getMessage());
+ return getContainerResponse(msg, INVALID_ARGUMENT, ex.getMessage()).build();
+ }
+
+ /**
+ * We found a command type but no associated payload for the command. Hence
+ * return malformed Command as response.
+ *
+ * @param msg - Protobuf message.
+ * @return ContainerCommandResponseProto - MALFORMED_REQUEST.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ malformedRequest(ContainerProtos.ContainerCommandRequestProto msg) {
+ return getContainerResponse(msg, ContainerProtos.Result.MALFORMED_REQUEST,
+ "Cmd type does not match the payload.").build();
+ }
+
+ /**
+ * We found a command type that is not supported yet.
+ *
+ * @param msg - Protobuf message.
+ * @return ContainerCommandResponseProto - MALFORMED_REQUEST.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ unsupportedRequest(ContainerProtos.ContainerCommandRequestProto msg) {
+ return getContainerResponse(msg, ContainerProtos.Result.UNSUPPORTED_REQUEST,
+ "Server does not support this command yet.").build();
+ }
+
+ /**
+ * get containerName from a container file.
+ *
+ * @param containerFile - File
+ * @return Name of the container.
+ */
+ public static String getContainerNameFromFile(File containerFile) {
+ Preconditions.checkNotNull(containerFile);
+ return Paths.get(containerFile.getParent()).resolve(
+ removeExtension(containerFile.getName())).toString();
+ }
+
+ /**
+ * Verifies that this in indeed a new container.
+ *
+ * @param containerFile - Container File to verify
+ * @param metadataFile - metadata File to verify
+ * @throws IOException
+ */
+ public static void verifyIsNewContainer(File containerFile, File metadataFile)
+ throws IOException {
+ Logger log = LoggerFactory.getLogger(ContainerManagerImpl.class);
+ if (containerFile.exists()) {
+ log.error("container already exists on disk. File: {}",
+ containerFile.toPath());
+ throw new FileAlreadyExistsException("container already exists on " +
+ "disk.");
+ }
+
+ if (metadataFile.exists()) {
+ log.error("metadata found on disk, but missing container. Refusing to" +
+ " write this container. File: {} ", metadataFile.toPath());
+ throw new FileAlreadyExistsException(("metadata found on disk, but " +
+ "missing container. Refusing to write this container."));
+ }
+
+ File parentPath = new File(containerFile.getParent());
+
+ if (!parentPath.exists() && !parentPath.mkdirs()) {
+ log.error("Unable to create parent path. Path: {}",
+ parentPath.toString());
+ throw new IOException("Unable to create container directory.");
+ }
+
+ if (!containerFile.createNewFile()) {
+ log.error("creation of a new container file failed. File: {}",
+ containerFile.toPath());
+ throw new IOException("creation of a new container file failed.");
+ }
+
+ if (!metadataFile.createNewFile()) {
+ log.error("creation of the metadata file failed. File: {}",
+ metadataFile.toPath());
+ throw new IOException("creation of a new container file failed.");
+ }
+ }
+
+ public static String getContainerDbFileName(String containerName) {
+ return containerName + OzoneConsts.DN_CONTAINER_DB;
+ }
+
+ /**
+ * creates a Metadata DB for the specified container.
+ *
+ * @param containerPath - Container Path.
+ * @throws IOException
+ */
+ public static Path createMetadata(Path containerPath, String containerName,
+ Configuration conf)
+ throws IOException {
+ Logger log = LoggerFactory.getLogger(ContainerManagerImpl.class);
+ Preconditions.checkNotNull(containerPath);
+ Path metadataPath = containerPath.resolve(OzoneConsts.CONTAINER_META_PATH);
+ if (!metadataPath.toFile().mkdirs()) {
+ log.error("Unable to create directory for metadata storage. Path: {}",
+ metadataPath);
+ throw new IOException("Unable to create directory for metadata storage." +
+ " Path: " + metadataPath);
+ }
+ MetadataStore store = MetadataStoreBuilder.newBuilder()
+ .setConf(conf)
+ .setCreateIfMissing(true)
+ .setDbFile(metadataPath
+ .resolve(getContainerDbFileName(containerName)).toFile())
+ .build();
+
+ // we close since the SCM pre-creates containers.
+ // we will open and put Db handle into a cache when keys are being created
+ // in a container.
+
+ store.close();
+
+ Path dataPath = containerPath.resolve(OzoneConsts.CONTAINER_DATA_PATH);
+ if (!dataPath.toFile().mkdirs()) {
+
+ // If we failed to create data directory, we cleanup the
+ // metadata directory completely. That is, we will delete the
+ // whole directory including LevelDB file.
+ log.error("Unable to create directory for data storage. cleaning up the" +
+ " container path: {} dataPath: {}",
+ containerPath, dataPath);
+ FileUtils.deleteDirectory(containerPath.toFile());
+ throw new IOException("Unable to create directory for data storage." +
+ " Path: " + dataPath);
+ }
+ return metadataPath;
+ }
+
+ /**
+ * Returns Metadata location.
+ *
+ * @param containerData - Data
+ * @param location - Path
+ * @return Path
+ */
+ public static File getMetadataFile(ContainerData containerData,
+ Path location) {
+ return location.resolve(containerData
+ .getContainerName().concat(CONTAINER_META))
+ .toFile();
+ }
+
+ /**
+ * Returns container file location.
+ *
+ * @param containerData - Data
+ * @param location - Root path
+ * @return Path
+ */
+ public static File getContainerFile(ContainerData containerData,
+ Path location) {
+ return location.resolve(containerData
+ .getContainerName().concat(CONTAINER_EXTENSION))
+ .toFile();
+ }
+
+ /**
+ * Container metadata directory -- here is where the level DB lives.
+ *
+ * @param cData - cData.
+ * @return Path to the parent directory where the DB lives.
+ */
+ public static Path getMetadataDirectory(ContainerData cData) {
+ Path dbPath = Paths.get(cData.getDBPath());
+ Preconditions.checkNotNull(dbPath);
+ Preconditions.checkState(dbPath.toString().length() > 0);
+ return dbPath.getParent();
+ }
+
+ /**
+ * Returns the path where data or chunks live for a given container.
+ *
+ * @param cData - cData container
+ * @return - Path
+ * @throws StorageContainerException
+ */
+ public static Path getDataDirectory(ContainerData cData)
+ throws StorageContainerException {
+ Path path = getMetadataDirectory(cData);
+ Preconditions.checkNotNull(path);
+ Path parentPath = path.getParent();
+ if (parentPath == null) {
+ throw new StorageContainerException("Unable to get Data directory."
+ + path, UNABLE_TO_FIND_DATA_DIR);
+ }
+ return parentPath.resolve(OzoneConsts.CONTAINER_DATA_PATH);
+ }
+
+ /**
+ * remove Container if it is empty.
+ * <p/>
+ * There are three things we need to delete.
+ * <p/>
+ * 1. Container file and metadata file. 2. The Level DB file 3. The path that
+ * we created on the data location.
+ *
+ * @param containerData - Data of the container to remove.
+ * @param conf - configuration of the cluster.
+ * @param forceDelete - whether this container should be deleted forcibly.
+ * @throws IOException
+ */
+ public static void removeContainer(ContainerData containerData,
+ Configuration conf, boolean forceDelete) throws IOException {
+ Preconditions.checkNotNull(containerData);
+ Path dbPath = Paths.get(containerData.getDBPath());
+
+ MetadataStore db = KeyUtils.getDB(containerData, conf);
+ // If the container is not empty and cannot be deleted forcibly,
+ // then throw a SCE to stop deleting.
+ if(!forceDelete && !db.isEmpty()) {
+ throw new StorageContainerException(
+ "Container cannot be deleted because it is not empty.",
+ ContainerProtos.Result.ERROR_CONTAINER_NOT_EMPTY);
+ }
+ // Close the DB connection and remove the DB handler from cache
+ KeyUtils.removeDB(containerData, conf);
+
+ // Delete the DB File.
+ FileUtils.forceDelete(dbPath.toFile());
+ dbPath = dbPath.getParent();
+
+ // Delete all Metadata in the Data directories for this containers.
+ if (dbPath != null) {
+ FileUtils.deleteDirectory(dbPath.toFile());
+ dbPath = dbPath.getParent();
+ }
+
+ // now delete the container directory, this means that all key data dirs
+ // will be removed too.
+ if (dbPath != null) {
+ FileUtils.deleteDirectory(dbPath.toFile());
+ }
+
+ // Delete the container metadata from the metadata locations.
+ String rootPath = getContainerNameFromFile(new File(containerData
+ .getContainerPath()));
+ Path containerPath = Paths.get(rootPath.concat(CONTAINER_EXTENSION));
+ Path metaPath = Paths.get(rootPath.concat(CONTAINER_META));
+
+ FileUtils.forceDelete(containerPath.toFile());
+ FileUtils.forceDelete(metaPath.toFile());
+ }
+
+ /**
+ * Persistent a {@link DatanodeDetails} to a local file.
+ *
+ * @throws IOException when read/write error occurs
+ */
+ public synchronized static void writeDatanodeDetailsTo(
+ DatanodeDetails datanodeDetails, File path) throws IOException {
+ if (path.exists()) {
+ if (!path.delete() || !path.createNewFile()) {
+ throw new IOException("Unable to overwrite the datanode ID file.");
+ }
+ } else {
+ if(!path.getParentFile().exists() &&
+ !path.getParentFile().mkdirs()) {
+ throw new IOException("Unable to create datanode ID directories.");
+ }
+ }
+ try (FileOutputStream out = new FileOutputStream(path)) {
+ HddsProtos.DatanodeDetailsProto proto =
+ datanodeDetails.getProtoBufMessage();
+ proto.writeTo(out);
+ }
+ }
+
+ /**
+ * Read {@link DatanodeDetails} from a local ID file.
+ *
+ * @param path ID file local path
+ * @return {@link DatanodeDetails}
+ * @throws IOException If the id file is malformed or other I/O exceptions
+ */
+ public synchronized static DatanodeDetails readDatanodeDetailsFrom(File path)
+ throws IOException {
+ if (!path.exists()) {
+ throw new IOException("Datanode ID file not found.");
+ }
+ try(FileInputStream in = new FileInputStream(path)) {
+ return DatanodeDetails.getFromProtoBuf(
+ HddsProtos.DatanodeDetailsProto.parseFrom(in));
+ } catch (IOException e) {
+ throw new IOException("Failed to parse DatanodeDetails from "
+ + path.getAbsolutePath(), e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DeletedContainerBlocksSummary.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DeletedContainerBlocksSummary.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DeletedContainerBlocksSummary.java
new file mode 100644
index 0000000..ade162a
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DeletedContainerBlocksSummary.java
@@ -0,0 +1,103 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.hadoop.ozone.container.common.helpers;
+
+import com.google.common.collect.Maps;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A helper class to wrap the info about under deletion container blocks.
+ */
+public final class DeletedContainerBlocksSummary {
+
+ private final List<DeletedBlocksTransaction> blocks;
+ // key : txID
+ // value : times of this tx has been processed
+ private final Map<Long, Integer> txSummary;
+ // key : container name
+ // value : the number of blocks need to be deleted in this container
+ // if the message contains multiple entries for same block,
+ // blocks will be merged
+ private final Map<String, Integer> blockSummary;
+ // total number of blocks in this message
+ private int numOfBlocks;
+
+ private DeletedContainerBlocksSummary(List<DeletedBlocksTransaction> blocks) {
+ this.blocks = blocks;
+ txSummary = Maps.newHashMap();
+ blockSummary = Maps.newHashMap();
+ blocks.forEach(entry -> {
+ txSummary.put(entry.getTxID(), entry.getCount());
+ if (blockSummary.containsKey(entry.getContainerName())) {
+ blockSummary.put(entry.getContainerName(),
+ blockSummary.get(entry.getContainerName())
+ + entry.getBlockIDCount());
+ } else {
+ blockSummary.put(entry.getContainerName(), entry.getBlockIDCount());
+ }
+ numOfBlocks += entry.getBlockIDCount();
+ });
+ }
+
+ public static DeletedContainerBlocksSummary getFrom(
+ List<DeletedBlocksTransaction> blocks) {
+ return new DeletedContainerBlocksSummary(blocks);
+ }
+
+ public int getNumOfBlocks() {
+ return numOfBlocks;
+ }
+
+ public int getNumOfContainers() {
+ return blockSummary.size();
+ }
+
+ public String getTXIDs() {
+ return String.join(",", txSummary.keySet()
+ .stream().map(String::valueOf).collect(Collectors.toList()));
+ }
+
+ public String getTxIDSummary() {
+ List<String> txSummaryEntry = txSummary.entrySet().stream()
+ .map(entry -> entry.getKey() + "(" + entry.getValue() + ")")
+ .collect(Collectors.toList());
+ return "[" + String.join(",", txSummaryEntry) + "]";
+ }
+
+ @Override public String toString() {
+ StringBuffer sb = new StringBuffer();
+ for (DeletedBlocksTransaction blks : blocks) {
+ sb.append(" ")
+ .append("TXID=")
+ .append(blks.getTxID())
+ .append(", ")
+ .append("TimesProceed=")
+ .append(blks.getCount())
+ .append(", ")
+ .append(blks.getContainerName())
+ .append(" : [")
+ .append(String.join(",", blks.getBlockIDList())).append("]")
+ .append("\n");
+ }
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/651a05a1/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/FileUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/FileUtils.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/FileUtils.java
new file mode 100644
index 0000000..566db02
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/FileUtils.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.container.common.helpers;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.ByteString;
+import org.apache.hadoop.hdds.protocol.proto.ContainerProtos;
+
+/**
+ * File Utils are helper routines used by putSmallFile and getSmallFile
+ * RPCs.
+ */
+public final class FileUtils {
+ /**
+ * Never Constructed.
+ */
+ private FileUtils() {
+ }
+
+ /**
+ * Gets a response for the putSmallFile RPC.
+ * @param msg - ContainerCommandRequestProto
+ * @return - ContainerCommandResponseProto
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ getPutFileResponse(ContainerProtos.ContainerCommandRequestProto msg) {
+ ContainerProtos.PutSmallFileResponseProto.Builder getResponse =
+ ContainerProtos.PutSmallFileResponseProto.newBuilder();
+ ContainerProtos.ContainerCommandResponseProto.Builder builder =
+ ContainerUtils.getContainerResponse(msg, ContainerProtos.Result
+ .SUCCESS, "");
+ builder.setCmdType(ContainerProtos.Type.PutSmallFile);
+ builder.setPutSmallFile(getResponse);
+ return builder.build();
+ }
+
+ /**
+ * Gets a response to the read small file call.
+ * @param msg - Msg
+ * @param data - Data
+ * @param info - Info
+ * @return Response.
+ */
+ public static ContainerProtos.ContainerCommandResponseProto
+ getGetSmallFileResponse(ContainerProtos.ContainerCommandRequestProto msg,
+ byte[] data, ChunkInfo info) {
+ Preconditions.checkNotNull(msg);
+
+ ContainerProtos.ReadChunkResponseProto.Builder readChunkresponse =
+ ContainerProtos.ReadChunkResponseProto.newBuilder();
+ readChunkresponse.setChunkData(info.getProtoBufMessage());
+ readChunkresponse.setData(ByteString.copyFrom(data));
+ readChunkresponse.setPipeline(msg.getGetSmallFile().getKey().getPipeline());
+
+ ContainerProtos.GetSmallFileResponseProto.Builder getSmallFile =
+ ContainerProtos.GetSmallFileResponseProto.newBuilder();
+ getSmallFile.setData(readChunkresponse.build());
+ ContainerProtos.ContainerCommandResponseProto.Builder builder =
+ ContainerUtils.getContainerResponse(msg, ContainerProtos.Result
+ .SUCCESS, "");
+ builder.setCmdType(ContainerProtos.Type.GetSmallFile);
+ builder.setGetSmallFile(getSmallFile);
+ return builder.build();
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org