From e24a443b1779afa24b385b6e638c2e70a1d1796c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 18 Apr 2026 20:03:23 -0700 Subject: [PATCH] peer chunk sharing 2/8: filer mount registry (#9131) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * proto: define MountRegister/MountList and MountPeer service Adds the wire types for peer chunk sharing between weed mount clients: * filer.proto: MountRegister / MountList RPCs so each mount can heartbeat its peer-serve address into a filer-hosted registry, and refresh the list of peers. Tiny payload; the filer stores only O(fleet_size) state. * mount_peer.proto (new): ChunkAnnounce / ChunkLookup RPCs for the mount-to-mount chunk directory. Each fid's directory entry lives on an HRW-assigned mount; announces and lookups route to that mount. No behavior yet — later PRs wire the RPCs into the filer and mount. See design-weed-mount-peer-chunk-sharing.md for the full design. * filer: add mount-server registry behind -peer.registry.enable Implements tier 1 of the peer chunk sharing design: an in-memory registry of live weed mount servers, keyed by peer address, refreshed by MountRegister heartbeats and served by MountList. * weed/filer/peer_registry.go: thread-safe map with TTL eviction; lazy sweep on List plus a background sweeper goroutine for bounded memory. * weed/server/filer_grpc_server_peer.go: MountRegister / MountList RPC handlers. When -peer.registry.enable is false (the default), both RPCs are silent no-ops so probing older filers is harmless. * -peer.registry.enable flag on weed filer; FilerOption.PeerRegistryEnabled wires it through. Phase 1 is single-filer (no cross-filer replication of the registry); mounts that fail over to another filer will re-register on the next heartbeat, so the registry self-heals within one TTL cycle. Part of the peer-chunk-sharing design; no behavior change at runtime until a later PR enables the flag on both filer and mount. * filer: nil-safe peerRegistryEnable + registry hardening Addresses review feedback on PR #9131. * Fix: nil pointer deref in the mini cluster. FilerOptions instances constructed outside weed/command/filer.go (e.g. miniFilerOptions in mini.go) do not populate peerRegistryEnable, so dereferencing the pointer panics at Filer startup. Use the same `nil && deref` idiom already used for distributedLock / writebackCache. * Hardening (gemini review): registry now enforces three invariants: - empty peer_addr is silently rejected (no client-controlled sentinel mass-inserts) - TTL is capped at 1 hour so a runaway client cannot pin entries - new-entry count is capped at 10000 to bound memory; renewals of existing entries are always honored, so a full registry still heartbeats its existing members correctly Covered by new unit tests. * filer: rename -peer.registry.enable flag to -mount.p2p Per review feedback: the old name "peer.registry.enable" leaked the implementation ("registry") into the CLI surface. "mount.p2p" is shorter and describes what it actually controls — whether this filer participates in mount-to-mount peer chunk sharing. Flag renames (all three keep default=true, idle cost is near-zero): -peer.registry.enable -> -mount.p2p (weed filer) -filer.peer.registry.enable -> -filer.mount.p2p (weed mini, weed server) Internal variable names (mountPeerRegistryEnable, MountPeerRegistry) keep their longer form — they describe the component, not the knob. * filer: MountList returns DataCenter + List uses RLock Two review follow-ups on the mount peer registry: * weed/server/filer_grpc_server_mount_peer.go: MountList was dropping the DataCenter on the wire. The whole point of carrying DC separately from Rack is letting the mount-side fetcher re-rank peers by the two-level locality hierarchy (same-rack > same-DC > cross-DC); without DC in the response every remote peer collapsed to "unknown locality." * weed/filer/mount_peer_registry.go: List() was taking a write lock so it could lazy-delete expired entries inline. But MountList is a read-heavy RPC hit on every mount's 30 s refresh loop, and Sweep is already wired as the sole reclamation path (same pattern as the mount-side PeerDirectory). Switch List to RLock + filter, let Sweep do the map mutation, so concurrent MountList callers don't serialize on each other. Test updated to reflect the new contract (List no longer mutates the map; Sweep is what drops expired entries). --- weed/command/filer.go | 3 + weed/command/mini.go | 1 + weed/command/server.go | 1 + weed/filer/mount_peer_registry.go | 141 +++++++++++++++ weed/filer/mount_peer_registry_test.go | 169 ++++++++++++++++++ weed/server/filer_grpc_server_mount_peer.go | 65 +++++++ .../filer_grpc_server_mount_peer_test.go | 84 +++++++++ weed/server/filer_server.go | 9 + 8 files changed, 473 insertions(+) create mode 100644 weed/filer/mount_peer_registry.go create mode 100644 weed/filer/mount_peer_registry_test.go create mode 100644 weed/server/filer_grpc_server_mount_peer.go create mode 100644 weed/server/filer_grpc_server_mount_peer_test.go diff --git a/weed/command/filer.go b/weed/command/filer.go index 08f4b3399..0781aae0d 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -84,6 +84,7 @@ type FilerOptions struct { tusBasePath *string certProvider certprovider.Provider s3ConfigFile *string // optional path to static S3 identity config + mountPeerRegistryEnable *bool // accept MountRegister/MountList RPCs (peer chunk sharing tier 1) // shutdownCtx, when non-nil, tells startFiler to gracefully shut down its // HTTP/gRPC servers once the ctx is cancelled. Used by integration tests // and by weed mini; nil for standalone weed filer. @@ -126,6 +127,7 @@ func init() { f.allowedOrigins = cmdFiler.Flag.String("allowedOrigins", "*", "comma separated list of allowed origins") f.exposeDirectoryData = cmdFiler.Flag.Bool("exposeDirectoryData", true, "whether to return directory metadata and content in Filer UI") f.tusBasePath = cmdFiler.Flag.String("tusBasePath", "/.tus", "TUS resumable upload endpoint base path (e.g., /.tus)") + f.mountPeerRegistryEnable = cmdFiler.Flag.Bool("mount.p2p", true, "accept MountRegister/MountList RPCs from weed mount clients for peer chunk sharing (tier 1). Idle cost is near-zero; set false to disable.") // start s3 on filer filerStartS3 = cmdFiler.Flag.Bool("s3", false, "whether to start S3 gateway") @@ -386,6 +388,7 @@ func (fo *FilerOptions) startFiler() { AllowedOrigins: strings.Split(*fo.allowedOrigins, ","), TusBasePath: *fo.tusBasePath, CredentialManager: credentialManager, + MountPeerRegistryEnabled: fo.mountPeerRegistryEnable != nil && *fo.mountPeerRegistryEnable, }) if nfs_err != nil { glog.Fatalf("Filer startup error: %v", nfs_err) diff --git a/weed/command/mini.go b/weed/command/mini.go index 9b25ae33e..950037402 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -311,6 +311,7 @@ func initMiniFilerFlags() { miniFilerOptions.allowedOrigins = cmdMini.Flag.String("filer.allowedOrigins", "*", "comma separated list of allowed origins") miniFilerOptions.exposeDirectoryData = cmdMini.Flag.Bool("filer.exposeDirectoryData", true, "whether to return directory metadata and content in Filer UI") miniFilerOptions.tusBasePath = cmdMini.Flag.String("filer.tusBasePath", "/.tus", "TUS resumable upload endpoint base path") + miniFilerOptions.mountPeerRegistryEnable = cmdMini.Flag.Bool("filer.mount.p2p", true, "accept MountRegister/MountList RPCs from weed mount clients for peer chunk sharing (tier 1). Idle cost is near-zero; set false to disable.") } // initMiniVolumeFlags initializes Volume server flag options diff --git a/weed/command/server.go b/weed/command/server.go index c6e42d9a4..4ae3260db 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -130,6 +130,7 @@ func init() { filerOptions.diskType = cmdServer.Flag.String("filer.disk", "", "[hdd|ssd|] hard drive or solid state drive or any tag") filerOptions.exposeDirectoryData = cmdServer.Flag.Bool("filer.exposeDirectoryData", true, "expose directory data via filer. If false, filer UI will be innaccessible.") filerOptions.tusBasePath = cmdServer.Flag.String("filer.tusBasePath", "/.tus", "TUS resumable upload endpoint base path (e.g., /.tus)") + filerOptions.mountPeerRegistryEnable = cmdServer.Flag.Bool("filer.mount.p2p", true, "accept MountRegister/MountList RPCs from weed mount clients for peer chunk sharing (tier 1). Idle cost is near-zero; set false to disable.") serverOptions.v.port = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port") serverOptions.v.portGrpc = cmdServer.Flag.Int("volume.port.grpc", 0, "volume server grpc listen port") diff --git a/weed/filer/mount_peer_registry.go b/weed/filer/mount_peer_registry.go new file mode 100644 index 000000000..0385afaba --- /dev/null +++ b/weed/filer/mount_peer_registry.go @@ -0,0 +1,141 @@ +package filer + +import ( + "sync" + "time" +) + +// maxMountPeerRegistryEntries caps the number of mounts the filer will track to +// prevent a burst of buggy or malicious clients from exhausting memory. At +// ~80 B per entry (map + struct + strings) a 10k cap costs ~1 MB, which is +// ample headroom for real fleets while being well under any filer's budget. +// A full registry silently rejects new registrations until expiry frees room. +const maxMountPeerRegistryEntries = 10000 + +// maxMountPeerRegistryTTL caps a single heartbeat's requested TTL. Prevents a +// misconfigured or malicious client from pinning an entry indefinitely. +const maxMountPeerRegistryTTL = time.Hour + +// MountPeerRegistry is the in-memory mount-server registry (tier 1 of the peer +// chunk sharing design). The filer holds a map of mount-server address -> +// metadata with TTL-bounded entries refreshed by MountRegister heartbeats. +// +// The registry is small (O(fleet_size)) and slow-changing; fid-level state +// is NOT stored here — that lives on the mount fleet itself (tier 2). +// +// See design-weed-mount-peer-chunk-sharing.md §4.2.1. +type MountPeerRegistry struct { + mu sync.RWMutex + entries map[string]*mountPeerRegistryEntry + clock func() time.Time // injectable for tests +} + +type mountPeerRegistryEntry struct { + peerAddr string + dataCenter string + rack string + expiry time.Time + lastSeen time.Time +} + +// MountPeerInfo is the public view of a registered mount. DataCenter and +// Rack are carried as a two-level locality hierarchy: a peer in the same +// DC but a different rack is still a much better fetch target than a peer +// in a different DC, so both are worth distinguishing for ranking. +type MountPeerInfo struct { + PeerAddr string + DataCenter string + Rack string + LastSeenNs int64 +} + +// NewMountPeerRegistry constructs an empty registry using the real wall clock. +func NewMountPeerRegistry() *MountPeerRegistry { + return newMountPeerRegistryWithClock(time.Now) +} + +func newMountPeerRegistryWithClock(clock func() time.Time) *MountPeerRegistry { + return &MountPeerRegistry{ + entries: make(map[string]*mountPeerRegistryEntry), + clock: clock, + } +} + +// Register inserts or renews an entry. A zero or negative ttl is treated as +// "use a sane default" (60 s); a ttl exceeding maxMountPeerRegistryTTL is capped. +// An empty peerAddr is rejected silently. When the registry is at capacity, +// a *new* entry is rejected; renewals of existing entries always succeed. +func (r *MountPeerRegistry) Register(peerAddr, dataCenter, rack string, ttl time.Duration) { + if peerAddr == "" { + return + } + if ttl <= 0 { + ttl = 60 * time.Second + } + if ttl > maxMountPeerRegistryTTL { + ttl = maxMountPeerRegistryTTL + } + now := r.clock() + r.mu.Lock() + defer r.mu.Unlock() + entry, ok := r.entries[peerAddr] + if !ok { + if len(r.entries) >= maxMountPeerRegistryEntries { + return + } + entry = &mountPeerRegistryEntry{peerAddr: peerAddr} + r.entries[peerAddr] = entry + } + entry.dataCenter = dataCenter + entry.rack = rack + entry.lastSeen = now + entry.expiry = now.Add(ttl) +} + +// List returns all entries that have not yet expired, in no particular +// order. Expired entries are filtered out of the response but NOT deleted +// here — Sweep handles that under a write lock on its own schedule. List +// is called on every mount's MountList refresh (30 s cadence per mount) +// so keeping it RLock-only lets concurrent callers proceed in parallel. +func (r *MountPeerRegistry) List() []MountPeerInfo { + now := r.clock() + r.mu.RLock() + defer r.mu.RUnlock() + result := make([]MountPeerInfo, 0, len(r.entries)) + for _, entry := range r.entries { + if !entry.expiry.After(now) { + continue // Sweep will clean this up + } + result = append(result, MountPeerInfo{ + PeerAddr: entry.peerAddr, + DataCenter: entry.dataCenter, + Rack: entry.rack, + LastSeenNs: entry.lastSeen.UnixNano(), + }) + } + return result +} + +// Len returns the current entry count (including entries that may have +// expired but not yet been swept). +func (r *MountPeerRegistry) Len() int { + r.mu.RLock() + defer r.mu.RUnlock() + return len(r.entries) +} + +// Sweep removes expired entries. Safe to call periodically. Returns the +// number of entries evicted. +func (r *MountPeerRegistry) Sweep() int { + now := r.clock() + r.mu.Lock() + defer r.mu.Unlock() + evicted := 0 + for addr, entry := range r.entries { + if !entry.expiry.After(now) { + delete(r.entries, addr) + evicted++ + } + } + return evicted +} diff --git a/weed/filer/mount_peer_registry_test.go b/weed/filer/mount_peer_registry_test.go new file mode 100644 index 000000000..1bbc04b38 --- /dev/null +++ b/weed/filer/mount_peer_registry_test.go @@ -0,0 +1,169 @@ +package filer + +import ( + "fmt" + "sort" + "testing" + "time" +) + +func testClock(t time.Time) func() time.Time { + return func() time.Time { return t } +} + +func TestMountPeerRegistry_RegisterAndList(t *testing.T) { + start := time.Unix(1000, 0) + current := start + r := newMountPeerRegistryWithClock(func() time.Time { return current }) + + r.Register("mount-a:18080", "", "rack1", 30*time.Second) + r.Register("mount-b:18080", "", "rack2", 30*time.Second) + + list := r.List() + if len(list) != 2 { + t.Fatalf("expected 2 entries, got %d", len(list)) + } + + sort.Slice(list, func(i, j int) bool { return list[i].PeerAddr < list[j].PeerAddr }) + if list[0].PeerAddr != "mount-a:18080" || list[0].Rack != "rack1" { + t.Errorf("entry 0 unexpected: %+v", list[0]) + } + if list[1].PeerAddr != "mount-b:18080" || list[1].Rack != "rack2" { + t.Errorf("entry 1 unexpected: %+v", list[1]) + } +} + +func TestMountPeerRegistry_RenewExtendsExpiry(t *testing.T) { + current := time.Unix(1000, 0) + r := newMountPeerRegistryWithClock(func() time.Time { return current }) + + r.Register("mount-a:18080", "", "rack1", 30*time.Second) + + // Advance time past original expiry. + current = current.Add(25 * time.Second) + // Renew — should push expiry to now+30. + r.Register("mount-a:18080", "", "rack1-updated", 30*time.Second) + + // Advance to where original expiry would have triggered eviction. + current = current.Add(10 * time.Second) + list := r.List() + if len(list) != 1 { + t.Fatalf("expected 1 entry after renew, got %d", len(list)) + } + if list[0].Rack != "rack1-updated" { + t.Errorf("rack not updated on renew: %q", list[0].Rack) + } +} + +func TestMountPeerRegistry_ExpirationDropsEntry(t *testing.T) { + current := time.Unix(1000, 0) + r := newMountPeerRegistryWithClock(func() time.Time { return current }) + + r.Register("mount-a:18080", "", "", 10*time.Second) + if n := r.Len(); n != 1 { + t.Fatalf("expected 1 entry, got %d", n) + } + + current = current.Add(15 * time.Second) + list := r.List() + if len(list) != 0 { + t.Errorf("expected 0 entries after expiry, got %d", len(list)) + } + // List no longer deletes — it's RLock-only so concurrent callers can + // proceed in parallel. Sweep is the sole reclamation path. + if n := r.Len(); n != 1 { + t.Errorf("List should not delete expired entries; Len=%d want 1", n) + } + if evicted := r.Sweep(); evicted != 1 { + t.Errorf("Sweep should have evicted the expired entry; got %d", evicted) + } + if n := r.Len(); n != 0 { + t.Errorf("after Sweep, Len=%d want 0", n) + } +} + +func TestMountPeerRegistry_SweepCountsEvictions(t *testing.T) { + current := time.Unix(1000, 0) + r := newMountPeerRegistryWithClock(func() time.Time { return current }) + + r.Register("mount-a:18080", "", "", 10*time.Second) + r.Register("mount-b:18080", "", "", 60*time.Second) + + current = current.Add(30 * time.Second) + got := r.Sweep() + if got != 1 { + t.Errorf("expected 1 eviction, got %d", got) + } + if n := r.Len(); n != 1 { + t.Errorf("expected 1 surviving entry, got %d", n) + } +} + +func TestMountPeerRegistry_NegativeTTLFallsBackToDefault(t *testing.T) { + current := time.Unix(1000, 0) + r := newMountPeerRegistryWithClock(func() time.Time { return current }) + + r.Register("mount-a:18080", "", "", -5*time.Second) + if n := r.Len(); n != 1 { + t.Fatalf("expected 1 entry even with bad ttl, got %d", n) + } + + // Advance past the default (60s) and confirm it expires. + current = current.Add(61 * time.Second) + list := r.List() + if len(list) != 0 { + t.Errorf("expected entry to expire after default ttl, got %d entries", len(list)) + } +} + +func TestMountPeerRegistry_EmptyList(t *testing.T) { + r := NewMountPeerRegistry() + list := r.List() + if len(list) != 0 { + t.Errorf("expected empty list, got %d", len(list)) + } + _ = testClock // keep helper exported for future tests +} + +func TestMountPeerRegistry_EmptyPeerAddrRejected(t *testing.T) { + r := NewMountPeerRegistry() + r.Register("", "", "rack", 30*time.Second) + if n := r.Len(); n != 0 { + t.Errorf("empty peer_addr should not insert; Len=%d", n) + } +} + +func TestMountPeerRegistry_TTLCapped(t *testing.T) { + current := time.Unix(1000, 0) + r := newMountPeerRegistryWithClock(func() time.Time { return current }) + + // Request a ridiculous TTL; should be capped to maxMountPeerRegistryTTL. + r.Register("mount-a:18080", "", "", 24*time.Hour) + + // Advance past the cap; entry should now be expired. + current = current.Add(maxMountPeerRegistryTTL + time.Second) + if list := r.List(); len(list) != 0 { + t.Errorf("expected entry to expire after maxMountPeerRegistryTTL, got %d", len(list)) + } +} + +func TestMountPeerRegistry_CapacityLimit(t *testing.T) { + r := NewMountPeerRegistry() + // Fill to capacity. + for i := 0; i < maxMountPeerRegistryEntries; i++ { + r.Register(fmt.Sprintf("mount-%d:18080", i), "", "", 60*time.Second) + } + if n := r.Len(); n != maxMountPeerRegistryEntries { + t.Fatalf("expected %d entries, got %d", maxMountPeerRegistryEntries, n) + } + // A brand-new address beyond the cap is rejected. + r.Register("new-mount:18080", "", "", 60*time.Second) + if n := r.Len(); n != maxMountPeerRegistryEntries { + t.Errorf("new entry past cap should be rejected; Len=%d", n) + } + // A renewal of an existing entry still succeeds. + r.Register("mount-0:18080", "", "rack-renewed", 60*time.Second) + if n := r.Len(); n != maxMountPeerRegistryEntries { + t.Errorf("renewal should not increase size; Len=%d", n) + } +} diff --git a/weed/server/filer_grpc_server_mount_peer.go b/weed/server/filer_grpc_server_mount_peer.go new file mode 100644 index 000000000..e8eedec70 --- /dev/null +++ b/weed/server/filer_grpc_server_mount_peer.go @@ -0,0 +1,65 @@ +package weed_server + +import ( + "context" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// mountPeerRegistrySweepInterval is how often the filer evicts expired mount +// registry entries. Eviction is also done lazily on List; the sweep keeps +// memory bounded on long-running filers with high churn. +const mountPeerRegistrySweepInterval = 60 * time.Second + +// runMountPeerRegistrySweeper runs for the lifetime of the FilerServer when +// peer registry is enabled. +func (fs *FilerServer) runMountPeerRegistrySweeper() { + ticker := time.NewTicker(mountPeerRegistrySweepInterval) + defer ticker.Stop() + for range ticker.C { + if fs.mountPeerRegistry == nil { + return + } + if evicted := fs.mountPeerRegistry.Sweep(); evicted > 0 { + glog.V(2).Infof("peer registry: evicted %d stale entries", evicted) + } + } +} + +// MountRegister records (or refreshes) the caller as a live mount server in +// the filer's peer registry. Returns an empty response; the caller is +// expected to heartbeat before the TTL expires. +// +// Requests are silently dropped when the registry is disabled (default), so +// clients can safely probe without breaking older filers. +func (fs *FilerServer) MountRegister(ctx context.Context, req *filer_pb.MountRegisterRequest) (*filer_pb.MountRegisterResponse, error) { + if fs.mountPeerRegistry == nil { + return &filer_pb.MountRegisterResponse{}, nil + } + ttl := time.Duration(req.TtlSeconds) * time.Second + fs.mountPeerRegistry.Register(req.PeerAddr, req.DataCenter, req.Rack, ttl) + return &filer_pb.MountRegisterResponse{}, nil +} + +// MountList returns the current set of live mounts for callers building +// their HRW seed view. +func (fs *FilerServer) MountList(ctx context.Context, req *filer_pb.MountListRequest) (*filer_pb.MountListResponse, error) { + if fs.mountPeerRegistry == nil { + return &filer_pb.MountListResponse{}, nil + } + entries := fs.mountPeerRegistry.List() + resp := &filer_pb.MountListResponse{ + Mounts: make([]*filer_pb.MountInfo, 0, len(entries)), + } + for _, e := range entries { + resp.Mounts = append(resp.Mounts, &filer_pb.MountInfo{ + PeerAddr: e.PeerAddr, + DataCenter: e.DataCenter, + Rack: e.Rack, + LastSeenNs: e.LastSeenNs, + }) + } + return resp, nil +} diff --git a/weed/server/filer_grpc_server_mount_peer_test.go b/weed/server/filer_grpc_server_mount_peer_test.go new file mode 100644 index 000000000..7a688aad7 --- /dev/null +++ b/weed/server/filer_grpc_server_mount_peer_test.go @@ -0,0 +1,84 @@ +package weed_server + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +func TestMountRegister_DisabledIsNoOp(t *testing.T) { + fs := &FilerServer{} // mountPeerRegistry nil → registry disabled + _, err := fs.MountRegister(context.Background(), &filer_pb.MountRegisterRequest{ + PeerAddr: "mount-a:18080", + TtlSeconds: 30, + }) + if err != nil { + t.Fatalf("MountRegister returned error with disabled registry: %v", err) + } + resp, err := fs.MountList(context.Background(), &filer_pb.MountListRequest{}) + if err != nil { + t.Fatalf("MountList returned error with disabled registry: %v", err) + } + if len(resp.Mounts) != 0 { + t.Errorf("expected empty list from disabled registry, got %d", len(resp.Mounts)) + } +} + +func TestMountRegister_EnabledStoresAndLists(t *testing.T) { + fs := &FilerServer{mountPeerRegistry: filer.NewMountPeerRegistry()} + + if _, err := fs.MountRegister(context.Background(), &filer_pb.MountRegisterRequest{ + PeerAddr: "mount-a:18080", + Rack: "rack-a", + TtlSeconds: 60, + }); err != nil { + t.Fatalf("MountRegister: %v", err) + } + if _, err := fs.MountRegister(context.Background(), &filer_pb.MountRegisterRequest{ + PeerAddr: "mount-b:18080", + Rack: "rack-b", + TtlSeconds: 60, + }); err != nil { + t.Fatalf("MountRegister: %v", err) + } + + resp, err := fs.MountList(context.Background(), &filer_pb.MountListRequest{}) + if err != nil { + t.Fatalf("MountList: %v", err) + } + if len(resp.Mounts) != 2 { + t.Fatalf("expected 2 mounts, got %d", len(resp.Mounts)) + } + + byAddr := map[string]*filer_pb.MountInfo{} + for _, m := range resp.Mounts { + byAddr[m.PeerAddr] = m + } + if byAddr["mount-a:18080"].Rack != "rack-a" { + t.Errorf("unexpected rack for mount-a: %+v", byAddr["mount-a:18080"]) + } + if byAddr["mount-b:18080"].Rack != "rack-b" { + t.Errorf("unexpected rack for mount-b: %+v", byAddr["mount-b:18080"]) + } + if byAddr["mount-a:18080"].LastSeenNs == 0 { + t.Errorf("last_seen_ns should be populated") + } +} + +func TestMountRegister_RenewIsIdempotent(t *testing.T) { + fs := &FilerServer{mountPeerRegistry: filer.NewMountPeerRegistry()} + for i := 0; i < 5; i++ { + if _, err := fs.MountRegister(context.Background(), &filer_pb.MountRegisterRequest{ + PeerAddr: "mount-a:18080", + TtlSeconds: 30, + }); err != nil { + t.Fatalf("MountRegister iteration %d: %v", i, err) + } + } + resp, _ := fs.MountList(context.Background(), &filer_pb.MountListRequest{}) + if len(resp.Mounts) != 1 { + t.Errorf("repeated renew should still show 1 entry, got %d", len(resp.Mounts)) + } +} diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 0365401be..63634ce4d 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -84,6 +84,7 @@ type FilerOption struct { TusBasePath string S3ConfigFile string // optional path to static S3 identity config file CredentialManager *credential.CredentialManager + MountPeerRegistryEnabled bool // opt-in: accept MountRegister/MountList RPCs } type FilerServer struct { @@ -120,6 +121,10 @@ type FilerServer struct { // credential manager for IAM operations CredentialManager *credential.CredentialManager + + // mountPeerRegistry is nil unless FilerOption.MountPeerRegistryEnabled is true. + // When populated, it backs the MountRegister / MountList RPCs. + mountPeerRegistry *filer.MountPeerRegistry } func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) { @@ -159,6 +164,10 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) recentCopyRequests: make(map[string]recentCopyRequest), CredentialManager: option.CredentialManager, } + if option.MountPeerRegistryEnabled { + fs.mountPeerRegistry = filer.NewMountPeerRegistry() + go fs.runMountPeerRegistrySweeper() + } fs.listenersCond = sync.NewCond(&fs.listenersLock) option.Masters.RefreshBySrvIfAvailable()