mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 23:14:21 +00:00
peer chunk sharing 2/8: filer mount registry (#9131)
* 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).
This commit is contained in:
Chris Lu
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -130,6 +130,7 @@ func init() {
|
||||
filerOptions.diskType = cmdServer.Flag.String("filer.disk", "", "[hdd|ssd|<tag>] 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")
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user