HDDS-1391 : Add ability in OM to serve delta updates through an API.

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/DBStore.java

@@ -172,4 +172,13 @@ <KEY, VALUE> void move(KEY sourceKey, KEY destKey, VALUE value,
    * @return codec registry.
    */
   CodecRegistry getCodecRegistry();
+
+  /**
+   * Get data written to DB since a specific sequence number.
+   * @param sequenceNumber
+   * @return
+   * @throws DataNotFoundException
+   */
+  DBUpdatesWrapper getUpdatesSince(long sequenceNumber)
+      throws DataNotFoundException;
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/DBUpdatesWrapper.java

@@ -0,0 +1,48 @@
+/*
+ * 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.utils.db;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Wrapper class to hold DB data read from the RocksDB log file.
+ */
+public class DBUpdatesWrapper {
+
+  private List<byte[]> dataList = new ArrayList<>();
+  private long currentSequenceNumber = 0;
+
+  public void addWriteBatch(byte[] data, long sequenceNumber) {
+    dataList.add(data);
+    if (currentSequenceNumber < sequenceNumber) {
+      currentSequenceNumber = sequenceNumber;
+    }
+  }
+
+  public List<byte[]> getData() {
+    return dataList;
+  }
+
+  public long getCurrentSequenceNumber() {
+    return currentSequenceNumber;
+  }
+}
+

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/DataNotFoundException.java

@@ -0,0 +1,37 @@
+/*
+ * 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.utils.db;
+
+import java.io.IOException;
+
+/**
+ * Thrown if RocksDB is unable to find requested data from WAL file.
+ */
+public class DataNotFoundException extends IOException {
+
+  public DataNotFoundException() {
+    super();
+  }
+
+  public DataNotFoundException(String message) {
+    super(message);
+  }
+
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStore.java

@@ -45,6 +45,8 @@
 import org.rocksdb.FlushOptions;
 import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
+import org.rocksdb.TransactionLogIterator;
+import org.rocksdb.WriteBatch;
 import org.rocksdb.WriteOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -318,6 +320,44 @@ public CodecRegistry getCodecRegistry() {
     return codecRegistry;
   }
 
+  @Override
+  public DBUpdatesWrapper getUpdatesSince(long sequenceNumber)
+      throws DataNotFoundException {
+
+    DBUpdatesWrapper dbUpdatesWrapper = new DBUpdatesWrapper();
+    try {
+      TransactionLogIterator transactionLogIterator =
+          db.getUpdatesSince(sequenceNumber);
+
+      boolean flag = true;
+
+      while (transactionLogIterator.isValid()) {
+        TransactionLogIterator.BatchResult result =
+            transactionLogIterator.getBatch();
+        long currSequenceNumber = result.sequenceNumber();
+        if (flag && currSequenceNumber > 1 + sequenceNumber) {
+          throw new DataNotFoundException("Unable to read data from " +
+              "RocksDB wal to get delta updates. It may have already been" +
+              "flushed to SSTs.");
+        }
+        flag = false;
+        if (currSequenceNumber == sequenceNumber) {
+          transactionLogIterator.next();
+          continue;
+        }
+        WriteBatch writeBatch = result.writeBatch();
+        byte[] writeBatchData = writeBatch.data();
+        dbUpdatesWrapper.addWriteBatch(writeBatchData,
+            result.sequenceNumber());
+        transactionLogIterator.next();
+      }
+    } catch (RocksDBException e) {
+      LOG.error("Unable to get delta updates since sequenceNumber {} ",
+          sequenceNumber, e);
+    }
+    return dbUpdatesWrapper;
+  }
+
   @VisibleForTesting
   public RocksDB getDb() {
     return db;

hadoop-hdds/common/src/test/java/org/apache/hadoop/utils/db/TestRDBStore.java

@@ -325,4 +325,25 @@ public void testRocksDBKeyMayExistApi() throws Exception {
     }
   }
 
+  @Test
+  public void testGetDBUpdatesSince() throws Exception {
+
+    try (RDBStore newStore =
+             new RDBStore(folder.newFolder(), options, configSet)) {
+
+      try (Table firstTable = newStore.getTable(families.get(1))) {
+        firstTable.put(StringUtils.getBytesUtf16("Key1"), StringUtils
+            .getBytesUtf16("Value1"));
+        firstTable.put(StringUtils.getBytesUtf16("Key2"), StringUtils
+            .getBytesUtf16("Value2"));
+      }
+      Assert.assertTrue(
+          newStore.getDb().getLatestSequenceNumber() == 2);
+
+      DBUpdatesWrapper dbUpdatesSince = newStore.getUpdatesSince(0);
+      Assert.assertEquals(2, dbUpdatesSince.getData().size());
+    }
+  }
+
+
 }

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

@@ -75,6 +75,7 @@ enum Type {
   ServiceList = 51;
 
   ApplyInitiateMultiPartUpload = 52;
+  DBUpdates = 53;
 
   GetDelegationToken = 61;
   RenewDelegationToken = 62;
@@ -139,6 +140,7 @@ message OMRequest {
 
   optional ServiceListRequest               serviceListRequest             = 51;
   optional MultipartInfoApplyInitiateRequest initiateMultiPartUploadApplyRequest = 52;
+  optional DBUpdatesRequest                  dbUpdatesRequest              = 53;
 
   optional hadoop.common.GetDelegationTokenRequestProto getDelegationTokenRequest = 61;
   optional hadoop.common.RenewDelegationTokenRequestProto renewDelegationTokenRequest= 62;
@@ -205,6 +207,7 @@ message OMResponse {
   optional MultipartUploadListPartsResponse listMultipartUploadPartsResponse = 50;
 
   optional ServiceListResponse               ServiceListResponse           = 51;
+  optional DBUpdatesResponse                 dbUpdatesResponse             = 52;
 
   optional GetDelegationTokenResponseProto getDelegationTokenResponse = 61;
   optional RenewDelegationTokenResponseProto renewDelegationTokenResponse = 62;
@@ -842,11 +845,20 @@ message AllocateBlockResponse {
 message ServiceListRequest {
 }
 
+message DBUpdatesRequest {
+    required uint64 sequenceNumber = 1;
+}
+
 message ServiceListResponse {
 
     repeated ServiceInfo serviceInfo = 2;
 }
 
+message DBUpdatesResponse {
+    required uint64 sequenceNumber = 1;
+    repeated bytes data = 2;
+}
+
 message ServicePort {
     enum Type {
         RPC = 1;

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

@@ -59,6 +59,7 @@
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.helpers.OpenKeySession;
 import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DBUpdatesRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServicePort;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.VolumeList;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
@@ -81,6 +82,7 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.Time;
+import org.apache.hadoop.utils.db.DBUpdatesWrapper;
 import org.apache.hadoop.utils.db.RDBStore;
 import org.apache.hadoop.utils.db.Table;
 import org.apache.hadoop.utils.db.Table.KeyValue;
@@ -1395,8 +1397,41 @@ public void testDBKeyMayExist() throws Exception {
     RDBStore rdbStore = (RDBStore) cluster.getOzoneManager()
         .getMetadataManager().getStore();
     RocksDB db = rdbStore.getDb();
-    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
 
+    OmKeyInfo keyInfo = getNewOmKeyInfo();
+    OmKeyInfoCodec omKeyInfoCodec = new OmKeyInfoCodec();
+
+    db.put(StringUtils.getBytesUtf16("OMKey1"),
+        omKeyInfoCodec.toPersistedFormat(keyInfo));
+
+    StringBuilder sb = new StringBuilder();
+    Assert.assertTrue(db.keyMayExist(StringUtils.getBytesUtf16("OMKey1"),
+        sb));
+    Assert.assertTrue(sb.length() > 0);
+  }
+
+
+  @Test
+  public void testGetOMDBUpdates() throws IOException {
+
+    DBUpdatesRequest dbUpdatesRequest =
+        DBUpdatesRequest.newBuilder().setSequenceNumber(0).build();
+
+    DBUpdatesWrapper dbUpdates =
+        cluster.getOzoneManager().getDBUpdates(dbUpdatesRequest);
+    Assert.assertTrue(dbUpdates.getData().isEmpty());
+
+    //Write data to OM.
+    OmKeyInfo keyInfo = getNewOmKeyInfo();
+    Assert.assertNotNull(keyInfo);
+    dbUpdates =
+        cluster.getOzoneManager().getDBUpdates(dbUpdatesRequest);
+    Assert.assertFalse(dbUpdates.getData().isEmpty());
+
+  }
+
+  private OmKeyInfo getNewOmKeyInfo() throws IOException {
+    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
     OmVolumeArgs volumeArgs = OmVolumeArgs.newBuilder()
         .setVolume("vol1")
         .setAdminName("bilbo")
@@ -1423,16 +1458,6 @@ public void testDBKeyMayExist() throws Exception {
         .build();
     OpenKeySession keySession = cluster.getOzoneManager().getKeyManager()
         .openKey(keyArgs);
-    OmKeyInfo keyInfo = keySession.getKeyInfo();
-    OmKeyInfoCodec omKeyInfoCodec = new OmKeyInfoCodec();
-
-    db.put(StringUtils.getBytesUtf16("OMKey1"),
-        omKeyInfoCodec.toPersistedFormat(keyInfo));
-
-    StringBuilder sb = new StringBuilder();
-    Assert.assertTrue(db.keyMayExist(StringUtils.getBytesUtf16("OMKey1"),
-        sb));
-    Assert.assertTrue(sb.length() > 0);
+    return  keySession.getKeyInfo();
   }
-
 }

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

@@ -82,6 +82,7 @@
 import org.apache.hadoop.ozone.om.helpers.S3SecretValue;
 import org.apache.hadoop.ozone.om.protocol.OzoneManagerServerProtocol;
 import org.apache.hadoop.ozone.om.snapshot.OzoneManagerSnapshotProvider;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DBUpdatesRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
     .KeyArgs;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
@@ -147,6 +148,8 @@
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.ShutdownHookManager;
 import org.apache.hadoop.utils.RetriableTask;
+import org.apache.hadoop.utils.db.DBUpdatesWrapper;
+import org.apache.hadoop.utils.db.DataNotFoundException;
 import org.apache.ratis.util.LifeCycle;
 import org.bouncycastle.pkcs.PKCS10CertificationRequest;
 import org.slf4j.Logger;
@@ -3238,4 +3241,18 @@ public boolean isLeader() {
   public boolean isRatisEnabled() {
     return isRatisEnabled;
   }
+
+  /**
+   * Get DB updates since a specific sequence number.
+   * @param dbUpdatesRequest request that encapsulates a sequence number.
+   * @return Wrapper containing the updates.
+   * @throws DataNotFoundException if the DB is unable to read the data.
+   */
+  public DBUpdatesWrapper getDBUpdates(
+      DBUpdatesRequest dbUpdatesRequest)
+      throws DataNotFoundException {
+    return metadataManager.getStore()
+        .getUpdatesSince(dbUpdatesRequest.getSequenceNumber());
+
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java

@@ -124,6 +124,9 @@
 import org.apache.hadoop.security.token.Token;
 
 import com.google.common.collect.Lists;
+
+import org.apache.hadoop.utils.db.DBUpdatesWrapper;
+import org.apache.hadoop.utils.db.DataNotFoundException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -307,6 +310,11 @@ public OMResponse handle(OMRequest request) {
             request.getServiceListRequest());
         responseBuilder.setServiceListResponse(serviceListResponse);
         break;
+      case DBUpdates:
+        DBUpdatesResponse dbUpdatesResponse = getOMDBUpdates(
+            request.getDbUpdatesRequest());
+        responseBuilder.setDbUpdatesResponse(dbUpdatesResponse);
+        break;
       case GetDelegationToken:
         GetDelegationTokenResponseProto getDtResp = getDelegationToken(
             request.getGetDelegationTokenRequest());
@@ -386,6 +394,21 @@ public OMResponse handle(OMRequest request) {
     return responseBuilder.build();
   }
 
+  private DBUpdatesResponse getOMDBUpdates(
+      DBUpdatesRequest dbUpdatesRequest)
+      throws DataNotFoundException {
+
+    DBUpdatesResponse.Builder builder = DBUpdatesResponse
+        .newBuilder();
+    DBUpdatesWrapper dbUpdatesWrapper =
+        impl.getDBUpdates(dbUpdatesRequest);
+    for (int i = 0; i < dbUpdatesWrapper.getData().size(); i++) {
+      builder.setData(i,
+          OMPBHelper.getByteString(dbUpdatesWrapper.getData().get(i)));
+    }
+    return builder.build();
+  }
+
   private GetAclResponse getAcl(GetAclRequest req) throws IOException {
     List<OzoneAclInfo> acls = new ArrayList<>();