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/OzoneManagerCompletedRequestInfoProvider.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
+ *
+ *      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.ratis;
+
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo.OperationArgs;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class when given a request will determine if the request is one of
+ * a subset of write requests which we want to save to the
+ * OmCompletedRequestInfo ledger and if so it will return the
+ * appropriate instance.
+ *
+ * NOTE: as per elsewhere - this class may not need to exist if we go
+ * with the approach of the subset of response handlers which we want to
+ * populate a OmCompletedRequestInfo ledger row can do so themselves
+ * (e.g. in OMKeyCommitResponse::addToBatch)
+ */
+public final class OzoneManagerCompletedRequestInfoProvider {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OzoneManagerCompletedRequestInfoProvider.class);
+
+  public OmCompletedRequestInfo get(long trxLogIndex, OzoneManagerProtocolProtos.OMRequest omRequest) {
+
+    switch (omRequest.getCmdType()) {
+    case CreateVolume:
+      logRequest("CreateVolume", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
+          .setVolumeName(omRequest.getCreateVolumeRequest().getVolumeInfo().getVolume())
+          .setOpArgs(new OperationArgs.CreateVolumeArgs())
+          .build();
+
+    case DeleteVolume:
+      logRequest("DeleteVolume", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
+          .setVolumeName(omRequest.getDeleteVolumeRequest().getVolumeName())
+          .setOpArgs(new OperationArgs.DeleteVolumeArgs())
+          .build();
+
+    case CreateBucket:
+      logRequest("CreateBucket", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
+          .setVolumeName(omRequest.getCreateBucketRequest().getBucketInfo().getVolumeName())
+          .setBucketName(omRequest.getCreateBucketRequest().getBucketInfo().getBucketName())
+          .setOpArgs(new OperationArgs.CreateBucketArgs())
+          .build();
+
+    case DeleteBucket:
+      logRequest("DeleteBucket", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
+          .setVolumeName(omRequest.getDeleteBucketRequest().getVolumeName())
+          .setBucketName(omRequest.getDeleteBucketRequest().getBucketName())
+          .setOpArgs(new OperationArgs.DeleteBucketArgs())
+          .build();
+
+    case CreateKey:
+      logRequest("CreateKey", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCreateKeyRequest().getKeyArgs())
+          .setOpArgs(new OperationArgs.CreateKeyArgs())
+          .build();
+
+    case RenameKey:
+      logRequest("RenameKey", omRequest);
+      OzoneManagerProtocolProtos.RenameKeyRequest renameReq
+          = (OzoneManagerProtocolProtos.RenameKeyRequest) omRequest.getRenameKeyRequest();
+
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getRenameKeyRequest().getKeyArgs())
+          .setOpArgs(new OperationArgs.RenameKeyArgs(renameReq.getToKeyName()))
+          .build();
+
+    case DeleteKey:
+      logRequest("DeleteKey", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getDeleteKeyRequest().getKeyArgs())
+          .setOpArgs(new OperationArgs.DeleteKeyArgs())
+          .build();
+
+    case CommitKey:
+      logRequest("CommitKey", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCommitKeyRequest().getKeyArgs())
+          .setOpArgs(new OperationArgs.CommitKeyArgs())
+          .build();
+
+    case CreateDirectory:
+      logRequest("CreateDirectory", omRequest);
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCreateDirectoryRequest().getKeyArgs())
+          .setOpArgs(new OperationArgs.CreateDirectoryArgs())
+          .build();
+
+    case CreateFile:
+      logRequest("CreateFile", omRequest);
+
+      OzoneManagerProtocolProtos.CreateFileRequest createFileReq
+          = (OzoneManagerProtocolProtos.CreateFileRequest) omRequest.getCreateFileRequest();
+
+      return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCreateFileRequest().getKeyArgs())
+          .setOpArgs(new OperationArgs.CreateFileArgs(createFileReq.getIsRecursive(),
+                                                      createFileReq.getIsOverwrite()))
+          .build();
+
+    default:
+      LOG.debug("Unhandled cmdType={}", omRequest.getCmdType());
+      return null;
+    }
+  }
+
+  private static void logRequest(String label, OzoneManagerProtocolProtos.OMRequest omRequest) {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("---> {} {}", label, omRequest);
+    }
+  }
+
+  private OmCompletedRequestInfo.Builder requestInfoBuilder(long trxLogIndex,
+                                                            OzoneManagerProtocolProtos.Type cmdType,
+                                                            OzoneManagerProtocolProtos.KeyArgs keyArgs) {
+
+    OmCompletedRequestInfo.Builder builder = OmCompletedRequestInfo.newBuilder()
+        .setTrxLogIndex(trxLogIndex)
+        .setCmdType(cmdType)
+        .setCreationTime(System.currentTimeMillis());
+
+    if (keyArgs != null) {
+      builder.setVolumeName(keyArgs.getVolumeName());
+      builder.setBucketName(keyArgs.getBucketName());
+      builder.setKeyName(keyArgs.getKeyName());
+    }
+
+    return builder;
+  }
+}

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

