Client and Server AlignmentContext

Author: ivandika3

hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/AlignmentContext.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 
+import com.google.protobuf.Message;
 import org.apache.hadoop.ipc_.protobuf.RpcHeaderProtos.RpcRequestHeaderProto;
 import org.apache.hadoop.ipc_.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
 
@@ -86,7 +87,8 @@ long receiveRequestState(RpcRequestHeaderProto header, long threshold)
    *
    * @param protocolName the name of the protocol
    * @param method the method call to check
+   * @param payload the payload to check
    * @return true if this method is async, false otherwise.
    */
-  boolean isCoordinatedCall(String protocolName, String method);
+  boolean isCoordinatedCall(String protocolName, String method, Message payload);
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/ProtobufRpcEngine.java

@@ -582,6 +582,10 @@ RequestHeaderProto getRequestHeader() throws IOException {
       return requestHeader;
     }
 
+    Message getPayload() throws IOException {
+      return payload;
+    }
+
     @Override
     public void writeTo(ResponseBuffer out) throws IOException {
       requestHeader.writeDelimitedTo(out);

hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/Server.java

@@ -2699,16 +2699,18 @@ private void processRpcRequest(RpcRequestHeaderProto header,
           (call.rpcRequest instanceof ProtobufRpcEngine.RpcProtobufRequest)) {
         // if call.rpcRequest is not RpcProtobufRequest, will skip the following
         // step and treat the call as uncoordinated. As currently only certain
-        // ClientProtocol methods request made through RPC protobuf needs to be
+        // OzoneManagerProtocol methods request made through RPC protobuf needs to be
         // coordinated.
         String methodName;
         String protoName;
         ProtobufRpcEngine.RpcProtobufRequest req =
             (ProtobufRpcEngine.RpcProtobufRequest) call.rpcRequest;
+        Message payload;
         try {
           methodName = req.getRequestHeader().getMethodName();
           protoName = req.getRequestHeader().getDeclaringClassProtocolName();
-          if (alignmentContext.isCoordinatedCall(protoName, methodName)) {
+          payload = req.getPayload();
+          if (alignmentContext.isCoordinatedCall(protoName, methodName, payload)) {
             call.markCallCoordinated(true);
             long stateId;
             stateId = alignmentContext.receiveRequestState(

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

@@ -51,6 +51,10 @@ public final class OMConfigKeys {
   public static final int OZONE_OM_DB_MAX_OPEN_FILES_DEFAULT
       = -1;
 
+  public static final String OZONE_OM_STATE_CONTEXT_ENABLED_KEY =
+      "ozone.om.state.context.enabled";
+  public static final boolean OZONE_OM_STATE_CONTEXT_ENABLED_DEFAULT = false;
+
   public static final String OZONE_OM_INTERNAL_SERVICE_ID =
       "ozone.om.internal.service.id";
 

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/ClientAlignmentContext.java

@@ -0,0 +1,89 @@
+/*
+ * 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.ha;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.LongAccumulator;
+import org.apache.hadoop.ipc.AlignmentContext;
+import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto;
+import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
+
+/**
+ * Global State Id context for the client.
+ * <p>
+ * This is the client side implementation responsible for receiving
+ * state alignment info from server(s).
+ */
+public class ClientAlignmentContext implements AlignmentContext {
+
+  private final LongAccumulator lastSeenStateId;
+
+  public ClientAlignmentContext(LongAccumulator lastSeenStateId) {
+    this.lastSeenStateId = lastSeenStateId;
+  }
+
+  @Override
+  public long getLastSeenStateId() {
+    return lastSeenStateId.get();
+  }
+
+  @Override
+  public boolean isCoordinatedCall(String protocolName, String method) {
+    throw new UnsupportedOperationException(
+        "Client should not be checking uncoordinated call");
+  }
+
+  /**
+   * Client side implementation only receives state alignment info.
+   * It does not provide state alignment info therefore this does nothing.
+   */
+  @Override
+  public void updateResponseState(RpcResponseHeaderProto.Builder header) {
+    // Do nothing.
+  }
+
+  /**
+   * Client side implementation for receiving state alignment info
+   * in responses.
+   */
+  @Override
+  public synchronized void receiveResponseState(RpcResponseHeaderProto header) {
+    lastSeenStateId.accumulate(header.getStateId());
+  }
+
+  /**
+   * Client side implementation for providing state alignment info in requests.
+   */
+  @Override
+  public synchronized void updateRequestState(RpcRequestHeaderProto.Builder header) {
+    if (lastSeenStateId.get() != Long.MIN_VALUE) {
+      header.setStateId(lastSeenStateId.get());
+    }
+  }
+
+  /**
+   * Client side implementation only provides state alignment info in requests.
+   * Client does not receive RPC requests therefore this does nothing.
+   */
+  @Override
+  public long receiveRequestState(RpcRequestHeaderProto header, long threshold)
+      throws IOException {
+    // Do nothing.
+    return 0;
+  }
+}

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProviderBase.java

@@ -146,6 +146,7 @@ protected T createOMProxy(InetSocketAddress omAddress) throws IOException {
         NetUtils.getDefaultSocketFactory(hadoopConf),
         (int) OmUtils.getOMClientRpcTimeOut(getConf()),
         connectionRetryPolicy
+        // TODO: Implement and add alignment context here
     ).getProxy();
   }
 

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

@@ -1158,4 +1158,22 @@ void testOMRatisSnapshot() throws Exception {
             "snapshot indices");
 
   }
+
+  /**
+   * This test checks if after a client writes we can see the state id in
+   * updated via the response.
+   */
+  @Test
+  void testStateTransferOnWrite() throws Exception {
+
+  }
+
+  /**
+   * This test checks if after a client reads we can see the state id in
+   * updated via the response.
+   */
+  @Test
+  void testStateTransferOnRead() throws Exception {
+    
+  }
 }

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

@@ -0,0 +1,184 @@
+/*
+ * 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;
+
+import com.google.protobuf.Message;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.ipc_.AlignmentContext;
+import org.apache.hadoop.ipc_.RetriableException;
+import org.apache.hadoop.ipc_.protobuf.RpcHeaderProtos.RpcRequestHeaderProto;
+import org.apache.hadoop.ipc_.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
+import org.apache.hadoop.ozone.OmUtils;
+import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.ratis.proto.RaftProtos.RaftPeerRole;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is the server side implementation responsible for passing
+ * state alignment info to clients.
+ * <p>
+ * Unlike HDFS's ClientNamenodeProtocol that has a RPC method for each
+ * distinct call, OM (OzoneManagerService) only contains a single RPC method
+ * (i.e. submitRequest(OMRequest)). Therefore, we need to query the OMRequest
+ * to get the OMRequest parameter from the RPC method get the corresponding cmdType
+ * to check whether we can the request to be run on non-leader OMs.
+ */
+public class OmAlignmentContext implements AlignmentContext {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OmAlignmentContext.class);
+  /**
+   * Estimated number of journal transactions a typical OM can execute
+   * per second. The number is used to estimate how long a client's
+   * RPC request will wait in the call queue before the Observer catches up
+   * with its state id.
+   */
+  private static final long ESTIMATED_TRANSACTIONS_PER_SECOND = 10000L;
+
+  /**
+   * The client wait time on an RPC request is composed of
+   * the server execution time plus the communication time.
+   * This is an expected fraction of the total wait time spent on
+   * server execution.
+   */
+  private static final float ESTIMATED_SERVER_TIME_MULTIPLIER = 0.8f;
+
+  private final OzoneManager ozoneManager;
+
+  /**
+   * Server side constructor.
+   * @param ozoneManager server side state provider
+   */
+  OmAlignmentContext(OzoneManager ozoneManager) {
+    this.ozoneManager = ozoneManager;
+  }
+
+  /**
+   * Server side implementation for providing state alignment info in responses.
+   */
+  @Override
+  public void updateResponseState(RpcResponseHeaderProto.Builder header) {
+    // Using getCorrectLastAppliedOrWrittenTxId will acquire the lock on
+    // FSEditLog. This is needed so that ANN will return the correct state id
+    // it currently has. But this may not be necessary for Observer, may want
+    // revisit for optimization. Same goes to receiveRequestState.
+    header.setStateId(getLastSeenStateId());
+  }
+
+  /**
+   * Server side implementation only provides state alignment info.
+   * It does not receive state alignment info therefore this does nothing.
+   */
+  @Override
+  public void receiveResponseState(RpcResponseHeaderProto header) {
+    // Do nothing.
+  }
+
+  /**
+   * Server side implementation only receives state alignment info.
+   * It does not build RPC requests therefore this does nothing.
+   */
+  @Override
+  public void updateRequestState(RpcRequestHeaderProto.Builder header) {
+    // Do nothing.
+  }
+
+  /**
+   * Server-side implementation for processing state alignment info in
+   * requests.
+   * For Follower/Listener it compares the client and the server states and determines
+   * if it makes sense to wait until the server catches up with the client
+   * state. If not the server throws RetriableException so that the client
+   * could retry the call according to the retry policy with another Follower/Listener
+   * or the Leader.
+   *
+   * @param header The RPC request header.
+   * @param clientWaitTime time in milliseconds indicating how long client
+   *    waits for the server response. It is used to verify if the client's
+   *    state is too far ahead of the server's
+   * @return the minimum of the state ids of the client or the server.
+   * @throws RetriableException if Observer is too far behind.
+   */
+  @Override
+  public long receiveRequestState(RpcRequestHeaderProto header,
+      long clientWaitTime) throws IOException {
+    RaftPeerRole selfRole;
+    if (ozoneManager.getOmRatisServer() == null) {
+      selfRole = RaftPeerRole.LEADER;
+    } else {
+      selfRole = ozoneManager.getSelfRole(ozoneManager.getOmRatisServer().getLeaderId());
+    }
+
+    if (!header.hasStateId() && !RaftPeerRole.LEADER.equals(selfRole)) {
+      // This could happen if client configured with non-follower proxy provider
+      // (e.g., ConfiguredFailoverProxyProvider) is accessing a cluster with.
+      // In this case, we should let the client failover to the
+      // leader node, rather than potentially serving stale result (client
+      // stateId is 0 if not set).
+      throw new IOException("Node received request without "
+          + "stateId. This mostly likely is because client is not configured "
+          + "with ObserverReadProxyProvider");
+    }
+    long serverStateId = getLastSeenStateId();
+    long clientStateId = header.getStateId();
+    LOG.trace("Client State ID= {} and Server State ID= {}",
+        clientStateId, serverStateId);
+
+    if (clientStateId > serverStateId &&
+        RaftPeerRole.LEADER.equals(selfRole)) {
+      LOG.warn("The client stateId: {} is greater than "
+              + "the server stateId: {} This is unexpected. "
+              + "Resetting client stateId to server stateId",
+          clientStateId, serverStateId);
+      return serverStateId;
+    }
+    if ((RaftPeerRole.FOLLOWER.equals(selfRole) || RaftPeerRole.LISTENER.equals(selfRole)) &&
+        clientStateId - serverStateId >
+            ESTIMATED_TRANSACTIONS_PER_SECOND
+                * TimeUnit.MILLISECONDS.toSeconds(clientWaitTime)
+                * ESTIMATED_SERVER_TIME_MULTIPLIER) {
+      throw new RetriableException(
+          "Follower / Listener Node is too far behind: serverStateId = "
+              + serverStateId + " clientStateId = " + clientStateId);
+    }
+    return clientStateId;
+  }
+
+  @Override
+  public long getLastSeenStateId() {
+    return ozoneManager.getOmRatisServer().getLastAppliedTermIndex().getIndex();
+  }
+
+  @Override
+  public boolean isCoordinatedCall(String protocolName, String methodName, Message payload) {
+    // See OzoneManagerProtocolPB @ProtocolInfo annotation for the protocol name
+    if (!protocolName.equals(OzoneManagerProtocol.class.getCanonicalName()) ||
+        !(methodName.equals("submitRequest")) ||
+        !(payload instanceof OMRequest)) {
+      return false;
+    }
+
+    OMRequest omRequest = (OMRequest) payload;
+
+    return OmUtils.isReadOnly(omRequest);
+  }
+
+}