diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStoreIterator.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStoreIterator.java
index 7b62f7ad43eda..92051dd6619c8 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStoreIterator.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/LevelDBStoreIterator.java
@@ -61,4 +61,9 @@ public void seekToFirst() {
   public void seekToLast() {
     levelDBIterator.seekToLast();
   }
+
+  @Override
+  public void prefixSeek(byte[] prefix) {
+    levelDBIterator.seek(prefix);
+  }
 }
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetaStoreIterator.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetaStoreIterator.java
index 52d0a3efd3bf5..15ded0d64300f 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetaStoreIterator.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/MetaStoreIterator.java
@@ -36,4 +36,9 @@ public interface MetaStoreIterator<T> extends Iterator<T> {
    */
   void seekToLast();
 
+  /**
+   * seek with prefix.
+   */
+  void prefixSeek(byte[] prefix);
+
 }
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStoreIterator.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStoreIterator.java
index 6e9b6958da179..161d5de491772 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStoreIterator.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStoreIterator.java
@@ -63,4 +63,9 @@ public void seekToLast() {
     rocksDBIterator.seekToLast();
   }
 
+  @Override
+  public void prefixSeek(byte[] prefix) {
+    rocksDBIterator.seek(prefix);
+  }
+
 }
diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestMetadataStore.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestMetadataStore.java
index 5da8fbc1c9243..96d818b522f0f 100644
--- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestMetadataStore.java
+++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/TestMetadataStore.java
@@ -163,6 +163,58 @@ public void testIterator() throws Exception {
 
   }
 
