diff --git a/weed/cluster/master_client.go b/weed/cluster/master_client.go index ba0ed4f3d..70a391712 100644 --- a/weed/cluster/master_client.go +++ b/weed/cluster/master_client.go @@ -10,10 +10,13 @@ import ( "google.golang.org/grpc" ) -func ListExistingPeerUpdates(master pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string, clientType string) (existingNodes []*master_pb.ClusterNodeUpdate) { +// ListExistingPeerUpdates lists the cluster nodes one master currently tracks. +// An empty result is only a successful empty answer; a failed RPC returns the +// error so callers do not mistake "unknown" for "absent". +func ListExistingPeerUpdates(ctx context.Context, master pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string, clientType string) (existingNodes []*master_pb.ClusterNodeUpdate, err error) { - if grpcErr := pb.WithMasterClient(context.Background(), false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error { - resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ + err = pb.WithMasterClient(ctx, false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error { + resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ ClientType: clientType, FilerGroup: filerGroup, }) @@ -31,8 +34,24 @@ func ListExistingPeerUpdates(master pb.ServerAddress, grpcDialOption grpc.DialOp }) } return nil - }); grpcErr != nil { - glog.V(0).Infof("connect to %s: %v", master, grpcErr) + }) + if err != nil { + glog.V(0).Infof("connect to %s: %v", master, err) } return } + +// LookupClusterLeader asks one master for the raft leader it currently sees. +// Any reachable master answers, including followers; it returns "" while the +// cluster is mid-election. +func LookupClusterLeader(ctx context.Context, master pb.ServerAddress, grpcDialOption grpc.DialOption) (leader pb.ServerAddress, err error) { + err = pb.WithMasterClient(ctx, false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error { + resp, err := client.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{}) + if err != nil { + return err + } + leader = pb.ServerAddress(resp.Leader) + return nil + }) + return +} diff --git a/weed/command/filer.go b/weed/command/filer.go index 317e960b8..03d2f11be 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -382,8 +382,10 @@ func (fo *FilerOptions) startFiler() { } } + announce := make(chan struct{}) fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{ Masters: fo.masters, + AnnounceCh: announce, FilerGroup: *fo.filerGroup, Collection: *fo.collection, DefaultReplication: *fo.defaultReplicaPlacement, @@ -500,6 +502,13 @@ func (fo *FilerOptions) startFiler() { go grpcS.Serve(grpcL) pb.ServeGrpcOnLocalSocket(grpcS, grpcPort) + // Membership in the master's filer list routes S3 and peer traffic here; + // announce once the gRPC port is serving and the metadata replay has + // caught up with the peers found at startup, so the filer does not take + // lock-ring ownership while a peer's older events can still land on top + // of its fresh writes. A cap bounds the wait. + go fs.AnnounceOnceCaughtUp(announce, time.Now()) + // Helper to gracefully stop the gRPC server, waiting for active RPCs. gracefulTimeout := fo.gracefulStopTimeout if gracefulTimeout <= 0 { diff --git a/weed/filer/filer.go b/weed/filer/filer.go index bee16acdf..204659372 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -254,8 +254,96 @@ func (f *Filer) AggregateFromPeers(self pb.ServerAddress, existingNodes []*maste } -func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*master_pb.ClusterNodeUpdate) { - return cluster.ListExistingPeerUpdates(f.GetMaster(ctx), f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType) +// peerLookupTimeout bounds each master query so one unresponsive master +// cannot stall the peer scan. +const peerLookupTimeout = 10 * time.Second + +// leaderSettleTimeout bounds how long an empty peer list from the leader is +// re-polled: a freshly elected leader has not yet received the filers' +// reconnects, so an immediate empty answer can just mean membership has not +// migrated yet. A persistent empty answer is then checked against the other +// masters below before it is accepted as an empty cluster. +const leaderSettleTimeout = 10 * time.Second +const leaderSettleInterval = time.Second + +// ListExistingPeerUpdates returns the peer list to bootstrap from. The +// leader's list is authoritative: after a leader change an old master keeps +// its registrations until its streams end, so the longest answer is not +// necessarily the current one. The leader's answer wins whenever it can be +// found and queried; only when the leader is unreachable does the scan fall +// back to the fullest answer from the other masters. It returns an error when +// no master answers, so a failed lookup is not mistaken for an empty cluster. +func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*master_pb.ClusterNodeUpdate, err error) { + masters := f.MasterClient.ListMasters() + var queryErrs []error + + var leader pb.ServerAddress + for _, master := range masters { + if ctx.Err() != nil { + break + } + lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout) + l, lookErr := cluster.LookupClusterLeader(lookupCtx, master, f.GrpcDialOption) + cancel() + if lookErr != nil { + queryErrs = append(queryErrs, lookErr) + continue + } + if l != "" { + leader = l + break + } + } + if leader != "" { + settleDeadline := time.Now().Add(leaderSettleTimeout) + for { + lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout) + nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, leader, f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType) + cancel() + if listErr != nil { + queryErrs = append(queryErrs, fmt.Errorf("list peers from leader %s: %w", leader, listErr)) + break + } + if len(nodes) > 0 { + return nodes, nil + } + if ctx.Err() != nil || !time.Now().Before(settleDeadline) { + break + } + select { + case <-ctx.Done(): + case <-time.After(leaderSettleInterval): + } + } + } + + answered := false + for _, master := range masters { + if ctx.Err() != nil { + break + } + lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout) + nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, master, f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType) + cancel() + if listErr != nil { + queryErrs = append(queryErrs, listErr) + continue + } + answered = true + if len(nodes) > len(existingNodes) { + existingNodes = nodes + } + } + if !answered { + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + if err = errors.Join(queryErrs...); err == nil { + err = fmt.Errorf("no masters configured") + } + return nil, fmt.Errorf("list existing peers: %w", err) + } + return existingNodes, nil } func (f *Filer) SetStore(store FilerStore) (isFresh bool) { diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index f993d68dc..0f8e16d1b 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -273,6 +273,21 @@ func lowWatermarkOf(watermarks map[pb.ServerAddress]int64) int64 { return low } +// PeersCaughtUpTo reports whether every listed peer's stream has delivered +// an event or idle heartbeat at or past tsNs, i.e. no pre-tsNs backlog can +// still arrive from it. A peer no longer tracked counts as caught up: it +// cannot deliver a backlog anymore. +func (ma *MetaAggregator) PeersCaughtUpTo(peers []pb.ServerAddress, tsNs int64) bool { + ma.peerWatermarksLock.Lock() + defer ma.peerWatermarksLock.Unlock() + for _, peer := range peers { + if ts, found := ma.peerWatermarks[peer]; found && ts < tsNs { + return false + } + } + return true +} + func (ma *MetaAggregator) HasRemotePeers() bool { ma.peerChansLock.Lock() defer ma.peerChansLock.Unlock() diff --git a/weed/mq/broker/broker_server.go b/weed/mq/broker/broker_server.go index a3e658c35..e9b367617 100644 --- a/weed/mq/broker/broker_server.go +++ b/weed/mq/broker/broker_server.go @@ -125,7 +125,10 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial ) glog.V(0).Info("Started idle partition cleanup task (check: 1m, timeout: 5m)") - existingNodes := cluster.ListExistingPeerUpdates(mqBroker.MasterClient.GetMaster(context.Background()), grpcDialOption, option.FilerGroup, cluster.FilerType) + existingNodes, listErr := cluster.ListExistingPeerUpdates(context.Background(), mqBroker.MasterClient.GetMaster(context.Background()), grpcDialOption, option.FilerGroup, cluster.FilerType) + if listErr != nil { + glog.Warningf("broker failed to list existing peers: %v", listErr) + } for _, newNode := range existingNodes { mqBroker.OnBrokerUpdate(newNode, time.Now()) } diff --git a/weed/pb/server_discovery.go b/weed/pb/server_discovery.go index 59a333f0a..b869c5100 100644 --- a/weed/pb/server_discovery.go +++ b/weed/pb/server_discovery.go @@ -24,8 +24,19 @@ func NewServiceDiscoveryFromMap(m map[string]ServerAddress) (sd *ServerDiscovery // RefreshBySrvIfAvailable performs a DNS SRV lookup and updates list with the results // of the lookup func (sd *ServerDiscovery) RefreshBySrvIfAvailable() { + if newList := sd.LookupSrvInstances(); newList != nil { + sd.SetInstances(newList) + } +} + +// LookupSrvInstances resolves the SRV record without touching the stored +// list. It returns nil when there is no SRV record, the lookup fails, or the +// result has no well-formed names. Running it without holding a lock keeps a +// slow resolver from blocking GetInstances callers; callers serialize the +// SetInstances that saves the result. +func (sd *ServerDiscovery) LookupSrvInstances() []ServerAddress { if sd.srvRecord == nil { - return + return nil } newList, err := sd.srvRecord.LookUp() if err != nil { @@ -33,8 +44,14 @@ func (sd *ServerDiscovery) RefreshBySrvIfAvailable() { } if newList == nil || len(newList) == 0 { glog.V(0).Infof("looked up SRV for %s, but found no well-formed names", *sd.srvRecord) - return + return nil } + return newList +} + +// SetInstances saves a resolved address list. Callers racing a refresh must +// serialize this with their GetInstances reads. +func (sd *ServerDiscovery) SetInstances(newList []ServerAddress) { if !reflect.DeepEqual(sd.list, newList) { sd.list = newList } diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index a823bfd5b..e4871f914 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -91,6 +91,10 @@ type FilerOption struct { TusSessionExpiry time.Duration S3ConfigFile string // optional path to static S3 identity config file CredentialManager *credential.CredentialManager + // AnnounceCh, when set, must be closed before the filer registers on the + // master; it joins the filer list only once the gRPC port is serving and + // the metadata replay has caught up with the peers found at startup. + AnnounceCh chan struct{} // AllowUntrustedRemoteEndpoints lets a read of a remote-only entry dial a // mounted endpoint that resolves to a loopback / private / metadata host. AllowUntrustedRemoteEndpoints bool @@ -155,6 +159,16 @@ type FilerServer struct { ringPeerIPs atomic.Pointer[ringPeerIPs] ringResolveGroup singleflight.Group + // masterCtx drives KeepConnectedToMaster; cancelling it ends the stream + // so the master drops this filer from its list before gRPC stops. + masterCtx context.Context + masterCancel context.CancelFunc + + // startupPeers are the filers discovered during NewFilerServer, before + // this filer announced itself; AnnounceOnceCaughtUp waits for their + // replayed backlogs before the filer joins the master's list. + startupPeers []pb.ServerAddress + // entryLockTable serializes mutations to the same entry path on this filer. // CreateEntry takes it today; UpdateEntry and DeleteEntry are intended to take // it too as their callers route a key's writes to this node, making it the @@ -272,7 +286,11 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) fs.checkWithMaster() go stats.LoopPushingMetric("filer", string(fs.option.Host), fs.metricsAddress, fs.metricsIntervalSec) - go fs.filer.MasterClient.KeepConnectedToMaster(context.Background()) + if option.AnnounceCh != nil { + fs.filer.MasterClient.SetAnnounceCh(option.AnnounceCh) + } + fs.masterCtx, fs.masterCancel = context.WithCancel(context.Background()) + go fs.filer.MasterClient.KeepConnectedToMaster(fs.masterCtx) fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete") v.SetDefault("filer.options.buckets_folder", "/buckets") @@ -321,7 +339,20 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) readonlyMux.HandleFunc("/", fs.filerGuard.WhiteList(requestIDMiddleware(fs.readonlyFilerHandler))) } - existingNodes := fs.filer.ListExistingPeerUpdates(context.Background()) + // A failed peer scan is not an empty cluster: a fresh filer that skipped + // bootstrap here would serve without the existing files, so retry until a + // master answers. This runs before the announce gate opens, so the filer + // is not yet advertised. + var existingNodes []*master_pb.ClusterNodeUpdate + for { + var listErr error + existingNodes, listErr = fs.filer.ListExistingPeerUpdates(context.Background()) + if listErr == nil { + break + } + glog.Warningf("%s cannot list existing peers: %v; retrying", option.Host, listErr) + time.Sleep(2 * time.Second) + } startFromTime := time.Now().Add(-filer.LogFlushInterval) if isFresh { glog.V(0).Infof("%s bootstrap from peers %+v", option.Host, existingNodes) @@ -335,6 +366,15 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) } fs.filer.AggregateFromPeers(option.Host, existingNodes, startFromTime) + // A peer replaying its backlog applies the old owner's events over this + // filer's fresh writes if it takes lock-ring ownership before the replay + // finishes, so the announce gate waits for these peers to catch up. + for _, node := range existingNodes { + if addr := pb.ServerAddress(node.Address); !addr.Equals(option.Host) { + fs.startupPeers = append(fs.startupPeers, addr) + } + } + fs.filer.LoadFilerConf() fs.filer.LoadRemoteStorageConfAndMapping() @@ -390,6 +430,12 @@ func (fs *FilerServer) Shutdown() { fs.remoteCacheEvictCancel() } fs.filer.Shutdown() + // LeaveLockRing already ended the master stream in the normal path; this + // covers callers that skip it, and runs after the filer's own shutdown so + // the final metadata-log flush can still reach the master. + if fs.masterCancel != nil { + fs.masterCancel() + } } // priorOwnerWindowSkew covers peers and gateways that start their prior-owner @@ -417,6 +463,106 @@ func (fs *FilerServer) leaveLockRingWithin(removalTimeout, budget time.Duration) case <-time.After(budget): glog.Warningf("LockRing: %s did not finish leaving within %v, shutting down anyway", fs.option.Host, budget) } + // End the master stream so the master drops this filer from the cluster + // list before gRPC stops accepting connections. + if fs.masterCancel != nil { + fs.masterCancel() + } + fs.waitForClusterRemoval() +} + +// announceCatchUpCap bounds how long startup holds the master registration +// while the metadata replay catches up with the peers found at startup. Past +// it the filer announces anyway, as it did when the gate opened on serve. +const announceCatchUpCap = time.Minute + +// AnnounceOnceCaughtUp closes announce once every peer found at startup has +// replayed its backlog. Each peer's stream advances a watermark per event or +// idle heartbeat, so a watermark past waitStart means nothing older can still +// land on top of this filer's fresh writes. Called once gRPC is serving, so +// routed traffic is answerable the moment the filer is listed. +func (fs *FilerServer) AnnounceOnceCaughtUp(announce chan struct{}, waitStart time.Time) { + defer close(announce) + if len(fs.startupPeers) == 0 || fs.filer.MetaAggregator == nil { + return + } + deadline := time.Now().Add(announceCatchUpCap) + for { + // Take the channel before reading the watermarks so a rise in + // between still wakes this wait (see DeliveryWatermarkAdvancedChan). + advanced := fs.filer.MetaAggregator.DeliveryWatermarkAdvancedChan() + if fs.filer.MetaAggregator.PeersCaughtUpTo(fs.startupPeers, waitStart.UnixNano()) { + glog.V(0).Infof("%s caught up with %d startup peers, announcing to masters", fs.option.Host, len(fs.startupPeers)) + return + } + remaining := time.Until(deadline) + if remaining <= 0 { + glog.Warningf("%s metadata replay not caught up after %v, announcing anyway", fs.option.Host, announceCatchUpCap) + return + } + select { + case <-advanced: + case <-time.After(min(remaining, time.Second)): + } + } +} + +// masterRemovalWaitBudget bounds how long shutdown waits for masters to +// drop this filer before gRPC drains anyway. +const masterRemovalWaitBudget = 5 * time.Second + +// waitForClusterRemoval polls the configured masters until none still lists +// this filer. The master removes membership when the KeepConnected handler +// unwinds, which is asynchronous to cancel, so new requests could otherwise +// route to this filer after its gRPC listener has already closed. +func (fs *FilerServer) waitForClusterRemoval() { + ctx, cancel := context.WithTimeout(context.Background(), masterRemovalWaitBudget) + defer cancel() + confirmedAbsent := make(map[pb.ServerAddress]struct{}) + for ctx.Err() == nil { + pending := false + // The cached list keeps a slow SRV lookup from stretching the + // removal-wait budget. + for _, master := range fs.filer.MasterClient.ListCachedMasters() { + if _, ok := confirmedAbsent[master]; ok { + continue + } + if ctx.Err() != nil { + return + } + // The per-query timeout shares the overall deadline, so a stalled + // master cannot stretch the wait past the budget. + queryCtx, queryCancel := context.WithTimeout(ctx, 2*time.Second) + nodes, err := cluster.ListExistingPeerUpdates(queryCtx, master, fs.grpcDialOption, fs.filer.MasterClient.FilerGroup, cluster.FilerType) + queryCancel() + if err != nil { + // A failed query is not confirmed removal; retry until the + // master answers or the overall deadline expires. + pending = true + continue + } + listed := false + for _, node := range nodes { + if node.Address == string(fs.option.Host) { + listed = true + break + } + } + if listed { + pending = true + } else { + confirmedAbsent[master] = struct{}{} + } + } + if !pending { + return + } + select { + case <-ctx.Done(): + return + case <-time.After(200 * time.Millisecond): + } + } } func (fs *FilerServer) leaveLockRing(removalTimeout time.Duration) { diff --git a/weed/wdclient/filer_client.go b/weed/wdclient/filer_client.go index 957be5f0d..f41cfaa7f 100644 --- a/weed/wdclient/filer_client.go +++ b/weed/wdclient/filer_client.go @@ -341,7 +341,11 @@ func (fc *FilerClient) refreshFilerList() { generation := fc.peerUpdateGeneration() // Query master for filers in our group - updates := cluster.ListExistingPeerUpdates(currentMaster, fc.grpcDialOption, fc.filerGroup, cluster.FilerType) + updates, err := cluster.ListExistingPeerUpdates(context.Background(), currentMaster, fc.grpcDialOption, fc.filerGroup, cluster.FilerType) + if err != nil { + glog.V(1).Infof("FilerClient: failed to list filers in group '%s': %v", fc.filerGroup, err) + return + } if len(updates) == 0 { glog.V(2).Infof("FilerClient: no filers found in group '%s'", fc.filerGroup) diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index e5ab8075a..5e0e0134e 100644 --- a/weed/wdclient/masterclient.go +++ b/weed/wdclient/masterclient.go @@ -145,14 +145,18 @@ func (p *masterVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds [] type MasterClient struct { *vidMapClient // Embedded cache with shared logic - FilerGroup string - clientType string - clientHost pb.ServerAddress - rack string - currentMaster pb.ServerAddress - lastServedMaster pb.ServerAddress - currentMasterLock sync.RWMutex - masters pb.ServerDiscovery + FilerGroup string + clientType string + clientHost pb.ServerAddress + rack string + currentMaster pb.ServerAddress + lastServedMaster pb.ServerAddress + currentMasterLock sync.RWMutex + masters pb.ServerDiscovery + mastersLock sync.Mutex + // refreshLock serializes SRV refreshes so an older lookup finishing + // second cannot overwrite a newer saved list. + refreshLock sync.Mutex grpcDialOption grpc.DialOption grpcTimeout time.Duration // Timeout for gRPC calls to master OnPeerUpdate func(update *master_pb.ClusterNodeUpdate, startFrom time.Time) @@ -167,6 +171,55 @@ type MasterClient struct { streamLock sync.Mutex stream master_pb.Seaweed_KeepConnectedClient leavingLockRing bool + + // announceCh, when non-nil, gates the registration send that adds this + // client to the master's list, so a server can connect and query masters + // before it is ready to be routed to. + announceCh chan struct{} +} + +// SetAnnounceCh sets a channel that must be closed before this client +// registers itself on the master. +func (mc *MasterClient) SetAnnounceCh(ch chan struct{}) { + mc.announceCh = ch +} + +// ListMasters returns the configured master addresses without waiting for a +// connection, for callers that run before KeepConnectedToMaster is connected. +func (mc *MasterClient) ListMasters() []pb.ServerAddress { + mc.refreshMasters() + return mc.masterInstances() +} + +// ListCachedMasters returns the most recently resolved master addresses +// without a fresh SRV lookup, for callers on a deadline that cannot wait for +// DNS. +func (mc *MasterClient) ListCachedMasters() []pb.ServerAddress { + return mc.masterInstances() +} + +// refreshMasters resolves the SRV record outside mastersLock so a slow DNS +// lookup does not stall masterInstances readers; the lock only covers saving +// the resolved list. refreshLock is held across lookup and save so +// overlapping refreshes keep resolution order. +func (mc *MasterClient) refreshMasters() { + mc.refreshLock.Lock() + defer mc.refreshLock.Unlock() + newList := mc.masters.LookupSrvInstances() + if newList == nil { + return + } + mc.mastersLock.Lock() + defer mc.mastersLock.Unlock() + mc.masters.SetInstances(newList) +} + +// masterInstances returns a consistent copy of the configured masters, +// serialized against SRV refreshes. +func (mc *MasterClient) masterInstances() []pb.ServerAddress { + mc.mastersLock.Lock() + defer mc.mastersLock.Unlock() + return mc.masters.GetInstances() } func NewMasterClient(grpcDialOption grpc.DialOption, filerGroup string, clientType string, clientHost pb.ServerAddress, clientDataCenter string, rack string, masters pb.ServerDiscovery) *MasterClient { @@ -210,8 +263,9 @@ func (mc *MasterClient) SetOnMasterChangeFn(fn func(previous, current pb.ServerA func (mc *MasterClient) tryAllMasters(ctx context.Context) { var nextHintedLeader pb.ServerAddress failedMasters := make(map[pb.ServerAddress]struct{}) - mc.masters.RefreshBySrvIfAvailable() - for _, master := range mc.masters.GetInstances() { + mc.refreshMasters() + masterAddrs := mc.masterInstances() + for _, master := range masterAddrs { if _, failed := failedMasters[master]; failed { continue } @@ -234,7 +288,12 @@ func (mc *MasterClient) tryAllMasters(ctx context.Context) { } } } - mc.setCurrentMaster("") + // After a canceled disconnect the last master still answers unary + // queries (e.g. the metadata-log flush during filer shutdown), so + // only clear the serving address on an unintended disconnect. + if ctx.Err() == nil { + mc.setCurrentMaster("") + } } } @@ -254,6 +313,19 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server } glog.V(1).Infof("%s.%s masterClient gRPC stream established to %s in %v", mc.FilerGroup, mc.clientType, master, time.Since(connectStartTime)) + // The stream is usable for queries before registration, so callers + // blocked in GetMaster (e.g. loading a chunked filer.conf during + // startup) proceed while the announce gate still holds. + mc.setCurrentMaster(master) + + if mc.announceCh != nil { + select { + case <-mc.announceCh: + case <-ctx.Done(): + return ctx.Err() + } + } + mc.streamLock.Lock() err = stream.Send(&master_pb.KeepConnectedRequest{ FilerGroup: mc.FilerGroup, @@ -588,14 +660,14 @@ func (mc *MasterClient) GetMaster(ctx context.Context) pb.ServerAddress { // See GetMaster() for important initialization contract details. func (mc *MasterClient) GetMasters(ctx context.Context) []pb.ServerAddress { mc.WaitUntilConnected(ctx) - return mc.masters.GetInstances() + return mc.masterInstances() } // ListMasterSet returns a set of configured master addresses keyed by their // canonical http form. Unlike GetMasters this does not wait for a connection, // so it is safe to call from admission paths that must stay non-blocking. func (mc *MasterClient) ListMasterSet() map[string]struct{} { - addrs := mc.masters.GetInstances() + addrs := mc.masterInstances() set := make(map[string]struct{}, len(addrs)) for _, a := range addrs { set[a.ToHttpAddress()] = struct{}{} @@ -659,7 +731,7 @@ func (mc *MasterClient) KeepConnectedToMaster(ctx context.Context) { } func (mc *MasterClient) FindLeaderFromOtherPeers(myMasterAddress pb.ServerAddress) (leader string) { - for _, master := range mc.masters.GetInstances() { + for _, master := range mc.masterInstances() { if master == myMasterAddress { continue } diff --git a/weed/wdclient/masterclient_announce_test.go b/weed/wdclient/masterclient_announce_test.go new file mode 100644 index 000000000..c467aaea8 --- /dev/null +++ b/weed/wdclient/masterclient_announce_test.go @@ -0,0 +1,64 @@ +package wdclient + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// A filer connects to the master before its gRPC port serves so it can query +// masters during startup; the registration send that adds it to the master's +// filer list must wait until it is actually serving. +func TestAnnounceGateDelaysRegistration(t *testing.T) { + srv := &fakeKeepConnectedServer{requests: make(chan *master_pb.KeepConnectedRequest, 4)} + addr := startFakeMasterServer(t, srv) + mc := NewMasterClient( + grpc.WithTransportCredentials(insecure.NewCredentials()), + "", "filer", "filer1:18888", "", "", + *pb.NewServiceDiscoveryFromMap(map[string]pb.ServerAddress{"m": addr}), + ) + gate := make(chan struct{}) + mc.SetAnnounceCh(gate) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go mc.KeepConnectedToMaster(ctx) + + // WaitUntilConnected returns once the stream is established, before any + // registration: queries (e.g. loading a chunked filer.conf during + // startup) must not block behind the gate. Bound the wait so a + // regression fails fast instead of hanging the test run. + waitCtx, waitCancel := context.WithTimeout(ctx, 3*time.Second) + defer waitCancel() + mc.WaitUntilConnected(waitCtx) + if err := waitCtx.Err(); err != nil { + t.Fatalf("master stream not established within 3s: %v", err) + } + got := make(chan pb.ServerAddress, 1) + go func() { got <- mc.GetMaster(ctx) }() + select { + case m := <-got: + if m != addr { + t.Fatalf("GetMaster = %q, want %q", m, addr) + } + case <-time.After(3 * time.Second): + t.Fatal("GetMaster blocked behind the announce gate") + } + + // Registration must not have been sent yet. + select { + case req := <-srv.requests: + t.Fatalf("client registered before the gate opened: %+v", req) + default: + } + + close(gate) + req := nextRequest(t, srv.requests) + if req.ClientAddress != "filer1:18888" { + t.Fatalf("registration ClientAddress = %q, want filer1:18888", req.ClientAddress) + } +}