diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 1da5e908a8..92ae99182e 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -53,6 +53,12 @@ fn needle_not_found(needle_id: NeedleId) -> Status { Status::not_found(format!("needle not found {}", needle_id.0)) } +/// Mirrors Go's `isNotLeaderErr`: the master returns raft's NotLeaderError, +/// which arrives as an Unknown-status message. +fn is_not_leader_status(e: &Status) -> bool { + e.message().contains("Not current leader") +} + /// Deadline from the caller's `grpc-timeout` header (TimeoutValue + /// H|M|S|m|u|n unit, per the gRPC-over-HTTP2 spec). Tonic's GrpcTimeout layer /// drops an over-deadline handler future anyway; parsing it here lets a queued @@ -344,22 +350,46 @@ impl VolumeGrpcService { is_readonly: bool, ) -> Result<(), Status> { // VolumeMarkReadonly mutates raft-replicated master topology, so it must - // reach the leader. Prefer the live leader the heartbeat is talking to - // (current_master_url), falling back to the static seed before the first - // heartbeat — mirrors store_ec.rs and Go's vs.GetMaster(). Sending it to - // the static seed fails "not current leader" after any master failover. - let master_url = { - let live = self.state.current_master_url.read().await.clone(); - if !live.is_empty() { - live - } else { - self.state.master_url.clone() - } - }; + // reach the leader the heartbeat confirmed. On "Not current leader", + // ask the seed masters who leads and retry there — mirrors Go's + // notifyMasterVolumeReadonly. + let master_url = self.state.current_master().await; if master_url.is_empty() { - return Ok(()); + if !self.state.has_master { + return Ok(()); + } + return Err(Status::unavailable("no master leader known")); } - let grpc_addr = parse_grpc_address(&master_url).map_err(|e| { + match self + .volume_mark_readonly_on_master(&master_url, info, is_readonly) + .await + { + Err(e) if is_not_leader_status(&e) => { + let leader = super::heartbeat::lookup_leader_from_peers( + &self.state.master_urls, + self.state.outgoing_grpc_tls.as_ref(), + ) + .await + .ok_or_else(|| { + Status::unavailable(format!( + "master {} rejected mark-readonly and no peer named a leader", + master_url + )) + })?; + self.volume_mark_readonly_on_master(&leader, info, is_readonly) + .await + } + result => result, + } + } + + async fn volume_mark_readonly_on_master( + &self, + master_url: &str, + info: &MasterVolumeInfo, + is_readonly: bool, + ) -> Result<(), Status> { + let grpc_addr = parse_grpc_address(master_url).map_err(|e| { Status::internal(format!("invalid master address {}: {}", master_url, e)) })?; let channel = connect_channel( @@ -5776,7 +5806,8 @@ impl VolumeCopyJob { (&self.dest.data_base_name, self.dest.has_remote_dat); let mut preallocate_size: i64 = 0; if !has_remote_dat { - let grpc_addr = super::heartbeat::to_grpc_address(&state.master_url); + let master = state.current_master().await; + let grpc_addr = super::heartbeat::to_grpc_address(&master); // Race the master-configuration RPC against the caller's // response channel: a stalled master (or a slow leader // election) would otherwise hold the task and its .note @@ -5801,7 +5832,7 @@ impl VolumeCopyJob { } } Err(e) => { - tracing::warn!("get master {} configuration: {}", state.master_url, e); + tracing::warn!("get master {} configuration: {}", master, e); } } diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 737fdfafbc..4eb390f314 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -603,7 +603,7 @@ async fn batch_delete_file_ids( let locations = lookup_volume( &state.http_client, &state.outgoing_http_scheme, - &state.master_url, + &state.current_master().await, volume_id, ) .await @@ -675,7 +675,7 @@ async fn do_replicated_request( let locations = lookup_volume( &state.http_client, &state.outgoing_http_scheme, - &state.master_url, + &state.current_master().await, vid, ) .await @@ -832,7 +832,7 @@ async fn proxy_or_redirect_to_target( let locations = match lookup_volume( &state.http_client, &state.outgoing_http_scheme, - &state.master_url, + &state.current_master().await, vid.0, ) .await @@ -1238,7 +1238,7 @@ async fn proxy_missing_volume( let query_string = uri.query().unwrap_or("").to_string(); if is_proxied_query(&query_string) || state.read_mode == ReadMode::Local - || state.master_url.is_empty() + || state.current_master().await.is_empty() { return StatusCode::NOT_FOUND.into_response(); } @@ -1282,7 +1282,7 @@ async fn check_download_limit( // Go tries proxy to replica once before the wait loop // (checkDownloadLimit); it does not retry on each wakeup. let should_try_replica = - !is_proxied_query(&query_string) && !state.master_url.is_empty() && { + !is_proxied_query(&query_string) && !state.current_master().await.is_empty() && { let store = state.store.read().unwrap(); store .find_volume(vid) @@ -2960,7 +2960,9 @@ pub async fn post_handler( // Replicate to remote volume servers if this volume has replicas. // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1. - if !is_replicate && write_result.is_ok() && !state.master_url.is_empty() { + // No leader-known gate: when the lookup cannot run, the write must fail + // rather than be acknowledged on fewer copies than required. + if !is_replicate && write_result.is_ok() { let needs_replication = { let store = state.store.read().unwrap(); store @@ -3358,7 +3360,7 @@ pub async fn delete_handler( }; let is_replicate = del_query.split('&').any(|p| p == "type=replicate"); - if !is_replicate && delete_result.is_ok() && !state.master_url.is_empty() { + if !is_replicate && delete_result.is_ok() { let needs_replication = { let store = state.store.read().unwrap(); store @@ -4232,7 +4234,7 @@ async fn read_remote_chunk_needle( let locations = lookup_volume( &state.http_client, &state.outgoing_http_scheme, - &state.master_url, + &state.current_master().await, vid.0, ) .await?; @@ -6671,7 +6673,8 @@ mod tests { let s = Arc::get_mut(&mut state).unwrap(); // Empty self_url makes `url.contains(self_url)` true for every peer. s.self_url = "198.51.100.8:8080".to_string(); - s.master_url = master_url; + s.master_url = master_url.clone(); + s.current_master_url = tokio::sync::RwLock::new(master_url); state } diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 081ff71a25..9d56440a75 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -155,7 +155,11 @@ pub async fn run_heartbeat_with_state( } PostAction::SleepPulse => { new_leader = None; - tokio::time::sleep(pulse).await; + // 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; + tokio::time::sleep_until(retry_at).await; } } @@ -254,10 +258,56 @@ pub async fn try_get_master_configuration( Ok(resp.into_inner()) } +/// Publish the leader the configured masters report, so lookups stop going +/// 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. +async fn refresh_leader_from_peers( + config: &HeartbeatConfig, + state: &Arc, + timeout: Duration, +) { + let lookup = lookup_leader_from_peers( + &config.master_addresses, + 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(), + } +} + +/// 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`. +pub async fn lookup_leader_from_peers( + master_addrs: &[String], + tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>, +) -> Option { + for master_addr in master_addrs { + let grpc_addr = to_grpc_address(master_addr); + if let Ok(resp) = try_get_master_configuration(&grpc_addr, tls).await + && !resp.leader.is_empty() + { + return Some(resp.leader); + } + } + None +} + fn is_stopping(state: &VolumeServerState) -> bool { *state.is_stopping.read().unwrap() } +/// Store the master lookups should go to, in the address form it was +/// announced in — `host:port` or `host:port.grpcPort` — so an explicit gRPC +/// port survives to the gRPC callers, mirroring Go's setCurrentMaster. +/// The heartbeat calls this only for a master known to lead. +async fn publish_current_master(state: &Arc, master: &str) { + *state.current_master_url.write().await = master.to_string(); +} + fn duplicate_uuid_retry_delay(retry_count: u32) -> Duration { Duration::from_secs((1u64 << retry_count) * 2) } @@ -430,14 +480,6 @@ async fn do_heartbeat( let mut response_stream = client.send_heartbeat(stream).await?.into_inner(); info!("Heartbeat stream established with {}", grpc_addr); - // Publish the master we're now talking to in canonical http host:port - // form so Ping admission can recognise it once a leader change moves us - // off the seed list. Mirrors Go's vs.setCurrentMaster(masterAddress). - { - let normalised = super::volume_server::to_http_address(current_master).into_owned(); - let mut guard = state.current_master_url.write().await; - *guard = normalised; - } if is_stopping(state) { state.is_heartbeating.store(false, Ordering::Relaxed); send_deregister_heartbeat(config, state, &tx).await; @@ -451,13 +493,31 @@ async fn do_heartbeat( volume_tick.tick().await; ec_tick.tick().await; + let mut confirmed_leader = false; loop { tokio::select! { resp = response_stream.message() => { match resp { Ok(Some(hb_resp)) => { - // Match Go ordering: DuplicatedUuids first, then volume - // options, then leader redirect. + // A reply names the leader or comes from it, so it + // settles where lookups go before anything else on the + // reply is handled: a follower reply carries no other + // fields, and a leader stays the lookup target even + // when it rejects this registration. Mirrors Go. + if !hb_resp.leader.is_empty() { + if super::volume_server::to_http_address(current_master) + != super::volume_server::to_http_address(&hb_resp.leader) + { + publish_current_master(state, &hb_resp.leader).await; + return Ok(Some(hb_resp.leader)); + } + } else if !confirmed_leader { + // The leader answers every new stream without + // naming a leader, so this peer is the one lookups + // go to. + confirmed_leader = true; + publish_current_master(state, current_master).await; + } if !hb_resp.duplicated_uuids.is_empty() { let uuids = hb_resp.duplicated_uuids.clone(); let duplicate_dirs = off_runtime(config, state, move |_, state| { @@ -502,11 +562,6 @@ async fn do_heartbeat( if metrics_changed { state.metrics_notify.notify_waiters(); } - // Match Go: only redirect if leader is non-empty AND - // different from the current master we're connected to. - if !hb_resp.leader.is_empty() && current_master != hb_resp.leader { - return Ok(Some(hb_resp.leader)); - } } Ok(None) => return Ok(None), Err(e) => return Err(Box::new(e)), diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 35b76ca4d7..bf973d7fa5 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -1139,16 +1139,9 @@ async fn cached_lookup_ec_shard_locations( state: &Arc, vid: VolumeId, ) -> io::Result>> { - let master = { - let live = state.current_master_url.read().await.clone(); - if !live.is_empty() { - live - } else { - state.master_url.clone() - } - }; + let master = state.current_master().await; if master.is_empty() { - return Err(io::Error::other("no master configured for ec shard lookup")); + return Err(io::Error::other("no master leader known for ec shard lookup")); } let grpc_addr = diff --git a/seaweed-volume/src/server/volume_server.rs b/seaweed-volume/src/server/volume_server.rs index a440c2a9e1..eab2e3617a 100644 --- a/seaweed-volume/src/server/volume_server.rs +++ b/seaweed-volume/src/server/volume_server.rs @@ -86,7 +86,8 @@ pub struct VolumeServerState { /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation, /// allowing arbitrary (incl. loopback / link-local / metadata) hosts. pub allow_untrusted_remote_endpoints: bool, - /// First master address for volume lookups (e.g., "localhost:9333"). + /// First configured master address (e.g., "localhost:9333"). A seed, not a + /// lookup target: lookups go to `current_master_url`. pub master_url: String, /// Seed master addresses for UI rendering. pub master_urls: Vec, @@ -94,10 +95,12 @@ pub struct VolumeServerState { /// Built once at construction so Ping admission stays O(1). Mirrors /// Go's `seedMasterSet` on `VolumeServer`. pub seed_master_set: std::collections::HashSet, - /// Current master this server is heartbeating with, in canonical http - /// `host:port` form. Empty when no heartbeat connection is active. The - /// heartbeat goroutine writes; admission reads — the lock keeps them - /// from racing on a leader change. Mirrors Go's `currentMaster` plus + /// Master this server takes as leader, in its announced + /// `host:port[.grpcPort]` form — callers pick http or gRPC form per use. + /// Empty until the heartbeat confirms one — a peer that answered a + /// stream without naming a leader, or the leader a follower named — and + /// cleared again when no master reports one. The heartbeat writes; + /// admission and lookups read. Mirrors Go's `currentMaster` plus /// `currentMasterLock`. pub current_master_url: tokio::sync::RwLock, /// This server's own address (ip:port) for filtering self from lookup results. @@ -140,6 +143,13 @@ pub struct VolumeServerState { } impl VolumeServerState { + /// The master lookups should go to: the leader the heartbeat confirmed, + /// or empty while none is known, in its announced `host:port[.grpcPort]` + /// form. Mirrors Go's `vs.getCurrentMaster()`. + pub async fn current_master(&self) -> String { + self.current_master_url.read().await.clone() + } + /// Check if the server is in maintenance mode; return gRPC error if so. pub fn check_maintenance(&self) -> Result<(), tonic::Status> { if self.maintenance.load(Ordering::Relaxed) { @@ -209,7 +219,7 @@ impl VolumeServerState { return false; } let current = self.current_master_url.read().await.clone(); - if !current.is_empty() && current == key { + if !current.is_empty() && to_http_address(¤t) == key { return true; } self.seed_master_set.contains(&key) diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index 5c17a3b0ca..639bb5ef23 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -54,6 +54,22 @@ 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 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) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + leader, err := vs.lookupRaftLeaderMaster(ctx) + if err != nil { + vs.setCurrentMaster("") + return + } + vs.setCurrentMaster(leader) +} + func isContextDoneErr(err error) bool { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return true @@ -74,7 +90,10 @@ func (vs *VolumeServer) getCurrentMaster() pb.ServerAddress { } // setCurrentMaster updates vs.currentMaster under a write lock. The -// heartbeat goroutine calls this whenever it (re)connects to a master. +// 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, on +// the heartbeat stream or, after the stream failed, in its configuration. +// After a failed stream with no leader named, it is cleared to "". func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) { vs.currentMasterLock.Lock() vs.currentMaster = master @@ -122,7 +141,6 @@ func (vs *VolumeServer) heartbeat() { time.Sleep(3 * time.Second) master = newLeader } - vs.store.MasterAddress = master newLeader, err = vs.doHeartbeatWithRetry(master, grpcDialOption, vs.pulsePeriod, duplicateRetryCount) if err != nil { glog.V(0).Infof("heartbeat to %s error: %v", master, err) @@ -136,12 +154,15 @@ 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() newLeader = "" - vs.store.MasterAddress = "" } else { // Successful connection, reset retry count duplicateRetryCount = 0 @@ -180,17 +201,42 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp return "", err } glog.V(0).Infof("Heartbeat to: %v", masterAddress) - vs.setCurrentMaster(masterAddress) doneChan := make(chan error, 1) + // The receiver asks for a fresh heartbeat here rather than sending it, + // since a gRPC stream allows only one sender at a time. + resendChan := make(chan struct{}, 1) go func() { + confirmedLeader := false for { in, err := stream.Recv() if err != nil { doneChan <- err return } + // A reply names the leader or comes from it, so it settles where + // lookups go before anything else on the reply is handled: a + // follower reply carries no other fields, and a leader stays the + // lookup target even when it rejects this registration. + if in.GetLeader() != "" { + // Only a follower names the leader. Publish the named leader right + // away, so lookups stop landing on this follower while the loop + // reconnects. + leader := pb.ServerAddress(in.GetLeader()) + if !masterAddress.Equals(leader) { + glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", leader, masterAddress) + vs.setCurrentMaster(leader) + newLeader = leader + doneChan <- nil + return + } + } else if !confirmedLeader { + // The leader answers every new heartbeat stream without naming a + // leader, so this peer is the one lookups should go to. + confirmedLeader = true + vs.setCurrentMaster(masterAddress) + } if len(in.DuplicatedUuids) > 0 { var duplicateDir []string for _, loc := range vs.store.Locations { @@ -231,9 +277,9 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp } if volumeOptsChanged { if vs.store.MaybeAdjustVolumeMax() { - if err = stream.Send(vs.store.CollectHeartbeat()); err != nil { - glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", vs.getCurrentMaster(), err) - return + select { + case resendChan <- struct{}{}: + default: } } } @@ -244,15 +290,6 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp glog.V(0).Infof("master %s asked for the full volume list", masterAddress) vs.store.RequestFullVolumeList() } - if in.GetLeader() != "" { - current := vs.getCurrentMaster() - if !current.Equals(pb.ServerAddress(in.GetLeader())) { - glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), current) - newLeader = pb.ServerAddress(in.GetLeader()) - doneChan <- nil - return - } - } } }() @@ -359,6 +396,11 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err) return "", err } + case <-resendChan: + if err = stream.Send(vs.store.CollectHeartbeat()); err != nil { + glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err) + return "", err + } case <-volumeTickChan.C: glog.V(4).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port) vs.store.MaybeAdjustVolumeMax() diff --git a/weed/server/volume_grpc_client_to_master_test.go b/weed/server/volume_grpc_client_to_master_test.go index 90d1ba18f4..b235e66ca5 100644 --- a/weed/server/volume_grpc_client_to_master_test.go +++ b/weed/server/volume_grpc_client_to_master_test.go @@ -4,11 +4,17 @@ import ( "context" "fmt" "net" + "strings" + "sync" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) @@ -60,3 +66,279 @@ func startFakeMasterServerForLeaderLookup(t *testing.T, srv master_pb.SeaweedSer _, port, _ := net.SplitHostPort(lis.Addr().String()) return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port)) } + +// fakeHeartbeatMaster answers SendHeartbeat the way a master does: a follower +// names the leader on every heartbeat, the leader answers the first one with +// its volume options and never names itself. +type fakeHeartbeatMaster struct { + master_pb.UnimplementedSeaweedServer + leader string // set on a follower + release chan struct{} // a follower waits on it before answering + heartbeats chan *master_pb.Heartbeat +} + +func (m *fakeHeartbeatMaster) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error { + answered := false + for { + hb, err := stream.Recv() + if err != nil { + return err + } + if m.heartbeats != nil { + m.heartbeats <- hb + } + if m.leader != "" { + if m.release != nil { + <-m.release + } + if err := stream.Send(&master_pb.HeartbeatResponse{Leader: m.leader}); err != nil { + return err + } + continue + } + if !answered { + answered = true + if err := stream.Send(&master_pb.HeartbeatResponse{VolumeSizeLimit: 1 << 30}); err != nil { + return err + } + } + } +} + +func newHeartbeatTestVolumeServer(t *testing.T) *VolumeServer { + t.Helper() + store := storage.NewStore(nil, "127.0.0.1", 8080, 18080, "http://127.0.0.1:8080", "store-id", + []string{t.TempDir()}, + []int32{0}, + []util.MinFreeSpace{{}}, + "", + storage.NeedleMapInMemory, + []types.DiskType{types.HardDriveType}, + nil, + 3, + stats.DefaultDiskIOProbeConfig(), + ) + t.Cleanup(store.Close) + vs := &VolumeServer{ + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: store, + stopChan: make(chan bool), + isHeartbeating: true, + } + store.MasterFn = vs.getCurrentMaster + return vs +} + +func waitForCurrentMaster(t *testing.T, vs *VolumeServer, want pb.ServerAddress) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !vs.getCurrentMaster().Equals(want) { + if time.Now().After(deadline) { + t.Fatalf("currentMaster = %q, want %q", vs.getCurrentMaster(), want) + } + time.Sleep(10 * time.Millisecond) + } +} + +// heartbeatRun is one doHeartbeatWithRetry call running in the background. +type heartbeatRun struct { + done chan heartbeatResult // receives the call's result once + exited chan struct{} + stop func() +} + +type heartbeatResult struct { + newLeader pb.ServerAddress + err error +} + +// startHeartbeat runs doHeartbeatWithRetry against master until it returns or +// is stopped. Its cleanup stops the heartbeat and waits for it, and runs before +// the cleanups registered earlier, so a failed assertion cannot leave the stream +// open for the fake master's GracefulStop to wait on. +func startHeartbeat(t *testing.T, vs *VolumeServer, master pb.ServerAddress) *heartbeatRun { + t.Helper() + var stopOnce sync.Once + run := &heartbeatRun{ + done: make(chan heartbeatResult, 1), + exited: make(chan struct{}), + stop: func() { stopOnce.Do(func() { close(vs.stopChan) }) }, + } + go func() { + defer close(run.exited) + newLeader, err := vs.doHeartbeatWithRetry(master, vs.grpcDialOption, time.Hour, 0) + run.done <- heartbeatResult{newLeader, err} + }() + t.Cleanup(func() { + run.stop() + select { + case <-run.exited: + case <-time.After(5 * time.Second): + t.Error("heartbeat did not stop") + } + }) + return run +} + +// A follower must never become the master lookups go to: until it names the +// leader the previous one stays, and then the named leader takes over before +// the heartbeat loop has reconnected. +func TestHeartbeatToFollowerPublishesNamedLeaderOnly(t *testing.T) { + const leader = pb.ServerAddress("leader.example.com:9333.19333") + follower := &fakeHeartbeatMaster{ + leader: string(leader), + release: make(chan struct{}), + heartbeats: make(chan *master_pb.Heartbeat, 8), + } + followerAddr := startFakeMasterServerForLeaderLookup(t, follower) + // cleanups run last-in first-out, so a failed assertion still lets the + // follower answer before its server stops gracefully + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(follower.release) }) } + t.Cleanup(release) + + vs := newHeartbeatTestVolumeServer(t) + previous := pb.ServerAddress("previous.example.com:9333.19333") + vs.setCurrentMaster(previous) + run := startHeartbeat(t, vs, followerAddr) + + select { + case <-follower.heartbeats: + case <-time.After(5 * time.Second): + t.Fatal("follower got no heartbeat") + } + if got := vs.getCurrentMaster(); got != previous { + t.Fatalf("currentMaster = %q while connected to follower %q, want unchanged %q", string(got), string(followerAddr), string(previous)) + } + if got := vs.store.MasterFn(); got != previous { + t.Fatalf("store master = %q, want %q", got, previous) + } + release() + + select { + case r := <-run.done: + if r.err != nil { + t.Fatalf("heartbeat: %v", r.err) + } + if r.newLeader != leader { + t.Fatalf("newLeader = %q, want %q", r.newLeader, leader) + } + case <-time.After(5 * time.Second): + t.Fatal("heartbeat did not return after the follower named the leader") + } + if got := vs.getCurrentMaster(); got != leader { + t.Fatalf("currentMaster = %q, want named leader %q", got, leader) + } +} + +// The leader's first answer confirms it, and the size limit that answer carries +// makes the receiver ask for a fresh heartbeat, which the send loop delivers. +func TestHeartbeatToLeaderPublishesItAndResendsFromSendLoop(t *testing.T) { + leader := &fakeHeartbeatMaster{heartbeats: make(chan *master_pb.Heartbeat, 8)} + leaderAddr := startFakeMasterServerForLeaderLookup(t, leader) + + vs := newHeartbeatTestVolumeServer(t) + run := startHeartbeat(t, vs, leaderAddr) + + waitForCurrentMaster(t, vs, leaderAddr) + if got := vs.store.MasterFn(); got != leaderAddr { + t.Fatalf("store master = %q, want %q", got, leaderAddr) + } + + // the initial volume and ec heartbeats, then the resend + for i := 0; i < 3; i++ { + select { + case <-leader.heartbeats: + case <-time.After(5 * time.Second): + t.Fatalf("leader got %d heartbeats, want 3", i) + } + } + + run.stop() + select { + case r := <-run.done: + if r.err != nil { + t.Fatalf("heartbeat: %v", r.err) + } + case <-time.After(5 * time.Second): + t.Fatal("heartbeat did not stop") + } +} + +// duplicateUUIDMaster answers every heartbeat the way a leader holding a stale +// registration does: a duplicate-uuid response and nothing else. +type duplicateUUIDMaster struct { + master_pb.UnimplementedSeaweedServer +} + +func (m *duplicateUUIDMaster) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error { + for { + if _, err := stream.Recv(); err != nil { + return err + } + if err := stream.Send(&master_pb.HeartbeatResponse{DuplicatedUuids: []string{"uuid-1"}}); err != nil { + return err + } + } +} + +// A leader that rejects the registration over duplicate uuids still answers +// lookups, so it is published before the retry error goes out. +func TestHeartbeatPublishesLeaderBeforeDuplicateUuidError(t *testing.T) { + masterAddr := startFakeMasterServerForLeaderLookup(t, &duplicateUUIDMaster{}) + vs := newHeartbeatTestVolumeServer(t) + vs.setCurrentMaster("previous.example.com:9333.19333") + + _, err := vs.doHeartbeatWithRetry(masterAddr, vs.grpcDialOption, time.Hour, 0) + if err == nil || !strings.Contains(err.Error(), "duplicate UUIDs") { + t.Fatalf("err = %v, want duplicate UUIDs", err) + } + if got := vs.getCurrentMaster(); got != masterAddr { + t.Fatalf("currentMaster = %q, want the answering leader %q", got, masterAddr) + } +} + +// 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 lost master is dead or now a +// follower that would answer lookups from its own cache, so none is published. +func TestRefreshLeaderFromPeersClearsLeaderWithoutAnswer(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}, + } + vs.store.MasterFn = vs.getCurrentMaster + vs.setCurrentMaster("lost.example.com:9333.19333") + + vs.refreshLeaderFromPeers(5 * time.Second) + + if got := vs.getCurrentMaster(); got != "" { + t.Fatalf("currentMaster = %q, want none", string(got)) + } + if got := vs.store.MasterFn(); got != "" { + t.Fatalf("store master = %q, want none", string(got)) + } +} diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index a544d44513..ff1c85d044 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -148,6 +148,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, vs.checkWithMaster() vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout, diskProbeConfig) + vs.store.MasterFn = vs.getCurrentMaster vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec) handleStaticResources(adminMux) diff --git a/weed/storage/store.go b/weed/storage/store.go index b4035444ae..1f1e880430 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -61,14 +61,25 @@ type ReadOption struct { ReadBufferSize int } +// masterAddress returns the master to ask for volume and shard locations, or +// "" while no leader is known. +func (s *Store) masterAddress() pb.ServerAddress { + if s.MasterFn == nil { + return "" + } + return s.MasterFn() +} + /* * A VolumeServer contains one Store */ type Store struct { - MasterAddress pb.ServerAddress + // MasterFn reports the master the volume server currently takes as leader. + // The heartbeat loop owns that answer; the store only reads it. + MasterFn func() pb.ServerAddress grpcDialOption grpc.DialOption - volumeSizeLimit uint64 // read from the master - preallocate atomic.Bool // read from the master + volumeSizeLimit atomic.Uint64 // read from the master + preallocate atomic.Bool // read from the master Ip string Port int GrpcPort int @@ -1302,11 +1313,11 @@ func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error { } func (s *Store) SetVolumeSizeLimit(x uint64) { - atomic.StoreUint64(&s.volumeSizeLimit, x) + s.volumeSizeLimit.Store(x) } func (s *Store) GetVolumeSizeLimit() uint64 { - return atomic.LoadUint64(&s.volumeSizeLimit) + return s.volumeSizeLimit.Load() } func (s *Store) SetPreallocate(x bool) { diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 396a1619a7..b66592f2bd 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -568,7 +568,7 @@ const ecIntervalReadConcurrency = 8 func (s *Store) readEcShardIntervals(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, is_deleted bool, err error) { if err = s.cachedLookupEcShardLocations(ecVolume); err != nil { - return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err) + return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err) } var totalSize int @@ -721,7 +721,7 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) glog.V(3).Infof("lookup and cache ec volume %d locations", ecVolume.VolumeId) - err = operation.WithMasterServerClient(context.Background(), false, s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + err = operation.WithMasterServerClient(context.Background(), false, s.masterAddress(), s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { req := &master_pb.LookupEcVolumeRequest{ VolumeId: uint32(ecVolume.VolumeId), } diff --git a/weed/storage/store_ec_location_cache_test.go b/weed/storage/store_ec_location_cache_test.go index c2bb1b5c54..60affb146c 100644 --- a/weed/storage/store_ec_location_cache_test.go +++ b/weed/storage/store_ec_location_cache_test.go @@ -78,7 +78,7 @@ func rewindShardLocationsRefresh(ecVolume *erasure_coding.EcVolume, d time.Durat func TestCachedLookupEcShardLocationsRefreshesAfterAFailedRead(t *testing.T) { master, masterAddr := startCountingMaster(t) store := &Store{ - MasterAddress: masterAddr, + MasterFn: func() pb.ServerAddress { return masterAddr }, grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), } ecVolume := &erasure_coding.EcVolume{ @@ -127,7 +127,7 @@ func TestCachedLookupEcShardLocationsRefreshesAfterAFailedRead(t *testing.T) { func TestCachedLookupEcShardLocationsSettlesAfterRefresh(t *testing.T) { master, masterAddr := startCountingMaster(t) store := &Store{ - MasterAddress: masterAddr, + MasterFn: func() pb.ServerAddress { return masterAddr }, grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), } ecVolume := &erasure_coding.EcVolume{ @@ -257,7 +257,7 @@ func TestReadOneEcShardIntervalMarksTheMapAfterADirectReadFails(t *testing.T) { func TestCachedLookupEcShardLocationsKeepsAMarkRaisedDuringTheLookup(t *testing.T) { master, masterAddr := startCountingMaster(t) store := &Store{ - MasterAddress: masterAddr, + MasterFn: func() pb.ServerAddress { return masterAddr }, grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), } ecVolume := &erasure_coding.EcVolume{ @@ -295,7 +295,7 @@ func TestCachedLookupEcShardLocationsKeepsAMarkRaisedDuringTheLookup(t *testing. func TestCachedLookupEcShardLocationsKeepsTheMarkWhenTheLookupFails(t *testing.T) { master, masterAddr := startCountingMaster(t) store := &Store{ - MasterAddress: masterAddr, + MasterFn: func() pb.ServerAddress { return masterAddr }, grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), } ecVolume := &erasure_coding.EcVolume{ diff --git a/weed/storage/store_ec_scrub.go b/weed/storage/store_ec_scrub.go index ce236cf3f9..25f2f88427 100644 --- a/weed/storage/store_ec_scrub.go +++ b/weed/storage/store_ec_scrub.go @@ -23,7 +23,7 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId, mode volume_server_pb.VolumeS return 0, nil, []error{fmt.Errorf("EC volume id %d not found", vid)} } if err := s.cachedLookupEcShardLocations(ecv); err != nil { - return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)} + return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)} } return s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck) } @@ -56,7 +56,7 @@ func (s *Store) ScrubEcVolumeMerged(merged *erasure_coding.MergedEcRuntimes, mod } } if err := s.cachedLookupEcShardLocations(ecv); err != nil { - return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)} + return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err)} } files, shardInfos, errs := s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck)