Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

HDDS-4766: Recon resets the Operational State of datanodes to IN_SERVICE #1857

Merged
merged 2 commits into from
Feb 2, 2021
Merged

HDDS-4766: Recon resets the Operational State of datanodes to IN_SERVICE #1857

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9391921
Fix for HDDS-4766
Jan 29, 2021
85a27a4
Filter commands which are not allowed from the heartbeat response and…
Feb 1, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -417,16 +417,17 @@ private void updateDatanodeOpState(DatanodeDetails reportedDn)
|| scmStatus.getOpStateExpiryEpochSeconds()
!= reportedDn.getPersistedOpStateExpiryEpochSec()) {
LOG.info("Scheduling a command to update the operationalState " +
"persisted on the datanode as the reported value ({}, {}) does not " +
"persisted on {} as the reported value does not " +
"match the value stored in SCM ({}, {})",
reportedDn.getPersistedOpState(),
reportedDn.getPersistedOpStateExpiryEpochSec(),
reportedDn,
scmStatus.getOperationalState(),
scmStatus.getOpStateExpiryEpochSeconds());
commandQueue.addCommand(reportedDn.getUuid(),

onMessage(new CommandForDatanode(reportedDn.getUuid(),
new SetNodeOperationalStateCommand(
Time.monotonicNow(), scmStatus.getOperationalState(),
scmStatus.getOpStateExpiryEpochSeconds()));
scmStatus.getOpStateExpiryEpochSeconds())
), null);
}
DatanodeDetails scmDnd = nodeStateManager.getNode(reportedDn);
scmDnd.setPersistedOpStateExpiryEpochSec(
Expand Down
8 changes: 7 additions & 1 deletion hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconNodeManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@
import org.apache.hadoop.util.Time;

import com.google.common.collect.ImmutableSet;

import static java.util.stream.Collectors.toList;
import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reregisterCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -129,6 +131,10 @@ public void onMessage(CommandForDatanode commandForDatanode,
public List<SCMCommand> processHeartbeat(DatanodeDetails datanodeDetails) {
// Update heartbeat map with current time
datanodeHeartbeatMap.put(datanodeDetails.getUuid(), Time.now());
return super.processHeartbeat(datanodeDetails);

List<SCMCommand> cmds = super.processHeartbeat(datanodeDetails);
return cmds.stream()
.filter(c -> ALLOWED_COMMANDS.contains(c.getType()))
.collect(toList());
}
}
25 changes: 25 additions & 0 deletions hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconNodeManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,16 +26,22 @@
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.util.List;
import java.util.UUID;

import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
import org.apache.hadoop.hdds.server.events.EventQueue;
import org.apache.hadoop.hdds.utils.db.DBStore;
import org.apache.hadoop.hdds.utils.db.DBStoreBuilder;
import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.protocol.commands.ReregisterCommand;
import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
import org.apache.hadoop.ozone.protocol.commands.SetNodeOperationalStateCommand;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
Expand Down Expand Up @@ -91,6 +97,25 @@ public void testReconNodeDB() throws IOException {
assertEquals(1, reconNodeManager.getAllNodes().size());
assertNotNull(reconNodeManager.getNodeByUuid(uuidString));

// If any commands are added to the eventQueue without using the onMessage
// interface, then they should be filtered out and not returned to the DN
// when it heartbeats.
// This command should never be returned by Recon
reconNodeManager.addDatanodeCommand(datanodeDetails.getUuid(),
new SetNodeOperationalStateCommand(1234,
HddsProtos.NodeOperationalState.DECOMMISSIONING, 0));

// This one should be returned
reconNodeManager.addDatanodeCommand(datanodeDetails.getUuid(),
new ReregisterCommand());

// Upon processing the heartbeat, the illegal command should be filtered out
List<SCMCommand> returnedCmds =
reconNodeManager.processHeartbeat(datanodeDetails);
assertEquals(1, returnedCmds.size());
assertEquals(SCMCommandProto.Type.reregisterCommand,
returnedCmds.get(0).getType());

// Close the DB, and recreate the instance of Recon Node Manager.
eventQueue.close();
reconNodeManager.close();
Expand Down