+
+  @Test
+  public void testIteratorPrefixSeek() throws Exception {
+    Configuration conf = new OzoneConfiguration();
+    conf.set(OzoneConfigKeys.OZONE_METADATA_STORE_IMPL, storeImpl);
+    File dbDir = GenericTestUtils.getRandomizedTestDir();
+    MetadataStore dbStore = MetadataStoreBuilder.newBuilder()
+        .setConf(conf)
+        .setCreateIfMissing(true)
+        .setDbFile(dbDir)
+        .build();
+
+    for (int i = 0; i < 5; i++) {
+      dbStore.put(getBytes("a" + i), getBytes("a-value" + i));
+    }
+
+    for (int i = 0; i < 5; i++) {
+      dbStore.put(getBytes("b" + i), getBytes("b-value" + i));
+    }
+
+    for (int i = 0; i < 5; i++) {
+      dbStore.put(getBytes("c" + i), getBytes("c-value" + i));
+    }
+
+    for (int i = 5; i < 10; i++) {
+      dbStore.put(getBytes("b" + i), getBytes("b-value" + i));
+    }
+
+    for (int i = 5; i < 10; i++) {
+      dbStore.put(getBytes("a" + i), getBytes("a-value" + i));
+    }
+
+
+    MetaStoreIterator<KeyValue> metaStoreIterator = dbStore.iterator();
+    metaStoreIterator.prefixSeek(getBytes("b"));
+    int i = 0;
+    while (metaStoreIterator.hasNext()) {
+      KeyValue val = metaStoreIterator.next();
+      String key = getString(val.getKey());
+      if (key.startsWith("b")) {
+        assertEquals("b-value" + i, getString(val.getValue()));
+      } else {
+        break;
+      }
+      i++;
+    }
+    assertTrue(i == 10);
+    dbStore.close();
+    dbStore.destroy();
+    FileUtils.deleteDirectory(dbDir);
+  }
+
   @Test
   public void testMetaStoreConfigDifferentFromType() throws IOException {
 
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/ServerUtils.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/ServerUtils.java
index fae68397c25a4..ab5d2ecd8238f 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/ServerUtils.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/ServerUtils.java
@@ -125,13 +125,29 @@ public static void releaseConnection(HttpRequestBase request) {
    * @return
    */
   public static File getScmDbDir(Configuration conf) {
-    final Collection<String> metadirs = conf.getTrimmedStringCollection(
-        ScmConfigKeys.OZONE_SCM_DB_DIRS);
+
+    File metadataDir = getDirWithFallBackToOzoneMetadata(conf, ScmConfigKeys
+        .OZONE_SCM_DB_DIRS, "SCM");
+    if (metadataDir != null) {
+      return metadataDir;
+    }
+
+    LOG.warn("{} is not configured. We recommend adding this setting. " +
+        "Falling back to {} instead.",
+        ScmConfigKeys.OZONE_SCM_DB_DIRS, HddsConfigKeys.OZONE_METADATA_DIRS);
+    return getOzoneMetaDirPath(conf);
+  }
+
+  public static File getDirWithFallBackToOzoneMetadata(Configuration conf,
+                                                       String key,
+                                                       String componentName) {
+    final Collection<String> metadirs = conf.getTrimmedStringCollection(key);
 
     if (metadirs.size() > 1) {
       throw new IllegalArgumentException(
-          "Bad config setting " + ScmConfigKeys.OZONE_SCM_DB_DIRS +
-          ". SCM does not support multiple metadata dirs currently");
+          "Bad config setting " + key +
+              ". " + componentName +
+              " does not support multiple metadata dirs currently");
     }
 
     if (metadirs.size() == 1) {
@@ -143,11 +159,7 @@ public static File getScmDbDir(Configuration conf) {
       }
       return dbDirPath;
     }
-
-    LOG.warn("{} is not configured. We recommend adding this setting. " +
-        "Falling back to {} instead.",
-        ScmConfigKeys.OZONE_SCM_DB_DIRS, HddsConfigKeys.OZONE_METADATA_DIRS);
-    return getOzoneMetaDirPath(conf);
+    return null;
   }
 
   /**
diff --git a/hadoop-ozone/ozone-recon/pom.xml b/hadoop-ozone/ozone-recon/pom.xml
index b8e41872935f3..b278033db2780 100644
--- a/hadoop-ozone/ozone-recon/pom.xml
+++ b/hadoop-ozone/ozone-recon/pom.xml
@@ -22,7 +22,7 @@
   </parent>
   <name>Apache Hadoop Ozone Recon</name>
   <modelVersion>4.0.0</modelVersion>
-  <artifactId>ozone-recon</artifactId>
+  <artifactId>hadoop-ozone-ozone-recon</artifactId>
   <dependencies>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
@@ -50,5 +50,16 @@
       <artifactId>guice-assistedinject</artifactId>
       <version>4.1.0</version>
     </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+      <version>1.10.19</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>
\ No newline at end of file
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
index 1a90e70264497..fe318561e2215 100644
--- a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
@@ -18,6 +18,10 @@
 package org.apache.hadoop.ozone.recon;
 
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.recon.spi.MetadataStoreProvider;
+import org.apache.hadoop.ozone.recon.spi.ContainerDBServiceProvider;
+import org.apache.hadoop.ozone.recon.spi.impl.ContainerDBServiceProviderImpl;
+import org.apache.hadoop.utils.MetadataStore;
 
 import com.google.inject.AbstractModule;
 import com.google.inject.Singleton;
@@ -30,6 +34,9 @@ public class ReconControllerModule extends AbstractModule {
   protected void configure() {
     bind(OzoneConfiguration.class).toProvider(OzoneConfigurationProvider.class);
     bind(ReconHttpServer.class).in(Singleton.class);
+    bind(MetadataStore.class).toProvider(MetadataStoreProvider.class);
+    bind(ContainerDBServiceProvider.class)
+        .to(ContainerDBServiceProviderImpl.class);
   }
 
 
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconHttpServer.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconHttpServer.java
index 72818c57ed696..e7dcb0cc4d6e6 100644
--- a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconHttpServer.java
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconHttpServer.java
@@ -37,52 +37,52 @@ public class ReconHttpServer extends BaseHttpServer {
 
   @Override
   protected String getHttpAddressKey() {
-    return ReconServerConfiguration.OZONE_RECON_HTTP_ADDRESS_KEY;
+    return ReconServerConfigKeys.OZONE_RECON_HTTP_ADDRESS_KEY;
   }
 
   @Override
   protected String getHttpsAddressKey() {
-    return ReconServerConfiguration.OZONE_RECON_HTTPS_ADDRESS_KEY;
+    return ReconServerConfigKeys.OZONE_RECON_HTTPS_ADDRESS_KEY;
   }
 
   @Override
   protected String getHttpBindHostKey() {
-    return ReconServerConfiguration.OZONE_RECON_HTTP_BIND_HOST_KEY;
+    return ReconServerConfigKeys.OZONE_RECON_HTTP_BIND_HOST_KEY;
   }
 
   @Override
   protected String getHttpsBindHostKey() {
-    return ReconServerConfiguration.OZONE_RECON_HTTPS_BIND_HOST_KEY;
+    return ReconServerConfigKeys.OZONE_RECON_HTTPS_BIND_HOST_KEY;
   }
 
   @Override
   protected String getBindHostDefault() {
-    return ReconServerConfiguration.OZONE_RECON_HTTP_BIND_HOST_DEFAULT;
+    return ReconServerConfigKeys.OZONE_RECON_HTTP_BIND_HOST_DEFAULT;
   }
 
   @Override
   protected int getHttpBindPortDefault() {
-    return ReconServerConfiguration.OZONE_RECON_HTTP_BIND_PORT_DEFAULT;
+    return ReconServerConfigKeys.OZONE_RECON_HTTP_BIND_PORT_DEFAULT;
   }
 
   @Override
   protected int getHttpsBindPortDefault() {
-    return ReconServerConfiguration.OZONE_RECON_HTTPS_BIND_PORT_DEFAULT;
+    return ReconServerConfigKeys.OZONE_RECON_HTTPS_BIND_PORT_DEFAULT;
   }
 
   @Override
   protected String getKeytabFile() {
-    return ReconServerConfiguration.OZONE_RECON_KEYTAB_FILE;
+    return ReconServerConfigKeys.OZONE_RECON_KEYTAB_FILE;
   }
 
   @Override
   protected String getSpnegoPrincipal() {
-    return ReconServerConfiguration
+    return ReconServerConfigKeys
         .OZONE_RECON_WEB_AUTHENTICATION_KERBEROS_PRINCIPAL;
   }
 
   @Override
   protected String getEnabledKey() {
-    return ReconServerConfiguration.OZONE_RECON_HTTP_ENABLED_KEY;
+    return ReconServerConfigKeys.OZONE_RECON_HTTP_ENABLED_KEY;
   }
 }
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServerConfiguration.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServerConfigKeys.java
similarity index 81%
rename from hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServerConfiguration.java
rename to hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServerConfigKeys.java
index 78281bcac265d..a33ff975a4f89 100644
--- a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServerConfiguration.java
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServerConfigKeys.java
@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.ozone.recon;
 
+import static org.apache.hadoop.ozone.OzoneConsts.CONTAINER_DB_SUFFIX;
+
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 
@@ -25,7 +27,7 @@
  */
 @InterfaceAudience.Public
 @InterfaceStability.Unstable
-public final class ReconServerConfiguration {
+public final class ReconServerConfigKeys {
 
   public static final String OZONE_RECON_HTTP_ENABLED_KEY =
       "ozone.recon.http.enabled";
@@ -48,9 +50,17 @@ public final class ReconServerConfiguration {
   public static final String OZONE_RECON_DOMAIN_NAME =
       "ozone.recon.domain.name";
 
+  public static final String OZONE_RECON_CONTAINER_DB_CACHE_SIZE_MB =
+      "ozone.recon.container.db.cache.size.mb";
+  public static final int OZONE_RECON_CONTAINER_DB_CACHE_SIZE_DEFAULT = 128;
+
+  public static final String OZONE_RECON_DB_DIRS = "ozone.recon.db.dirs";
+  public static final String RECON_CONTAINER_DB = "recon-" +
+      CONTAINER_DB_SUFFIX;
+
   /**
    * Private constructor for utility class.
    */
-  private ReconServerConfiguration() {
+  private ReconServerConfigKeys() {
   }
 }
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/api/types/ContainerKeyPrefix.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/api/types/ContainerKeyPrefix.java
new file mode 100644
index 0000000000000..064dc5c2f952d
--- /dev/null
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/api/types/ContainerKeyPrefix.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.ozone.recon.api.types;
+
+/**
+ * Class to encapsulate the Key information needed for the Recon container DB.
+ * Currently, it is containerId and key prefix.
+ */
+public class ContainerKeyPrefix {
+
+  private long containerId;
+  private String keyPrefix;
+
+  public ContainerKeyPrefix(long containerId, String keyPrefix) {
+    this.containerId = containerId;
+    this.keyPrefix = keyPrefix;
+  }
+
+  public long getContainerId() {
+    return containerId;
+  }
+
+  public void setContainerId(long containerId) {
+    this.containerId = containerId;
+  }
+
+  public String getKeyPrefix() {
+    return keyPrefix;
+  }
+
+  public void setKeyPrefix(String keyPrefix) {
+    this.keyPrefix = keyPrefix;
+  }
+}
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/ContainerDBServiceProvider.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/ContainerDBServiceProvider.java
new file mode 100644
index 0000000000000..15491e8b15ee8
--- /dev/null
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/ContainerDBServiceProvider.java
@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.recon.spi;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.ozone.recon.api.types.ContainerKeyPrefix;
+
+/**
+ * The Recon Container DB Service interface.
+ */
+@InterfaceStability.Evolving
+public interface ContainerDBServiceProvider {
+
+  /**
+   * Store the container to Key prefix mapping into the Recon Container DB.
+   *
+   * @param containerKeyPrefix the containerId, key-prefix tuple.
+   * @param count              Count of Keys with that prefix.
+   */
+  void storeContainerKeyMapping(ContainerKeyPrefix containerKeyPrefix,
+                                Integer count) throws IOException;
+
+  /**
+   * Get the stored key prefix count for the given containerId, key prefix.
+   *
+   * @param containerKeyPrefix the containerId, key-prefix tuple.
+   * @return count of keys with that prefix.
+   */
+  Integer getCountForForContainerKeyPrefix(
+      ContainerKeyPrefix containerKeyPrefix) throws IOException;
+
+  /**
+   * Get the stored key prefixes for the given containerId.
+   *
+   * @param containerId the given containerId.
+   * @return Map of Key prefix -> count.
+   */
+  Map<String, Integer> getKeyPrefixesForContainer(long containerId);
+}
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/MetadataStoreProvider.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/MetadataStoreProvider.java
new file mode 100644
index 0000000000000..acf1f9a34b655
--- /dev/null
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/MetadataStoreProvider.java
@@ -0,0 +1,77 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.recon.spi;
+
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.
+    OZONE_RECON_CONTAINER_DB_CACHE_SIZE_DEFAULT;
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.
+    OZONE_RECON_CONTAINER_DB_CACHE_SIZE_MB;
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.
+    OZONE_RECON_DB_DIRS;
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.
+    RECON_CONTAINER_DB;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.server.ServerUtils;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.utils.MetadataStore;
+import org.apache.hadoop.utils.MetadataStoreBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+
+/**
+ * Provider for the Recon container DB (Metadata store).
+ */
+public class MetadataStoreProvider implements
+    Provider<MetadataStore> {
+
+  @VisibleForTesting
+  private static final Logger LOG =
+      LoggerFactory.getLogger(MetadataStoreProvider.class);
+
+  @Inject
+  private OzoneConfiguration configuration;
+
+  @Override
+  public MetadataStore get() {
+    File metaDir = ServerUtils.getDirWithFallBackToOzoneMetadata(configuration,
+        OZONE_RECON_DB_DIRS, "Recon");
+    File containerDBPath = new File(metaDir, RECON_CONTAINER_DB);
+    int cacheSize = configuration.getInt(OZONE_RECON_CONTAINER_DB_CACHE_SIZE_MB,
+        OZONE_RECON_CONTAINER_DB_CACHE_SIZE_DEFAULT);
+
+    try {
+      return MetadataStoreBuilder.newBuilder()
+          .setConf(configuration)
+          .setDbFile(containerDBPath)
+          .setCacheSize(cacheSize * OzoneConsts.MB)
+          .build();
+    } catch (IOException ioEx) {
+      LOG.error("Unable to initialize Recon container metadata store.", ioEx);
+    }
+    return null;
+  }
+}
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ContainerDBServiceProviderImpl.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ContainerDBServiceProviderImpl.java
new file mode 100644
index 0000000000000..055e2e4f708cd
--- /dev/null
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ContainerDBServiceProviderImpl.java
@@ -0,0 +1,138 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.recon.spi.impl;
+
+import static org.apache.commons.compress.utils.CharsetNames.UTF_8;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.hadoop.ozone.recon.api.types.ContainerKeyPrefix;
+import org.apache.hadoop.ozone.recon.spi.ContainerDBServiceProvider;
+import org.apache.hadoop.utils.MetaStoreIterator;
+import org.apache.hadoop.utils.MetadataStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.primitives.Longs;
+
+/**
+ * Implementation of the Recon Container DB Service.
+ */
+@Singleton
+public class ContainerDBServiceProviderImpl
+    implements ContainerDBServiceProvider {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(ContainerDBServiceProviderImpl.class);
+  private final static String KEY_DELIMITER = "_";
+
+  @Inject
+  private MetadataStore containerDBStore;
+
+  /**
+   * Concatenate the containerId and Key Prefix using a delimiter and store the
+   * count into the container DB store.
+   *
+   * @param containerKeyPrefix the containerId, key-prefix tuple.
+   * @param count Count of the keys matching that prefix.
+   * @throws IOException
+   */
+  @Override
+  public void storeContainerKeyMapping(ContainerKeyPrefix containerKeyPrefix,
+                                       Integer count)
+      throws IOException {
+    byte[] containerIdBytes = Longs.toByteArray(containerKeyPrefix
+        .getContainerId());
+    byte[] keyPrefixBytes = (KEY_DELIMITER + containerKeyPrefix.getKeyPrefix())
+        .getBytes(UTF_8);
+    byte[] dbKey = ArrayUtils.addAll(containerIdBytes, keyPrefixBytes);
+    byte[] dbValue = ByteBuffer.allocate(Integer.BYTES).putInt(count).array();
+    containerDBStore.put(dbKey, dbValue);
+  }
+
+  /**
+   * Put together the key from the passed in object and get the count from
+   * the container DB store.
+   *
+   * @param containerKeyPrefix the containerId, key-prefix tuple.
+   * @return count of keys matching the containerId, key-prefix.
+   * @throws IOException
+   */
+  @Override
+  public Integer getCountForForContainerKeyPrefix(
+      ContainerKeyPrefix containerKeyPrefix) throws IOException {
+    byte[] containerIdBytes = Longs.toByteArray(containerKeyPrefix
+        .getContainerId());
+    byte[] keyPrefixBytes = (KEY_DELIMITER + containerKeyPrefix
+        .getKeyPrefix()).getBytes(UTF_8);
+    byte[] dbKey = ArrayUtils.addAll(containerIdBytes, keyPrefixBytes);
+    byte[] dbValue = containerDBStore.get(dbKey);
+    return ByteBuffer.wrap(dbValue).getInt();
+  }
+
+  /**
+   * Use the DB's prefix seek iterator to start the scan from the given
+   * container ID prefix.
+   *
+   * @param containerId the given containerId.
+   * @return Map of (Key-Prefix,Count of Keys).
+   */
+  @Override
+  public Map<String, Integer> getKeyPrefixesForContainer(long containerId) {
+
+    Map<String, Integer> prefixes = new HashMap<>();
+    MetaStoreIterator<MetadataStore.KeyValue> containerIterator =
+        containerDBStore.iterator();
+    byte[] containerIdPrefixBytes = Longs.toByteArray(containerId);
+    containerIterator.prefixSeek(containerIdPrefixBytes);
+    while (containerIterator.hasNext()) {
+      MetadataStore.KeyValue keyValue = containerIterator.next();
+      byte[] containerKey = keyValue.getKey();
+      long containerIdFromDB = ByteBuffer.wrap(ArrayUtils.subarray
+          (containerKey, 0, Long.BYTES)).getLong();
+
+      //The prefix seek only guarantees that the iterator's head will be
+      // positioned at the first prefix match. We still have to check the key
+      // prefix.
+      if (containerIdFromDB == containerId) {
+        byte[] keyPrefix = ArrayUtils.subarray(containerKey,
+            containerIdPrefixBytes.length + 1,
+            containerKey.length);
+        try {
+          prefixes.put(new String(keyPrefix, UTF_8),
+              ByteBuffer.wrap(keyValue.getValue()).getInt());
+        } catch (UnsupportedEncodingException e) {
+          LOG.warn("Unable to read key prefix from container DB.", e);
+        }
+      } else {
+        break; //Break when the first mismatch occurs.
+      }
+    }
+    return prefixes;
+  }
+
+}
\ No newline at end of file
diff --git a/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/package-info.java b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/package-info.java
new file mode 100644
index 0000000000000..1ed44294d1eff
--- /dev/null
+++ b/hadoop-ozone/ozone-recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.
+ */
+/**
+ * The classes in this package define the Service Provider implementations for
+ * Recon. This provides connectivity to underlying Ozone subsystems.
+ */
+package org.apache.hadoop.ozone.recon.spi.impl;
\ No newline at end of file
diff --git a/hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/ContainerDBServiceProviderImplTest.java b/hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/ContainerDBServiceProviderImplTest.java
new file mode 100644
index 0000000000000..71a05021e9a08
--- /dev/null
+++ b/hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/ContainerDBServiceProviderImplTest.java
@@ -0,0 +1,148 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.recon.spi.impl;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.recon.api.types.ContainerKeyPrefix;
+import org.apache.hadoop.ozone.recon.spi.ContainerDBServiceProvider;
+import org.apache.hadoop.utils.MetaStoreIterator;
+import org.apache.hadoop.utils.MetadataStore;
+import org.apache.hadoop.utils.MetadataStoreBuilder;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+
+/**
+ * Test ContainerDBServiceProviderImpl.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ContainerDBServiceProviderImplTest {
+
+  @Rule
+  public TemporaryFolder tempFolder = new TemporaryFolder();
+
+  private MetadataStore containerDBStore;
+  private ContainerDBServiceProvider containerDbServiceProvider
+      = new ContainerDBServiceProviderImpl();
+  private Injector injector;
+
+  @Before
+  public void setUp() throws IOException {
+    tempFolder.create();
+    File dbDir = tempFolder.getRoot();
+    containerDBStore = MetadataStoreBuilder.newBuilder()
+        .setConf(new OzoneConfiguration())
+        .setCreateIfMissing(true)
+        .setDbFile(dbDir)
+        .build();
+    injector = Guice.createInjector(new AbstractModule() {
+      @Override
+      protected void configure() {
+        bind(MetadataStore.class).toInstance(containerDBStore);
+        bind(ContainerDBServiceProvider.class)
+            .toInstance(containerDbServiceProvider);
+      }
+    });
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    tempFolder.delete();
+  }
+
+  @Test
+  public void testStoreContainerKeyMapping() throws Exception {
+
+    long containerId = System.currentTimeMillis();
+    Map<String, Integer> prefixCounts = new HashMap<>();
+    prefixCounts.put("V1/B1/K1", 1);
+    prefixCounts.put("V1/B1/K2", 2);
+    prefixCounts.put("V1/B2/K3", 3);
+
+    for (String prefix : prefixCounts.keySet()) {
+      ContainerKeyPrefix containerKeyPrefix = new ContainerKeyPrefix(
+          containerId, prefix);
+      containerDbServiceProvider.storeContainerKeyMapping(
+          containerKeyPrefix, prefixCounts.get(prefix));
+    }
+
+    int count = 0;
+    MetaStoreIterator<MetadataStore.KeyValue> iterator =
+        containerDBStore.iterator();
+    while (iterator.hasNext()) {
+      iterator.next();
+      count++;
+    }
+    assertTrue(count == 3);
+  }
+
+  @Test
+  public void testGetCountForForContainerKeyPrefix() throws Exception {
+    long containerId = System.currentTimeMillis();
+
+    containerDbServiceProvider.storeContainerKeyMapping(new
+        ContainerKeyPrefix(containerId, "V1/B1/K1"), 2);
+
+    Integer count = containerDbServiceProvider.
+        getCountForForContainerKeyPrefix(new ContainerKeyPrefix(containerId,
+            "V1/B1/K1"));
+    assertTrue(count == 2);
+  }
+
+  @Test
+  public void testGetKeyPrefixesForContainer() throws Exception {
+    long containerId = System.currentTimeMillis();
+
+    containerDbServiceProvider.storeContainerKeyMapping(new
+        ContainerKeyPrefix(containerId, "V1/B1/K1"), 1);
+
+    containerDbServiceProvider.storeContainerKeyMapping(new
+        ContainerKeyPrefix(containerId, "V1/B1/K2"), 2);
+
+    long nextContainerId = System.currentTimeMillis();
+    containerDbServiceProvider.storeContainerKeyMapping(new
+        ContainerKeyPrefix(nextContainerId, "V1/B2/K1"), 3);
+
+    Map<String, Integer> keyPrefixMap = containerDbServiceProvider
+        .getKeyPrefixesForContainer(containerId);
+    assertTrue(keyPrefixMap.size() == 2);
+    assertTrue(keyPrefixMap.get("V1/B1/K1") == 1);
+    assertTrue(keyPrefixMap.get("V1/B1/K2") == 2);
+
+    keyPrefixMap = containerDbServiceProvider
+        .getKeyPrefixesForContainer(nextContainerId);
+    assertTrue(keyPrefixMap.size() == 1);
+    assertTrue(keyPrefixMap.get("V1/B2/K1") == 3);
+  }
+}
diff --git a/hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/package-info.java b/hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/package-info.java
new file mode 100644
index 0000000000000..932c4375c8c9e
--- /dev/null
+++ b/hadoop-ozone/ozone-recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/package-info.java
@@ -0,0 +1,21 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <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 for recon server impl tests.
+ */
+package org.apache.hadoop.ozone.recon.spi.impl;
\ No newline at end of file