mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-22 06:22:38 +00:00
volume: ask the masters for the leader after losing the heartbeat
When the heartbeat stream to the leader failed, currentMaster kept the lost master until a new stream answered, so lookups on a cache miss kept dialing it through the retry delay even with a new leader already elected. On a regular heartbeat error, ask the configured masters for the raft leader and publish it, within the retry delay rather than on top of it. Without an answer the last known leader stays: clearing it would only trade a dead address for an empty one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
e680a78d68
commit
a5925906c8
@@ -54,6 +54,19 @@ func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAd
|
||||
return leader, nil
|
||||
}
|
||||
|
||||
// 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 the last known leader stays.
|
||||
func (vs *VolumeServer) refreshLeaderFromPeers(timeout time.Duration) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
leader, err := vs.lookupRaftLeaderMaster(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
vs.setCurrentMaster(leader)
|
||||
}
|
||||
|
||||
func isContextDoneErr(err error) bool {
|
||||
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
||||
return true
|
||||
@@ -75,7 +88,8 @@ func (vs *VolumeServer) getCurrentMaster() pb.ServerAddress {
|
||||
|
||||
// setCurrentMaster updates vs.currentMaster under a write lock. The
|
||||
// heartbeat goroutine calls this only with a master known to be the leader:
|
||||
// one that answered a heartbeat as leader, or the one a follower named.
|
||||
// one that answered a heartbeat as leader, or the one a follower named, on
|
||||
// the heartbeat stream or, after the stream failed, in its configuration.
|
||||
func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) {
|
||||
vs.currentMasterLock.Lock()
|
||||
vs.currentMaster = master
|
||||
@@ -136,7 +150,11 @@ func (vs *VolumeServer) heartbeat() {
|
||||
} else {
|
||||
// Regular error, reset duplicate retry count
|
||||
duplicateRetryCount = 0
|
||||
time.Sleep(vs.pulsePeriod)
|
||||
// The lookup spends part of the retry delay rather than
|
||||
// adding to it.
|
||||
retryAt := time.Now().Add(vs.pulsePeriod)
|
||||
vs.refreshLeaderFromPeers(vs.pulsePeriod)
|
||||
time.Sleep(time.Until(retryAt))
|
||||
}
|
||||
|
||||
stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc()
|
||||
|
||||
@@ -237,3 +237,45 @@ func TestHeartbeatToLeaderPublishesItAndResendsFromSendLoop(t *testing.T) {
|
||||
t.Fatal("heartbeat did not stop")
|
||||
}
|
||||
}
|
||||
|
||||
// After the heartbeat loses its master, lookups move to the leader the other
|
||||
// masters report instead of dialing the lost one until a new stream answers.
|
||||
func TestRefreshLeaderFromPeersPublishesReportedLeader(t *testing.T) {
|
||||
const leader = pb.ServerAddress("leader.example.com:9333.19333")
|
||||
peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{leader: string(leader)})
|
||||
vs := &VolumeServer{
|
||||
SeedMasterNodes: []pb.ServerAddress{peer},
|
||||
grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
store: &storage.Store{Ip: "127.0.0.1", Port: 8080},
|
||||
}
|
||||
vs.store.MasterFn = vs.getCurrentMaster
|
||||
vs.setCurrentMaster("lost.example.com:9333.19333")
|
||||
|
||||
vs.refreshLeaderFromPeers(5 * time.Second)
|
||||
|
||||
if got := vs.getCurrentMaster(); got != leader {
|
||||
t.Fatalf("currentMaster = %q, want reported leader %q", string(got), string(leader))
|
||||
}
|
||||
if got := vs.store.MasterFn(); got != leader {
|
||||
t.Fatalf("store master = %q, want %q", string(got), string(leader))
|
||||
}
|
||||
}
|
||||
|
||||
// Mid-election no master names a leader; the last known one stays rather
|
||||
// than an address nobody vouched for.
|
||||
func TestRefreshLeaderFromPeersKeepsLastLeaderWithoutAnswer(t *testing.T) {
|
||||
peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{})
|
||||
vs := &VolumeServer{
|
||||
SeedMasterNodes: []pb.ServerAddress{peer},
|
||||
grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
store: &storage.Store{Ip: "127.0.0.1", Port: 8080},
|
||||
}
|
||||
last := pb.ServerAddress("last.example.com:9333.19333")
|
||||
vs.setCurrentMaster(last)
|
||||
|
||||
vs.refreshLeaderFromPeers(5 * time.Second)
|
||||
|
||||
if got := vs.getCurrentMaster(); got != last {
|
||||
t.Fatalf("currentMaster = %q, want unchanged %q", string(got), string(last))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user