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()