HDDS-1638. Implement Key Write Requests to use Cache and DoubleBuffer.

hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/audit/DummyAction.java

@@ -24,7 +24,6 @@ public enum DummyAction implements AuditAction {
 
   CREATE_VOLUME,
   CREATE_BUCKET,
-  CREATE_KEY,
   READ_VOLUME,
   READ_BUCKET,
   READ_KEY,

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/audit/OMAction.java

@@ -30,7 +30,6 @@ public enum OMAction implements AuditAction {
   COMMIT_KEY,
   CREATE_VOLUME,
   CREATE_BUCKET,
-  CREATE_KEY,
   DELETE_VOLUME,
   DELETE_BUCKET,
   DELETE_KEY,

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java

@@ -165,28 +165,34 @@ public void updateLocationInfoList(List<OmKeyLocationInfo> locationInfoList) {
    * part of the latest version, not a new version.
    *
    * @param newLocationList the list of new blocks to be added.
+   * @param updateTime if true, will update modification time.
    * @throws IOException
    */
   public synchronized void appendNewBlocks(
-      List<OmKeyLocationInfo> newLocationList) throws IOException {
+      List<OmKeyLocationInfo> newLocationList, boolean updateTime)
+      throws IOException {
     if (keyLocationVersions.size() == 0) {
       throw new IOException("Appending new block, but no version exist");
     }
     OmKeyLocationInfoGroup currentLatestVersion =
         keyLocationVersions.get(keyLocationVersions.size() - 1);
     currentLatestVersion.appendNewBlocks(newLocationList);
-    setModificationTime(Time.now());
+    if (updateTime) {
+      setModificationTime(Time.now());
+    }
   }
 
   /**
    * Add a new set of blocks. The new blocks will be added as appending a new
    * version to the all version list.
    *
    * @param newLocationList the list of new blocks to be added.
+   * @param updateTime - if true, updates modification time.
    * @throws IOException
    */
   public synchronized long addNewVersion(
-      List<OmKeyLocationInfo> newLocationList) throws IOException {
+      List<OmKeyLocationInfo> newLocationList, boolean updateTime)
+      throws IOException {
     long latestVersionNum;
     if (keyLocationVersions.size() == 0) {
       // no version exist, these blocks are the very first version.
@@ -202,7 +208,10 @@ public synchronized long addNewVersion(
       keyLocationVersions.add(newVersion);
       latestVersionNum = newVersion.getVersion();
     }
-    setModificationTime(Time.now());
+
+    if (updateTime) {
+      setModificationTime(Time.now());
+    }
     return latestVersionNum;
   }
 

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java

@@ -23,6 +23,8 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation;
 import org.apache.hadoop.security.token.Token;
 
+import java.util.Objects;
+
 /**
  * One key can be too huge to fit in one container. In which case it gets split
  * into a number of subkeys. This class represents one such subkey instance.
@@ -202,4 +204,27 @@ public String toString() {
         ", pipeline=" + pipeline +
         ", createVersion=" + createVersion + '}';
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    OmKeyLocationInfo that = (OmKeyLocationInfo) o;
+    return length == that.length &&
+        offset == that.offset &&
+        createVersion == that.createVersion &&
+        Objects.equals(blockID, that.blockID) &&
+        Objects.equals(token, that.token) &&
+        Objects.equals(pipeline, that.pipeline);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(blockID, length, offset, token, createVersion,
+        pipeline);
+  }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -726,11 +726,20 @@ private OMResponse handleError(OMResponse resp) throws OMException {
   public OmKeyLocationInfo allocateBlock(OmKeyArgs args, long clientId,
       ExcludeList excludeList) throws IOException {
     AllocateBlockRequest.Builder req = AllocateBlockRequest.newBuilder();
-    KeyArgs keyArgs = KeyArgs.newBuilder()
+    KeyArgs.Builder keyArgs = KeyArgs.newBuilder()
         .setVolumeName(args.getVolumeName())
         .setBucketName(args.getBucketName())
         .setKeyName(args.getKeyName())
-        .setDataSize(args.getDataSize()).build();
+        .setDataSize(args.getDataSize());
+
+    if (args.getFactor() != null) {
+      keyArgs.setFactor(args.getFactor());
+    }
+
+    if (args.getType() != null) {
+      keyArgs.setType(args.getType());
+    }
+
     req.setKeyArgs(keyArgs);
     req.setClientID(clientId);
     req.setExcludeList(excludeList.getProtoBuf());

hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto

@@ -617,6 +617,10 @@ message KeyArgs {
     optional uint32 multipartNumber = 10;
     repeated hadoop.hdds.KeyValue metadata = 11;
     repeated OzoneAclInfo acls = 12;
+    // This will be set when the request is received in pre-Execute. This
+    // value is used in setting creation/modification time depending on the
+    // request type.
+    optional uint64 modificationTime = 13;
 }
 
 message KeyLocation {
@@ -712,6 +716,9 @@ message ListStatusResponse {
 
 message CreateKeyRequest {
     required KeyArgs keyArgs = 1;
+    // Set in OM HA during preExecute step. This way all OM's use same ID in
+    // OM HA.
+    optional uint64 clientID = 2;
 }
 
 message CreateKeyResponse {

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java

@@ -103,7 +103,7 @@ public void restartOzoneManager() throws IOException {
   public void stop() {
     for (OzoneManager ozoneManager : ozoneManagers) {
       if (ozoneManager != null) {
-        LOG.info("Stopping the OzoneManager " + ozoneManager.getOMNodId());
+        LOG.info("Stopping the OzoneManager " + ozoneManager.getOMNodeId());
         ozoneManager.stop();
         ozoneManager.join();
       }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java

@@ -479,7 +479,7 @@ public void testOMProxyProviderInitialization() throws Exception {
         }
       }
       Assert.assertTrue("There is no OM Client Proxy corresponding to OM " +
-              "node" + cluster.getOzoneManager(i).getOMNodId(),
+              "node" + cluster.getOzoneManager(i).getOMNodeId(),
           omClientProxyExists);
     }
   }
@@ -604,7 +604,7 @@ public void testReadRequest() throws Exception {
 
       // Failover to the OM node that the objectStore points to
       omFailoverProxyProvider.performFailoverIfRequired(
-          ozoneManager.getOMNodId());
+          ozoneManager.getOMNodeId());
 
       // A read request should result in the proxyProvider failing over to
       // leader node.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java

@@ -282,7 +282,7 @@ public OmKeyLocationInfo addAllocatedBlock(OmKeyArgs args, long clientID,
 
     OmKeyLocationInfo omKeyLocationInfo =
         OmKeyLocationInfo.getFromProtobuf(keyLocation);
-    keyInfo.appendNewBlocks(Collections.singletonList(omKeyLocationInfo));
+    keyInfo.appendNewBlocks(Collections.singletonList(omKeyLocationInfo), true);
     keyInfo.updateModifcationTime();
     metadataManager.getOpenKeyTable().put(openKey, keyInfo);
     return omKeyLocationInfo;
@@ -317,7 +317,7 @@ public OmKeyLocationInfo allocateBlock(OmKeyArgs args, long clientID,
     // If om is not managing via ratis, write to db, otherwise write to DB
     // will happen via ratis apply transaction.
     if (!isRatisEnabled) {
-      keyInfo.appendNewBlocks(locationInfos);
+      keyInfo.appendNewBlocks(locationInfos, true);
       keyInfo.updateModifcationTime();
       metadataManager.getOpenKeyTable().put(openKey, keyInfo);
     }
@@ -486,7 +486,7 @@ private void allocateBlockInKey(OmKeyInfo keyInfo, long size, long sessionId)
     if (size > 0) {
       List<OmKeyLocationInfo> locationInfos =
           allocateBlock(keyInfo, new ExcludeList(), size);
-      keyInfo.appendNewBlocks(locationInfos);
+      keyInfo.appendNewBlocks(locationInfos, true);
     }
 
     // When OM is not managed via ratis we should write in to Om db in
@@ -509,7 +509,7 @@ private OmKeyInfo prepareKeyInfo(
       // the key already exist, the new blocks will be added as new version
       // when locations.size = 0, the new version will have identical blocks
       // as its previous version
-      keyInfo.addNewVersion(locations);
+      keyInfo.addNewVersion(locations, true);
       keyInfo.setDataSize(size + keyInfo.getDataSize());
     }
     return keyInfo;
@@ -632,8 +632,8 @@ public void commitKey(OmKeyArgs args, long clientID) throws IOException {
       validateBucket(volumeName, bucketName);
       OmKeyInfo keyInfo = metadataManager.getOpenKeyTable().get(openKey);
       if (keyInfo == null) {
-        throw new OMException("Commit a key without corresponding entry " +
-            objectKey, KEY_NOT_FOUND);
+        throw new OMException("Failed to commit key, as " + openKey + "entry " +
+            "is not found in the openKey table", KEY_NOT_FOUND);
       }
       keyInfo.setDataSize(args.getDataSize());
       keyInfo.setModificationTime(Time.now());

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java

@@ -473,10 +473,7 @@ public boolean isVolumeEmpty(String volume) throws IOException {
     try (TableIterator<String, ? extends KeyValue<String, OmBucketInfo>>
         bucketIter = bucketTable.iterator()) {
       KeyValue<String, OmBucketInfo> kv = bucketIter.seek(volumePrefix);
-      // During iteration from DB, check in mean time if this bucket is not
-      // marked for delete.
-      if (kv != null && kv.getKey().startsWith(volumePrefix) &&
-          bucketTable.get(kv.getKey()) != null) {
+      if (kv != null && kv.getKey().startsWith(volumePrefix)) {
         return false; // we found at least one bucket with this volume prefix.
       }
     }
@@ -495,8 +492,20 @@ public boolean isVolumeEmpty(String volume) throws IOException {
   public boolean isBucketEmpty(String volume, String bucket)
       throws IOException {
     String keyPrefix = getBucketKey(volume, bucket);
-    //TODO: When Key ops are converted in to HA model, use cache also to
-    // determine bucket is empty or not.
+
+    // First check in key table cache.
+    Iterator<Map.Entry<CacheKey<String>, CacheValue<OmKeyInfo>>> iterator =
+        ((TypedTable< String, OmKeyInfo>) keyTable).cacheIterator();
+    while (iterator.hasNext()) {
+      Map.Entry< CacheKey<String>, CacheValue<OmKeyInfo>> entry =
+          iterator.next();
+      String key = entry.getKey().getCacheKey();
+      OmKeyInfo omKeyInfo = entry.getValue().getCacheValue();
+      // Making sure that entry is not for delete key request.
+      if (key.startsWith(keyPrefix) && omKeyInfo != null) {
+        return false;
+      }
+    }
     try (TableIterator<String, ? extends KeyValue<String, OmKeyInfo>> keyIter =
         keyTable.iterator()) {
       KeyValue<String, OmKeyInfo> kv = keyIter.seek(keyPrefix);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -33,6 +33,7 @@
 import java.util.Objects;
 
 import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.StorageUnit;
 import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
@@ -101,6 +102,7 @@
 import org.apache.hadoop.ozone.audit.AuditMessage;
 import org.apache.hadoop.ozone.audit.Auditor;
 import org.apache.hadoop.ozone.audit.OMAction;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.common.Storage.StorageState;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes;
@@ -170,16 +172,24 @@
 import java.util.TimerTask;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED;
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED_DEFAULT;
 import static org.apache.hadoop.hdds.HddsUtils.getScmAddressForBlockClients;
 import static org.apache.hadoop.hdds.HddsUtils.getScmAddressForClients;
 import static org.apache.hadoop.hdds.HddsUtils.getScmAddressForSecurityProtocol;
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY;
 import static org.apache.hadoop.hdds.security.x509.certificates.utils.CertificateSignRequest.getEncodedString;
 import static org.apache.hadoop.hdds.server.ServerUtils.updateRPCListenAddress;
 import static org.apache.hadoop.io.retry.RetryPolicies.retryUpToMaximumCountWithFixedSleep;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_CONTAINER_RATIS_ENABLED_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.DFS_CONTAINER_RATIS_ENABLED_KEY;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_ENABLED;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_ENABLED_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_KEY_PREALLOCATION_BLOCKS_MAX;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_KEY_PREALLOCATION_BLOCKS_MAX_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_METRICS_FILE;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_METRICS_TEMP_FILE;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_RATIS_SNAPSHOT_INDEX;
@@ -275,6 +285,13 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
   private long maxUserVolumeCount;
 
 
+  private final ScmClient scmClient;
+  private final long scmBlockSize;
+  private final int preallocateBlocksMax;
+  private final boolean grpcBlockTokenEnabled;
+  private final boolean useRatisForReplication;
+
+
   private OzoneManager(OzoneConfiguration conf) throws IOException,
       AuthenticationException {
     super(OzoneVersionInfo.OZONE_VERSION_INFO);
@@ -385,8 +402,8 @@ private OzoneManager(OzoneConfiguration conf) throws IOException,
     omRpcServer = getRpcServer(conf);
     omRpcAddress = updateRPCListenAddress(configuration,
         OZONE_OM_ADDRESS_KEY, omNodeRpcAddr, omRpcServer);
-    keyManager = new KeyManagerImpl(
-        new ScmClient(scmBlockClient, scmContainerClient), metadataManager,
+    this.scmClient = new ScmClient(scmBlockClient, scmContainerClient);
+    keyManager = new KeyManagerImpl(scmClient, metadataManager,
         configuration, omStorage.getOmId(), blockTokenMgr, getKmsProvider());
 
     prefixManager = new PrefixManagerImpl(metadataManager);
@@ -404,6 +421,63 @@ private OzoneManager(OzoneConfiguration conf) throws IOException,
       accessAuthorizer = null;
     }
     omMetaDir = OmUtils.getOmDbDir(configuration);
+
+    this.scmBlockSize = (long) conf
+        .getStorageSize(OZONE_SCM_BLOCK_SIZE, OZONE_SCM_BLOCK_SIZE_DEFAULT,
+            StorageUnit.BYTES);
+    this.preallocateBlocksMax = conf.getInt(
+        OZONE_KEY_PREALLOCATION_BLOCKS_MAX,
+        OZONE_KEY_PREALLOCATION_BLOCKS_MAX_DEFAULT);
+    this.grpcBlockTokenEnabled = conf.getBoolean(
+        HDDS_BLOCK_TOKEN_ENABLED,
+        HDDS_BLOCK_TOKEN_ENABLED_DEFAULT);
+    this.useRatisForReplication = conf.getBoolean(
+        DFS_CONTAINER_RATIS_ENABLED_KEY, DFS_CONTAINER_RATIS_ENABLED_DEFAULT);
+  }
+
+  /**
+   * Return configuration value of
+   * {@link OzoneConfigKeys#DFS_CONTAINER_RATIS_ENABLED_KEY}.
+   */
+  public boolean shouldUseRatis() {
+    return useRatisForReplication;
+  }
+
+  /**
+   * Return scmClient.
+   */
+  public ScmClient getScmClient() {
+    return scmClient;
+  }
+
+  /**
+   * Return SecretManager for OM.
+   */
+  public OzoneBlockTokenSecretManager getBlockTokenSecretManager() {
+    return blockTokenMgr;
+  }
+
+  /**
+   * Return config value of {@link OzoneConfigKeys#OZONE_SCM_BLOCK_SIZE}.
+   */
+  public long getScmBlockSize() {
+    return scmBlockSize;
+  }
+
+  /**
+   * Return config value of
+   * {@link OzoneConfigKeys#OZONE_KEY_PREALLOCATION_BLOCKS_MAX}.
+   */
+  public int getPreallocateBlocksMax() {
+    return preallocateBlocksMax;
+  }
+
+  /**
+   * Return config value of
+   * {@link HddsConfigKeys#HDDS_BLOCK_TOKEN_ENABLED}.
+   */
+  public boolean isGrpcBlockTokenEnabled() {
+    return grpcBlockTokenEnabled;
   }
 
   /**
@@ -3057,7 +3131,7 @@ public static void setTestSecureOmFlag(boolean testSecureOmFlag) {
     OzoneManager.testSecureOmFlag = testSecureOmFlag;
   }
 
-  public String getOMNodId() {
+  public String getOMNodeId() {
     return omNodeDetails.getOMNodeId();
   }