HDDS-15089. Support S3 per request read consistency

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -62,6 +62,7 @@
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.helpers.OzoneFileStatus;
 import org.apache.hadoop.ozone.om.helpers.OzoneFileStatusLight;
+import org.apache.hadoop.ozone.om.helpers.ReadConsistency;
 import org.apache.hadoop.ozone.om.helpers.S3SecretValue;
 import org.apache.hadoop.ozone.om.helpers.S3VolumeContext;
 import org.apache.hadoop.ozone.om.helpers.TenantStateList;
@@ -1378,17 +1379,44 @@ OzoneKey headObject(String volumeName, String bucketName,
    */
   void setThreadLocalS3Auth(S3Auth s3Auth);
 
+  /**
+   * Sets the read consistency hint for the current request thread.
+   * @param readConsistency read consistency selected by the client.
+   */
+  void setThreadLocalReadConsistency(ReadConsistency readConsistency);
+
+  /**
+   * Sets the read consistency hint and optional local lease context for the
+   * current request thread.
+   * @param readConsistency read consistency selected by the client.
+   * @param localLeaseLogLimit optional local lease log limit.
+   * @param localLeaseTimeMs optional local lease duration in milliseconds.
+   */
+  void setThreadLocalReadConsistency(ReadConsistency readConsistency,
+      Long localLeaseLogLimit, Long localLeaseTimeMs);
+
   /**
    * Gets the S3 Authentication information that is attached to the thread.
    * @return S3 Authentication information.
    */
   S3Auth getThreadLocalS3Auth();
 
+  /**
+   * Gets the read consistency hint that is attached to the thread.
+   * @return request-local read consistency.
+   */
+  ReadConsistency getThreadLocalReadConsistency();
+
   /**
    * Clears the S3 Authentication information attached to the thread.
    */
   void clearThreadLocalS3Auth();
 
+  /**
+   * Clears the read consistency hint attached to the thread.
+   */
+  void clearThreadLocalReadConsistency();
+
   default ThreadLocal<S3Auth> getS3CredentialsProvider() {
     return null;
   }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -162,6 +162,7 @@
 import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils;
 import org.apache.hadoop.ozone.om.helpers.OzoneFileStatus;
 import org.apache.hadoop.ozone.om.helpers.OzoneFileStatusLight;
+import org.apache.hadoop.ozone.om.helpers.ReadConsistency;
 import org.apache.hadoop.ozone.om.helpers.S3SecretValue;
 import org.apache.hadoop.ozone.om.helpers.S3VolumeContext;
 import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
@@ -2922,16 +2923,38 @@ public void setThreadLocalS3Auth(
     this.s3gUgi = UserGroupInformation.createRemoteUser(getThreadLocalS3Auth().getUserPrincipal());
   }
 
+  @Override
+  public void setThreadLocalReadConsistency(ReadConsistency readConsistency) {
+    ozoneManagerClient.setThreadLocalReadConsistency(readConsistency);
+  }
+
+  @Override
+  public void setThreadLocalReadConsistency(ReadConsistency readConsistency,
+      Long localLeaseLogLimit, Long localLeaseTimeMs) {
+    ozoneManagerClient.setThreadLocalReadConsistency(readConsistency,
+        localLeaseLogLimit, localLeaseTimeMs);
+  }
+
   @Override
   public S3Auth getThreadLocalS3Auth() {
     return ozoneManagerClient.getThreadLocalS3Auth();
   }
 
+  @Override
+  public ReadConsistency getThreadLocalReadConsistency() {
+    return ozoneManagerClient.getThreadLocalReadConsistency();
+  }
+
   @Override
   public void clearThreadLocalS3Auth() {
     ozoneManagerClient.clearThreadLocalS3Auth();
   }
 
+  @Override
+  public void clearThreadLocalReadConsistency() {
+    ozoneManagerClient.clearThreadLocalReadConsistency();
+  }
+
   @Override
   public ThreadLocal<S3Auth> getS3CredentialsProvider() {
     return ozoneManagerClient.getS3CredentialsProvider();

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

@@ -76,11 +76,11 @@ public class HadoopRpcOMFollowerReadFailoverProxyProvider implements FailoverPro
   /** The combined proxy which redirects to other proxies as necessary. */
   private final ProxyInfo<OzoneManagerProtocolPB> combinedProxy;
 
-  /**
-   * Whether reading from follower is enabled. If this is false, all read
-   * requests will still go to OM leader.
-   */
-  private volatile boolean useFollowerRead;
+  /** Whether follower reads are supported by the OM service. */
+  private volatile boolean omServiceSupportsFollowerRead;
+
+  /** Whether eligible reads without an explicit consistency hint use followers. */
+  private final boolean defaultFollowerReadEnabled;
 
   /**
    * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used.
@@ -96,17 +96,11 @@ public class HadoopRpcOMFollowerReadFailoverProxyProvider implements FailoverPro
   /** The read consistency hint used when follower read is disabled or when follower read fails. */
   private final ReadConsistencyHint leaderReadConsistency;
 
-  public HadoopRpcOMFollowerReadFailoverProxyProvider(
-      HadoopRpcOMFailoverProxyProvider<OzoneManagerProtocolPB> leaderProxy
-  ) {
-    this(leaderProxy, ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER, ReadConsistency.DEFAULT, true);
-  }
-
   public HadoopRpcOMFollowerReadFailoverProxyProvider(
       HadoopRpcOMFailoverProxyProvider<OzoneManagerProtocolPB> leaderProxy,
       ReadConsistency followerReadConsistencyType,
       ReadConsistency leaderReadConsistencyType,
-      boolean useFollowerRead) {
+      boolean defaultFollowerReadEnabled) {
     Preconditions.assertTrue(followerReadConsistencyType.allowFollowerRead(),
         "Invalid follower read consistency " + followerReadConsistencyType);
     Preconditions.assertTrue(!leaderReadConsistencyType.allowFollowerRead(),
@@ -121,7 +115,8 @@ public HadoopRpcOMFollowerReadFailoverProxyProvider(
         FollowerReadInvocationHandler.class.getClassLoader(),
         new Class<?>[] {OzoneManagerProtocolPB.class}, new FollowerReadInvocationHandler());
     combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo);
-    this.useFollowerRead = useFollowerRead;
+    this.omServiceSupportsFollowerRead = true;
+    this.defaultFollowerReadEnabled = defaultFollowerReadEnabled;
     this.followerReadConsistency = followerReadConsistencyType.getHint();
     this.leaderReadConsistency = leaderReadConsistencyType.getHint();
   }
@@ -178,6 +173,19 @@ private static OMRequest parseOMRequest(Object[] args) throws ServiceException {
     return (OMRequest) args[1];
   }
 
+  private static boolean allowFollowerRead(OMRequest omRequest) {
+    return !omRequest.hasReadConsistencyHint()
+        || ReadConsistency.fromProto(omRequest.getReadConsistencyHint()
+            .getReadConsistency()).allowFollowerRead();
+  }
+
+  private static ReadConsistency getReadConsistency(OMRequest omRequest) {
+    return omRequest.hasReadConsistencyHint()
+        ? ReadConsistency.fromProto(omRequest.getReadConsistencyHint()
+            .getReadConsistency())
+        : ReadConsistency.DEFAULT;
+  }
+
   @VisibleForTesting
   public ProxyInfo<OzoneManagerProtocolPB> getLastProxy() {
     return lastProxy;
@@ -218,6 +226,23 @@ private synchronized OMProxyInfo<OzoneManagerProtocolPB> changeProxy(OMProxyInfo
     return currentProxy;
   }
 
+  private OMProxyInfo<OzoneManagerProtocolPB> getCurrentProxy(
+      ReadConsistency readConsistency) {
+    OMProxyInfo<OzoneManagerProtocolPB> current = getCurrentProxy();
+    if (readConsistency != ReadConsistency.LOCAL_LEASE) {
+      return current;
+    }
+
+    String leaderNodeId = leaderProxy.getCurrentProxyOMNodeId();
+    for (int i = 0; i < leaderProxy.getOMProxyMap().size(); i++) {
+      if (!current.getNodeId().equals(leaderNodeId)) {
+        return current;
+      }
+      current = changeProxy(current);
+    }
+    return null;
+  }
+
   /**
    * An InvocationHandler to handle incoming requests. This class's invoke
    * method contains the primary logic for redirecting to followers.
@@ -242,12 +267,10 @@ public Object invoke(Object proxy, final Method method, final Object[] args)
       OMRequest omRequest = parseOMRequest(args);
 
       // Apply default consistency hint once, before any routing decision.
-      // In the future, we will support per-request hints which allows client (e.g. S3 clients)
-      // to specify a custom request header (e.g. x-ozone-read-consistency) as a consistency hint
-      // for read requests.
-      boolean isFollowerReadEligible = useFollowerRead && OmUtils.shouldSendToFollower(omRequest);
+      boolean isReadRequest = OmUtils.shouldSendToFollower(omRequest);
       if (!omRequest.hasReadConsistencyHint()) {
-        final ReadConsistencyHint defaultReadConsistency = isFollowerReadEligible
+        final ReadConsistencyHint defaultReadConsistency = omServiceSupportsFollowerRead
+            && defaultFollowerReadEnabled && isReadRequest
             ? followerReadConsistency : leaderReadConsistency;
         if (defaultReadConsistency != null) {
           omRequest = omRequest.toBuilder()
@@ -256,11 +279,22 @@ public Object invoke(Object proxy, final Method method, final Object[] args)
           args[1] = omRequest;
         }
       }
+      ReadConsistency readConsistency = getReadConsistency(omRequest);
+      // Requests with explicit hints (for example S3 read consistency headers)
+      // can narrow routing below, such as forcing leader-only reads.
+      boolean isExplicitFollowerRead = omRequest.hasReadConsistencyHint()
+          && allowFollowerRead(omRequest);
+      boolean isFollowerReadEligible = omServiceSupportsFollowerRead && isReadRequest
+          && (defaultFollowerReadEnabled || isExplicitFollowerRead);
 
       if (isFollowerReadEligible) {
         int failedCount = 0;
-        for (int i = 0; useFollowerRead && i < leaderProxy.getOMProxyMap().size(); i++) {
-          OMProxyInfo<OzoneManagerProtocolPB> current = getCurrentProxy();
+        for (int i = 0; i < leaderProxy.getOMProxyMap().size(); i++) {
+          OMProxyInfo<OzoneManagerProtocolPB> current =
+              getCurrentProxy(readConsistency);
+          if (current == null) {
+            break;
+          }
           LOG.debug("Attempting to service {} with cmdType {} using proxy {}",
               method.getName(), omRequest.getCmdType(), current.proxyInfo);
           try {
@@ -292,7 +326,7 @@ public Object invoke(Object proxy, final Method method, final Object[] args)
                 // the OM follower does not support / disable follower read or something is misconfigured
                 LOG.debug("Encountered OMNotLeaderException from {}. " +
                     "Disable OM follower read and retry OM leader directly.", current.proxyInfo);
-                useFollowerRead = false;
+                omServiceSupportsFollowerRead = false;
                 // Break here instead of throwing exception so that it is not counted
                 // as a failover
                 break;
@@ -363,19 +397,24 @@ public Object invoke(Object proxy, final Method method, final Object[] args)
       // Either all followers have failed, follower reads are disabled,
       // or this is a write request. In any case, forward the request to
       // the leader OM.
+      return invokeLeader(method, args);
+    }
+
+    private Object invokeLeader(Method method, Object[] args) throws Throwable {
       LOG.debug("Using leader-based failoverProxy to service {}", method.getName());
-      final OMProxyInfo<OzoneManagerProtocolPB> currentLeaderProxy = leaderProxy.getProxy();
-      Object retVal = null;
+      final OMProxyInfo<OzoneManagerProtocolPB> currentLeaderProxy =
+          leaderProxy.getProxy();
       try {
-        retVal = method.invoke(currentLeaderProxy.getProxy(), args);
+        Object retVal = method.invoke(currentLeaderProxy.getProxy(), args);
+        lastProxy = currentLeaderProxy;
+        return retVal;
       } catch (InvocationTargetException e) {
         LOG.debug("Exception thrown from leader-based failoverProxy", e.getCause());
         // This exception will be handled by the OMFailoverProxyProviderBase#getRetryPolicy
         // (see getRetryPolicy). This ensures that the leader-only failover should still work.
         throwServiceException(e.getCause());
+        return null;
       }
-      lastProxy = currentLeaderProxy;
-      return retVal;
     }
 
     @Override
@@ -395,7 +434,7 @@ public ConnectionId getConnectionId() {
       // visibility hazard, not a tearing one -- but the outcome is the
       // same: a stale proxy whose underlying connection has been
       // stopped is dialed instead of the live replacement.
-      return RPC.getConnectionIdForProxy(useFollowerRead
+      return RPC.getConnectionIdForProxy(omServiceSupportsFollowerRead
           ? getCurrentProxy().getProxy() : leaderProxy.getProxy().getProxy());
     }
   }
@@ -408,8 +447,8 @@ public synchronized void close() throws IOException {
   }
 
   @VisibleForTesting
-  public boolean isUseFollowerRead() {
-    return useFollowerRead;
+  public boolean isOmServiceSupportsFollowerRead() {
+    return omServiceSupportsFollowerRead;
   }
 
   @VisibleForTesting

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

@@ -99,7 +99,8 @@ public class GrpcOmTransport implements OmTransport {
   private RetryPolicy retryPolicy;
   private final GrpcOMFailoverProxyProvider<OzoneManagerProtocolPB>
       omFailoverProxyProvider;
-  private volatile boolean useFollowerRead;
+  private final boolean defaultFollowerReadEnabled;
+  private volatile boolean omServiceSupportsFollowerRead;
   private final ReadConsistencyHint followerReadConsistency;
   private final ReadConsistencyHint leaderReadConsistency;
   private int currentFollowerReadIndex = -1;
@@ -128,9 +129,10 @@ public GrpcOmTransport(ConfigurationSource conf,
         omServiceId,
         OzoneManagerProtocolPB.class);
 
-    this.useFollowerRead = conf.getBoolean(
+    this.defaultFollowerReadEnabled = conf.getBoolean(
         OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY,
         OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_DEFAULT);
+    this.omServiceSupportsFollowerRead = true;
     String defaultFollowerReadConsistencyStr = conf.get(
         OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_DEFAULT_CONSISTENCY_KEY,
         OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_DEFAULT_CONSISTENCY_DEFAULT
@@ -206,20 +208,29 @@ public void start() throws IOException {
 
   @Override
   public OMResponse submitRequest(OMRequest payload) throws IOException {
-    if (useFollowerRead && OmUtils.shouldSendToFollower(payload)) {
+    if (shouldUseFollowerRead(payload)) {
       return submitRequestWithFollowerRead(payload);
     }
     return submitRequestToLeader(addReadConsistencyHint(payload,
         leaderReadConsistency));
   }
 
+  private boolean shouldUseFollowerRead(OMRequest payload) {
+    if (!omServiceSupportsFollowerRead || !OmUtils.shouldSendToFollower(payload)) {
+      return false;
+    }
+    return defaultFollowerReadEnabled || payload.hasReadConsistencyHint()
+        && ReadConsistency.fromProto(payload.getReadConsistencyHint()
+            .getReadConsistency()).allowFollowerRead();
+  }
+
   private OMResponse submitRequestWithFollowerRead(OMRequest payload)
       throws IOException {
     OMRequest followerPayload = addReadConsistencyHint(payload,
         followerReadConsistency);
     int failedCount = 0;
-    for (int i = 0; useFollowerRead &&
-        i < omFailoverProxyProvider.getOMProxyMap().getNodeIds().size(); i++) {
+    for (int i = 0;
+         i < omFailoverProxyProvider.getOMProxyMap().getNodeIds().size(); i++) {
       String nodeId = getCurrentFollowerReadNodeId();
       String followerHost = omFailoverProxyProvider.getGrpcProxyAddress(nodeId);
       try {
@@ -234,7 +245,7 @@ private OMResponse submitRequestWithFollowerRead(OMRequest payload)
         if (OMFailoverProxyProviderBase.getNotLeaderException(unwrapped) != null) {
           LOG.debug("Encountered OMNotLeaderException from {}. Disable OM follower read and retry OM leader directly.",
               followerHost);
-          useFollowerRead = false;
+          omServiceSupportsFollowerRead = false;
           break;
         }
         if (OMFailoverProxyProviderBase.getLeaderNotReadyException(unwrapped) != null) {

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

@@ -64,7 +64,7 @@ public Hadoop3OmTransport(ConfigurationSource conf,
     this.omFailoverProxyProvider = new HadoopRpcOMFailoverProxyProvider<>(
             conf, ugi, omServiceId, OzoneManagerProtocolPB.class);
 
-    boolean followerReadEnabled = conf.getBoolean(
+    boolean defaultFollowerReadEnabled = conf.getBoolean(
         OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY,
         OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_DEFAULT);
 
@@ -90,7 +90,7 @@ public Hadoop3OmTransport(ConfigurationSource conf,
         new HadoopRpcOMFollowerReadFailoverProxyProvider(omFailoverProxyProvider,
             defaultFollowerReadConsistency,
             defaultLeaderReadConsistency,
-            followerReadEnabled);
+            defaultFollowerReadEnabled);
     this.rpcProxy = OzoneManagerProtocolPB.newProxy(followerReadFailoverProxyProvider, maxFailovers);
   }
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.om.protocolPB;
 
+import org.apache.hadoop.ozone.om.helpers.ReadConsistency;
 import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
 import org.apache.hadoop.ozone.om.protocol.S3Auth;
 
@@ -33,9 +34,18 @@ public interface OzoneManagerClientProtocol extends OzoneManagerProtocol {
    */
   void setThreadLocalS3Auth(S3Auth s3Auth);
 
+  void setThreadLocalReadConsistency(ReadConsistency readConsistency);
+
+  void setThreadLocalReadConsistency(ReadConsistency readConsistency,
+      Long localLeaseLogLimit, Long localLeaseTimeMs);
+
   S3Auth getThreadLocalS3Auth();
 
+  ReadConsistency getThreadLocalReadConsistency();
+
   void clearThreadLocalS3Auth();
 
+  void clearThreadLocalReadConsistency();
+
   ThreadLocal<S3Auth> getS3CredentialsProvider();
 }