@@ -44,6 +44,7 @@
 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.OMClientResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
@@ -106,10 +107,13 @@ public final class OzoneManagerDoubleBuffer {
   private static class Entry {
     private final TermIndex termIndex;
     private final OMClientResponse response;
+    // strawman approach: see comments in addToBatch()
+    private final OmCompletedRequestInfo completedRequestInfo;
 
-    Entry(TermIndex termIndex, OMClientResponse response) {
+    Entry(TermIndex termIndex, OMClientResponse response, OmCompletedRequestInfo completedRequestInfo) {
       this.termIndex = termIndex;
       this.response = response;
+      this.completedRequestInfo = completedRequestInfo;
     }
 
     TermIndex getTermIndex() {
@@ -119,6 +123,10 @@ TermIndex getTermIndex() {
     OMClientResponse getResponse() {
       return response;
     }
+
+    OmCompletedRequestInfo getCompletedRequestInfo() {
+      return completedRequestInfo;
+    }
   }
 
   /**
@@ -404,11 +412,29 @@ private String addToBatch(Queue<Entry> buffer, BatchOperation batchOperation) {
     for (Entry entry: buffer) {
       OMClientResponse response = entry.getResponse();
       OMResponse omResponse = response.getOMResponse();
+      OmCompletedRequestInfo completedRequestInfo = entry.getCompletedRequestInfo();
       lastTraceId = omResponse.getTraceID();
 
       try {
         addToBatchWithTrace(omResponse,
             () -> response.checkAndUpdateDB(omMetadataManager, batchOperation));
+
+        // This is a strawman approach and requires some discussion
+        // with the community on approach.
+        //
+        // TODO: would it be better to have each type of request we want
+        // to capture populate a row in the ledger in the addToBatch
+        // callback of the response (triggered above) e.g.
+        // OMKeyCommitResponse::addToBatch could generate the suitable
+        // OmCompletedRequestInfo row and populate (Q: if we went with
+        // that approach then would it have access to all the necessary
+        // request parameters? e.g. termIndex.getIndex(), cmdType and
+        // all captured request parameters
+        if (completedRequestInfo != null) {
+          omMetadataManager.getCompletedRequestInfoTable().putWithBatch(
+              batchOperation, completedRequestInfo.getTrxLogIndex(), completedRequestInfo);
+        }
+
       } catch (IOException ex) {
         // During Adding to RocksDB batch entry got an exception.
         // We should terminate the OM.
@@ -554,8 +580,9 @@ private void terminate(Throwable t, int status, OMResponse omResponse) {
   /**
    * Add OmResponseBufferEntry to buffer.
    */
-  public synchronized void add(OMClientResponse response, TermIndex termIndex) {
-    currentBuffer.add(new Entry(termIndex, response));
+  public synchronized void add(OMClientResponse response, TermIndex termIndex,
+                               OmCompletedRequestInfo completedRequestInfo) {
+    currentBuffer.add(new Entry(termIndex, response, completedRequestInfo));
     notify();
   }
 

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

@@ -629,7 +629,7 @@ private OMResponse createErrorResponse(
     }
     OMResponse omResponse = omResponseBuilder.build();
     OMClientResponse omClientResponse = new DummyOMClientResponse(omResponse);
-    ozoneManagerDoubleBuffer.add(omClientResponse, termIndex);
+    ozoneManagerDoubleBuffer.add(omClientResponse, termIndex, null);
     return omResponse;
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMPrepareRequest.java

@@ -103,7 +103,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       // the snapshot index in the prepared state.
       OzoneManagerDoubleBuffer doubleBuffer =
           ozoneManager.getOmRatisServer().getOmStateMachine().getOzoneManagerDoubleBuffer();
-      doubleBuffer.add(response, context.getTermIndex());
+      doubleBuffer.add(response, context.getTermIndex(), null);
 
       OzoneManagerRatisServer omRatisServer = ozoneManager.getOmRatisServer();
       final RaftServer.Division division = omRatisServer.getServerDivision();

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

@@ -75,6 +75,7 @@
 import org.apache.hadoop.ozone.om.helpers.ListOpenFilesResult;
 import org.apache.hadoop.ozone.om.helpers.OMAuditLogger;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadList;
@@ -91,6 +92,7 @@
 import org.apache.hadoop.ozone.om.helpers.TenantStateList;
 import org.apache.hadoop.ozone.om.helpers.TenantUserInfoValue;
 import org.apache.hadoop.ozone.om.helpers.TenantUserList;
+import org.apache.hadoop.ozone.om.ratis.OzoneManagerCompletedRequestInfoProvider;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils;
 import org.apache.hadoop.ozone.om.request.OMClientRequest;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
@@ -166,6 +168,7 @@
 import org.apache.hadoop.ozone.upgrade.UpgradeFinalization.StatusAndMessages;
 import org.apache.hadoop.ozone.util.PayloadUtils;
 import org.apache.hadoop.ozone.util.ProtobufUtils;
+import org.apache.ratis.server.protocol.TermIndex;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -177,10 +180,12 @@ public class OzoneManagerRequestHandler implements RequestHandler {
   static final Logger LOG =
       LoggerFactory.getLogger(OzoneManagerRequestHandler.class);
   private final OzoneManager impl;
+  private final OzoneManagerCompletedRequestInfoProvider completedRequestInfoProvider;
   private FaultInjector injector;
 
   public OzoneManagerRequestHandler(OzoneManager om) {
     this.impl = om;
+    this.completedRequestInfoProvider = new OzoneManagerCompletedRequestInfoProvider();
   }
 
   //TODO simplify it to make it shorter
@@ -426,6 +431,12 @@ public OMClientResponse handleWriteRequestImpl(OMRequest omRequest, ExecutionCon
     }
   }
 
+  @Override
+  public OmCompletedRequestInfo handleSuccessfulWriteRequestImpl(TermIndex termIndex, OMRequest omRequest)
+          throws IOException {
+    return completedRequestInfoProvider.get(termIndex.getIndex(), omRequest);
+  }
+
   @VisibleForTesting
   public void setInjector(FaultInjector injector) {
     this.injector = injector;

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

@@ -20,11 +20,13 @@
 import java.io.IOException;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.execution.flowcontrol.ExecutionContext;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
 import org.apache.hadoop.ozone.om.ratis.OzoneManagerDoubleBuffer;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.apache.ratis.server.protocol.TermIndex;
 
 /**
  * Handler to handleRequest the OmRequests.
@@ -61,7 +63,11 @@ default OMClientResponse handleWriteRequest(OMRequest omRequest, ExecutionContex
       OzoneManagerDoubleBuffer ozoneManagerDoubleBuffer) throws IOException {
     final OMClientResponse response = handleWriteRequestImpl(omRequest, context);
     if (omRequest.getCmdType() != Type.Prepare) {
-      ozoneManagerDoubleBuffer.add(response, context.getTermIndex());
+
+      final OmCompletedRequestInfo omCompletedRequestInfo = handleSuccessfulWriteRequestImpl(
+          context.getTermIndex(), omRequest);
+
+      ozoneManagerDoubleBuffer.add(response, context.getTermIndex(), omCompletedRequestInfo);
     }
     return response;
   }
@@ -74,4 +80,6 @@ default OMClientResponse handleWriteRequest(OMRequest omRequest, ExecutionContex
    * @return OMClientResponse
    */
   OMClientResponse handleWriteRequestImpl(OMRequest omRequest, ExecutionContext context) throws IOException;
+
+  OmCompletedRequestInfo handleSuccessfulWriteRequestImpl(TermIndex index, OMRequest omRequest) throws IOException;
 }