Skip to content

Commit

Permalink
raftstore: Revert remove peer check (#16479)
Browse files Browse the repository at this point in the history 
close #16465

revert the remove peer check. Without this revert, the scale-in will be blocked when there're one slow or down peer in a region of the scale-in node.

Signed-off-by: tonyxuqqi <[email protected]>
  • Loading branch information
@tonyxuqqi
tonyxuqqi committed Feb 1, 2024
1 parent 13bbe32 commit 4653fc6
Show file tree
Hide file tree
Showing 5 changed files with 0 additions and 220 deletions.
1 change: 0 additions & 1 deletion components/raftstore-v2/src/operation/command/admin/conf_change.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,6 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
changes.as_ref(),
&cc,
false,
self.get_peer_heartbeats(),
)?;

// TODO: check if the new peer is already in history record.
Expand Down
5 changes: 0 additions & 5 deletions components/raftstore-v2/src/raft/peer.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -412,11 +412,6 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
self.peer_heartbeats.remove(&peer_id);
}

#[inline]
pub fn get_peer_heartbeats(&self) -> &HashMap<u64, Instant> {
&self.peer_heartbeats
}

pub fn collect_down_peers(&self, max_duration: Duration) -> Vec<pdpb::PeerStats> {
let mut down_peers = Vec::new();
let now = Instant::now();
Expand Down
1 change: 0 additions & 1 deletion components/raftstore/src/store/peer.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4847,7 +4847,6 @@ where
changes.as_ref(),
&cc,
self.is_force_leader(),
&self.peer_heartbeats,
)?;

ctx.raft_metrics.propose.conf_change.inc();
Expand Down
177 changes: 0 additions & 177 deletions components/raftstore/src/store/util.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -909,7 +909,6 @@ pub fn check_conf_change(
change_peers: &[ChangePeerRequest],
cc: &impl ConfChangeI,
ignore_safety: bool,
peer_heartbeat: &collections::HashMap<u64, std::time::Instant>,
) -> Result<()> {
let current_progress = node.status().progress.unwrap().clone();
let mut after_progress = current_progress.clone();
Expand Down Expand Up @@ -993,7 +992,6 @@ pub fn check_conf_change(
return Err(box_err!("multiple changes that only effect learner"));
}

check_remove_or_demote_voter(region, cfg, change_peers, leader.get_id(), peer_heartbeat)?;
if !ignore_safety {
let promoted_commit_index = after_progress.maximal_committed_index().0;
let first_index = node.raft.raft_log.first_index();
Expand All @@ -1020,82 +1018,6 @@ pub fn check_conf_change(
}
}

fn check_remove_or_demote_voter(
region: &metapb::Region,
cfg: &Config,
change_peers: &[ChangePeerRequest],
leader_id: u64,
peer_heartbeat: &collections::HashMap<u64, std::time::Instant>,
) -> Result<()> {
let mut slow_peer_count = 0;
let mut normal_peer_count = 0;
// Here we assume if the last beartbeat is within 2 election timeout, the peer
// is healthy. When a region is hibernate, we expect all its peers are *slow*
// and it would still allow the operation
let slow_peer_threshold =
2 * cfg.raft_base_tick_interval.0 * cfg.raft_max_election_timeout_ticks as u32;
for (id, last_heartbeat) in peer_heartbeat {
// for slow and normal peer calculation, we only count voter role
if region
.get_peers()
.iter()
.find(|p| p.get_id() == *id)
.map_or(false, |p| p.role == PeerRole::Voter)
{
// leader itself is not a slow peer
if *id == leader_id || last_heartbeat.elapsed() <= slow_peer_threshold {
normal_peer_count += 1;
} else {
slow_peer_count += 1;
}
}
}

let mut normal_peers_to_remove = vec![];
for cp in change_peers {
let (change_type, peer) = (cp.get_change_type(), cp.get_peer());
if change_type == ConfChangeType::RemoveNode
|| change_type == ConfChangeType::AddLearnerNode
{
let is_voter = region
.get_peers()
.iter()
.find(|p| p.get_id() == peer.get_id())
.map_or(false, |p| p.role == PeerRole::Voter);

// If the change_type is AddLearnerNode and the last heartbeat is found, it
// means it's a demote from voter as AddLearnerNode on existing learner node is
// not allowed.
if is_voter && let Some(last_heartbeat) = peer_heartbeat.get(&peer.get_id()) {
// peer itself is *not* slow peer, but current slow peer is >= total peers/2
if last_heartbeat.elapsed() <= slow_peer_threshold {
normal_peer_count -= 1;
normal_peers_to_remove.push(peer.clone());
}
}
}
}

// only block the conf change when there's chance to improve the availability
// For example, if there's no normal peers actually, then we still allow the
// option to finish as there's no choice.
// We only block the operation when normal peers are going to be removed and it
// could lead to slow peers more than normal peers
if !normal_peers_to_remove.is_empty()
&& slow_peer_count > 0
&& slow_peer_count >= normal_peer_count
{
return Err(box_err!(
"Ignore conf change command on region {} because RemoveNode or Demote a voter on peers {:?} may lead to unavailability. There're {} slow peers and {} normal peers",
region.get_id(),
&normal_peers_to_remove,
slow_peer_count,
normal_peer_count
));
}

Ok(())
}
pub struct MsgType<'a>(pub &'a RaftMessage);

