This is an automated email from the ASF dual-hosted git repository.
ivandika3 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new ba05c2e3ad8 HDDS-16497. Support OM ThreadLocal context propagation for
Ratis read (#11257)
ba05c2e3ad8 is described below
commit ba05c2e3ad8151251f61732a00f98bccefe33f3b
Author: Ivan Andika <[email protected]>
AuthorDate: Tue Sep 29 16:27:23 2026 +0800
HDDS-16497. Support OM ThreadLocal context propagation for Ratis read
(#11257)
---
.../ozone/om/OzoneManagerHAFollowerReadTests.java | 8 +-
...stOzoneManagerHAFollowerReadWithAllRunning.java | 43 ++++
.../hadoop/ozone/om/execution/OMExecutionFlow.java | 20 +-
.../hadoop/ozone/om/lock/OMLockDetailsUtil.java | 48 +++++
.../hadoop/ozone/om/lock/OzoneManagerLock.java | 7 +-
.../hadoop/ozone/om/ratis/OMThreadContext.java | 198 ++++++++++++++++++
.../ozone/om/ratis/OzoneManagerStateMachine.java | 52 ++---
.../hadoop/ozone/om/request/OMClientRequest.java | 19 +-
...OzoneManagerProtocolServerSideTranslatorPB.java | 15 +-
.../ozone/om/execution/TestOMExecutionFlow.java | 165 +++++++++++++++
.../ozone/om/lock/TestOMLockDetailsUtil.java | 78 +++++++
.../hadoop/ozone/om/ratis/TestOMThreadContext.java | 207 +++++++++++++++++++
.../om/ratis/TestOzoneManagerStateMachine.java | 224 +++++++++++++++++++++
.../hadoop/ozone/om/response/TestOMResponse.java | 51 ++++-
14 files changed, 1078 insertions(+), 57 deletions(-)
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/OzoneManagerHAFollowerReadTests.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/OzoneManagerHAFollowerReadTests.java
index 2db23fdbe7e..74a3959d9c9 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/OzoneManagerHAFollowerReadTests.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/OzoneManagerHAFollowerReadTests.java
@@ -17,6 +17,9 @@
package org.apache.hadoop.ozone.om;
+import static
org.apache.hadoop.hdds.security.SecurityConfig.OZONE_TEST_AUTHORIZATION_ENABLED;
+import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS;
+import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS_NATIVE;
import static org.assertj.core.api.Assertions.assertThat;
import java.io.IOException;
@@ -32,7 +35,10 @@ public abstract class OzoneManagerHAFollowerReadTests
extends AbstractOzoneManag
@BeforeAll
public static void init() throws Exception {
- initCluster(true);
+ initCluster(true, conf -> {
+ conf.setBoolean(OZONE_TEST_AUTHORIZATION_ENABLED, true);
+ conf.set(OZONE_ACL_AUTHORIZER_CLASS, OZONE_ACL_AUTHORIZER_CLASS_NATIVE);
+ });
}
protected void listVolumes(boolean checkSuccess)
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java
index 0e97b249420..25573262dd0 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java
@@ -21,6 +21,7 @@
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_DEFAULT_CONSISTENCY_KEY;
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY;
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_LEADER_READ_DEFAULT_CONSISTENCY_KEY;
+import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_READ_BLACKLIST_USERS;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_TRANSPORT_CLASS;
import static
org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.DIRECTORY_NOT_FOUND;
import static
org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.FILE_ALREADY_EXISTS;
@@ -39,6 +40,7 @@
import com.google.protobuf.ServiceException;
import java.net.InetSocketAddress;
+import java.security.PrivilegedExceptionAction;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
@@ -78,6 +80,7 @@
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.VolumeInfo;
import
org.apache.hadoop.ozone.protocolPB.OzoneManagerProtocolServerSideTranslatorPB;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.ozone.test.tag.Flaky;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -281,6 +284,46 @@ void testLinearizableReadConsistency() throws Exception {
}
}
+ @Test
+ void testRemoteUserPreservedForLinearizableRead() throws Exception {
+ OzoneBucket bucket = setupBucket();
+ String key = createKey(bucket);
+ String deniedUserName = uniqueObjectName("denied-user");
+ UserGroupInformation deniedUser =
UserGroupInformation.createRemoteUser(deniedUserName);
+ OzoneManager followerOM = getCluster().getOzoneManagersList().stream()
+ .filter(om -> !om.isLeaderReady())
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("No follower OM
available"));
+
+ for (OzoneManager om : getCluster().getOzoneManagersList()) {
+
om.getReconfigurationHandler().reconfigureProperty(OZONE_READ_BLACKLIST_USERS,
deniedUserName);
+ }
+
+ try {
+ long previousLinearizableRead =
followerOM.getMetrics().getNumLinearizableRead();
+ OMException exception =
deniedUser.doAs((PrivilegedExceptionAction<OMException>) () -> {
+ OzoneConfiguration clientConf = new OzoneConfiguration(getConf());
+ clientConf.setBoolean(OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, true);
+ clientConf.set(OZONE_CLIENT_FOLLOWER_READ_DEFAULT_CONSISTENCY_KEY,
"LINEARIZABLE_ALLOW_FOLLOWER");
+ clientConf.set(OZONE_OM_TRANSPORT_CLASS,
Hadoop3OmTransportFactory.class.getName());
+ try (OzoneClient deniedClient =
OzoneClientFactory.getRpcClient(getOmServiceId(), clientConf)) {
+ ObjectStore deniedObjectStore = deniedClient.getObjectStore();
+ changeFollowerReadInitialProxy(deniedObjectStore,
Hadoop3OmTransportFactory.class,
+ getCluster().getOMLeader().getOMNodeId(),
followerOM.getOMNodeId());
+ return assertThrows(OMException.class, () ->
deniedObjectStore.getClientProxy().headObject(
+ bucket.getVolumeName(), bucket.getName(), key));
+ }
+ });
+
+ assertEquals(OMException.ResultCodes.PERMISSION_DENIED,
exception.getResult());
+
assertThat(followerOM.getMetrics().getNumLinearizableRead()).isGreaterThan(previousLinearizableRead);
+ } finally {
+ for (OzoneManager om : getCluster().getOzoneManagersList()) {
+
om.getReconfigurationHandler().reconfigureProperty(OZONE_READ_BLACKLIST_USERS,
"");
+ }
+ }
+ }
+
@Test
void testFileOperationsWithRecursive() throws Exception {
OzoneBucket ozoneBucket = setupBucket();
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/execution/OMExecutionFlow.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/execution/OMExecutionFlow.java
index de45d7f5e46..63127d62c96 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/execution/OMExecutionFlow.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/execution/OMExecutionFlow.java
@@ -28,6 +28,7 @@
import org.apache.hadoop.ozone.om.request.OMClientRequest;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.security.STSSecurityUtil;
/**
* entry for execution flow for write request.
@@ -73,7 +74,11 @@ private OMResponse submitExecutionToRatis(OMRequest request,
boolean isWrite) th
return OzoneManagerRatisUtils.createErrorResponse(request, ex);
}
} else {
- requestToSubmit = request;
+ try {
+ requestToSubmit = captureReadContext(request);
+ } catch (IOException ex) {
+ return OzoneManagerRatisUtils.createErrorResponse(request, ex);
+ }
}
// 2. submit request to ratis
@@ -83,4 +88,17 @@ private OMResponse submitExecutionToRatis(OMRequest request,
boolean isWrite) th
}
return response;
}
+
+ /**
+ * Captures authenticated request context on the RPC thread before
submitting a read request to Ratis.
+ */
+ private OMRequest captureReadContext(OMRequest request) throws IOException {
+ OMRequest.Builder requestBuilder = request.toBuilder()
+ .setUserInfo(OMClientRequest.getAuthenticatedUserInfo(request));
+ if (requestBuilder.hasS3Authentication()) {
+ requestBuilder.setS3Authentication(
+
STSSecurityUtil.resolveS3Authentication(requestBuilder.getS3Authentication(),
ozoneManager));
+ }
+ return requestBuilder.build();
+ }
}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OMLockDetailsUtil.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OMLockDetailsUtil.java
new file mode 100644
index 00000000000..7721152b566
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OMLockDetailsUtil.java
@@ -0,0 +1,48 @@
+/*
+ * 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.lock;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.ipc_.ProcessingDetails;
+import org.apache.hadoop.ipc_.ProcessingDetails.Timing;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMLockDetailsProto;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+
+/** Converts OM lock details between internal request processing and protocol
responses. */
+public final class OMLockDetailsUtil {
+
+ private OMLockDetailsUtil() {
+ }
+
+ public static OMResponse addToResponse(OMResponse response,
ProcessingDetails processingDetails) {
+ long waitNanos = processingDetails.get(Timing.LOCKWAIT,
TimeUnit.NANOSECONDS);
+ long readNanos = processingDetails.get(Timing.LOCKSHARED,
TimeUnit.NANOSECONDS);
+ long writeNanos = processingDetails.get(Timing.LOCKEXCLUSIVE,
TimeUnit.NANOSECONDS);
+ if (waitNanos == 0 && readNanos == 0 && writeNanos == 0) {
+ return response;
+ }
+
+ OMLockDetailsProto.Builder lockDetails = response.hasOmLockDetails()
+ ? response.getOmLockDetails().toBuilder() :
OMLockDetailsProto.newBuilder();
+ lockDetails.setWaitLockNanos(lockDetails.getWaitLockNanos() + waitNanos);
+ lockDetails.setReadLockNanos(lockDetails.getReadLockNanos() + readNanos);
+ lockDetails.setWriteLockNanos(lockDetails.getWriteLockNanos() +
writeNanos);
+ return response.toBuilder().setOmLockDetails(lockDetails).build();
+ }
+
+}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java
index 1847dd31b7c..3f875d586af 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java
@@ -723,10 +723,9 @@ short getMask() {
/**
* Update the processing details.
*
- * If Server.getCurCall() is null, which means it's write operation on Ratis,
- * then we need to update the omLockDetails.
- * If not null, it's read operation, or write operation on non-Ratis cluster,
- * we can update ThreadLocal variable directly.
+ * If Server.getCurCall() is null, the operation is detached from a Hadoop
+ * RPC call, such as a write operation on Ratis, so update omLockDetails for
+ * propagation to the original call. Otherwise, update the call directly.
* @param type IPC Timing types
* @param deltaNanos consumed time
*/
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OMThreadContext.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OMThreadContext.java
new file mode 100644
index 00000000000..b869916c2ad
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OMThreadContext.java
@@ -0,0 +1,198 @@
+/*
+ * 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 java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Objects;
+import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ipc_.RpcConstants;
+import org.apache.hadoop.ipc_.Server;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.lock.OMLockDetailsUtil;
+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.S3Authentication;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo;
+import org.apache.hadoop.ozone.security.STSSecurityUtil;
+import org.apache.hadoop.ozone.security.STSTokenIdentifier;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Request-derived context that can be applied while OM executes a Ratis
request.
+ *
+ * <p>Ratis may execute a read or write on the submitting thread or on a
separate thread. Applying this context replaces
+ * request-scoped state on whichever thread performs the operation, and
closing the returned {@link Scope} restores the
+ * state that was previously present on that thread.</p>
+ */
+public final class OMThreadContext {
+ private static final Logger LOG =
LoggerFactory.getLogger(OMThreadContext.class);
+
+ private final UserInfo userInfo;
+ private final S3Authentication s3Authentication;
+ private final STSTokenIdentifier stsTokenIdentifier;
+
+ private OMThreadContext(UserInfo userInfo, S3Authentication s3Authentication,
+ STSTokenIdentifier stsTokenIdentifier) {
+ this.userInfo = userInfo;
+ this.s3Authentication = s3Authentication;
+ this.stsTokenIdentifier = stsTokenIdentifier;
+ }
+
+ /**
+ * Creates context for a Ratis read. Applying it creates a fresh synthetic
Hadoop RPC call for the request.
+ */
+ public static OMThreadContext forRead(OMRequest request, OzoneManager
ozoneManager) throws IOException {
+ Objects.requireNonNull(request, "request");
+ Objects.requireNonNull(ozoneManager, "ozoneManager");
+ UserInfo requestUserInfo = request.hasUserInfo() ? request.getUserInfo() :
null;
+ return fromRequest(requestUserInfo, request, ozoneManager);
+ }
+
+ /**
+ * Creates context for a Ratis write. Writes do not install a synthetic
Hadoop RPC call, preserving lock accounting
+ * through ResourceLockTracker.
+ */
+ public static OMThreadContext forWrite(OMRequest request, OzoneManager
ozoneManager) throws IOException {
+ Objects.requireNonNull(request, "request");
+ Objects.requireNonNull(ozoneManager, "ozoneManager");
+ return fromRequest(null, request, ozoneManager);
+ }
+
+ private static OMThreadContext fromRequest(UserInfo userInfo, OMRequest
request,
+ OzoneManager ozoneManager) throws IOException {
+ if (!ozoneManager.isSecurityEnabled() || !request.hasS3Authentication()) {
+ return new OMThreadContext(userInfo, null, null);
+ }
+
+ STSSecurityUtil.ensureResolvedStsFieldsInvariants(request);
+ S3Authentication s3Authentication = request.getS3Authentication();
+ return new OMThreadContext(userInfo, s3Authentication,
+ STSSecurityUtil.rehydrateStsTokenIdentifier(s3Authentication));
+ }
+
+ /**
+ * Replaces request-scoped state on the current thread until the returned
scope is closed.
+ */
+ public Scope applyToCurrentThread() {
+ Server.Call currentCall = createCall(userInfo);
+ Scope scope = new Scope(Server.getCurCall().get(),
OzoneManager.getS3Auth(),
+ OzoneManager.getStsTokenIdentifier(), currentCall);
+ install(currentCall, s3Authentication, stsTokenIdentifier);
+ return scope;
+ }
+
+ private static Server.Call createCall(UserInfo userInfo) {
+ if (userInfo == null) {
+ return null;
+ }
+
+ UserGroupInformation remoteUser = userInfo.hasUserName() &&
!userInfo.getUserName().isEmpty()
+ ? UserGroupInformation.createRemoteUser(userInfo.getUserName()) : null;
+ InetAddress remoteAddress = createRemoteAddress(userInfo);
+ if (remoteUser == null && remoteAddress == null) {
+ return null;
+ }
+
+ return new Server.Call(RpcConstants.INVALID_CALL_ID,
RpcConstants.INVALID_RETRY_COUNT,
+ null, null, RPC.RpcKind.RPC_PROTOCOL_BUFFER,
RpcConstants.DUMMY_CLIENT_ID) {
+ @Override
+ public UserGroupInformation getRemoteUser() {
+ return remoteUser;
+ }
+
+ @Override
+ public InetAddress getHostInetAddress() {
+ return remoteAddress;
+ }
+ };
+ }
+
+ private static InetAddress createRemoteAddress(UserInfo userInfo) {
+ if (!userInfo.hasRemoteAddress() || userInfo.getRemoteAddress().isEmpty())
{
+ return null;
+ }
+
+ try {
+ InetAddress address = InetAddress.getByName(userInfo.getRemoteAddress());
+ return userInfo.hasHostName() && !userInfo.getHostName().isEmpty()
+ ? InetAddress.getByAddress(userInfo.getHostName(),
address.getAddress()) : address;
+ } catch (UnknownHostException ex) {
+ LOG.debug("Unable to restore remote address from Ratis request context:
{}",
+ userInfo.getRemoteAddress(), ex);
+ return null;
+ }
+ }
+
+ private static void install(Server.Call call, S3Authentication
s3Authentication,
+ STSTokenIdentifier stsTokenIdentifier) {
+ if (call == null) {
+ Server.getCurCall().remove();
+ } else {
+ Server.getCurCall().set(call);
+ }
+ OzoneManager.setS3Auth(s3Authentication);
+ OzoneManager.setStsTokenIdentifier(stsTokenIdentifier);
+ }
+
+ /**
+ * Applied OM request context. The scope must be closed on the thread where
it was created.
+ */
+ public static final class Scope implements AutoCloseable {
+ private final Thread ownerThread;
+ private final Server.Call previousCall;
+ private final S3Authentication previousS3Authentication;
+ private final STSTokenIdentifier previousStsTokenIdentifier;
+ private final Server.Call currentCall;
+ private boolean closed;
+
+ private Scope(Server.Call previousCall, S3Authentication
previousS3Authentication,
+ STSTokenIdentifier previousStsTokenIdentifier, Server.Call
currentCall) {
+ this.ownerThread = Thread.currentThread();
+ this.previousCall = previousCall;
+ this.previousS3Authentication = previousS3Authentication;
+ this.previousStsTokenIdentifier = previousStsTokenIdentifier;
+ this.currentCall = currentCall;
+ }
+
+ /**
+ * Adds read lock timings from the synthetic Hadoop RPC call to the
response.
+ */
+ public OMResponse addLockDetails(OMResponse response) {
+ return currentCall == null ? response
+ : OMLockDetailsUtil.addToResponse(response,
currentCall.getProcessingDetails());
+ }
+
+ @Override
+ public void close() {
+ if (Thread.currentThread() != ownerThread) {
+ throw new IllegalStateException(
+ "OM thread context must be closed on the thread where it was
applied");
+ }
+ if (closed) {
+ return;
+ }
+
+ install(previousCall, previousS3Authentication,
previousStsTokenIdentifier);
+ closed = true;
+ }
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
index 8353df01fb1..5e4bb9ca360 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
@@ -56,8 +56,6 @@
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
import org.apache.hadoop.ozone.protocolPB.OzoneManagerRequestHandler;
import org.apache.hadoop.ozone.protocolPB.RequestHandler;
-import org.apache.hadoop.ozone.security.STSSecurityUtil;
-import org.apache.hadoop.ozone.security.STSTokenIdentifier;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.util.concurrent.HadoopExecutors;
@@ -697,37 +695,15 @@ public void close() {
*/
@VisibleForTesting
OMResponse runCommand(OMRequest request, TermIndex termIndex) {
- boolean isS3AuthThreadLocalSet = false;
- boolean isStsThreadLocalSet = false;
- try {
- if (ozoneManager.isSecurityEnabled() && request.hasS3Authentication()) {
- // STS token verification runs on the leader RPC path so we don't need
to recheck here on the apply
- // after the log is committed
- STSSecurityUtil.ensureResolvedStsFieldsInvariants(request);
-
- final OzoneManagerProtocolProtos.S3Authentication s3Auth =
request.getS3Authentication();
- // ThreadLocal carries S3 action for OmMetadataReader.
- OzoneManager.setS3Auth(s3Auth);
- isS3AuthThreadLocalSet = true;
-
- // ThreadLocal carries session policy for OmMetadataReader
- final STSTokenIdentifier rehydratedTokenIdentifier =
STSSecurityUtil.rehydrateStsTokenIdentifier(s3Auth);
- if (rehydratedTokenIdentifier != null) {
- OzoneManager.setStsTokenIdentifier(rehydratedTokenIdentifier);
- isStsThreadLocalSet = true;
- }
- }
+ try (OMThreadContext.Scope ignored =
+ OMThreadContext.forWrite(request,
ozoneManager).applyToCurrentThread()) {
ExecutionContext context = ExecutionContext.of(termIndex.getIndex(),
termIndex);
final OMClientResponse omClientResponse = handler.handleWriteRequest(
request, context, ozoneManagerDoubleBuffer);
- OMLockDetails omLockDetails = omClientResponse.getOmLockDetails();
- OMResponse omResponse = omClientResponse.getOMResponse();
- if (omLockDetails != null) {
- return omResponse.toBuilder()
- .setOmLockDetails(omLockDetails.toProtobufBuilder()).build();
- } else {
- return omResponse;
- }
+ OMLockDetails lockDetails = omClientResponse.getOmLockDetails();
+ OMResponse response = omClientResponse.getOMResponse();
+ return lockDetails == null ? response
+ :
response.toBuilder().setOmLockDetails(lockDetails.toProtobufBuilder()).build();
} catch (IOException e) {
LOG.warn("Failed to write, Exception occurred ", e);
return createErrorResponse(request, e, termIndex);
@@ -735,13 +711,6 @@ OMResponse runCommand(OMRequest request, TermIndex
termIndex) {
// For any Runtime exceptions, terminate OM.
String errorMessage = "Request " + request + " failed with exception";
ExitUtils.terminate(1, errorMessage, e, LOG);
- } finally {
- if (isS3AuthThreadLocalSet) {
- OzoneManager.setS3Auth(null);
- }
- if (isStsThreadLocalSet) {
- OzoneManager.setStsTokenIdentifier(null);
- }
}
return null;
}
@@ -785,9 +754,12 @@ public void loadSnapshotInfoFromDB() throws IOException {
* @param request OMRequest
* @return response from OM
*/
- private Message queryCommand(OMRequest request) {
- OMResponse response = handler.handleReadRequest(request);
- return OMRatisHelper.convertResponseToMessage(response);
+ private Message queryCommand(OMRequest request) throws IOException {
+ try (OMThreadContext.Scope context =
+ OMThreadContext.forRead(request,
ozoneManager).applyToCurrentThread()) {
+ OMResponse response = handler.handleReadRequest(request);
+ return
OMRatisHelper.convertResponseToMessage(context.addLockDetails(response));
+ }
}
private static <T> CompletableFuture<T> completeExceptionally(Exception e) {
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
index 62013f7c071..071a722bc42 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
@@ -183,6 +183,22 @@ public OMRequest getOmRequest() {
* @return User Info.
*/
public OzoneManagerProtocolProtos.UserInfo getUserInfo() throws IOException {
+ return getUserInfo(omRequest, true);
+ }
+
+ /**
+ * Get authenticated user information for a read request submitted to Ratis.
+ * Client-supplied user information is not trusted on this path.
+ * @param omRequest OM request
+ * @return User Info.
+ */
+ public static OzoneManagerProtocolProtos.UserInfo getAuthenticatedUserInfo(
+ OMRequest omRequest) throws IOException {
+ return getUserInfo(omRequest, false);
+ }
+
+ private static OzoneManagerProtocolProtos.UserInfo getUserInfo(
+ OMRequest omRequest, boolean allowClientUserInfo) throws IOException {
UserGroupInformation user = ProtobufRpcEngine.Server.getRemoteUser();
InetAddress remoteAddress = ProtobufRpcEngine.Server.getRemoteIp();
OzoneManagerProtocolProtos.UserInfo.Builder userInfo =
@@ -224,7 +240,8 @@ public OzoneManagerProtocolProtos.UserInfo getUserInfo()
throws IOException {
// client-supplied user name when no identity was established above:
// it is unauthenticated data and must never override the identity
// derived from S3 authentication or from the RPC user.
- if (user == null && !userInfo.hasUserName() && omRequest.hasUserInfo()) {
+ if (allowClientUserInfo && user == null && !userInfo.hasUserName() &&
+ omRequest.hasUserInfo()) {
userInfo.setUserName(omRequest.getUserInfo().getUserName());
}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerProtocolServerSideTranslatorPB.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerProtocolServerSideTranslatorPB.java
index b5d46fcdb17..12c2069f720 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerProtocolServerSideTranslatorPB.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerProtocolServerSideTranslatorPB.java
@@ -139,16 +139,15 @@ public OMResponse submitRequest(RpcController controller,
public OMResponse processRequest(OMRequest request) throws ServiceException {
OMResponse response = internalProcessRequest(request);
if (response.hasOmLockDetails()) {
- OzoneManagerProtocolProtos.OMLockDetailsProto omLockDetailsProto =
- response.getOmLockDetails();
Server.Call call = Server.getCurCall().get();
if (call != null) {
- call.getProcessingDetails().add(Timing.LOCKWAIT,
- omLockDetailsProto.getWaitLockNanos(), TimeUnit.NANOSECONDS);
- call.getProcessingDetails().add(Timing.LOCKSHARED,
- omLockDetailsProto.getReadLockNanos(), TimeUnit.NANOSECONDS);
- call.getProcessingDetails().add(Timing.LOCKEXCLUSIVE,
- omLockDetailsProto.getWriteLockNanos(), TimeUnit.NANOSECONDS);
+ OzoneManagerProtocolProtos.OMLockDetailsProto lockDetails =
response.getOmLockDetails();
+ call.getProcessingDetails().add(
+ Timing.LOCKWAIT, lockDetails.getWaitLockNanos(),
TimeUnit.NANOSECONDS);
+ call.getProcessingDetails().add(
+ Timing.LOCKSHARED, lockDetails.getReadLockNanos(),
TimeUnit.NANOSECONDS);
+ call.getProcessingDetails().add(
+ Timing.LOCKEXCLUSIVE, lockDetails.getWriteLockNanos(),
TimeUnit.NANOSECONDS);
}
}
return response;
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/execution/TestOMExecutionFlow.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/execution/TestOMExecutionFlow.java
new file mode 100644
index 00000000000..f327cdc7fe9
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/execution/TestOMExecutionFlow.java
@@ -0,0 +1,165 @@
+/*
+ * 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.security.UserGroupInformation.AuthenticationMethod.KERBEROS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.net.InetAddress;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ipc_.Server;
+import org.apache.hadoop.ozone.om.OzoneAclUtils;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
+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.S3Authentication;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo;
+import org.apache.hadoop.ozone.security.STSTokenIdentifier;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+class TestOMExecutionFlow {
+
+ @AfterEach
+ void tearDown() {
+ Server.getCurCall().remove();
+ OzoneManager.setS3Auth(null);
+ OzoneManager.setStsTokenIdentifier(null);
+ }
+
+ @Test
+ void readRequestUsesAuthenticatedCallerContext() throws Exception {
+ OzoneManager ozoneManager = mock(OzoneManager.class);
+ OzoneManagerRatisServer ratisServer = mock(OzoneManagerRatisServer.class);
+ AtomicReference<OMRequest> submittedRequest = new AtomicReference<>();
+ when(ozoneManager.getOmRatisServer()).thenReturn(ratisServer);
+ when(ratisServer.submitRequest(any(OMRequest.class),
eq(false))).thenAnswer(invocation -> {
+ submittedRequest.set(invocation.getArgument(0));
+ return successResponse();
+ });
+
+ UserGroupInformation realUser =
UserGroupInformation.createRemoteUser("kerberos-user");
+ realUser.setAuthenticationMethod(KERBEROS);
+ UserGroupInformation remoteUser =
UserGroupInformation.createProxyUser("authenticated-user", realUser);
+ Credentials credentials = new Credentials();
+ credentials.addSecretKey(new Text("credential-key"), new byte[] {1, 2, 3});
+ remoteUser.addCredentials(credentials);
+ remoteUser.addToken(new Token<TokenIdentifier>(
+ new byte[] {4}, new byte[] {5}, new Text("token-kind"), new
Text("token-service")));
+ InetAddress remoteAddress = InetAddress.getByAddress("client.example.com",
new byte[] {10, 20, 30, 40});
+ Server.getCurCall().set(createCall(remoteUser, remoteAddress));
+
+ OMRequest request = OMRequest.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setClientId("client-id")
+ .setUserInfo(UserInfo.newBuilder()
+ .setUserName("forged-user")
+ .setHostName("forged-host")
+ .setRemoteAddress("192.0.2.1"))
+ .build();
+
+ new OMExecutionFlow(ozoneManager).submit(request, false);
+
+ UserInfo userInfo = submittedRequest.get().getUserInfo();
+ assertEquals("authenticated-user", userInfo.getUserName());
+ assertEquals("client.example.com", userInfo.getHostName());
+ assertEquals("10.20.30.40", userInfo.getRemoteAddress());
+ }
+
+ @Test
+ void readRequestCarriesResolvedStsContext() throws Exception {
+ OzoneManager ozoneManager = mock(OzoneManager.class);
+ OzoneManagerRatisServer ratisServer = mock(OzoneManagerRatisServer.class);
+ AtomicReference<OMRequest> submittedRequest = new AtomicReference<>();
+ when(ozoneManager.getOmRatisServer()).thenReturn(ratisServer);
+ when(ozoneManager.isSecurityEnabled()).thenReturn(true);
+ when(ratisServer.submitRequest(any(OMRequest.class),
eq(false))).thenAnswer(invocation -> {
+ submittedRequest.set(invocation.getArgument(0));
+ return successResponse();
+ });
+
+ S3Authentication s3Authentication = S3Authentication.newBuilder()
+ .setAccessId("temp-access-id")
+ .setSessionToken("session-token")
+ .setS3Action("GetObject")
+ .build();
+ STSTokenIdentifier stsToken = mock(STSTokenIdentifier.class);
+ UUID secretKeyId = UUID.randomUUID();
+ when(stsToken.getSessionPolicy()).thenReturn("session-policy");
+ when(stsToken.getRoleArn()).thenReturn("role-arn");
+ when(stsToken.getOriginalAccessKeyId()).thenReturn("original-access-id");
+ when(stsToken.getTempAccessKeyId()).thenReturn("temp-access-id");
+ when(stsToken.getSecretKeyId()).thenReturn(secretKeyId);
+ OzoneManager.setS3Auth(s3Authentication);
+ OzoneManager.setStsTokenIdentifier(stsToken);
+
+ OMRequest request = OMRequest.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setClientId("client-id")
+ .setS3Authentication(s3Authentication)
+ .setUserInfo(UserInfo.newBuilder().setUserName("forged-user"))
+ .build();
+
+ new OMExecutionFlow(ozoneManager).submit(request, false);
+
+ OMRequest submitted = submittedRequest.get();
+ assertEquals(OzoneAclUtils.accessIdToUserPrincipal("original-access-id"),
+ submitted.getUserInfo().getUserName());
+ assertEquals("session-policy",
submitted.getS3Authentication().getResolvedStsSessionPolicy());
+ assertEquals("role-arn",
submitted.getS3Authentication().getResolvedStsRoleArn());
+ assertEquals("original-access-id",
submitted.getS3Authentication().getResolvedStsOriginalAccessKeyId());
+ assertEquals("temp-access-id",
submitted.getS3Authentication().getResolvedStsTempAccessKeyId());
+ assertEquals(secretKeyId.toString(),
submitted.getS3Authentication().getResolvedStsSecretKeyId());
+ }
+
+ private static Server.Call createCall(UserGroupInformation user, InetAddress
remoteAddress) {
+ return new Server.Call(1, 0, null, null, RPC.RpcKind.RPC_PROTOCOL_BUFFER,
new byte[0]) {
+ @Override
+ public UserGroupInformation getRemoteUser() {
+ return user;
+ }
+
+ @Override
+ public InetAddress getHostInetAddress() {
+ return remoteAddress;
+ }
+ };
+ }
+
+ private static OMResponse successResponse() {
+ return OMResponse.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/lock/TestOMLockDetailsUtil.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/lock/TestOMLockDetailsUtil.java
new file mode 100644
index 00000000000..82d5548d556
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/lock/TestOMLockDetailsUtil.java
@@ -0,0 +1,78 @@
+/*
+ * 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.lock;
+
+import static org.apache.hadoop.ipc_.ProcessingDetails.Timing.LOCKEXCLUSIVE;
+import static org.apache.hadoop.ipc_.ProcessingDetails.Timing.LOCKSHARED;
+import static org.apache.hadoop.ipc_.ProcessingDetails.Timing.LOCKWAIT;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ipc_.RpcConstants;
+import org.apache.hadoop.ipc_.Server;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMLockDetailsProto;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.junit.jupiter.api.Test;
+
+class TestOMLockDetailsUtil {
+
+ @Test
+ void testAddProcessingDetailsToResponse() {
+ Server.Call call = newCall();
+ call.getProcessingDetails().add(LOCKWAIT, 11, TimeUnit.NANOSECONDS);
+ call.getProcessingDetails().add(LOCKSHARED, 22, TimeUnit.NANOSECONDS);
+ call.getProcessingDetails().add(LOCKEXCLUSIVE, 33, TimeUnit.NANOSECONDS);
+ OMResponse response = newResponseBuilder()
+ .setOmLockDetails(OMLockDetailsProto.newBuilder()
+ .setIsLockAcquired(true)
+ .setWaitLockNanos(1)
+ .setReadLockNanos(2)
+ .setWriteLockNanos(3))
+ .build();
+
+ OMResponse result = OMLockDetailsUtil.addToResponse(response,
call.getProcessingDetails());
+
+ assertTrue(result.getOmLockDetails().getIsLockAcquired());
+ assertEquals(12, result.getOmLockDetails().getWaitLockNanos());
+ assertEquals(24, result.getOmLockDetails().getReadLockNanos());
+ assertEquals(36, result.getOmLockDetails().getWriteLockNanos());
+ }
+
+ @Test
+ void testEmptyProcessingDetailsDoesNotChangeResponse() {
+ OMResponse response = OMResponse.getDefaultInstance();
+
+ assertSame(response, OMLockDetailsUtil.addToResponse(response,
newCall().getProcessingDetails()));
+ }
+
+ private static Server.Call newCall() {
+ return new Server.Call(RpcConstants.INVALID_CALL_ID,
RpcConstants.INVALID_RETRY_COUNT,
+ null, null, RPC.RpcKind.RPC_PROTOCOL_BUFFER,
RpcConstants.DUMMY_CLIENT_ID);
+ }
+
+ private static OMResponse.Builder newResponseBuilder() {
+ return OMResponse.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setStatus(Status.OK);
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOMThreadContext.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOMThreadContext.java
new file mode 100644
index 00000000000..bd7c8e8c7ed
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOMThreadContext.java
@@ -0,0 +1,207 @@
+/*
+ * 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 static org.apache.hadoop.ipc_.ProcessingDetails.Timing.LOCKSHARED;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.net.InetAddress;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ipc_.RpcConstants;
+import org.apache.hadoop.ipc_.Server;
+import org.apache.hadoop.ozone.om.OzoneManager;
+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.S3Authentication;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo;
+import org.apache.hadoop.ozone.security.STSTokenIdentifier;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+class TestOMThreadContext {
+
+ @AfterEach
+ void tearDown() {
+ Server.getCurCall().remove();
+ OzoneManager.setS3Auth(null);
+ OzoneManager.setStsTokenIdentifier(null);
+ }
+
+ @Test
+ void readContextAppliesRequestAndRestoresPreviousContext() throws Exception {
+ OzoneManager ozoneManager = secureOzoneManager();
+ S3Authentication requestS3Auth =
S3Authentication.newBuilder().setAccessId("request-access-id").build();
+ OMRequest request = requestWithContext("request-user", requestS3Auth);
+ OMThreadContext context = OMThreadContext.forRead(request, ozoneManager);
+
+ Server.Call previousCall = createCall("previous-user");
+ S3Authentication previousS3Auth =
S3Authentication.newBuilder().setAccessId("previous-access-id").build();
+ STSTokenIdentifier previousStsToken = mock(STSTokenIdentifier.class);
+ Server.getCurCall().set(previousCall);
+ OzoneManager.setS3Auth(previousS3Auth);
+ OzoneManager.setStsTokenIdentifier(previousStsToken);
+
+ try (OMThreadContext.Scope ignored = context.applyToCurrentThread()) {
+ assertEquals("request-user", Server.getRemoteUser().getUserName());
+ assertEquals("127.0.0.1", Server.getRemoteIp().getHostAddress());
+ assertEquals(requestS3Auth, OzoneManager.getS3Auth());
+ assertNull(OzoneManager.getStsTokenIdentifier());
+ }
+
+ assertSame(previousCall, Server.getCurCall().get());
+ assertSame(previousS3Auth, OzoneManager.getS3Auth());
+ assertSame(previousStsToken, OzoneManager.getStsTokenIdentifier());
+ }
+
+ @Test
+ void writeContextAppliesRequestAndRestoresPreviousContext() throws Exception
{
+ OzoneManager ozoneManager = secureOzoneManager();
+ S3Authentication requestS3Auth =
S3Authentication.newBuilder().setAccessId("request-access-id").build();
+ OMThreadContext context =
OMThreadContext.forWrite(requestWithContext("request-user", requestS3Auth),
ozoneManager);
+
+ Server.Call previousCall = createCall("previous-user");
+ S3Authentication previousS3Auth =
S3Authentication.newBuilder().setAccessId("previous-access-id").build();
+ STSTokenIdentifier previousStsToken = mock(STSTokenIdentifier.class);
+ Server.getCurCall().set(previousCall);
+ OzoneManager.setS3Auth(previousS3Auth);
+ OzoneManager.setStsTokenIdentifier(previousStsToken);
+
+ try (OMThreadContext.Scope ignored = context.applyToCurrentThread()) {
+ assertNull(Server.getCurCall().get());
+ assertEquals(requestS3Auth, OzoneManager.getS3Auth());
+ assertNull(OzoneManager.getStsTokenIdentifier());
+ }
+
+ assertSame(previousCall, Server.getCurCall().get());
+ assertSame(previousS3Auth, OzoneManager.getS3Auth());
+ assertSame(previousStsToken, OzoneManager.getStsTokenIdentifier());
+ }
+
+ @Test
+ void eachReadScopeUsesFreshCallAndLockDetails() throws Exception {
+ OMThreadContext context = OMThreadContext.forRead(
+ requestWithContext("request-user", null), secureOzoneManager());
+ OMResponse response = OMResponse.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+
+ Server.Call firstCall;
+ try (OMThreadContext.Scope scope = context.applyToCurrentThread()) {
+ firstCall = Server.getCurCall().get();
+ firstCall.getProcessingDetails().add(LOCKSHARED, 11,
TimeUnit.NANOSECONDS);
+ assertEquals(11,
scope.addLockDetails(response).getOmLockDetails().getReadLockNanos());
+ }
+
+ try (OMThreadContext.Scope scope = context.applyToCurrentThread()) {
+ assertNotSame(firstCall, Server.getCurCall().get());
+ assertSame(response, scope.addLockDetails(response));
+ }
+ }
+
+ @Test
+ void closeIsIdempotent() throws Exception {
+ Server.Call previousCall = createCall("previous-user");
+ Server.getCurCall().set(previousCall);
+ OMThreadContext.Scope scope = OMThreadContext.forRead(
+ requestWithContext("request-user", null),
secureOzoneManager()).applyToCurrentThread();
+
+ scope.close();
+ scope.close();
+
+ assertSame(previousCall, Server.getCurCall().get());
+ }
+
+ @Test
+ void closeRejectsDifferentThread() throws Exception {
+ Server.Call previousCall = createCall("previous-user");
+ Server.getCurCall().set(previousCall);
+ OMThreadContext.Scope scope = OMThreadContext.forRead(
+ requestWithContext("request-user", null),
secureOzoneManager()).applyToCurrentThread();
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<Void> close = executor.submit(() -> {
+ scope.close();
+ return null;
+ });
+ ExecutionException exception = assertThrows(ExecutionException.class,
close::get);
+ assertInstanceOf(IllegalStateException.class, exception.getCause());
+ assertEquals("OM thread context must be closed on the thread where it
was applied",
+ exception.getCause().getMessage());
+ } finally {
+ executor.shutdownNow();
+ scope.close();
+ }
+
+ assertSame(previousCall, Server.getCurCall().get());
+ }
+
+ private static OzoneManager secureOzoneManager() {
+ OzoneManager ozoneManager = mock(OzoneManager.class);
+ when(ozoneManager.isSecurityEnabled()).thenReturn(true);
+ return ozoneManager;
+ }
+
+ private static OMRequest requestWithContext(String userName,
S3Authentication s3Authentication) {
+ OMRequest.Builder builder = OMRequest.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setClientId("client-id")
+ .setUserInfo(UserInfo.newBuilder()
+ .setUserName(userName)
+ .setHostName("localhost")
+ .setRemoteAddress("127.0.0.1"));
+ if (s3Authentication != null) {
+ builder.setS3Authentication(s3Authentication);
+ }
+ return builder.build();
+ }
+
+ private static Server.Call createCall(String userName) throws Exception {
+ UserGroupInformation user =
UserGroupInformation.createRemoteUser(userName);
+ InetAddress address = InetAddress.getByAddress("localhost", new byte[]
{127, 0, 0, 1});
+ return new Server.Call(RpcConstants.INVALID_CALL_ID,
RpcConstants.INVALID_RETRY_COUNT,
+ null, null, RPC.RpcKind.RPC_PROTOCOL_BUFFER,
RpcConstants.DUMMY_CLIENT_ID) {
+ @Override
+ public UserGroupInformation getRemoteUser() {
+ return user;
+ }
+
+ @Override
+ public InetAddress getHostInetAddress() {
+ return address;
+ }
+ };
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
index 40975d5af61..c5640b92c01 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
@@ -17,6 +17,8 @@
package org.apache.hadoop.ozone.om.ratis;
+import static
org.apache.hadoop.ozone.om.lock.OzoneManagerLock.LeveledResource.VOLUME_LOCK;
+import static
org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod.SIMPLE;
import static org.apache.ozone.test.GenericTestUtils.waitFor;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -38,9 +40,12 @@
import static org.mockito.Mockito.when;
import java.io.IOException;
+import java.net.InetAddress;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@@ -49,9 +54,14 @@
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+import java.util.stream.Collectors;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.hdds.utils.db.DBStore;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ipc_.Server;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.audit.AuditEventStatus;
import org.apache.hadoop.ozone.audit.AuditMessage;
@@ -68,6 +78,7 @@
import org.apache.hadoop.ozone.om.ha.OMServiceManager;
import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
import org.apache.hadoop.ozone.om.lock.OMLockDetails;
+import org.apache.hadoop.ozone.om.lock.OzoneManagerLock;
import org.apache.hadoop.ozone.om.ratis_snapshot.OmRatisSnapshotProvider;
import org.apache.hadoop.ozone.om.response.DummyOMClientResponse;
import org.apache.hadoop.ozone.om.response.OMClientResponse;
@@ -79,10 +90,12 @@
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareRequest;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareRequestArgs;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.S3Authentication;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo;
import org.apache.hadoop.ozone.protocolPB.RequestHandler;
+import org.apache.hadoop.ozone.security.STSTokenIdentifier;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.ratis.proto.RaftProtos;
import org.apache.ratis.protocol.ClientId;
@@ -134,9 +147,23 @@ public void setup() {
@AfterEach
public void tearDown() {
sm.stop();
+ Server.getCurCall().remove();
+ OzoneManager.setS3Auth(null);
OzoneManager.setStsTokenIdentifier(null);
}
+ @Test
+ public void testOzoneManagerThreadLocalsAreReviewedForRatisPropagation() {
+ List<String> threadLocalFields =
Arrays.stream(OzoneManager.class.getDeclaredFields())
+ .filter(field -> ThreadLocal.class.isAssignableFrom(field.getType()))
+ .map(field -> field.getName())
+ .sorted()
+ .collect(Collectors.toList());
+
+ assertEquals(Arrays.asList("S3_AUTH", "STS_TOKEN"), threadLocalFields,
+ "Update OMThreadContext when adding request-scoped ThreadLocal fields
to OzoneManager");
+ }
+
// --- startTransaction tests ---
@Test
@@ -455,6 +482,40 @@ public void testRunCommandSetsAndClearsStsThreadLocal()
throws Exception {
assertNull(OzoneManager.getStsTokenIdentifier(), "Expected STS ThreadLocal
to be cleared after runCommand");
}
+ @Test
+ public void testRunCommandRestoresPreviousRequestContext() throws Exception {
+ OMRequest request = sampleWriteRequest();
+ TermIndex ti = TermIndex.valueOf(1, 5);
+ OMResponse expectedResponse = OMResponse.newBuilder()
+ .setCmdType(Type.CreateKey)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ OMClientResponse clientResponse = mock(OMClientResponse.class);
+ when(clientResponse.getOMResponse()).thenReturn(expectedResponse);
+ when(clientResponse.getOmLockDetails()).thenReturn(null);
+ when(handler.handleWriteRequest(eq(request), any(),
eq(doubleBuffer))).thenAnswer(invocation -> {
+ assertNull(Server.getCurCall().get());
+ assertNull(OzoneManager.getS3Auth());
+ assertNull(OzoneManager.getStsTokenIdentifier());
+ return clientResponse;
+ });
+
+ Server.Call previousCall = createCall("previous-user",
"previous.example.com", new byte[] {10, 0, 0, 1});
+ S3Authentication previousS3Auth =
S3Authentication.newBuilder().setAccessId("previous-access-id").build();
+ STSTokenIdentifier previousStsToken = mock(STSTokenIdentifier.class);
+ Server.getCurCall().set(previousCall);
+ OzoneManager.setS3Auth(previousS3Auth);
+ OzoneManager.setStsTokenIdentifier(previousStsToken);
+
+ OMResponse result = sm.runCommand(request, ti);
+
+ assertTrue(result.getSuccess());
+ assertSame(previousCall, Server.getCurCall().get());
+ assertSame(previousS3Auth, OzoneManager.getS3Auth());
+ assertSame(previousStsToken, OzoneManager.getStsTokenIdentifier());
+ }
+
@Test
public void testRunCommandSetsStsThreadLocalWithLegacyResolvedFields()
throws Exception {
when(om.isSecurityEnabled()).thenReturn(true);
@@ -635,6 +696,153 @@ public void testQueryIOException() {
assertInstanceOf(IOException.class, ex.getCause());
}
+ @Test
+ public void testQueryInstallsAndRestoresCallerContext() throws Exception {
+ UserInfo userInfo = UserInfo.newBuilder()
+ .setUserName("ratis-user")
+ .setHostName("client.example.com")
+ .setRemoteAddress("10.20.30.40")
+ .build();
+ OMRequest request = sampleReadRequest().toBuilder()
+ .setUserInfo(userInfo)
+ .build();
+ OMResponse expectedResponse = OMResponse.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ AtomicReference<UserGroupInformation> observedUser = new
AtomicReference<>();
+ AtomicReference<InetAddress> observedAddress = new AtomicReference<>();
+
when(handler.handleReadRequest(any(OMRequest.class))).thenAnswer(invocation -> {
+ observedUser.set(Server.getRemoteUser());
+ observedAddress.set(Server.getRemoteIp());
+ return expectedResponse;
+ });
+
+ Server.Call previousCall = createCall("previous-user",
"previous.example.com", new byte[] {10, 0, 0, 1});
+ Server.getCurCall().set(previousCall);
+
+
sm.query(Message.valueOf(OMRatisHelper.convertRequestToByteString(request))).get();
+
+ assertSame(previousCall, Server.getCurCall().get());
+ assertEquals("ratis-user", observedUser.get().getUserName());
+ assertEquals(SIMPLE, observedUser.get().getAuthenticationMethod());
+ assertNull(observedUser.get().getRealUser());
+ assertTrue(observedUser.get().getTokens().isEmpty());
+ assertNull(observedUser.get().getCredentials().getSecretKey(new
Text("credential-key")));
+ assertEquals("10.20.30.40", observedAddress.get().getHostAddress());
+ assertEquals("client.example.com", observedAddress.get().getHostName());
+ }
+
+ @Test
+ public void testQueryReturnsResourceLockDetails() throws Exception {
+ OMRequest request = sampleReadRequest().toBuilder()
+ .setUserInfo(UserInfo.newBuilder().setUserName("ratis-user"))
+ .build();
+ OMResponse expectedResponse = OMResponse.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ OzoneManagerLock lock = new OzoneManagerLock(new OzoneConfiguration());
+
when(handler.handleReadRequest(any(OMRequest.class))).thenAnswer(invocation -> {
+ lock.acquireReadLock(VOLUME_LOCK, "volume");
+ try {
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+ return expectedResponse;
+ } finally {
+ lock.releaseReadLock(VOLUME_LOCK, "volume");
+ }
+ });
+
+ OMResponse response;
+ try {
+ Message message = sm.query(Message.valueOf(
+ OMRatisHelper.convertRequestToByteString(request))).get();
+ response =
OMRatisHelper.convertByteStringToOMResponse(message.getContent());
+ } finally {
+ lock.cleanup();
+ }
+
+ assertTrue(response.hasOmLockDetails());
+ assertTrue(response.getOmLockDetails().getWaitLockNanos() > 0);
+ assertTrue(response.getOmLockDetails().getReadLockNanos() > 0);
+ assertEquals(0, response.getOmLockDetails().getWriteLockNanos());
+ }
+
+ @Test
+ public void testQueryInstallsAndRestoresS3Context() throws Exception {
+ when(om.isSecurityEnabled()).thenReturn(true);
+ S3Authentication requestS3Auth = S3Authentication.newBuilder()
+ .setAccessId("temp-access-id")
+ .setSessionToken("session-token")
+ .setS3Action("GetObject")
+ .setResolvedStsSessionPolicy("session-policy")
+ .setResolvedStsRoleArn("role-arn")
+ .setResolvedStsOriginalAccessKeyId("original-access-id")
+ .setResolvedStsTempAccessKeyId("temp-access-id")
+ .setResolvedStsSecretKeyId(UUID.randomUUID().toString())
+ .build();
+ OMRequest request = sampleReadRequest().toBuilder()
+ .setS3Authentication(requestS3Auth)
+ .build();
+ OMResponse expectedResponse = OMResponse.newBuilder()
+ .setCmdType(Type.ServiceList)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ AtomicReference<S3Authentication> observedS3Auth = new AtomicReference<>();
+ AtomicReference<STSTokenIdentifier> observedStsToken = new
AtomicReference<>();
+
when(handler.handleReadRequest(any(OMRequest.class))).thenAnswer(invocation -> {
+ observedS3Auth.set(OzoneManager.getS3Auth());
+ observedStsToken.set(OzoneManager.getStsTokenIdentifier());
+ return expectedResponse;
+ });
+
+ S3Authentication previousS3Auth =
S3Authentication.newBuilder().setAccessId("previous-access-id").build();
+ STSTokenIdentifier previousStsToken = mock(STSTokenIdentifier.class);
+ OzoneManager.setS3Auth(previousS3Auth);
+ OzoneManager.setStsTokenIdentifier(previousStsToken);
+
+
sm.query(Message.valueOf(OMRatisHelper.convertRequestToByteString(request))).get();
+
+ assertSame(previousS3Auth, OzoneManager.getS3Auth());
+ assertSame(previousStsToken, OzoneManager.getStsTokenIdentifier());
+ assertEquals(requestS3Auth, observedS3Auth.get());
+ assertEquals("session-policy", observedStsToken.get().getSessionPolicy());
+ assertEquals("role-arn", observedStsToken.get().getRoleArn());
+ assertEquals("original-access-id",
observedStsToken.get().getOriginalAccessKeyId());
+ assertEquals("temp-access-id",
observedStsToken.get().getTempAccessKeyId());
+ }
+
+ @Test
+ public void testQueryRestoresContextAfterFailure() throws Exception {
+ when(om.isSecurityEnabled()).thenReturn(true);
+ S3Authentication requestS3Auth = S3Authentication.newBuilder()
+ .setAccessId("request-access-id")
+ .build();
+ OMRequest request = sampleReadRequest().toBuilder()
+ .setUserInfo(UserInfo.newBuilder().setUserName("request-user"))
+ .setS3Authentication(requestS3Auth)
+ .build();
+ when(handler.handleReadRequest(any(OMRequest.class)))
+ .thenThrow(new IllegalStateException("read failed"));
+
+ Server.Call previousCall = createCall("previous-user",
"previous.example.com", new byte[] {10, 0, 0, 1});
+ S3Authentication previousS3Auth =
S3Authentication.newBuilder().setAccessId("previous-access-id").build();
+ STSTokenIdentifier previousStsToken = mock(STSTokenIdentifier.class);
+ Server.getCurCall().set(previousCall);
+ OzoneManager.setS3Auth(previousS3Auth);
+ OzoneManager.setStsTokenIdentifier(previousStsToken);
+
+ assertThrows(IllegalStateException.class,
+ () ->
sm.query(Message.valueOf(OMRatisHelper.convertRequestToByteString(request))));
+
+ assertSame(previousCall, Server.getCurCall().get());
+ assertSame(previousS3Auth, OzoneManager.getS3Auth());
+ assertSame(previousStsToken, OzoneManager.getStsTokenIdentifier());
+ }
+
// --- notifyTermIndexUpdated tests ---
@Test
@@ -1317,6 +1525,22 @@ private OMRequest sampleReadRequest() {
.build();
}
+ private static Server.Call createCall(String userName, String hostName,
byte[] address) throws IOException {
+ UserGroupInformation user =
UserGroupInformation.createRemoteUser(userName);
+ InetAddress remoteAddress = InetAddress.getByAddress(hostName, address);
+ return new Server.Call(1, 0, null, null, RPC.RpcKind.RPC_PROTOCOL_BUFFER,
new byte[0]) {
+ @Override
+ public UserGroupInformation getRemoteUser() {
+ return user;
+ }
+
+ @Override
+ public InetAddress getHostInetAddress() {
+ return remoteAddress;
+ }
+ };
+ }
+
private OzoneManagerProtocolProtos.S3Authentication stsS3Authentication() {
return OzoneManagerProtocolProtos.S3Authentication.newBuilder()
.setAccessId("accessId")
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestOMResponse.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestOMResponse.java
index bf7defdb4c1..7d0a0d0dc6d 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestOMResponse.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestOMResponse.java
@@ -18,21 +18,30 @@
package org.apache.hadoop.ozone.om.response;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.ipc_.ProcessingDetails.Timing;
+import org.apache.hadoop.ipc_.RPC;
+import org.apache.hadoop.ipc_.Server;
import org.apache.hadoop.ozone.om.OMConfigKeys;
import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OMMetrics;
import org.apache.hadoop.ozone.om.OmConfig;
import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.om.execution.OMExecutionFlow;
import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
+import
org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer.RaftServerStatus;
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.hadoop.ozone.protocolPB.OzoneManagerProtocolServerSideTranslatorPB;
import org.apache.ozone.test.GenericTestUtils;
@@ -49,6 +58,8 @@ public class TestOMResponse {
private Path folder;
private OzoneManagerProtocolServerSideTranslatorPB translator;
+ private OzoneManager ozoneManager;
+ private OzoneManagerRatisServer ratisServer;
private GenericTestUtils.LogCapturer logCapturer;
private static final long IPC_MAX_LEN = 1024 * 1024; // 1MB
private static final long WARN_THRESHOLD = IPC_MAX_LEN / 2; // 512KB
@@ -61,7 +72,7 @@ public void setup() throws Exception {
configuration.setLong(
"ipc.maximum.response.length", IPC_MAX_LEN);
- OzoneManager ozoneManager = mock(OzoneManager.class);
+ ozoneManager = mock(OzoneManager.class);
OMMetadataManager omMetadataManager = new OmMetadataManagerImpl(
configuration, ozoneManager);
when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
@@ -73,7 +84,7 @@ public void setup() throws Exception {
OMExecutionFlow omExecutionFlow = new OMExecutionFlow(ozoneManager);
when(ozoneManager.getOmExecutionFlow()).thenReturn(omExecutionFlow);
- OzoneManagerRatisServer ratisServer = mock(OzoneManagerRatisServer.class);
+ ratisServer = mock(OzoneManagerRatisServer.class);
when(ratisServer.checkRetryCache()).thenReturn(null);
ProtocolMessageMetrics<OzoneManagerProtocolProtos.Type>
protocolMessageMetrics =
@@ -146,4 +157,40 @@ public void testSmallResponseNoLogging() {
long responseSize = smallResponse.getSerializedSize();
assertThat(responseSize).isLessThan(WARN_THRESHOLD);
}
+
+ @Test
+ public void testRatisReadLockDetailsAddedToCurrentCall() throws Exception {
+ OMExecutionFlow executionFlow = mock(OMExecutionFlow.class);
+ when(ozoneManager.getOmExecutionFlow()).thenReturn(executionFlow);
+ when(ozoneManager.getMetrics()).thenReturn(mock(OMMetrics.class));
+
when(ratisServer.getLeaderStatus()).thenReturn(RaftServerStatus.LEADER_AND_READY);
+ when(ratisServer.isLinearizableRead()).thenReturn(true);
+
+ OMRequest request = OMRequest.newBuilder()
+ .setCmdType(OzoneManagerProtocolProtos.Type.ServiceList)
+ .setClientId("test-client")
+ .build();
+ OMResponse response = OMResponse.newBuilder()
+ .setCmdType(request.getCmdType())
+ .setStatus(OzoneManagerProtocolProtos.Status.OK)
+ .setSuccess(true)
+
.setOmLockDetails(OzoneManagerProtocolProtos.OMLockDetailsProto.newBuilder()
+ .setWaitLockNanos(11)
+ .setReadLockNanos(22)
+ .setWriteLockNanos(33))
+ .build();
+ when(executionFlow.submit(any(OMRequest.class),
eq(false))).thenReturn(response);
+
+ Server.Call call = new Server.Call(1, 0, null, null,
RPC.RpcKind.RPC_PROTOCOL_BUFFER, new byte[0]);
+ Server.getCurCall().set(call);
+ try {
+ translator.processRequest(request);
+
+ assertThat(call.getProcessingDetails().get(Timing.LOCKWAIT,
TimeUnit.NANOSECONDS)).isEqualTo(11);
+ assertThat(call.getProcessingDetails().get(Timing.LOCKSHARED,
TimeUnit.NANOSECONDS)).isEqualTo(22);
+ assertThat(call.getProcessingDetails().get(Timing.LOCKEXCLUSIVE,
TimeUnit.NANOSECONDS)).isEqualTo(33);
+ } finally {
+ Server.getCurCall().remove();
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
