HDDS-14004. EventNotification: Capture data to the completed operation ledger table

Author: Colm Dougan

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

@@ -44,7 +44,9 @@
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.S3SecretManager;
 import org.apache.hadoop.ozone.om.codec.OMDBDefinition;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
@@ -409,6 +411,46 @@ private String addToBatch(Queue<Entry> buffer, BatchOperation batchOperation) {
       try {
         addToBatchWithTrace(omResponse,
             () -> response.checkAndUpdateDB(omMetadataManager, batchOperation));
+
+        // This is a strawman approach and requires some discussion
+        // with the community on approach.
+        //
+        // At the moment any response type which we want to capture a
+        // OmCompletedRequestInfo for is required to implement the
+        // interface HasCompletedRequestInfo and the method
+        // getCompletedRequestInfo() and we then have this extra step
+        // here in the double buffer to capture the rows.
+        //
+        // It would seem ideal that the double buffer shouldn't have to
+        // know/care that there is the concept of capturing this
+        // OmCompletedRequestInfo row for certain response times but the
+        // approach described above seemed like the least invasive
+        // approach overall.  I am open to other views.
+        //
+        // Other approaches I considered:
+        // - adding a similar getCompletedRequestInfo method to each
+        // *request* type which want to capture a row for.  The downside
+        // of this was that since requests are not part of the
+        // addToBatch flow the OmCompletedRequestInfo instance then had
+        // to be passed through from the request to the relevant
+        // response constructors and this created quite a bit of code
+        // churn which I perceived as messy
+        //
+        // * in terms of the actual data capture, rather than this
+        // "instanceof" approach in this class I toyed with
+        // having each response type which we want to capture a row for
+        // capturing it itself in its on implementation of addDBBatch
+        // (i.e. no need for any new code in this class).  This
+        // seemed to be a bit messier to me in terms of code duplication
+        //
+        if (response instanceof HasCompletedRequestInfo) {
+          OmCompletedRequestInfo completedRequestInfo = ((HasCompletedRequestInfo) response).getCompletedRequestInfo(
+              entry.getTermIndex().getIndex());
+
+          omMetadataManager.getCompletedRequestInfoTable().putWithBatch(
+              batchOperation, completedRequestInfo.getTrxLogIndex(), completedRequestInfo);
+        }
+
       } catch (IOException ex) {
         // During Adding to RocksDB batch entry got an exception.
         // We should terminate the OM.

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

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.response;
+
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
+
+/**
+ * Interface to define that a response class provides a
+ * OmCompletedRequestInfo implementation.
+ */
+public interface HasCompletedRequestInfo {
+
+  OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex);
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/bucket/OMBucketCreateResponse.java

@@ -26,16 +26,19 @@
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 
 /**
  * Response for CreateBucket request.
  */
 @CleanupTableInfo(cleanupTables = {BUCKET_TABLE, VOLUME_TABLE})
-public final class OMBucketCreateResponse extends OMClientResponse {
+public final class OMBucketCreateResponse extends OMClientResponse implements HasCompletedRequestInfo {
 
   private final OmBucketInfo omBucketInfo;
   private final OmVolumeArgs omVolumeArgs;
@@ -87,5 +90,16 @@ public OmBucketInfo getOmBucketInfo() {
     return omBucketInfo;
   }
 
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.CreateBucket)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(omBucketInfo.getVolumeName())
+        .setBucketName(omBucketInfo.getBucketName())
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.CreateBucketArgs())
+        .build();
+  }
 }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/bucket/OMBucketDeleteResponse.java

@@ -24,16 +24,19 @@
 import java.io.IOException;
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 
 /**
  * Response for DeleteBucket request.
  */
 @CleanupTableInfo(cleanupTables = {BUCKET_TABLE, VOLUME_TABLE})
-public final class OMBucketDeleteResponse extends OMClientResponse {
+public final class OMBucketDeleteResponse extends OMClientResponse implements HasCompletedRequestInfo {
 
   private String volumeName;
   private String bucketName;
@@ -89,5 +92,15 @@ public String getBucketName() {
     return bucketName;
   }
 
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.DeleteBucket)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(volumeName)
+        .setBucketName(bucketName)
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.DeleteBucketArgs())
+        .build();
+  }
 }
-

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/file/OMDirectoryCreateResponse.java

@@ -26,19 +26,22 @@
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.request.file.OMDirectoryCreateRequest.Result;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.response.key.OmKeyResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * Response for create directory request.
  */
 @CleanupTableInfo(cleanupTables = {KEY_TABLE})
-public class OMDirectoryCreateResponse extends OmKeyResponse {
+public class OMDirectoryCreateResponse extends OmKeyResponse implements HasCompletedRequestInfo {
 
   private static final Logger LOG =
       LoggerFactory.getLogger(OMDirectoryCreateResponse.class);
@@ -98,4 +101,17 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
       LOG.debug("Directory already exists. addToDBBatch is a no-op");
     }
   }
