Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-11-01 09:50:39 +01:00
Code Issues Packages Projects Releases Wiki Activity

volume: after losing the leader, dial the leader the other masters name (#12259)

Browse Source 
* volume: after losing the leader, dial the leader the other masters name

When the heartbeat stream to the leader breaks and that master stays
unreachable (partitioned or powered off, so dials hang rather than being
refused), the volume server could take close to a minute to register with
the new leader, and the new leader answered assigns with "no volume server
capacity registered yet" in the meantime.

Two things in the retry path added up:

- refreshLeaderFromPeers asked the seed masters in order with up to 5s
  each, inside a budget of one pulse (5s by default). With the lost
  master listed first, the whole budget went to it and the live peers
  that already knew the new leader were never asked.
- Even when the refresh found the new leader, the loop moved on to the
  next seed master. When the lost leader had been reached through a
  redirect, the next seed can be that same master, so the loop dialed it
  again and waited out the 20s gRPC connect timeout.

Ask the lost master last, and when the refresh names a leader other than
the lost master, dial it next.

Assisted-by: Claude <noreply@anthropic.com>

* rust volume: dial the leader the other masters name after losing one

Mirror the Go heartbeat change: refresh_leader_from_peers asks the lost
master last and returns the leader it found, and the loop dials that
leader next instead of a seed that can be the unreachable master itself.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* rust volume: let a named leader consume the next seed slot

Restarting the seed scan on every redirect let an earlier seed that keeps
reporting the dead leader starve every seed after it: the loop bounced
between the redirect target and the next seed forever. Go consumes the
next range value for the redirect instead, so a pass always reaches every
seed. Drop the breaks so new_leader overrides the next slot the same way.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: try a named leader on top of the seed scan

A reported leader used to replace the next seed slot: with seeds A,B
and A persistently naming a dead leader X, X consumed B's slot on
every pass so B was never dialed. The hint is now a one-shot attempt
made before the seed it would have displaced. Its own result stays
pending when a seed attempt names nothing, so a live follower's
redirect chain is followed while every seed is still dialed on every
pass. Mirrors between the Go and Rust volume servers.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: follow a fresh leader hint before the pending seed

A named leader whose stream breaks can leave the peer refresh naming its
replacement; the loop then dialed the pending seed first and paid a dead
master's connect timeout ahead of the fresh name. Fresh hints are now
followed right away, bounded by the seed count and stopping once a name
repeats, so a peer that keeps naming the same dead leader still cannot
starve the seed scan. Go and Rust mirrored.

* volume: dial each named leader once per pass (Go + Rust)

A name that stays pending after the hint chain — the cap was hit, or the
refresh named one already dialed — was retried before every later seed in
the pass: each seed reset the hint budget, so a dead leader still cost a
sleep and a connect timeout per seed. Track the names dialed in a pass,
seeds included, and skip a hint that repeats one.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: let a seed's named leader replace a pending hint (Go + Rust)

The seed dial runs after the hint chain, so the leader it names is the
freshest answer the pass has — but it was dropped whenever a stale hint
was still pending, and the next seed then chased the older name first.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: keep an untried leader hint over the seed's stale answer (Go + Rust)

A seed that keeps naming a stale leader spends the per-pass hint budget
on it; its own answer then overwrote the live name the hint chain had
just reached, so that leader was never dialed. A pending hint now
survives the seed dial until it has been tried.

---------

Co-authored-by: user.email <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Youngju KIMuser.emailDevin
 authored and GitHub committed 2026-10-31 04:10:36 -07:00
1 parent e98adfd3bb
commit bed0d7401d
3 files changed
+519 -153
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

seaweed-volume/src/server/heartbeat.rs
+214 -104
View File
@@ -3,7 +3,7 @@
//! Implements the bidirectional streaming `SendHeartbeat` RPC to the master,
//! matching Go's `server/volume_grpc_client_to_master.go`.
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;
@@ -13,7 +13,7 @@ use tokio::sync::broadcast;
use tracing::{error, info, warn};
use super::grpc_client::{GrpcDialOptions, connect_channel, master_client};
use super::volume_server::VolumeServerState;
use super::volume_server::{VolumeServerState, to_http_address};
use crate::pb::master_pb;
use crate::pb::volume_server_pb;
use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig};
@@ -70,6 +70,7 @@ pub async fn run_heartbeat_with_state(
let mut confirmed_leader_once = false;
loop {
let mut tried_hints: HashSet<String> = HashSet::new();
for master_addr in &config.master_addresses {
if is_stopping(&state) {
state.is_heartbeating.store(false, Ordering::Relaxed);
@@ -82,112 +83,72 @@ pub async fn run_heartbeat_with_state(
return;
}
// If we have a leader redirect, sleep 3s then connect to the leader
// instead of iterating through the seed list
let target_addr = if let Some(ref leader) = new_leader {
// Named leaders are tried on top of the seed scan rather than in
// the next seed's place: a peer that keeps naming a dead leader
// would otherwise eat that seed's slot on every pass and starve
// it. A fresh name is followed right away — the pending seed may
// be the master the stream just lost — and a name already dialed
// this pass is not dialed again. A hint naming the upcoming seed
// is covered by the seed dial itself. Mirrors Go's heartbeat loop.
let mut hints = 0;
while let Some(leader) = new_leader.clone() {
let leader_http = to_http_address(&leader).into_owned();
if leader_http == to_http_address(master_addr).as_ref()
|| tried_hints.contains(&leader_http)
|| hints >= config.master_addresses.len()
{
break;
}
tried_hints.insert(leader_http);
hints += 1;
// the new leader may actually is the same master
// need to wait a bit before adding itself
tokio::time::sleep(Duration::from_secs(3)).await;
leader.clone()
} else {
master_addr.clone()
};
let grpc_addr = to_grpc_address(&target_addr);
info!("Connecting heartbeat to master {}", grpc_addr);
// Determine what action to take after the heartbeat attempt.
// We convert the error to a string immediately so the non-Send
// Box<dyn Error> is dropped before any .await point.
enum PostAction {
LeaderRedirect(String),
Done,
SleepDuplicate(Duration),
SleepPulse,
new_leader = heartbeat_once(
&config,
&state,
&leader,
pulse,
&mut shutdown_rx,
&mut confirmed_leader_once,
&mut duplicate_retry_count,
)
.await;
if is_stopping(&state) {
break;
}
}
let action = match do_heartbeat(
if is_stopping(&state) {
break;
}
if let Some(ref leader) = new_leader
&& to_http_address(leader) == to_http_address(master_addr)
{
// the seed dial covers a hint that names the seed itself
new_leader = None;
}
tried_hints.insert(to_http_address(master_addr).into_owned());
if let Some(named) = heartbeat_once(
&config,
&state,
&grpc_addr,
&target_addr,
master_addr,
pulse,
&mut shutdown_rx,
&mut confirmed_leader_once,
&mut duplicate_retry_count,
)
.await
{
Ok(Some(leader)) => {
info!("Master leader changed to {}", leader);
PostAction::LeaderRedirect(leader)
}
Ok(None) => {
duplicate_retry_count = 0;
PostAction::Done
}
Err(e) => {
let err_msg = e.to_string();
// Drop `e` (non-Send) before any .await
drop(e);
warn!("Heartbeat to {} error: {}", grpc_addr, err_msg);
if err_msg.contains(DUPLICATE_UUID_RETRY_MESSAGE) {
if duplicate_retry_count >= MAX_DUPLICATE_UUID_RETRIES {
error!(
"Shut down Volume Server due to persistent duplicate volume directories after 3 retries"
);
error!(
"Please check if another volume server is using the same directory"
);
std::process::exit(1);
}
let retry_delay = duplicate_uuid_retry_delay(duplicate_retry_count);
duplicate_retry_count += 1;
warn!(
"Waiting {:?} before retrying due to duplicate UUID detection (attempt {}/3)...",
retry_delay, duplicate_retry_count
);
PostAction::SleepDuplicate(retry_delay)
} else {
duplicate_retry_count = 0;
PostAction::SleepPulse
}
}
};
match action {
PostAction::LeaderRedirect(leader) => {
new_leader = Some(leader);
break;
}
PostAction::Done => {
new_leader = None;
}
PostAction::SleepDuplicate(delay) => {
new_leader = None;
tokio::time::sleep(delay).await;
}
PostAction::SleepPulse => {
new_leader = None;
// The lookup spends part of the retry delay rather than
// adding to it.
let retry_at = tokio::time::Instant::now() + pulse;
refresh_leader_from_peers(&config, &state, pulse).await;
let mut wait = retry_at.saturating_duration_since(tokio::time::Instant::now());
if wait > Duration::from_secs(1)
&& (!state.current_master().await.is_empty() || !confirmed_leader_once)
{
// A leader just named by the refresh usually accepts
// the next stream right away, and an unregistered
// server has nothing else to do, so reconnect
// promptly. Once the server was registered and no
// leader is known, keep the pulse backoff.
wait = Duration::from_secs(1);
}
tokio::time::sleep(wait).await;
// A hint the cap left untried stays pending: the seed's
// answer replaces it only once it was dialed.
let tried = new_leader
.as_ref()
.is_none_or(|p| tried_hints.contains(to_http_address(p).as_ref()));
if tried {
new_leader = Some(named);
}
}
// If we connected to a leader (not seed list), break out after one attempt
// so we either reconnect to the new leader or fall back to seed list
if new_leader.is_some() {
if is_stopping(&state) {
break;
}
}
@@ -198,6 +159,110 @@ pub async fn run_heartbeat_with_state(
}
}
/// One heartbeat attempt on `target_addr`; returns the leader the stream or
/// the peer refresh named — `None` when neither did. Mirrors Go's
/// `heartbeatOnce`.
async fn heartbeat_once(
config: &HeartbeatConfig,
state: &Arc<VolumeServerState>,
target_addr: &str,
pulse: Duration,
shutdown_rx: &mut broadcast::Receiver<()>,
confirmed_leader_once: &mut bool,
duplicate_retry_count: &mut u32,
) -> Option<String> {
let grpc_addr = to_grpc_address(target_addr);
info!("Connecting heartbeat to master {}", grpc_addr);
// Determine what action to take after the heartbeat attempt.
// We convert the error to a string immediately so the non-Send
// Box<dyn Error> is dropped before any .await point.
enum PostAction {
SleepDuplicate(Duration),
SleepPulse,
}
let action = match do_heartbeat(
config,
state,
&grpc_addr,
target_addr,
pulse,
shutdown_rx,
confirmed_leader_once,
)
.await
{
Ok(Some(leader)) => {
info!("Master leader changed to {}", leader);
return Some(leader);
}
Ok(None) => {
*duplicate_retry_count = 0;
return None;
}
Err(e) => {
let err_msg = e.to_string();
// Drop `e` (non-Send) before any .await
drop(e);
warn!("Heartbeat to {} error: {}", grpc_addr, err_msg);
if err_msg.contains(DUPLICATE_UUID_RETRY_MESSAGE) {
if *duplicate_retry_count >= MAX_DUPLICATE_UUID_RETRIES {
error!(
"Shut down Volume Server due to persistent duplicate volume directories after 3 retries"
);
error!("Please check if another volume server is using the same directory");
std::process::exit(1);
}
let retry_delay = duplicate_uuid_retry_delay(*duplicate_retry_count);
*duplicate_retry_count += 1;
warn!(
"Waiting {:?} before retrying due to duplicate UUID detection (attempt {}/3)...",
retry_delay, duplicate_retry_count
);
PostAction::SleepDuplicate(retry_delay)
} else {
*duplicate_retry_count = 0;
PostAction::SleepPulse
}
}
};
match action {
PostAction::SleepDuplicate(delay) => {
tokio::time::sleep(delay).await;
}
PostAction::SleepPulse => {
// The lookup spends part of the retry delay rather than adding
// to it.
let retry_at = tokio::time::Instant::now() + pulse;
let leader = refresh_leader_from_peers(config, state, target_addr, pulse).await;
let mut wait = retry_at.saturating_duration_since(tokio::time::Instant::now());
if wait > Duration::from_secs(1)
&& (!state.current_master().await.is_empty() || !*confirmed_leader_once)
{
// A leader just named by the refresh usually accepts the
// next stream right away, and an unregistered server has
// nothing else to do, so reconnect promptly. Once the
// server was registered and no leader is known, keep the
// pulse backoff.
wait = Duration::from_secs(1);
}
tokio::time::sleep(wait).await;
if let Some(leader) = leader
&& to_http_address(&leader) != to_http_address(target_addr)
{
// Dial the leader the other masters just named rather than
// the next seed, which can be the master this stream just
// lost: an unreachable master holds each dial for the
// connect timeout before the loop moves on.
return Some(leader);
}
}
}
None
}
/// Convert a master address to a gRPC `host:port` target.
///
/// Mirrors Go's `pb.ServerToGrpcAddress()`:
@@ -274,22 +339,47 @@ pub async fn try_get_master_configuration(
/// to the master the heartbeat just lost while the loop retries. Without an
/// answer inside `timeout` the published master is cleared: the lost master
/// is dead or no longer leader, and a follower answers lookups from its own
/// cache rather than refusing them. Mirrors Go's refreshLeaderFromPeers.
/// cache rather than refusing them, and returns it so the loop can dial it
/// next. The lost master is asked last: an unreachable one holds its lookup
/// open long enough to starve the peers that know the new leader. Mirrors
/// Go's refreshLeaderFromPeers.
async fn refresh_leader_from_peers(
config: &HeartbeatConfig,
state: &Arc<VolumeServerState>,
lost: &str,
timeout: Duration,
) {
let lookup = lookup_leader_from_peers(
&config.master_addresses,
state.outgoing_grpc_tls.as_ref(),
);
) -> Option<String> {
let ordered = masters_asking_last(&config.master_addresses, lost);
let lookup = lookup_leader_from_peers(&ordered, state.outgoing_grpc_tls.as_ref());
match tokio::time::timeout(timeout, lookup).await {
Ok(Some(leader)) => publish_current_master(state, &leader).await,
_ => state.current_master_url.write().await.clear(),
Ok(Some(leader)) => {
publish_current_master(state, &leader).await;
Some(leader)
}
_ => {
state.current_master_url.write().await.clear();
None
}
}
}
/// Return masters with every entry equal to `lost` (same HTTP address) moved
/// to the end, keeping the order of the rest. Mirrors Go's mastersAskingLast.
fn masters_asking_last(masters: &[String], lost: &str) -> Vec<String> {
let lost_http = to_http_address(lost);
let mut ordered = Vec::with_capacity(masters.len());
let mut tail = Vec::new();
for m in masters {
if !lost.is_empty() && to_http_address(m) == lost_http {
tail.push(m.clone());
} else {
ordered.push(m.clone());
}
}
ordered.extend(tail);
ordered
}
/// Ask each seed master for the raft leader and return the first answer, or
/// None when none replies with one. Mirrors Go's
/// `operation.LookupRaftLeaderMaster`.
@@ -1619,6 +1709,26 @@ mod tests {
assert_eq!(to_grpc_address("host:not-a-port"), "host:not-a-port");
}
// Mirrors Go's TestMastersAskingLast: the just-lost master is asked last
// so an unreachable one cannot spend the whole refresh budget.
#[test]
fn test_masters_asking_last() {
let seeds = vec![
"a:9333".to_string(),
"b:9333".to_string(),
"c:9333".to_string(),
];
assert_eq!(
masters_asking_last(&seeds, "a:9333.19333"),
vec![
"b:9333".to_string(),
"c:9333".to_string(),
"a:9333".to_string()
]
);
assert_eq!(masters_asking_last(&seeds[..2], ""), seeds[..2]);
}
#[test]
fn test_build_heartbeat_includes_store_identity_and_disk_metadata() {
let temp_dir = tempfile::tempdir().unwrap();
weed/server/volume_grpc_client_to_master.go
+110 -43
View File
@@ -35,10 +35,14 @@ func (vs *VolumeServer) GetMaster(ctx context.Context) pb.ServerAddress {
// the heartbeat loop owns that field and reconnects when the stream reports a
// different leader than the connected peer.
func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAddress, error) {
return vs.lookupRaftLeaderMasterAmong(ctx, vs.SeedMasterNodes)
}
func (vs *VolumeServer) lookupRaftLeaderMasterAmong(ctx context.Context, masters []pb.ServerAddress) (pb.ServerAddress, error) {
if ctx == nil {
ctx = context.Background()
}
leader, err := operation.LookupRaftLeaderMaster(ctx, vs.SeedMasterNodes, vs.grpcDialOption)
leader, err := operation.LookupRaftLeaderMaster(ctx, masters, vs.grpcDialOption)
if err != nil {
if isContextDoneErr(err) {
glog.V(1).Infof("volume server %s:%d: raft leader lookup: %v", vs.store.Ip, vs.store.Port, err)
@@ -56,18 +60,40 @@ func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAd
// refreshLeaderFromPeers publishes the leader the configured masters report,
// so lookups stop going to a master the heartbeat just lost while the loop
// retries. Without an answer within timeout no leader is published: the lost
// master is dead or no longer leader, and a follower answers lookups from its
// own cache rather than refusing them, so failing fast is the safer answer.
func (vs *VolumeServer) refreshLeaderFromPeers(timeout time.Duration) {
// retries, and returns it so the loop can dial it next. Without an answer
// within timeout no leader is published: the lost master is dead or no longer
// leader, and a follower answers lookups from its own cache rather than
// refusing them, so failing fast is the safer answer.
//
// The lost master is asked last. Each peer gets up to 5s, so when the lost
// master is unreachable rather than refusing connections (a partition or a
// powered-off host) asking it first spends the whole timeout and the live
// peers that know the new leader are never asked.
func (vs *VolumeServer) refreshLeaderFromPeers(lost pb.ServerAddress, timeout time.Duration) pb.ServerAddress {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
leader, err := vs.lookupRaftLeaderMaster(ctx)
leader, err := vs.lookupRaftLeaderMasterAmong(ctx, mastersAskingLast(vs.SeedMasterNodes, lost))
if err != nil {
vs.setCurrentMaster("")
return
return ""
}
vs.setCurrentMaster(leader)
return leader
}
// mastersAskingLast returns masters with every entry equal to last moved to
// the end, keeping the order of the rest.
func mastersAskingLast(masters []pb.ServerAddress, last pb.ServerAddress) []pb.ServerAddress {
ordered := make([]pb.ServerAddress, 0, len(masters))
var tail []pb.ServerAddress
for _, m := range masters {
if last != "" && m.Equals(last) {
tail = append(tail, m)
continue
}
ordered = append(ordered, m)
}
return append(ordered, tail...)
}
func isContextDoneErr(err error) bool {
@@ -135,51 +161,42 @@ func (vs *VolumeServer) heartbeat() {
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.volume")
var err error
var newLeader pb.ServerAddress
duplicateRetryCount := 0
for vs.isHeartbeating {
triedHints := make(map[string]bool)
for _, master := range vs.SeedMasterNodes {
if newLeader != "" {
// Named leaders are tried on top of the seed scan rather than in
// the next seed's place: a peer that keeps naming a dead leader
// would otherwise eat that seed's slot on every pass and starve
// it. A fresh name is followed right away — the pending seed may
// be the master the stream just lost — and a name already dialed
// this pass is not dialed again.
for hints := 0; newLeader != "" && !newLeader.Equals(master) && !triedHints[newLeader.ToHttpAddress()] && hints < len(vs.SeedMasterNodes); hints++ {
// the new leader may actually is the same master
// need to wait a bit before adding itself
time.Sleep(3 * time.Second)
master = newLeader
}
newLeader, err = vs.doHeartbeatWithRetry(master, grpcDialOption, vs.pulsePeriod, duplicateRetryCount)
if err != nil {
glog.V(0).Infof("heartbeat to %s error: %v", master, err)
// Check if this is a duplicate UUID retry error
if strings.Contains(err.Error(), "duplicate UUIDs detected, retrying connection") {
duplicateRetryCount++
retryDelay := time.Duration(1<<(duplicateRetryCount-1)) * 2 * time.Second // exponential backoff: 2s, 4s, 8s
glog.V(0).Infof("Waiting %v before retrying due to duplicate UUID detection...", retryDelay)
time.Sleep(retryDelay)
} else {
// Regular error, reset duplicate retry count
duplicateRetryCount = 0
// The lookup spends part of the retry delay rather than
// adding to it.
retryAt := time.Now().Add(vs.pulsePeriod)
vs.refreshLeaderFromPeers(vs.pulsePeriod)
wait := time.Until(retryAt)
if wait > time.Second && (vs.getCurrentMaster() != "" || !vs.confirmedLeaderOnce.Load()) {
// A leader just named by the refresh usually accepts the
// next stream right away, and an unregistered server has
// nothing else to do, so reconnect promptly. Once the
// server was registered and no leader is known, keep the
// pulse backoff.
wait = time.Second
}
time.Sleep(wait)
triedHints[newLeader.ToHttpAddress()] = true
newLeader, duplicateRetryCount = vs.heartbeatOnce(newLeader, grpcDialOption, duplicateRetryCount)
if !vs.isHeartbeating {
break
}
stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc()
}
if !vs.isHeartbeating {
break
}
if newLeader.Equals(master) {
// the seed dial covers a hint that names the seed itself
newLeader = ""
} else {
// Successful connection, reset retry count
duplicateRetryCount = 0
}
triedHints[master.ToHttpAddress()] = true
named, retryCount := vs.heartbeatOnce(master, grpcDialOption, duplicateRetryCount)
duplicateRetryCount = retryCount
// A hint the cap left untried stays pending: the seed's answer
// replaces it only once it was dialed, so a stale name cannot
// keep a live leader from ever being tried.
if named != "" && (newLeader == "" || triedHints[newLeader.ToHttpAddress()]) {
newLeader = named
}
if !vs.isHeartbeating {
break
@@ -188,6 +205,56 @@ func (vs *VolumeServer) heartbeat() {
}
}
// heartbeatOnce makes one heartbeat attempt on target and returns the leader
// the stream or the peer refresh named — "" when neither did — with the
// running duplicate-UUID retry count.
func (vs *VolumeServer) heartbeatOnce(target pb.ServerAddress, grpcDialOption grpc.DialOption, duplicateRetryCount int) (named pb.ServerAddress, retryCount int) {
newLeader, err := vs.doHeartbeatWithRetry(target, grpcDialOption, vs.pulsePeriod, duplicateRetryCount)
if err == nil {
// Successful connection, reset retry count
return newLeader, 0
}
glog.V(0).Infof("heartbeat to %s error: %v", target, err)
var leader pb.ServerAddress
retryCount = duplicateRetryCount
// Check if this is a duplicate UUID retry error
if strings.Contains(err.Error(), "duplicate UUIDs detected, retrying connection") {
retryCount++
retryDelay := time.Duration(1<<(retryCount-1)) * 2 * time.Second // exponential backoff: 2s, 4s, 8s
glog.V(0).Infof("Waiting %v before retrying due to duplicate UUID detection...", retryDelay)
time.Sleep(retryDelay)
} else {
// Regular error, reset duplicate retry count
retryCount = 0
// The lookup spends part of the retry delay rather than
// adding to it.
retryAt := time.Now().Add(vs.pulsePeriod)
leader = vs.refreshLeaderFromPeers(target, vs.pulsePeriod)
wait := time.Until(retryAt)
if wait > time.Second && (vs.getCurrentMaster() != "" || !vs.confirmedLeaderOnce.Load()) {
// A leader just named by the refresh usually accepts the
// next stream right away, and an unregistered server has
// nothing else to do, so reconnect promptly. Once the
// server was registered and no leader is known, keep the
// pulse backoff.
wait = time.Second
}
time.Sleep(wait)
}
stats.VolumeServerMasterDisconnections.WithLabelValues(target.String()).Inc()
if leader != "" && !leader.Equals(target) {
// Dial the leader the other masters just named rather than
// the next seed, which can be the master this stream just
// lost: an unreachable master holds each dial for the gRPC
// connect timeout before the loop moves on.
return leader, retryCount
}
return "", retryCount
}
func (vs *VolumeServer) StopHeartbeat() (isAlreadyStopping bool) {
if !vs.isHeartbeating {
return true
weed/server/volume_grpc_client_to_master_test.go
+195 -6
View File
@@ -63,8 +63,15 @@ func startFakeMasterServerForLeaderLookup(t *testing.T, srv master_pb.SeaweedSer
go func() { _ = grpcServer.Serve(lis) }()
t.Cleanup(grpcServer.GracefulStop)
_, port, _ := net.SplitHostPort(lis.Addr().String())
return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port))
return fakeMasterAddress(lis.Addr())
}
// fakeMasterAddress names a fake master by its listener port. The port doubles
// as the HTTP port so that ServerAddress.Equals, which compares HTTP
// addresses, tells two fake masters apart.
func fakeMasterAddress(addr net.Addr) pb.ServerAddress {
_, port, _ := net.SplitHostPort(addr.String())
return pb.ServerAddress(fmt.Sprintf("127.0.0.1:%s.%s", port, port))
}
// fakeHeartbeatMaster answers SendHeartbeat the way a master does: a follower
@@ -311,8 +318,9 @@ func TestRefreshLeaderFromPeersPublishesReportedLeader(t *testing.T) {
vs.store.MasterFn = vs.getCurrentMaster
vs.setCurrentMaster("lost.example.com:9333.19333")
vs.refreshLeaderFromPeers(5 * time.Second)
if got := vs.refreshLeaderFromPeers("lost.example.com:9333.19333", 5*time.Second); got != leader {
t.Fatalf("refresh returned %q, want reported leader %q", string(got), string(leader))
}
if got := vs.getCurrentMaster(); got != leader {
t.Fatalf("currentMaster = %q, want reported leader %q", string(got), string(leader))
}
@@ -333,8 +341,9 @@ func TestRefreshLeaderFromPeersClearsLeaderWithoutAnswer(t *testing.T) {
vs.store.MasterFn = vs.getCurrentMaster
vs.setCurrentMaster("lost.example.com:9333.19333")
vs.refreshLeaderFromPeers(5 * time.Second)
if got := vs.refreshLeaderFromPeers("lost.example.com:9333.19333", 5*time.Second); got != "" {
t.Fatalf("refresh returned %q, want none", string(got))
}
if got := vs.getCurrentMaster(); got != "" {
t.Fatalf("currentMaster = %q, want none", string(got))
}
@@ -342,3 +351,183 @@ func TestRefreshLeaderFromPeersClearsLeaderWithoutAnswer(t *testing.T) {
t.Fatalf("store master = %q, want none", string(got))
}
}
// startUnresponsiveMaster listens without ever answering, the way a master
// behind a partition or on a powered-off host looks to a dialer: no refusal,
// just a connection that never becomes usable.
func startUnresponsiveMaster(t *testing.T) pb.ServerAddress {
t.Helper()
lis, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
t.Cleanup(func() { _ = lis.Close() })
return fakeMasterAddress(lis.Addr())
}
// An unreachable lost master listed first must not use up the refresh before
// a live peer is asked who leads now.
func TestRefreshLeaderFromPeersAsksLostMasterLast(t *testing.T) {
const leader = pb.ServerAddress("leader.example.com:9333.19333")
lost := startUnresponsiveMaster(t)
peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{leader: string(leader)})
vs := &VolumeServer{
SeedMasterNodes: []pb.ServerAddress{lost, peer},
grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
store: &storage.Store{Ip: "127.0.0.1", Port: 8080},
}
vs.setCurrentMaster(lost)
if got := vs.refreshLeaderFromPeers(lost, 2*time.Second); got != leader {
t.Fatalf("refresh returned %q, want %q from the live peer", string(got), string(leader))
}
if got := vs.getCurrentMaster(); got != leader {
t.Fatalf("currentMaster = %q, want %q", string(got), string(leader))
}
}
func TestMastersAskingLast(t *testing.T) {
a, b, c := pb.ServerAddress("a:9333"), pb.ServerAddress("b:9333"), pb.ServerAddress("c:9333")
got := mastersAskingLast([]pb.ServerAddress{a, b, c}, "a:9333.19333")
want := []pb.ServerAddress{b, c, a}
if fmt.Sprint(got) != fmt.Sprint(want) {
t.Fatalf("got %v, want %v", got, want)
}
if got := mastersAskingLast([]pb.ServerAddress{a, b}, ""); fmt.Sprint(got) != fmt.Sprint([]pb.ServerAddress{a, b}) {
t.Fatalf("no lost master: got %v", got)
}
}
// switchableFollower names whichever master it currently believes leads, on
// heartbeat streams and in its configuration alike.
type switchableFollower struct {
master_pb.UnimplementedSeaweedServer
mu sync.Mutex
leader pb.ServerAddress
}
func (f *switchableFollower) setLeader(leader pb.ServerAddress) {
f.mu.Lock()
f.leader = leader
f.mu.Unlock()
}
func (f *switchableFollower) currentLeader() string {
f.mu.Lock()
defer f.mu.Unlock()
return string(f.leader)
}
func (f *switchableFollower) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
for {
if _, err := stream.Recv(); err != nil {
return err
}
if err := stream.Send(&master_pb.HeartbeatResponse{Leader: f.currentLeader()}); err != nil {
return err
}
}
}
func (f *switchableFollower) GetMasterConfiguration(_ context.Context, _ *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) {
return &master_pb.GetMasterConfigurationResponse{Leader: f.currentLeader()}, nil
}
// When the leader's stream breaks and the leader stays unreachable, the next
// stream goes to the leader the other masters name, not to the next seed. The
// seeds are ordered so that the next seed is the lost leader itself, which is
// how a volume server that reached the old leader through a redirect finds
// its seed list after the failure.
func TestHeartbeatAfterLosingLeaderDialsTheNewLeader(t *testing.T) {
oldLeader := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)}
oldLis, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
oldServer := grpc.NewServer()
master_pb.RegisterSeaweedServer(oldServer, oldLeader)
go func() { _ = oldServer.Serve(oldLis) }()
t.Cleanup(oldServer.Stop)
oldAddr := fakeMasterAddress(oldLis.Addr())
newLeader := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)}
newAddr := startFakeMasterServerForLeaderLookup(t, newLeader)
follower := &switchableFollower{leader: oldAddr}
followerAddr := startFakeMasterServerForLeaderLookup(t, follower)
vs := newHeartbeatTestVolumeServer(t)
vs.SeedMasterNodes = []pb.ServerAddress{followerAddr, newAddr, oldAddr}
vs.pulsePeriod = time.Second
exited := make(chan struct{})
go func() {
defer close(exited)
vs.heartbeat()
}()
t.Cleanup(func() {
vs.StopHeartbeat()
select {
case <-exited:
case <-time.After(10 * time.Second):
t.Error("heartbeat loop did not stop")
}
})
select {
case <-oldLeader.heartbeats:
case <-time.After(10 * time.Second):
t.Fatal("old leader got no heartbeat")
}
// The old leader becomes unreachable: its streams break, and its port
// accepts connections that never answer.
follower.setLeader(newAddr)
oldServer.Stop()
blackHole, err := net.Listen("tcp", oldLis.Addr().String())
if err != nil {
t.Fatalf("reopen %s: %v", oldLis.Addr(), err)
}
t.Cleanup(func() { _ = blackHole.Close() })
// A dial to the unreachable old leader alone would take the gRPC connect
// timeout (20s).
select {
case <-newLeader.heartbeats:
case <-time.After(10 * time.Second):
t.Fatalf("new leader got no heartbeat within 10s of losing the old one; currentMaster = %q", vs.getCurrentMaster())
}
}
// With a single seed the hint budget is one try per pass. A seed stuck on a
// stale name spends it every pass, so its answer must not drop the live name
// that answer-chain reached before it was tried.
func TestHeartbeatKeepsUntriedHintOverStaleSeedAnswer(t *testing.T) {
live := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)}
liveAddr := startFakeMasterServerForLeaderLookup(t, live)
redirects := &fakeHeartbeatMaster{leader: string(liveAddr)}
redirectAddr := startFakeMasterServerForLeaderLookup(t, redirects)
stale := &fakeHeartbeatMaster{leader: string(redirectAddr)}
staleAddr := startFakeMasterServerForLeaderLookup(t, stale)
vs := newHeartbeatTestVolumeServer(t)
vs.SeedMasterNodes = []pb.ServerAddress{staleAddr}
vs.pulsePeriod = time.Second
exited := make(chan struct{})
go func() {
defer close(exited)
vs.heartbeat()
}()
t.Cleanup(func() {
vs.StopHeartbeat()
select {
case <-exited:
case <-time.After(10 * time.Second):
t.Error("heartbeat loop did not stop")
}
})
select {
case <-live.heartbeats:
case <-time.After(15 * time.Second):
t.Fatalf("live leader got no heartbeat; currentMaster = %q", vs.getCurrentMaster())
}
}