HDDS-11901. use om managed index for handling request

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -303,6 +303,7 @@ message OMRequest {
   optional GetObjectTaggingRequest          getObjectTaggingRequest        = 140;
   optional PutObjectTaggingRequest          putObjectTaggingRequest        = 141;
   optional DeleteObjectTaggingRequest       deleteObjectTaggingRequest     = 142;
+  optional ExecutionControlRequest          executionControlRequest        = 143;
 }
 
 message OMResponse {
@@ -2300,6 +2301,10 @@ message DeleteObjectTaggingRequest {
 message DeleteObjectTaggingResponse {
 }
 
+message ExecutionControlRequest {
+  required uint64 index = 1;
+}
+
 /**
  The OM service that takes care of Ozone namespace.
 */

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

@@ -0,0 +1,116 @@
+/*
+ * 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.execution;
+
+import static org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature.MANAGED_INDEX;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.ozone.om.OzoneManager;
+
+/**
+ * Manages indexes for request handling and persist.
+ */
+public final class IndexManager {
+  public static final String OM_INDEX_KEY = "#OMINDEX";
+
+  private final AtomicLong index = new AtomicLong();
+  private final AtomicLong commitIndex = new AtomicLong();
+  private final OzoneManager ozoneManager;
+  private final AtomicBoolean enabled = new AtomicBoolean(true);
+
+  public IndexManager(OzoneManager ozoneManager) throws IOException {
+    this.ozoneManager = ozoneManager;
+    initialize();
+  }
+
+  public void initialize() throws IOException {
+    if (!ozoneManager.getVersionManager().isAllowed(MANAGED_INDEX)) {
+      enabled.set(false);
+      return;
+    }
+
+    // default first time starts with "0"
+    long initIndex = 0;
+    // retrieve last saved index
+    TransactionInfo transactionInfo = ozoneManager.getMetadataManager().getTransactionInfoTable().get(OM_INDEX_KEY);
+    if (null == transactionInfo) {
+      // use ratis transaction for first time upgrade
+      transactionInfo = TransactionInfo.readTransactionInfo(ozoneManager.getMetadataManager());
+    }
+    if (null != transactionInfo) {
+      initIndex = transactionInfo.getTransactionIndex();
+    }
+    index.set(initIndex);
+    commitIndex.set(initIndex);
+  }
+
+  public void finalizeFeature() throws IOException {
+    if (enabled.get()) {
+      return;
+    }
+
+    // reinit the feature on finalization
+    long initIndex = 0;
+    TransactionInfo transactionInfo = TransactionInfo.readTransactionInfo(ozoneManager.getMetadataManager());
+    if (null != transactionInfo) {
+      initIndex = transactionInfo.getTransactionIndex();
+    }
+    index.set(initIndex);
+    enabled.set(true);
+
+    try (BatchOperation batchOperation = ozoneManager.getMetadataManager().getStore()
+        .initBatchOperation()) {
+      saveIndex(batchOperation, initIndex);
+      ozoneManager.getMetadataManager().getStore().commitBatchOperation(batchOperation);
+    }
+  }
+
+  public long nextIndex() {
+    if (!enabled.get()) {
+      return -1;
+    }
+    return index.incrementAndGet();
+  }
+
+  /**
+   *  Follower on every transaction update the commit index via saveIndex().
+   *  When the follower becomes leader onLeaderChange(), it updates index with Max (commit index, current index).
+   *  Max is done for purpose where follower can have higher index (not yet sync to other nodes) and being discarded
+   *  for continuation on that node.
+   */
+  public void onLeaderChange() {
+    index.set(Math.max(commitIndex.get(), index.get()));
+  }
+
+  public synchronized void saveIndex(BatchOperation batchOperation, long idx) throws IOException {
+    if (!enabled.get()) {
+      return;
+    }
+    if (idx <= commitIndex.get()) {
+      return;
+    }
+
+    ozoneManager.getMetadataManager().getTransactionInfoTable().putWithBatch(batchOperation, OM_INDEX_KEY,
+        TransactionInfo.valueOf(-1, idx));
+    commitIndex.set(idx);
+  }
+}

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

@@ -21,13 +21,16 @@
 
 import com.google.protobuf.ServiceException;
 import java.io.IOException;
+import java.util.function.Supplier;
 import org.apache.hadoop.ozone.om.OMPerformanceMetrics;
 import org.apache.hadoop.ozone.om.OzoneManager;
 import org.apache.hadoop.ozone.om.helpers.OMAuditLogger;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils;
 import org.apache.hadoop.ozone.om.request.OMClientRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.ratis.protocol.ClientId;
 
 /**
  * entry for execution flow for write request.
@@ -36,10 +39,25 @@ public class OMExecutionFlow {
 
   private final OzoneManager ozoneManager;
   private final OMPerformanceMetrics perfMetrics;
+  private final Supplier<Long> indexGenerator;
 
-  public OMExecutionFlow(OzoneManager om) {
+  public OMExecutionFlow(OzoneManager om) throws IOException {
     this.ozoneManager = om;
     this.perfMetrics = ozoneManager.getPerfMetrics();
+    indexGenerator = om.getOmRatisServer().getOmStateMachine().getIndexManager()::nextIndex;
+  }
+
+  /**
+   * Internal request handling with defined clientId and callId.
+   *
+   * @param omRequest the request
+   * @param clientId the clientId
+   * @param callId the callId
+   * @return OMResponse the response of execution
+   * @throws ServiceException the exception on execution
+   */
+  public OMResponse submitInternal(OMRequest omRequest, ClientId clientId, long callId) throws ServiceException {
+    return ozoneManager.getOmRatisServer().submitRequest(updateControlRequest(omRequest), clientId, callId);
   }
 
   /**
@@ -57,7 +75,7 @@ public OMResponse submit(OMRequest omRequest) throws ServiceException {
   private OMResponse submitExecutionToRatis(OMRequest request) throws ServiceException {
     // 1. create client request and preExecute
     OMClientRequest omClientRequest = null;
-    final OMRequest requestToSubmit;
+    OMRequest requestToSubmit;
     try {
       omClientRequest = OzoneManagerRatisUtils.createClientRequest(request, ozoneManager);
       assert (omClientRequest != null);
@@ -73,10 +91,18 @@ private OMResponse submitExecutionToRatis(OMRequest request) throws ServiceExcep
     }
 
     // 2. submit request to ratis
+    requestToSubmit = updateControlRequest(requestToSubmit);
     OMResponse response = ozoneManager.getOmRatisServer().submitRequest(requestToSubmit);
     if (!response.getSuccess()) {
       omClientRequest.handleRequestFailure(ozoneManager);
     }
     return response;
   }
+
+  private OMRequest updateControlRequest(OMRequest requestToSubmit) {
+    OzoneManagerProtocolProtos.ExecutionControlRequest controlRequest =
+        OzoneManagerProtocolProtos.ExecutionControlRequest.newBuilder().setIndex(indexGenerator.get()).build();
+    requestToSubmit = requestToSubmit.toBuilder().setExecutionControlRequest(controlRequest).build();
+    return requestToSubmit;
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/execution/flowcontrol/ExecutionContext.java

@@ -27,10 +27,11 @@ public final class ExecutionContext {
   private final TermIndex termIndex;
 
   private ExecutionContext(long index, TermIndex termIndex) {
-    this.index = index;
     if (null == termIndex) {
+      // termIndex will be null for pre-ratis execution case which is before ratis transaction
       termIndex = TermIndex.valueOf(-1, index);
     }
+    this.index = index;
     this.termIndex = termIndex;
   }
 

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

@@ -27,9 +27,9 @@
 import org.apache.hadoop.ozone.audit.AuditMessage;
 import org.apache.hadoop.ozone.audit.OMAction;
 import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.execution.flowcontrol.ExecutionContext;
 import org.apache.hadoop.ozone.om.request.OMClientRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
-import org.apache.ratis.server.protocol.TermIndex;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -115,12 +115,12 @@ private static OMAction getAction(OzoneManagerProtocolProtos.OMRequest request)
     return omAction;
   }
 
-  public static void log(OMAuditLogger.Builder builder, TermIndex termIndex) {
+  public static void log(OMAuditLogger.Builder builder, ExecutionContext context) {
     if (builder.isLog.get()) {
       if (null == builder.getAuditMap()) {
         builder.setAuditMap(new HashMap<>());
       }
-      builder.getAuditMap().put("Transaction", String.valueOf(termIndex.getIndex()));
+      builder.getAuditMap().put("Transaction", context.getTermIndex().getIndex() + "::" + context.getIndex());
       builder.getMessageBuilder().withParams(builder.getAuditMap());
       builder.getAuditLogger().logWrite(builder.getMessageBuilder().build());
     }
@@ -134,7 +134,7 @@ public static void log(OMAuditLogger.Builder builder) {
   }
 
   public static void log(OMAuditLogger.Builder builder, OMClientRequest request, OzoneManager om,
-                         TermIndex termIndex, Throwable th) {
+                         ExecutionContext context, Throwable th) {
     if (builder.isLog.get()) {
       builder.getAuditLogger().logWrite(builder.getMessageBuilder().build());
       return;
@@ -150,7 +150,7 @@ public static void log(OMAuditLogger.Builder builder, OMClientRequest request, O
     }
     try {
       builder.getAuditMap().put("Command", request.getOmRequest().getCmdType().name());
-      builder.getAuditMap().put("Transaction", String.valueOf(termIndex.getIndex()));
+      builder.getAuditMap().put("Transaction", context.getTermIndex().getIndex() + "::" + context.getIndex());
       request.buildAuditMessage(action, builder.getAuditMap(),
           th, request.getUserInfo());
       builder.setLog(true);

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

@@ -54,6 +54,7 @@
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.ratis.util.ExitUtils;
 import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.util.function.CheckedBiConsumer;
 import org.apache.ratis.util.function.CheckedRunnable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -77,10 +78,12 @@ public final class OzoneManagerDoubleBuffer {
   private static class Entry {
     private final TermIndex termIndex;
     private final OMClientResponse response;
+    private final long index;
 
-    Entry(TermIndex termIndex, OMClientResponse response) {
+    Entry(TermIndex termIndex, OMClientResponse response, long index) {
       this.termIndex = termIndex;
       this.response = response;
+      this.index = index;
     }
 
     TermIndex getTermIndex() {
@@ -90,6 +93,10 @@ TermIndex getTermIndex() {
     OMClientResponse getResponse() {
       return response;
     }
+
+    long getIndex() {
+      return index;
+    }
   }
 
   /**
@@ -98,6 +105,7 @@ OMClientResponse getResponse() {
   public static final class Builder {
     private OMMetadataManager omMetadataManager;
     private Consumer<TermIndex> updateLastAppliedIndex = termIndex -> { };
+    private CheckedBiConsumer<BatchOperation, Long, IOException> updateOmCommitIndex = (m, n) -> { };
     private boolean isTracingEnabled = false;
     private int maxUnFlushedTransactionCount = 0;
     private FlushNotifier flushNotifier;
@@ -116,6 +124,11 @@ Builder setUpdateLastAppliedIndex(Consumer<TermIndex> updateLastAppliedIndex) {
       return this;
     }
 
+    Builder setUpdateOmCommitIndex(CheckedBiConsumer<BatchOperation, Long, IOException> updateOmCommitIndex) {
+      this.updateOmCommitIndex = updateOmCommitIndex;
+      return this;
+    }
+
     public Builder enableTracing(boolean enableTracing) {
       this.isTracingEnabled = enableTracing;
       return this;
@@ -177,6 +190,7 @@ static Semaphore newSemaphore(int permits) {
   private final OMMetadataManager omMetadataManager;
 
   private final Consumer<TermIndex> updateLastAppliedIndex;
+  private final CheckedBiConsumer<BatchOperation, Long, IOException> updateOmCommitIndex;
 
   private final S3SecretManager s3SecretManager;
 
@@ -196,6 +210,7 @@ private OzoneManagerDoubleBuffer(Builder b) {
     this.omMetadataManager = b.omMetadataManager;
     this.s3SecretManager = b.s3SecretManager;
     this.updateLastAppliedIndex = b.updateLastAppliedIndex;
+    this.updateOmCommitIndex = b.updateOmCommitIndex;
     this.flushNotifier = b.flushNotifier;
     this.unFlushedTransactions = newSemaphore(b.maxUnFlushedTransactionCount);
 
@@ -330,6 +345,7 @@ private void flushBatch(Queue<Entry> buffer) throws IOException {
         .map(Entry::getTermIndex)
         .sorted()
         .collect(Collectors.toList());
+    final long index = buffer.stream().mapToLong(Entry::getIndex).max().orElse(0);
     final int flushedTransactionsSize = flushedTransactions.size();
     final TermIndex lastTransaction = flushedTransactions.get(flushedTransactionsSize - 1);
 
@@ -347,6 +363,8 @@ private void flushBatch(Queue<Entry> buffer) throws IOException {
           () -> omMetadataManager.getTransactionInfoTable().putWithBatch(
               batchOperation, TRANSACTION_INFO_KEY, TransactionInfo.valueOf(lastTransaction)));
 
+      updateOmCommitIndex.accept(batchOperation, index);
+
       long startTime = Time.monotonicNow();
       flushBatchWithTrace(lastTraceId, buffer.size(),
           () -> omMetadataManager.getStore()
@@ -458,7 +476,7 @@ private void addCleanupEntry(Entry entry, Map<String, List<Long>> cleanupEpochs)
       }
       for (String table : cleanupTables) {
         cleanupEpochs.computeIfAbsent(table, list -> new ArrayList<>())
-            .add(entry.getTermIndex().getIndex());
+            .add(entry.getIndex());
       }
     } else {
       // This is to catch early errors, when a new response class missed to
@@ -527,7 +545,14 @@ 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));
+    add(response, termIndex, termIndex.getIndex());
+  }
+
+  /**
+   * Add OmResponseBufferEntry to buffer.
+   */
+  public synchronized void add(OMClientResponse response, TermIndex termIndex, long index) {
+    currentBuffer.add(new Entry(termIndex, response, index));
     notify();
   }