From af1e5712972cdc80f812336fdad265a48b23f273 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 18 Apr 2026 20:03:34 -0700 Subject: [PATCH] peer chunk sharing 3/8: mount peer-serve HTTP endpoint (#9132) 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). * mount: add peer chunk sharing options + advertise address resolver First cut at the peer chunk sharing wiring on the mount side. No functional behavior yet — this PR just introduces the option fields, the -peer.* flags, and the helper that resolves a reachable host:port from them. The server implementation arrives in PR #5 (gRPC service) and the fetcher in PR #7. * ResolvePeerAdvertiseAddr: an explicit -peer.advertise wins; else we use -peer.listen's bind host if specific; else util.DetectedHostAddress combined with the port. This is what gets registered with the filer and announced to peers, so wildcard binds no longer result in unreachable identities like "[::]:18080". * Option fields: PeerEnabled, PeerListen, PeerAdvertise, PeerRack. One port handles both directory RPCs and streaming chunk fetches (see PR #1 FetchChunk proto), so there is no second -peer.grpc.* flag — the old HTTP byte-transfer path is gone. * New flags on weed mount: -peer.enable, -peer.listen (default :18080), -peer.advertise (default auto), -peer.rack. --- weed/command/mount.go | 14 ++++++++ weed/command/mount_std.go | 13 ++++++++ weed/mount/peer_advertise.go | 52 +++++++++++++++++++++++++++++ weed/mount/peer_advertise_test.go | 55 +++++++++++++++++++++++++++++++ weed/mount/weedfs.go | 10 ++++++ 5 files changed, 144 insertions(+) create mode 100644 weed/mount/peer_advertise.go create mode 100644 weed/mount/peer_advertise_test.go diff --git a/weed/command/mount.go b/weed/command/mount.go index 5803cd2e5..1e7c926a3 100644 --- a/weed/command/mount.go +++ b/weed/command/mount.go @@ -52,6 +52,13 @@ type MountOptions struct { rdmaMaxConcurrent *int rdmaTimeoutMs *int + // Peer chunk sharing options (design-weed-mount-peer-chunk-sharing.md). + peerEnabled *bool + peerListen *string + peerAdvertise *string + peerDataCenter *string + peerRack *string + dirIdleEvictSec *int // Distributed lock for cross-mount write coordination @@ -129,6 +136,13 @@ func init() { mountOptions.rdmaMaxConcurrent = cmdMount.Flag.Int("rdma.maxConcurrent", 64, "max concurrent RDMA operations") mountOptions.rdmaTimeoutMs = cmdMount.Flag.Int("rdma.timeoutMs", 5000, "RDMA operation timeout in milliseconds") + // Peer chunk sharing flags. + mountOptions.peerEnabled = cmdMount.Flag.Bool("peer.enable", false, "opt in to peer chunk sharing — mount serves its chunk cache to other mounts and fetches from peers instead of volume servers when available") + mountOptions.peerListen = cmdMount.Flag.String("peer.listen", ":18080", "bind address for peer gRPC (directory RPCs + FetchChunk streaming)") + mountOptions.peerAdvertise = cmdMount.Flag.String("peer.advertise", "", "externally-reachable host:port other mounts use to reach this one (defaults to autodetected host + -peer.listen port)") + mountOptions.peerDataCenter = cmdMount.Flag.String("peer.dataCenter", "", "optional data-center label advertised to peers; used with -peer.rack for two-level locality ranking") + mountOptions.peerRack = cmdMount.Flag.String("peer.rack", "", "optional rack label advertised to peers") + mountOptions.dirIdleEvictSec = cmdMount.Flag.Int("dirIdleEvictSec", 600, "seconds to evict idle cached directories (0 to disable)") mountCpuProfile = cmdMount.Flag.String("cpuprofile", "", "cpu profile output file") diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go index 0ca6bb654..e7c5cfe83 100644 --- a/weed/command/mount_std.go +++ b/weed/command/mount_std.go @@ -356,6 +356,12 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { EnableDistributedLock: option.distributedLock != nil && *option.distributedLock, WritebackCache: option.writebackCache != nil && *option.writebackCache, PosixDirNlink: option.posixDirNlink != nil && *option.posixDirNlink, + // Peer chunk sharing + PeerEnabled: option.peerEnabled != nil && *option.peerEnabled, + PeerListen: peerStringOrEmpty(option.peerListen), + PeerAdvertise: peerStringOrEmpty(option.peerAdvertise), + PeerDataCenter: peerStringOrEmpty(option.peerDataCenter), + PeerRack: peerStringOrEmpty(option.peerRack), }) // create mount root @@ -411,3 +417,10 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { return true } + +func peerStringOrEmpty(p *string) string { + if p == nil { + return "" + } + return *p +} diff --git a/weed/mount/peer_advertise.go b/weed/mount/peer_advertise.go new file mode 100644 index 000000000..ce88d2801 --- /dev/null +++ b/weed/mount/peer_advertise.go @@ -0,0 +1,52 @@ +package mount + +import ( + "fmt" + "net" + + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// ResolvePeerAdvertiseAddr returns the host:port this mount should +// register with the filer and announce to peers. It handles three +// common cases: +// +// 1. The operator set -peer.advertise=host:port explicitly — use it. +// 2. -peer.listen contains a specific bind host — use that host with +// the bind port (e.g. "10.0.0.5:18080"). +// 3. -peer.listen is a wildcard bind (":18080", "0.0.0.0:18080", or +// "[::]:18080") — combine util.DetectedHostAddress() with the port. +// +// Returns an error if the listen string is unparseable or if case (3) +// hits and auto-detection turns up nothing. We deliberately do NOT fall +// back to "localhost": an advertised loopback gets registered with the +// filer and hands other mounts a useless address to dial (their own +// loopback). Better to fail loudly so the operator sets -peer.advertise. +func ResolvePeerAdvertiseAddr(listen, advertise string) (string, error) { + if advertise != "" { + return advertise, nil + } + if listen == "" { + return "", fmt.Errorf("peer listen address is empty") + } + host, port, err := net.SplitHostPort(listen) + if err != nil { + return "", fmt.Errorf("parse -peer.listen %q: %w", listen, err) + } + if !isWildcardHost(host) { + return net.JoinHostPort(host, port), nil + } + detected := util.DetectedHostAddress() + if detected == "" { + return "", fmt.Errorf("cannot auto-detect host for wildcard -peer.listen %q; set -peer.advertise=host:port explicitly", listen) + } + return net.JoinHostPort(detected, port), nil +} + +func isWildcardHost(h string) bool { + switch h { + case "", "0.0.0.0", "::", "[::]": + return true + } + return false +} diff --git a/weed/mount/peer_advertise_test.go b/weed/mount/peer_advertise_test.go new file mode 100644 index 000000000..64e7f3a4e --- /dev/null +++ b/weed/mount/peer_advertise_test.go @@ -0,0 +1,55 @@ +package mount + +import ( + "strings" + "testing" +) + +func TestResolvePeerAdvertiseAddr(t *testing.T) { + tests := []struct { + name string + listen string + advertise string + wantErr bool + // addr substring check — DetectedHostAddress is system-dependent so + // we don't assert an exact value for the wildcard path. + wantSuffix string + }{ + {"explicit advertise wins", ":18080", "10.1.1.9:20000", false, ":20000"}, + {"bind host used verbatim", "10.0.0.5:18080", "", false, "10.0.0.5:18080"}, + {"wildcard ipv4 bind", "0.0.0.0:18080", "", false, ":18080"}, + {"empty host bind", ":18080", "", false, ":18080"}, + {"ipv6 wildcard", "[::]:18080", "", false, ":18080"}, + {"unparseable listen errors", "garbage", "", true, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := ResolvePeerAdvertiseAddr(tt.listen, tt.advertise) + if tt.wantErr { + if err == nil { + t.Fatalf("expected error, got addr=%q", got) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !strings.HasSuffix(got, tt.wantSuffix) { + t.Errorf("addr=%q, expected suffix %q", got, tt.wantSuffix) + } + }) + } +} + +func TestIsWildcardHost(t *testing.T) { + for _, h := range []string{"", "0.0.0.0", "::", "[::]"} { + if !isWildcardHost(h) { + t.Errorf("%q should be wildcard", h) + } + } + for _, h := range []string{"10.0.0.5", "host.example", "localhost"} { + if isWildcardHost(h) { + t.Errorf("%q should NOT be wildcard", h) + } + } +} diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 969dfbba2..98845b90b 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -82,6 +82,16 @@ type Option struct { RdmaMaxConcurrent int RdmaTimeoutMs int + // Peer chunk sharing options (design-weed-mount-peer-chunk-sharing.md). + // When PeerEnabled is false (default), the mount runs exactly as today. + // One gRPC port carries everything: directory RPCs (ChunkAnnounce / + // ChunkLookup) and streaming FetchChunk byte transfers. + PeerEnabled bool + PeerListen string // host:port to bind the peer gRPC server + PeerAdvertise string // externally reachable host:port (optional; defaults to auto-detected host + PeerListen port) + PeerDataCenter string // optional data-center label advertised to peers + PeerRack string // optional rack label advertised to peers (finer than DC) + // Directory cache refresh/eviction controls DirIdleEvictSec int