impl Display for MsgType<'_> {
Expand Down Expand Up @@ -2351,103 +2273,4 @@ mod tests {
rrp.update_safe_ts(0, 700);
assert_eq!(pending_items_num(&rrp), 0);
}

#[test]
fn test_check_conf_change_upon_slow_peers() {
// Create a sample configuration
let mut cfg = Config::default();
cfg.raft_max_election_timeout_ticks = 10;
// Initialize change_peers
let change_peers = vec![
ChangePeerRequest {
change_type: eraftpb::ConfChangeType::RemoveNode,
peer: Some(metapb::Peer {
id: 2,
..Default::default()
})
.into(),
..Default::default()
},
ChangePeerRequest {
change_type: eraftpb::ConfChangeType::AddLearnerNode,
peer: Some(metapb::Peer {
id: 2,
..Default::default()
})
.into(),
..Default::default()
},
];

let mut region = Region::default();
for i in 1..4 {
region.mut_peers().push(metapb::Peer {
id: i,
..Default::default()
});
}
for i in 0..change_peers.len() {
// Call the function under test and assert that the function returns failed
let mut cp = vec![change_peers[i].clone()];
let mut peer_heartbeat = collections::HashMap::default();
peer_heartbeat.insert(
1,
std::time::Instant::now() - std::time::Duration::from_secs(1),
);
peer_heartbeat.insert(
2,
std::time::Instant::now() - std::time::Duration::from_secs(1),
);
peer_heartbeat.insert(
3,
std::time::Instant::now() - std::time::Duration::from_secs(1),
);
// Call the function under test and assert that the function returns Ok
check_remove_or_demote_voter(&region, &cfg, &cp, 1, &peer_heartbeat).unwrap();

// now make one peer slow
if let Some(peer_heartbeat) = peer_heartbeat.get_mut(&3) {
*peer_heartbeat = std::time::Instant::now() - std::time::Duration::from_secs(100);
}

// Call the function under test
let result = check_remove_or_demote_voter(&region, &cfg, &cp, 1, &peer_heartbeat);
// Assert that the function returns failed
assert!(result.is_err());

// remove the slow peer instead
cp[0].peer = Some(metapb::Peer {
id: 3,
..Default::default()
})
.into();
// Call the function under test
check_remove_or_demote_voter(&region, &cfg, &cp, 1, &peer_heartbeat).unwrap();

// make peer to learner and remove the peer 2
region.mut_peers()[1].set_role(metapb::PeerRole::Learner);
cp[0].peer = Some(metapb::Peer {
id: 2,
..Default::default()
})
.into();
// Call the function under test
check_remove_or_demote_voter(&region, &cfg, &cp, 1, &peer_heartbeat).unwrap();
// set peer 2 voter again
region.mut_peers()[1].set_role(metapb::PeerRole::Voter);

// there's no remove node, it's fine with slow peers.
cp[0] = ChangePeerRequest {
change_type: eraftpb::ConfChangeType::AddNode,
peer: Some(metapb::Peer {
id: 2,
..Default::default()
})
.into(),
..Default::default()
};
// Call the function under test
check_remove_or_demote_voter(&region, &cfg, &cp, 1, &peer_heartbeat).unwrap();
}
}
}
36 changes: 0 additions & 36 deletions tests/integrations/raftstore/test_conf_change.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -931,39 +931,3 @@ fn test_conf_change_fast() {
must_get_equal(&cluster.get_engine(2), b"k1", b"v1");
assert!(timer.saturating_elapsed() < Duration::from_secs(5));
}

#[test]
fn test_remove_node_on_partition() {
let count = 3;
let mut cluster = new_server_cluster(0, count);
let pd_client = Arc::clone(&cluster.pd_client);
// Disable default max peer number check.
pd_client.disable_default_operator();
cluster.cfg.raft_store.raft_heartbeat_ticks = 1;
cluster.cfg.raft_store.raft_base_tick_interval = ReadableDuration::millis(10);
cluster.cfg.raft_store.raft_election_timeout_ticks = 3;
cluster.cfg.raft_store.raft_store_max_leader_lease = ReadableDuration::millis(20);
let r1 = cluster.run_conf_change();

cluster.must_put(b"k0", b"v0");
pd_client.must_add_peer(r1, new_peer(2, 2));
must_get_equal(&cluster.get_engine(2), b"k0", b"v0");
pd_client.must_add_peer(r1, new_peer(3, 3));
must_get_equal(&cluster.get_engine(3), b"k0", b"v0");

// peer 3 isolation
cluster.add_send_filter(IsolationFilterFactory::new(3));
// sleep for 13 heartbeat interval (>12 should be ok)
let sleep_time = cluster.cfg.raft_store.raft_base_tick_interval.0
* (4 * cluster.cfg.raft_store.raft_election_timeout_ticks as u32 + 1);
thread::sleep(sleep_time);
pd_client.remove_peer(r1, new_peer(2, 2));
cluster.must_put(b"k1", b"v1");
thread::sleep(Duration::from_millis(500));
// remove peer 2 should not work
pd_client.must_have_peer(r1, new_peer(2, 2));

// remove peer 3 should work
pd_client.must_remove_peer(r1, new_peer(3, 3));
cluster.must_put(b"k3", b"v3");
}

0 comments on commit 4653fc6

Please sign in to comment.