+
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.CreateDirectory)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(dirKeyInfo.getVolumeName())
+        .setBucketName(dirKeyInfo.getBucketName())
+        .setKeyName(dirKeyInfo.getKeyName())
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.CreateDirectoryArgs())
+        .build();
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/file/OMFileCreateResponse.java

@@ -24,10 +24,12 @@
 import java.util.List;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
 import org.apache.hadoop.ozone.om.response.key.OMKeyCreateResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 
 /**
  * Response for crate file request.
@@ -53,4 +55,20 @@ public OMFileCreateResponse(@Nonnull OMResponse omResponse, @Nonnull
     checkStatusNotOK();
   }
 
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    OmKeyInfo keyInfo = getOmKeyInfo();
+    boolean isRecursive = false; // ???
+    boolean isOverwrite = false; // ???
+
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.CreateFile)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(keyInfo.getVolumeName())
+        .setBucketName(keyInfo.getBucketName())
+        .setKeyName(keyInfo.getKeyName())
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.CreateFileArgs(isRecursive, isOverwrite))
+        .build();
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponse.java

@@ -30,17 +30,20 @@
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 
 /**
  * Response for CommitKey request.
  */
 @CleanupTableInfo(cleanupTables = {OPEN_KEY_TABLE, KEY_TABLE, DELETED_TABLE,
     BUCKET_TABLE})
-public class OMKeyCommitResponse extends OmKeyResponse {
+public class OMKeyCommitResponse extends OmKeyResponse implements HasCompletedRequestInfo {
 
   private OmKeyInfo omKeyInfo;
   private String ozoneKeyName;
@@ -154,4 +157,17 @@ protected boolean isHSync() {
   public OmKeyInfo getNewOpenKeyInfo() {
     return newOpenKeyInfo;
   }
+
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.CommitKey)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(omKeyInfo.getVolumeName())
+        .setBucketName(omKeyInfo.getBucketName())
+        .setKeyName(omKeyInfo.getKeyName())
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.CommitKeyArgs())
+        .build();
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCreateResponse.java

@@ -28,17 +28,20 @@
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * Response for CreateKey request.
  */
 @CleanupTableInfo(cleanupTables = {OPEN_KEY_TABLE, KEY_TABLE, BUCKET_TABLE})
-public class OMKeyCreateResponse extends OmKeyResponse {
+public class OMKeyCreateResponse extends OmKeyResponse implements HasCompletedRequestInfo {
 
   protected static final Logger LOG =
       LoggerFactory.getLogger(OMKeyCreateResponse.class);
@@ -114,5 +117,18 @@ protected OmKeyInfo getOmKeyInfo() {
   protected OmBucketInfo getOmBucketInfo() {
     return omBucketInfo;
   }
+
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.CreateKey)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(omKeyInfo.getVolumeName())
+        .setBucketName(omKeyInfo.getBucketName())
+        .setKeyName(omKeyInfo.getKeyName())
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.CreateKeyArgs())
+        .build();
+  }
 }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyDeleteResponse.java

@@ -30,15 +30,18 @@
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 
 /**
  * Response for DeleteKey request.
  */
 @CleanupTableInfo(cleanupTables = {KEY_TABLE, OPEN_KEY_TABLE, DELETED_TABLE, BUCKET_TABLE})
-public class OMKeyDeleteResponse extends AbstractOMKeyDeleteResponse {
+public class OMKeyDeleteResponse extends AbstractOMKeyDeleteResponse implements HasCompletedRequestInfo {
 
   private OmKeyInfo omKeyInfo;
   private OmBucketInfo omBucketInfo;
@@ -107,4 +110,16 @@ protected OmBucketInfo getOmBucketInfo() {
     return omBucketInfo;
   }
 
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.DeleteKey)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(omKeyInfo.getVolumeName())
+        .setBucketName(omKeyInfo.getBucketName())
+        .setKeyName(omKeyInfo.getKeyName())
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.DeleteKeyArgs())
+        .build();
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyRenameResponse.java

@@ -25,16 +25,19 @@
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.request.OMClientRequestUtils;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.HasCompletedRequestInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 
 /**
  * Response for RenameKey request.
  */
 @CleanupTableInfo(cleanupTables = {KEY_TABLE, SNAPSHOT_RENAMED_TABLE})
-public class OMKeyRenameResponse extends OmKeyResponse {
+public class OMKeyRenameResponse extends OmKeyResponse implements HasCompletedRequestInfo {
 
   private String fromKeyName;
   private String toKeyName;
@@ -107,4 +110,17 @@ public String getFromKeyName() {
   public String getToKeyName() {
     return toKeyName;
   }
+
+  @Override
+  public OmCompletedRequestInfo getCompletedRequestInfo(long trxnLogIndex) {
+    return OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxnLogIndex)
+        .setCmdType(Type.RenameKey)
+        .setCreationTime(System.currentTimeMillis())
+        .setVolumeName(renameKeyInfo.getVolumeName())
+        .setBucketName(renameKeyInfo.getBucketName())
+        .setKeyName(renameKeyInfo.getKeyName())
+        .setOpArgs(new OmCompletedRequestInfo.OperationArgs.RenameKeyArgs(toKeyName))
+        .build();
+  }
 }