This is an automated email from the ASF dual-hosted git repository.
errose28 pushed a commit to branch HDDS-14496-zdu
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-14496-zdu by this push:
new b5c52d34c82 HDDS-15885. SCM should validate software versions of peers
before accepting finalize command (#10783)
b5c52d34c82 is described below
commit b5c52d34c8209cf86e47daefeeec097141a8b89c
Author: Ethan Rose <[email protected]>
AuthorDate: Mon Aug 10 19:08:26 2026 -0400
HDDS-15885. SCM should validate software versions of peers before accepting
finalize command (#10783)
---
.../protocol/StorageContainerLocationProtocol.java | 22 ++-
...inerLocationProtocolClientSideTranslatorPB.java | 25 ++-
.../src/main/proto/ScmAdminProtocol.proto | 15 ++
.../apache/hadoop/hdds/scm/node/NodeManager.java | 34 +++-
...inerLocationProtocolServerSideTranslatorPB.java | 21 ++-
.../hdds/scm/server/SCMClientProtocolServer.java | 86 ++++++++-
.../hadoop/hdds/scm/node/TestSCMNodeManager.java | 35 ++++
.../scm/server/TestSCMClientProtocolServer.java | 200 +++++++++++++++++++++
.../hadoop/hdds/upgrade/TestScmHAFinalization.java | 3 +-
.../upgrade/OMStartFinalizeUpgradeRequest.java | 6 +-
.../upgrade/TestOMStartFinalizeUpgradeRequest.java | 22 ++-
11 files changed, 456 insertions(+), 13 deletions(-)
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java
index 505341ffa18..e4254fcec1c 100644
---
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java
@@ -28,6 +28,7 @@
import java.util.Set;
import java.util.UUID;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.HDDSVersion;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -79,7 +80,8 @@ public interface StorageContainerLocationProtocol extends
Closeable {
*/
Set<Type> FOLLOWER_READABLE_COMMAND_TYPES =
Collections.unmodifiableSet(EnumSet.of(
Type.InSafeMode,
- Type.GetSafeModeRuleStatuses));
+ Type.GetSafeModeRuleStatuses,
+ Type.GetPeerUpgradeStatus));
/**
* Asks SCM where a container should be allocated. SCM responds with the
@@ -517,6 +519,15 @@ StatusAndMessages finalizeScmUpgrade(String
upgradeClientID)
void finalizeUpgrade() throws IOException;
+ /**
+ * Same as {@link #finalizeUpgrade()}, but SCM skips the peer SCM and
datanode software version
+ * checks before finalizing. Use this to finalize when a peer or datanode is
intentionally down or
+ * on a different version.
+ *
+ * @throws IOException If any error occurs.
+ */
+ void forceFinalizeUpgrade() throws IOException;
+
@Deprecated
StatusAndMessages queryUpgradeFinalizationProgress(
String upgradeClientID, boolean force, boolean readonly)
@@ -524,6 +535,15 @@ StatusAndMessages queryUpgradeFinalizationProgress(
HddsProtos.UpgradeStatus queryUpgradeStatus() throws IOException;
+ /**
+ * Returns the {@link HDDSVersion#SOFTWARE_VERSION} of the responding SCM
binary. Intended for the
+ * SCM leader to verify that all peer SCMs run a matching software version
before starting
+ * finalization. This is a node-local read (follower-readable): combined
with a {@code ScmNodeTarget}
+ * it queries one specific SCM without a leader round-trip. A version not
recognized by the calling
+ * binary deserializes to {@link HDDSVersion#UNKNOWN_VERSION}.
+ */
+ HDDSVersion getPeerUpgradeStatus() throws IOException;
+
/**
* Obtain a token which can be used to let datanodes verify authentication of
* commands operating on {@code containerID}.
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java
index a2a0b58ac25..292488f221d 100644
---
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java
@@ -38,6 +38,7 @@
import java.util.function.Consumer;
import java.util.stream.Collectors;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.HDDSVersion;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.client.ReplicatedReplicationConfig;
@@ -1198,8 +1199,17 @@ public StatusAndMessages finalizeScmUpgrade(String
upgradeClientID)
@Override
public void finalizeUpgrade() throws IOException {
+ finalizeUpgrade(false);
+ }
+
+ @Override
+ public void forceFinalizeUpgrade() throws IOException {
+ finalizeUpgrade(true);
+ }
+
+ private void finalizeUpgrade(boolean force) throws IOException {
FinalizeUpgradeRequestProto req = FinalizeUpgradeRequestProto.newBuilder()
- .build();
+ .setForce(force).build();
submitRequest(Type.FinalizeUpgrade, builder ->
builder.setFinalizeUpgradeRequest(req));
}
@@ -1240,6 +1250,19 @@ public HddsProtos.UpgradeStatus queryUpgradeStatus()
throws IOException {
return response.getStatus();
}
+ @Override
+ public HDDSVersion getPeerUpgradeStatus() throws IOException {
+ StorageContainerLocationProtocolProtos.GetPeerUpgradeStatusRequestProto
req =
+ StorageContainerLocationProtocolProtos.GetPeerUpgradeStatusRequestProto
+ .newBuilder()
+ .build();
+
+ StorageContainerLocationProtocolProtos.GetPeerUpgradeStatusResponseProto
response =
+ submitRequest(Type.GetPeerUpgradeStatus, builder ->
builder.setGetPeerUpgradeStatusRequest(req))
+ .getGetPeerUpgradeStatusResponse();
+ return HDDSVersion.deserialize(response.getScmSoftwareVersion());
+ }
+
@Override
public Token<?> getContainerToken(
ContainerID containerID) throws IOException {
diff --git a/hadoop-hdds/interface-admin/src/main/proto/ScmAdminProtocol.proto
b/hadoop-hdds/interface-admin/src/main/proto/ScmAdminProtocol.proto
index e9f7335f4b9..b61764d1481 100644
--- a/hadoop-hdds/interface-admin/src/main/proto/ScmAdminProtocol.proto
+++ b/hadoop-hdds/interface-admin/src/main/proto/ScmAdminProtocol.proto
@@ -91,6 +91,7 @@ message ScmContainerLocationRequest {
optional SuppressContainerRequestProto suppressContainerRequest = 52;
optional QueryUpgradeStatusRequestProto queryUpgradeStatusRequest = 53;
optional FinalizeUpgradeRequestProto finalizeUpgradeRequest = 54;
+ optional GetPeerUpgradeStatusRequestProto getPeerUpgradeStatusRequest = 55;
}
message ScmContainerLocationResponse {
@@ -153,6 +154,7 @@ message ScmContainerLocationResponse {
optional SuppressContainerResponseProto suppressContainerResponse = 52;
optional QueryUpgradeStatusResponseProto queryUpgradeStatusResponse = 53;
optional FinalizeUpgradeResponseProto finalizeUpgradeResponse = 54;
+ optional GetPeerUpgradeStatusResponseProto getPeerUpgradeStatusResponse = 55;
enum Status {
OK = 1;
@@ -214,6 +216,7 @@ enum Type {
SuppressContainer = 48;
QueryUpgradeStatus = 49;
FinalizeUpgrade = 50;
+ GetPeerUpgradeStatus = 51;
}
/**
@@ -603,11 +606,23 @@ message QueryUpgradeStatusResponseProto {
}
message FinalizeUpgradeRequestProto {
+ optional bool force = 1 [default = false];
}
message FinalizeUpgradeResponseProto {
}
+// Request from an SCM leader to a peer SCM to read its local upgrade status.
+// The resulting status returned currently only contains the SCM's software
version to use in pre-finalization
+// safety checks.
+message GetPeerUpgradeStatusRequestProto {
+}
+
+message GetPeerUpgradeStatusResponseProto {
+ // Serialized HDDSVersion.SOFTWARE_VERSION of the responding SCM binary.
+ optional int32 scmSoftwareVersion = 1;
+}
+
message ContainerTokenSecretProto {
required string ownerId = 1;
required ContainerID containerId = 2;
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
index 0678d992c20..c36419de6ec 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
@@ -28,6 +28,7 @@
import java.util.Set;
import java.util.function.BiConsumer;
import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.HDDSVersion;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.DatanodeID;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState;
@@ -147,11 +148,13 @@ default int getAllNodeCount() {
}
/**
- * @return DatanodeFinalizationCounts, finalized and total healthy node
counts
+ * @return DatanodeFinalizationCounts, finalized and total healthy node
counts, and whether all
+ * healthy datanodes run SCM's software version
*/
default DatanodeFinalizationCounts getDatanodeFinalizationCounts() {
int finalizedNodes = 0;
int totalHealthyNodes = 0;
+ boolean allSoftwareVersionsMatchScm = true;
int minApparentVersion = Integer.MAX_VALUE;
int maxApparentVersion = 0;
@@ -186,6 +189,15 @@ default DatanodeFinalizationCounts
getDatanodeFinalizationCounts() {
} else {
finalizedNodes++;
}
+
+ // Track whether every healthy datanode is running the same software
version as SCM. A
+ // datanode on a lower (or unknown) software version must not be
present when SCM finalizes.
+ if (!HDDSVersion.SOFTWARE_VERSION.equals(dnSoftwareVersion)) {
+ allSoftwareVersionsMatchScm = false;
+ // This is expected during a rolling upgrade. Do not flood the logs
with one message for every datanode.
+ LOG.debug("Datanode {} software version {} does not match SCM
software version {}.",
+ dn.getHostName(), dnSoftwareVersion,
HDDSVersion.SOFTWARE_VERSION);
+ }
} catch (NodeNotFoundException e) {
// Node was removed while we were iterating. This is OK, skip it.
LOG.debug("Node {} not found while waiting for finalization, " +
@@ -203,6 +215,7 @@ default DatanodeFinalizationCounts
getDatanodeFinalizationCounts() {
.setTotalHealthyDatanodes(totalHealthyNodes)
.setMinApparentVersion(minApparentVersion)
.setMaxApparentVersion(maxApparentVersion)
+ .setAllSoftwareVersionsMatchScm(allSoftwareVersionsMatchScm)
.build();
}
@@ -415,8 +428,7 @@ void processNodeReport(DatanodeDetails datanodeDetails,
* @param datanodeDetails
* @param versionReport
*/
- void processVersionReport(DatanodeDetails datanodeDetails,
- DatanodeVersionProto versionReport);
+ void processVersionReport(DatanodeDetails datanodeDetails,
DatanodeVersionProto versionReport);
/**
* Get the number of commands of the given type queued on the datanode at the
@@ -538,12 +550,14 @@ final class DatanodeFinalizationCounts {
private final int totalHealthyDatanodes;
private final int minApparentVersion;
private final int maxApparentVersion;
+ private final boolean allSoftwareVersionsMatchScm;
private DatanodeFinalizationCounts(Builder b) {
this.numFinalizedDatanodes = b.numFinalizedDatanodes;
this.totalHealthyDatanodes = b.totalHealthyDatanodes;
this.minApparentVersion = b.minApparentVersion;
this.maxApparentVersion = b.maxApparentVersion;
+ this.allSoftwareVersionsMatchScm = b.allSoftwareVersionsMatchScm;
}
public static Builder newBuilder() {
@@ -570,6 +584,14 @@ public int getMaxApparentVersion() {
return maxApparentVersion;
}
+ /**
+ * @return true if every healthy datanode reports the same software
version as SCM
+ * ({@link HDDSVersion#SOFTWARE_VERSION}), vacuously true when there are
no healthy datanodes
+ */
+ public boolean allSoftwareVersionsMatchScmVersion() {
+ return allSoftwareVersionsMatchScm;
+ }
+
/**
* Builder for {@link DatanodeFinalizationCounts}.
*/
@@ -578,6 +600,7 @@ public static final class Builder {
private int totalHealthyDatanodes;
private int minApparentVersion;
private int maxApparentVersion;
+ private boolean allSoftwareVersionsMatchScm;
public Builder setNumFinalizedDatanodes(int value) {
this.numFinalizedDatanodes = value;
@@ -599,6 +622,11 @@ public Builder setMaxApparentVersion(int value) {
return this;
}
+ public Builder setAllSoftwareVersionsMatchScm(boolean value) {
+ this.allSoftwareVersionsMatchScm = value;
+ return this;
+ }
+
public DatanodeFinalizationCounts build() {
return new DatanodeFinalizationCounts(this);
}
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java
index a348b49f276..c95d5145215 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java
@@ -774,10 +774,20 @@ public ScmContainerLocationResponse processRequest(
.setQueryUpgradeStatusResponse(getQueryUpgradeStatus(request.getQueryUpgradeStatusRequest()))
.build();
case FinalizeUpgrade:
- impl.finalizeUpgrade();
+ if (request.getFinalizeUpgradeRequest().getForce()) {
+ impl.forceFinalizeUpgrade();
+ } else {
+ impl.finalizeUpgrade();
+ }
+ return ScmContainerLocationResponse.newBuilder()
+ .setCmdType(request.getCmdType())
+ .setStatus(Status.OK)
+ .build();
+ case GetPeerUpgradeStatus:
return ScmContainerLocationResponse.newBuilder()
.setCmdType(request.getCmdType())
.setStatus(Status.OK)
+
.setGetPeerUpgradeStatusResponse(getPeerUpgradeStatus(request.getGetPeerUpgradeStatusRequest()))
.build();
default:
throw new IllegalArgumentException(
@@ -1132,6 +1142,15 @@ public
StorageContainerLocationProtocolProtos.QueryUpgradeStatusResponseProto ge
.build();
}
+ public
StorageContainerLocationProtocolProtos.GetPeerUpgradeStatusResponseProto
getPeerUpgradeStatus(
+ StorageContainerLocationProtocolProtos.GetPeerUpgradeStatusRequestProto
request) throws IOException {
+
+ return
StorageContainerLocationProtocolProtos.GetPeerUpgradeStatusResponseProto
+ .newBuilder()
+ .setScmSoftwareVersion(impl.getPeerUpgradeStatus().serialize())
+ .build();
+ }
+
public ForceExitSafeModeResponseProto forceExitSafeMode(
ForceExitSafeModeRequestProto request)
throws IOException {
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
index 54434de80ec..e78395fc44c 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
@@ -54,6 +54,7 @@
import java.util.stream.Stream;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.hdds.HDDSVersion;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.ReconfigurationHandler;
@@ -92,6 +93,7 @@
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeDetails;
import org.apache.hadoop.hdds.scm.ha.SCMRatisServer;
import org.apache.hadoop.hdds.scm.ha.SCMRatisServerImpl;
import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
@@ -105,8 +107,10 @@
import org.apache.hadoop.hdds.scm.pipeline.PipelineNotFoundException;
import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
import
org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocolServerSideTranslatorPB;
+import
org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB.ScmNodeTarget;
import
org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolPB;
import org.apache.hadoop.hdds.security.SecurityConfig;
+import org.apache.hadoop.hdds.utils.HAUtils;
import org.apache.hadoop.hdds.utils.HddsServerUtil;
import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
import org.apache.hadoop.io.IOUtils;
@@ -1166,9 +1170,24 @@ public StatusAndMessages finalizeScmUpgrade(String
upgradeClientID) {
@Override
public void finalizeUpgrade() throws IOException {
- final Map<String, String> auditMap = Collections.emptyMap();
+ finalizeUpgrade(false);
+ }
+
+ @Override
+ public void forceFinalizeUpgrade() throws IOException {
+ finalizeUpgrade(true);
+ }
+
+ private void finalizeUpgrade(boolean force) throws IOException {
+ final Map<String, String> auditMap = Collections.singletonMap("force",
String.valueOf(force));
try {
getScm().checkAdminAccess(getRemoteUser(), false);
+ if (force) {
+ LOG.warn("Forcing upgrade finalization by skipping SCM peer and
datanode software version checks");
+ } else {
+ validatePeerScmVersionsBeforeFinalize();
+ validateDatanodeVersionsBeforeFinalize();
+ }
scm.getFinalizationManager().finalizeUpgrade();
AUDIT.logWriteSuccess(buildAuditMessageForSuccess(SCMAction.FINALIZE_SCM_UPGRADE,
auditMap));
} catch (Exception ex) {
@@ -1177,6 +1196,71 @@ public void finalizeUpgrade() throws IOException {
}
}
+ @Override
+ public HDDSVersion getPeerUpgradeStatus() throws IOException {
+ return HDDSVersion.SOFTWARE_VERSION;
+ }
+
+ /**
+ * Verifies that every peer SCM in the Ratis group runs the same software
version as this leader
+ * before finalization begins. Rejects the finalize command if any peer
reports a differing
+ * version or cannot be reached. The resulting exception propagates back to
the OM (which triggered
+ * finalization) and on to the client, leaving nothing finalized.
+ */
+ private void validatePeerScmVersionsBeforeFinalize() throws SCMException {
+ List<SCMNodeDetails> peerNodes =
scm.getSCMHANodeDetails().getPeerNodeDetails();
+ if (peerNodes.isEmpty()) {
+ return;
+ }
+ HDDSVersion leaderVersion = HDDSVersion.SOFTWARE_VERSION;
+ OzoneConfiguration conf = scm.getConfiguration();
+ List<String> failedPeers = new ArrayList<>();
+ for (SCMNodeDetails peer : peerNodes) {
+ String peerId = peer.getNodeId();
+ ScmNodeTarget target = new ScmNodeTarget();
+ target.setNodeId(peerId);
+ StorageContainerLocationProtocol peerClient = null;
+ try {
+ peerClient = HAUtils.getScmContainerClientForNode(conf, target);
+ HDDSVersion peerVersion = peerClient.getPeerUpgradeStatus();
+ if (!peerVersion.equals(leaderVersion)) {
+ LOG.warn("SCM peer {} is running software version {} but leader is
running version {}. "
+ + "Rejecting finalize command.", peerId, peerVersion,
leaderVersion);
+ failedPeers.add(peerId + " (version: " + peerVersion + ")");
+ }
+ } catch (IOException e) {
+ LOG.warn("Failed to contact SCM peer {} to check software version
before finalize.", peerId, e);
+ failedPeers.add(peerId + " (unreachable: " + e.getMessage() + ")");
+ } finally {
+ IOUtils.cleanupWithLogger(LOG, peerClient);
+ }
+ }
+ if (!failedPeers.isEmpty()) {
+ throw new SCMException("Finalize rejected: the following SCM peers did
not confirm matching software "
+ + "version (expected version=" + leaderVersion + "): " +
String.join(", ", failedPeers),
+ ResultCodes.UNSUPPORTED_OPERATION);
+ }
+ }
+
+ /**
+ * Verifies that every healthy datanode runs the same software version as
this SCM before
+ * finalization begins. Datanodes finalize only after SCM instructs them to,
so this does not
+ * require them to be finalized; it requires their binaries to match SCM's
software version.
+ * Rejects the finalize command if any healthy datanode reports a differing
(or unknown) software
+ * version. The resulting exception propagates back to the OM (which
triggered finalization) and on
+ * to the client, leaving nothing finalized.
+ */
+ private void validateDatanodeVersionsBeforeFinalize() throws SCMException {
+ NodeManager.DatanodeFinalizationCounts counts =
+ scm.getScmNodeManager().getDatanodeFinalizationCounts();
+ if (!counts.allSoftwareVersionsMatchScmVersion()) {
+ LOG.warn("Rejecting finalize command: not all {} healthy datanodes are
running SCM's software "
+ + "version {}.", counts.getTotalHealthyDatanodes(),
HDDSVersion.SOFTWARE_VERSION);
+ throw new SCMException("Finalize rejected: not all healthy datanodes are
running the SCM software version "
+ + HDDSVersion.SOFTWARE_VERSION, ResultCodes.UNSUPPORTED_OPERATION);
+ }
+ }
+
@Override
@Deprecated
public StatusAndMessages queryUpgradeFinalizationProgress(
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java
index c689d30c3b6..eea8569b335 100644
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java
@@ -45,6 +45,7 @@
import static org.apache.ozone.test.MetricsAsserts.getMetrics;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -650,6 +651,40 @@ public void
testDatanodeFinalizedCounterTracksVersionReports()
}
}
+ @Test
+ public void testDatanodeFinalizationCountsSoftwareVersionMatch()
+ throws IOException, AuthenticationException {
+ try (SCMNodeManager nodeManager = createNodeManager(getConf())) {
+ // With no datanodes, all software versions trivially match SCM.
+
assertTrue(nodeManager.getDatanodeFinalizationCounts().allSoftwareVersionsMatchScmVersion(),
+ "With no datanodes, all software versions trivially match SCM");
+
+ // A finalized datanode runs SCM's software version.
+ registerWithCapacity(nodeManager, defaultVersionProto(), success);
+
assertTrue(nodeManager.getDatanodeFinalizationCounts().allSoftwareVersionsMatchScmVersion(),
+ "A finalized datanode runs SCM's software version");
+
+ // A pre-finalized datanode runs SCM's software version even though its
apparent version is lower,
+ // so all datanodes still match SCM's software version.
+ DatanodeDetails preFinalizedNode = registerWithCapacity(nodeManager,
+ toVersionProto(HDDSLayoutFeature.INITIAL_VERSION,
HDDSVersion.SOFTWARE_VERSION), success);
+
assertTrue(nodeManager.getDatanodeFinalizationCounts().allSoftwareVersionsMatchScmVersion(),
+ "A pre-finalized datanode still runs SCM's software version");
+
+ // Report a lower software version for one datanode: it no longer
matches SCM.
+ nodeManager.processVersionReport(preFinalizedNode,
+ toVersionProto(HDDSLayoutFeature.INITIAL_VERSION,
HDDSLayoutFeature.INITIAL_VERSION));
+
assertFalse(nodeManager.getDatanodeFinalizationCounts().allSoftwareVersionsMatchScmVersion(),
+ "A datanode on a lower software version should not match SCM");
+
+ // Restore the datanode to SCM's software version: all datanodes match
again.
+ nodeManager.processVersionReport(preFinalizedNode,
+ toVersionProto(HDDSLayoutFeature.INITIAL_VERSION,
HDDSVersion.SOFTWARE_VERSION));
+
assertTrue(nodeManager.getDatanodeFinalizationCounts().allSoftwareVersionsMatchScmVersion(),
+ "Restoring the datanode to SCM's software version should make all
datanodes match again");
+ }
+ }
+
@Test
public void testDatanodeFinalizedCounterTracksRegistrationAndRemoveNode()
throws IOException, AuthenticationException, NodeNotFoundException {
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/TestSCMClientProtocolServer.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/TestSCMClientProtocolServer.java
index f044656b203..c21fcf11877 100644
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/TestSCMClientProtocolServer.java
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/TestSCMClientProtocolServer.java
@@ -24,7 +24,9 @@
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -37,9 +39,11 @@
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.HDDSVersion;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.ReconfigurationHandler;
@@ -54,24 +58,30 @@
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub;
+import org.apache.hadoop.hdds.scm.ha.SCMHANodeDetails;
import org.apache.hadoop.hdds.scm.ha.SCMNodeDetails;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
import
org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocolServerSideTranslatorPB;
import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager;
import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager.SafeModeStatus;
import org.apache.hadoop.hdds.scm.server.upgrade.FinalizationManager;
import org.apache.hadoop.hdds.scm.server.upgrade.ScmVersionManager;
+import org.apache.hadoop.hdds.utils.HAUtils;
import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.ozone.audit.SCMAction;
import org.apache.hadoop.ozone.container.common.SCMTestUtils;
import org.apache.hadoop.ozone.upgrade.UpgradeFinalization.StatusAndMessages;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ozone.test.GenericTestUtils.LogCapturer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedStatic;
/**
* Unit tests to validate the SCMClientProtocolServer
@@ -279,6 +289,196 @@ private SCMClientProtocolServer
serverWithMockFinalization(
new OzoneConfiguration(), mockScm, mock(ReconfigurationHandler.class));
}
+ @Test
+ public void testGetPeerUpgradeStatusReturnsLocalVersion() throws IOException
{
+ assertEquals(HDDSVersion.SOFTWARE_VERSION, server.getPeerUpgradeStatus());
+ }
+
+ @Test
+ public void testFinalizeProceedsWhenNoPeers() throws IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ try (SCMClientProtocolServer testServer =
+ peerCheckServer(finalizationManager, Collections.emptyList())) {
+ testServer.finalizeUpgrade();
+ verify(finalizationManager).finalizeUpgrade();
+ }
+ }
+
+ @Test
+ public void testFinalizeProceedsWhenAllPeersMatch() throws IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ StorageContainerLocationProtocol matching =
peerClient(HDDSVersion.SOFTWARE_VERSION);
+
+ try (SCMClientProtocolServer testServer =
+ peerCheckServer(finalizationManager,
Arrays.asList(peerNode("scm2"), peerNode("scm3")));
+ MockedStatic<HAUtils> haUtils = mockStatic(HAUtils.class)) {
+ haUtils.when(() -> HAUtils.getScmContainerClientForNode(any(),
any())).thenReturn(matching);
+ testServer.finalizeUpgrade();
+ }
+ verify(finalizationManager).finalizeUpgrade();
+ }
+
+ @Test
+ public void testFinalizeRejectsOlderPeerUnlessForced() throws IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ StorageContainerLocationProtocol matching =
peerClient(HDDSVersion.SOFTWARE_VERSION);
+ StorageContainerLocationProtocol older =
peerClient(HDDSVersion.DEFAULT_VERSION);
+
+ try (SCMClientProtocolServer testServer =
+ peerCheckServer(finalizationManager,
Arrays.asList(peerNode("scm2"), peerNode("scm3")));
+ MockedStatic<HAUtils> haUtils = mockStatic(HAUtils.class)) {
+ haUtils.when(() -> HAUtils.getScmContainerClientForNode(any(),
any())).thenReturn(matching, older);
+ // A peer on an older version is rejected without force.
+ assertThrows(SCMException.class, testServer::finalizeUpgrade);
+ verify(finalizationManager, never()).finalizeUpgrade();
+ // With force the peer version check is skipped and finalization
proceeds.
+ testServer.forceFinalizeUpgrade();
+ }
+ verify(finalizationManager).finalizeUpgrade();
+ }
+
+ @Test
+ public void testFinalizeRejectsUnknownFuturePeerUnlessForced() throws
IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ StorageContainerLocationProtocol matching =
peerClient(HDDSVersion.SOFTWARE_VERSION);
+ // A version not recognized by this binary deserializes to UNKNOWN_VERSION
in the client translator.
+ StorageContainerLocationProtocol unknown =
peerClient(HDDSVersion.UNKNOWN_VERSION);
+
+ try (SCMClientProtocolServer testServer =
+ peerCheckServer(finalizationManager,
Arrays.asList(peerNode("scm2"), peerNode("scm3")));
+ MockedStatic<HAUtils> haUtils = mockStatic(HAUtils.class)) {
+ haUtils.when(() -> HAUtils.getScmContainerClientForNode(any(), any()))
+ .thenReturn(matching, unknown);
+ // A peer on an unrecognized future version is rejected without force.
+ assertThrows(SCMException.class, testServer::finalizeUpgrade);
+ verify(finalizationManager, never()).finalizeUpgrade();
+ // With force the peer version check is skipped and finalization
proceeds.
+ testServer.forceFinalizeUpgrade();
+ }
+ verify(finalizationManager).finalizeUpgrade();
+ }
+
+ @Test
+ public void testFinalizeRejectsUnreachablePeerUnlessForced() throws
IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ StorageContainerLocationProtocol unreachable =
mock(StorageContainerLocationProtocol.class);
+ when(unreachable.getPeerUpgradeStatus()).thenThrow(new
IOException("connection refused"));
+
+ try (SCMClientProtocolServer testServer =
+ peerCheckServer(finalizationManager,
Collections.singletonList(peerNode("scm2")));
+ MockedStatic<HAUtils> haUtils = mockStatic(HAUtils.class)) {
+ haUtils.when(() -> HAUtils.getScmContainerClientForNode(any(),
any())).thenReturn(unreachable);
+ // An unreachable peer is rejected without force.
+ assertThrows(SCMException.class, testServer::finalizeUpgrade);
+ verify(finalizationManager, never()).finalizeUpgrade();
+ // With force the peer version check is skipped and finalization
proceeds.
+ testServer.forceFinalizeUpgrade();
+ }
+ verify(finalizationManager).finalizeUpgrade();
+ }
+
+ @Test
+ public void testFinalizeProceedsWhenAllDatanodesMatchScmVersion() throws
IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ NodeManager.DatanodeFinalizationCounts datanodeCounts =
NodeManager.DatanodeFinalizationCounts.newBuilder()
+ .setNumFinalizedDatanodes(3)
+ .setTotalHealthyDatanodes(3)
+ .setMinApparentVersion(HDDSVersion.SOFTWARE_VERSION.serialize())
+ .setMaxApparentVersion(HDDSVersion.SOFTWARE_VERSION.serialize())
+ .setAllSoftwareVersionsMatchScm(true)
+ .build();
+ try (SCMClientProtocolServer testServer =
peerCheckServer(finalizationManager,
+ Collections.emptyList(), datanodeCounts)) {
+ testServer.finalizeUpgrade();
+ verify(finalizationManager).finalizeUpgrade();
+ }
+ }
+
+ @Test
+ public void testFinalizeRejectsDatanodeWithMismatchedVersionUnlessForced()
throws IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ NodeManager.DatanodeFinalizationCounts datanodeCounts =
NodeManager.DatanodeFinalizationCounts.newBuilder()
+ .setNumFinalizedDatanodes(3)
+ .setTotalHealthyDatanodes(3)
+ .setMinApparentVersion(HDDSVersion.DEFAULT_VERSION.serialize())
+ .setMaxApparentVersion(HDDSVersion.SOFTWARE_VERSION.serialize())
+ .setAllSoftwareVersionsMatchScm(false)
+ .build();
+ try (SCMClientProtocolServer testServer =
peerCheckServer(finalizationManager,
+ Collections.emptyList(), datanodeCounts)) {
+ // A datanode on a mismatched version is rejected without force.
+ assertThrows(SCMException.class, testServer::finalizeUpgrade);
+ verify(finalizationManager, never()).finalizeUpgrade();
+ // With force the datanode version check is skipped and finalization
proceeds.
+ testServer.forceFinalizeUpgrade();
+ verify(finalizationManager).finalizeUpgrade();
+ }
+ }
+
+ @Test
+ public void testForceFinalizeAuditRecordsForceFlag() throws IOException {
+ FinalizationManager finalizationManager = mock(FinalizationManager.class);
+ // Drive the failure audit path (logged at ERROR, captured by the default
log4j2 config) so the
+ // recorded audit map can be inspected.
+ doThrow(new
RuntimeException("test")).when(finalizationManager).finalizeUpgrade();
+
+ LogCapturer auditLog = LogCapturer.log4j2("SCMAudit");
+ try (SCMClientProtocolServer testServer =
peerCheckServer(finalizationManager, Collections.emptyList())) {
+ assertThrows(RuntimeException.class, testServer::forceFinalizeUpgrade);
+ } finally {
+ auditLog.stopCapturing();
+ }
+
+ String output = auditLog.getOutput();
+ assertTrue(output.contains(SCMAction.FINALIZE_SCM_UPGRADE.getAction()),
+ "audit log should record the finalize action: " + output);
+ assertTrue(output.contains("\"force\":\"true\""),
+ "audit log should record that force was passed: " + output);
+ }
+
+ private SCMClientProtocolServer peerCheckServer(
+ FinalizationManager finalizationManager, List<SCMNodeDetails> peers)
throws IOException {
+ // Default to all datanode versions matching SCM so the SCM peer checks
are exercised in isolation.
+ NodeManager.DatanodeFinalizationCounts datanodeCounts =
NodeManager.DatanodeFinalizationCounts.newBuilder()
+ .setNumFinalizedDatanodes(0)
+ .setTotalHealthyDatanodes(0)
+ .setMinApparentVersion(0)
+ .setMaxApparentVersion(0)
+ .setAllSoftwareVersionsMatchScm(true)
+ .build();
+ return peerCheckServer(finalizationManager, peers, datanodeCounts);
+ }
+
+ private SCMClientProtocolServer peerCheckServer(
+ FinalizationManager finalizationManager, List<SCMNodeDetails> peers,
+ NodeManager.DatanodeFinalizationCounts datanodeCounts) throws
IOException {
+
+ StorageContainerManager mockScm = mockStorageContainerManager();
+ when(mockScm.getFinalizationManager()).thenReturn(finalizationManager);
+ when(mockScm.getConfiguration()).thenReturn(new OzoneConfiguration());
+
+ SCMHANodeDetails haNodeDetails = mock(SCMHANodeDetails.class);
+ when(haNodeDetails.getPeerNodeDetails()).thenReturn(peers);
+ when(mockScm.getSCMHANodeDetails()).thenReturn(haNodeDetails);
+
+ NodeManager nodeManager = mock(NodeManager.class);
+
when(nodeManager.getDatanodeFinalizationCounts()).thenReturn(datanodeCounts);
+ when(mockScm.getScmNodeManager()).thenReturn(nodeManager);
+ return new SCMClientProtocolServer(new OzoneConfiguration(), mockScm,
mock(ReconfigurationHandler.class));
+ }
+
+ private StorageContainerLocationProtocol peerClient(HDDSVersion version)
throws IOException {
+ StorageContainerLocationProtocol client =
mock(StorageContainerLocationProtocol.class);
+ when(client.getPeerUpgradeStatus()).thenReturn(version);
+ return client;
+ }
+
+ private SCMNodeDetails peerNode(String nodeId) {
+ SCMNodeDetails node = mock(SCMNodeDetails.class);
+ when(node.getNodeId()).thenReturn(nodeId);
+ return node;
+ }
+
@Test
public void testQueryUpgradeStatus() throws Exception {
HddsProtos.UpgradeStatus status = server.queryUpgradeStatus();
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/upgrade/TestScmHAFinalization.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/upgrade/TestScmHAFinalization.java
index 802562f886a..cbf3bcece0d 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/upgrade/TestScmHAFinalization.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/upgrade/TestScmHAFinalization.java
@@ -167,7 +167,8 @@ public void testSnapshotFinalization() throws Exception {
}
// Wait for finalization from the client perspective.
- scmClient.finalizeUpgrade();
+ // Force finalize skips the peer version checks that require all peers to
be active.
+ scmClient.forceFinalizeUpgrade();
HddsUpgradeTestUtils.waitForFinalizationFromClient(scmClient);
// Wait for two running SCMs to finish finalization.
waitForScmsToFinalize(activeScms);
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMStartFinalizeUpgradeRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMStartFinalizeUpgradeRequest.java
index 29ad641a982..cce820afdc4 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMStartFinalizeUpgradeRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMStartFinalizeUpgradeRequest.java
@@ -79,7 +79,11 @@ public OMRequest preExecute(OzoneManager ozoneManager)
throws IOException {
}
try {
- ozoneManager.getScmClient().getContainerClient().finalizeUpgrade();
+ if (force) {
+
ozoneManager.getScmClient().getContainerClient().forceFinalizeUpgrade();
+ } else {
+ ozoneManager.getScmClient().getContainerClient().finalizeUpgrade();
+ }
} catch (SCMException e) {
if (e.getResult() == SCMException.ResultCodes.UNSUPPORTED_OPERATION) {
throw new OMException(e.getMessage(), e,
OMException.ResultCodes.NOT_SUPPORTED_OPERATION);
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/upgrade/TestOMStartFinalizeUpgradeRequest.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/upgrade/TestOMStartFinalizeUpgradeRequest.java
index 7ffb7d90cf1..88746f0308a 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/upgrade/TestOMStartFinalizeUpgradeRequest.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/upgrade/TestOMStartFinalizeUpgradeRequest.java
@@ -77,6 +77,18 @@ public void testPreExecuteCallsScmFinalizeUpgrade() throws
IOException {
// SCM must have been asked to begin finalization.
verify(scmContainerLocationProtocol).finalizeUpgrade();
+ verify(scmContainerLocationProtocol, never()).forceFinalizeUpgrade();
+ }
+
+ @Test
+ public void testForcePreExecuteCallsScmForceFinalizeUpgrade() throws
IOException {
+ OMStartFinalizeUpgradeRequest request = new
OMStartFinalizeUpgradeRequest(buildRequest(true));
+
+ request.preExecute(ozoneManager);
+
+ // A forced request must route to SCM's force path so SCM skips its own
version checks.
+ verify(scmContainerLocationProtocol).forceFinalizeUpgrade();
+ verify(scmContainerLocationProtocol, never()).finalizeUpgrade();
}
@Test
@@ -282,12 +294,13 @@ public void
testForceSkipsPeerVersionCheckForUnreachablePeer() throws IOExceptio
.thenReturn(unreachableClient);
// With force=true the peer version check is skipped, so an unreachable
peer does not
- // prevent finalization and SCM is still asked to begin finalizing.
+ // prevent finalization and SCM is still asked to force finalization
(skipping its own checks).
new
OMStartFinalizeUpgradeRequest(buildRequest(true)).preExecute(ozoneManager);
}
verify(unreachableClient, never()).getPeerUpgradeStatus();
- verify(scmContainerLocationProtocol).finalizeUpgrade();
+ verify(scmContainerLocationProtocol).forceFinalizeUpgrade();
+ verify(scmContainerLocationProtocol, never()).finalizeUpgrade();
}
@Test
@@ -302,12 +315,13 @@ public void
testForceSkipsPeerVersionCheckForMismatchedPeer() throws IOException
.thenReturn(olderClient);
// With force=true the peer version check is skipped, so a peer running
a different software
- // version does not prevent finalization and SCM is still asked to begin
finalizing.
+ // version does not prevent finalization and SCM is still asked to force
finalization.
new
OMStartFinalizeUpgradeRequest(buildRequest(true)).preExecute(ozoneManager);
}
verify(olderClient, never()).getPeerUpgradeStatus();
- verify(scmContainerLocationProtocol).finalizeUpgrade();
+ verify(scmContainerLocationProtocol).forceFinalizeUpgrade();
+ verify(scmContainerLocationProtocol, never()).finalizeUpgrade();
}
private static OMNodeDetails buildPeer(String nodeId) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
