mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-19 13:46:58 +00:00
peer chunk sharing 5/8: mount chunk-directory shard (#9134)
mount: tier-2 chunk directory + FetchChunk streaming on one gRPC port Collapses the old two-port design (HTTP peer-serve + separate gRPC directory) into a single gRPC service that handles every mount-to- mount exchange: ChunkAnnounce, ChunkLookup, and the new FetchChunk byte stream. * peer_directory.go: fid -> holders shard, HRW-gated; returns holders in LRU order; capacity-bounded; Sweep handles eviction under write-lock while Lookup runs under RLock (hot path is concurrent). * peer_grpc.go: single MountPeer gRPC server implementing all three RPCs. FetchChunk frames bytes at 1 MiB per Send so the default 4 MiB message cap does not constrain chunk size; cache miss returns gRPC NOT_FOUND so clients distinguish miss from transport error. Reuses pb.NewGrpcServer for consistent keepalive + msg-size tuning. * peer_bytepool.go: sync.Pool wrapper around *[]byte that the server uses to avoid a fresh 8 MiB allocation per FetchChunk call. * WFS wiring starts the gRPC server on option.PeerListen (the single peer port) using the advertise address resolved in PR #3 as the HRW identity. A background sweeper evicts expired directory entries every 60 s.
This commit is contained in:
Chris Lu
@@ -0,0 +1,283 @@
|
||||
package mount
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// maxPeerDirectoryEntries caps the total (fid, holder) edges a single mount
|
||||
// will hold in its shard of the chunk directory. Each edge is a
|
||||
// (uint32, time.Time) map entry (~40 B with Go's map overhead), plus the
|
||||
// per-peer metadata is interned — see peerMetadata below — so 500k edges
|
||||
// come out to roughly 20 MB regardless of fleet size. A full directory
|
||||
// rejects new announces until TTL frees room.
|
||||
const maxPeerDirectoryEntries = 500000
|
||||
|
||||
// peerID is the interned identifier for a peer. Peer addresses and locality
|
||||
// labels live once in the peers slice; the (fid -> holder) index references
|
||||
// peers by this fixed-size id so we avoid paying for the ~48 B worth of
|
||||
// string headers (addr, dc, rack) on every edge.
|
||||
type peerID uint32
|
||||
|
||||
// peerMetadata is the per-peer data that used to be inlined on every
|
||||
// holderEntry. Fields are updated in place on each announce so a peer
|
||||
// reconfiguring its dc/rack is picked up without churn.
|
||||
type peerMetadata struct {
|
||||
addr string
|
||||
dataCenter string
|
||||
rack string
|
||||
}
|
||||
|
||||
// PeerDirectory is the tier-2 chunk directory: an in-memory
|
||||
// `fid -> set<peer>` map, holding entries only for fids for which this
|
||||
// mount is the HRW-assigned owner. Populated by inbound ChunkAnnounce RPCs,
|
||||
// queried by inbound ChunkLookup RPCs. Entries expire on TTL.
|
||||
//
|
||||
// Peer addresses are interned into peerID. That way the hot index stores
|
||||
// `fid -> (peerID -> expiry)`, a map of two fixed-size values, instead of
|
||||
// a full (addr, dc, rack, expiry) struct per edge. The peer slice itself
|
||||
// is bounded by fleet size (typically a few hundred), so the interning
|
||||
// table's memory cost is negligible next to the per-edge savings.
|
||||
//
|
||||
// Lookup takes only a read lock; TTL eviction is the sole responsibility of
|
||||
// the Sweep goroutine, which runs on a periodic ticker. This keeps the hot
|
||||
// read path concurrent even under heavy announce traffic.
|
||||
//
|
||||
// See design-weed-mount-peer-chunk-sharing.md §4.2.2.
|
||||
type PeerDirectory struct {
|
||||
mu sync.RWMutex
|
||||
|
||||
// Interning table. peers[id] gives the peer's metadata; peerIDByAddr
|
||||
// maps an address back to its id when a new announce comes in.
|
||||
peers []peerMetadata
|
||||
peerIDByAddr map[string]peerID
|
||||
|
||||
// fid -> (peerID -> expiry). The inner map's value is a bare timestamp
|
||||
// — metadata is looked up via peers[id] on demand.
|
||||
index map[string]map[peerID]time.Time
|
||||
|
||||
// entryCount is maintained in lock-step with map mutations (under
|
||||
// mu), so Stats() can report O(1) instead of scanning the index.
|
||||
entryCount int
|
||||
|
||||
announces atomic.Int64
|
||||
lookups atomic.Int64
|
||||
rejected atomic.Int64
|
||||
evictions atomic.Int64
|
||||
|
||||
clock func() time.Time
|
||||
}
|
||||
|
||||
// NewPeerDirectory constructs a directory using the real wall clock.
|
||||
func NewPeerDirectory() *PeerDirectory {
|
||||
return newPeerDirectoryWithClock(time.Now)
|
||||
}
|
||||
|
||||
func newPeerDirectoryWithClock(clock func() time.Time) *PeerDirectory {
|
||||
return &PeerDirectory{
|
||||
index: map[string]map[peerID]time.Time{},
|
||||
peerIDByAddr: map[string]peerID{},
|
||||
clock: clock,
|
||||
}
|
||||
}
|
||||
|
||||
// AnnounceResult reports how an inbound announce was handled per-fid: fids
|
||||
// this receiver owns were recorded; others were rejected so the caller can
|
||||
// retry the correct owner after refreshing its seed view.
|
||||
type AnnounceResult struct {
|
||||
Rejected []string
|
||||
}
|
||||
|
||||
// OwnerCheck is a predicate provided by the caller (backed by HRW on the
|
||||
// current seed list) that says whether self is the owner for a given fid.
|
||||
type OwnerCheck func(fid string) bool
|
||||
|
||||
// internPeer returns the interned id for peerAddr, allocating a new slot
|
||||
// if this is the first announce from that peer. Must be called under mu
|
||||
// (write). dc/rack are refreshed on each call so reconfig is picked up.
|
||||
func (d *PeerDirectory) internPeer(addr, dc, rack string) peerID {
|
||||
if id, ok := d.peerIDByAddr[addr]; ok {
|
||||
p := &d.peers[id]
|
||||
p.dataCenter = dc
|
||||
p.rack = rack
|
||||
return id
|
||||
}
|
||||
id := peerID(len(d.peers))
|
||||
d.peers = append(d.peers, peerMetadata{addr: addr, dataCenter: dc, rack: rack})
|
||||
d.peerIDByAddr[addr] = id
|
||||
return id
|
||||
}
|
||||
|
||||
// Announce records the caller as a holder for every accepted fid. Any fid
|
||||
// for which ownerCheck returns false is rejected and surfaced back. A
|
||||
// directory at maxPeerDirectoryEntries silently drops *new* edges
|
||||
// (renewals of existing edges continue to succeed).
|
||||
func (d *PeerDirectory) Announce(peerAddr, dataCenter, rack string, fids []string, ttl time.Duration, ownerCheck OwnerCheck) AnnounceResult {
|
||||
if ttl <= 0 {
|
||||
ttl = 60 * time.Second
|
||||
}
|
||||
expiry := d.clock().Add(ttl)
|
||||
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
|
||||
pid := d.internPeer(peerAddr, dataCenter, rack)
|
||||
|
||||
result := AnnounceResult{}
|
||||
for _, fid := range fids {
|
||||
if ownerCheck != nil && !ownerCheck(fid) {
|
||||
result.Rejected = append(result.Rejected, fid)
|
||||
d.rejected.Add(1)
|
||||
continue
|
||||
}
|
||||
holders, ok := d.index[fid]
|
||||
if !ok {
|
||||
if d.entryCount >= maxPeerDirectoryEntries {
|
||||
continue
|
||||
}
|
||||
holders = map[peerID]time.Time{}
|
||||
d.index[fid] = holders
|
||||
}
|
||||
if _, exists := holders[pid]; !exists {
|
||||
if d.entryCount >= maxPeerDirectoryEntries {
|
||||
continue
|
||||
}
|
||||
d.entryCount++
|
||||
}
|
||||
holders[pid] = expiry
|
||||
d.announces.Add(1)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// LookupHolder is the public form of a holder entry. DataCenter and Rack
|
||||
// are carried through so the fetcher can re-sort by its own locality
|
||||
// before picking a peer to dial.
|
||||
type LookupHolder struct {
|
||||
PeerAddr string
|
||||
DataCenter string
|
||||
Rack string
|
||||
}
|
||||
|
||||
// maxLookupHolders caps how many holders a single ChunkLookup response
|
||||
// returns per fid. The directory stores a holder per mount caching the
|
||||
// fid, which on a large hot file can reach fleet size (hundreds). The
|
||||
// fetcher typically only tries the first 1-3 before succeeding, so
|
||||
// returning every entry is pure wire-overhead. LRU ordering ensures the
|
||||
// freshest holders are the ones kept.
|
||||
const maxLookupHolders = 16
|
||||
|
||||
// LookupResult holds the merged lookup response. Fids for which this
|
||||
// receiver is not the owner are split out so the caller can retry.
|
||||
type LookupResult struct {
|
||||
PeersByFid map[string][]LookupHolder
|
||||
NotOwnerFids []string
|
||||
}
|
||||
|
||||
// Lookup returns known holders for the requested fids. Expired edges are
|
||||
// filtered out of the response but NOT deleted here — Sweep handles that
|
||||
// under a write lock on its own schedule, so Lookup can stay read-only and
|
||||
// concurrent. A fid owned by this receiver with no known holders returns
|
||||
// an empty peer list.
|
||||
//
|
||||
// Holders are returned in LRU order (most recently announced first). Since
|
||||
// each Announce stamps a fresh `now + ttl` expiry, sorting by expiry
|
||||
// descending is equivalent to ordering by last-seen descending. Fetchers
|
||||
// should try these in the returned order — the most recent announcer is
|
||||
// the holder most likely to still have the chunk cached, given that each
|
||||
// mount's chunk_cache is itself LRU.
|
||||
func (d *PeerDirectory) Lookup(fids []string, ownerCheck OwnerCheck) LookupResult {
|
||||
now := d.clock()
|
||||
result := LookupResult{
|
||||
PeersByFid: map[string][]LookupHolder{},
|
||||
}
|
||||
|
||||
// Collect candidate holders under the read lock, including the peer
|
||||
// metadata copies — Announce mutates peers[id] in place when an
|
||||
// existing peer re-announces, so reading peers[pid] outside the lock
|
||||
// would race with that write. Sorting happens after the unlock.
|
||||
type candidate struct {
|
||||
peerMetadata
|
||||
expiry time.Time
|
||||
}
|
||||
perFid := make(map[string][]candidate, len(fids))
|
||||
|
||||
d.mu.RLock()
|
||||
for _, fid := range fids {
|
||||
d.lookups.Add(1)
|
||||
if ownerCheck != nil && !ownerCheck(fid) {
|
||||
result.NotOwnerFids = append(result.NotOwnerFids, fid)
|
||||
continue
|
||||
}
|
||||
holders, ok := d.index[fid]
|
||||
if !ok {
|
||||
result.PeersByFid[fid] = nil
|
||||
continue
|
||||
}
|
||||
cands := make([]candidate, 0, len(holders))
|
||||
for pid, expiry := range holders {
|
||||
if !expiry.After(now) {
|
||||
continue // Sweep will clean this up
|
||||
}
|
||||
cands = append(cands, candidate{peerMetadata: d.peers[pid], expiry: expiry})
|
||||
}
|
||||
perFid[fid] = cands
|
||||
}
|
||||
d.mu.RUnlock()
|
||||
|
||||
for fid, cands := range perFid {
|
||||
sort.Slice(cands, func(i, j int) bool {
|
||||
if !cands[i].expiry.Equal(cands[j].expiry) {
|
||||
return cands[i].expiry.After(cands[j].expiry) // newest first
|
||||
}
|
||||
return cands[i].addr < cands[j].addr // deterministic tiebreak
|
||||
})
|
||||
// Cap response at maxLookupHolders. The fetcher typically tries 1-3
|
||||
// holders before succeeding; shipping every holder for hot fids is
|
||||
// wire-overhead without hit-rate gain.
|
||||
if len(cands) > maxLookupHolders {
|
||||
cands = cands[:maxLookupHolders]
|
||||
}
|
||||
out := make([]LookupHolder, len(cands))
|
||||
for i, c := range cands {
|
||||
out[i] = LookupHolder{PeerAddr: c.addr, DataCenter: c.dataCenter, Rack: c.rack}
|
||||
}
|
||||
result.PeersByFid[fid] = out
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// Sweep purges expired edges. Safe to call periodically. Peer slots are
|
||||
// left in place — their count is bounded by fleet size (a few hundred at
|
||||
// most) and never grows faster than the fleet itself, so reclaiming them
|
||||
// isn't worth the bookkeeping.
|
||||
func (d *PeerDirectory) Sweep() int {
|
||||
now := d.clock()
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
n := 0
|
||||
for fid, holders := range d.index {
|
||||
for pid, expiry := range holders {
|
||||
if !expiry.After(now) {
|
||||
delete(holders, pid)
|
||||
d.entryCount--
|
||||
n++
|
||||
}
|
||||
}
|
||||
if len(holders) == 0 {
|
||||
delete(d.index, fid)
|
||||
}
|
||||
}
|
||||
d.evictions.Add(int64(n))
|
||||
return n
|
||||
}
|
||||
|
||||
// Stats exposes counters for observability. O(1) — entryCount is kept in
|
||||
// lock-step with map mutations.
|
||||
func (d *PeerDirectory) Stats() (announces, lookups, rejected, evictions int64, entries int) {
|
||||
d.mu.RLock()
|
||||
defer d.mu.RUnlock()
|
||||
return d.announces.Load(), d.lookups.Load(), d.rejected.Load(), d.evictions.Load(), d.entryCount
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
package mount
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// alwaysOwner is a trivial OwnerCheck used by tests that aren't exercising
|
||||
// ownership rejection.
|
||||
func alwaysOwner(string) bool { return true }
|
||||
|
||||
// ownerSet returns an OwnerCheck that accepts only the listed fids.
|
||||
func ownerSet(fids ...string) OwnerCheck {
|
||||
m := map[string]struct{}{}
|
||||
for _, f := range fids {
|
||||
m[f] = struct{}{}
|
||||
}
|
||||
return func(fid string) bool {
|
||||
_, ok := m[fid]
|
||||
return ok
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerDirectory_AnnounceAndLookup(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
d := newPeerDirectoryWithClock(func() time.Time { return now })
|
||||
|
||||
res := d.Announce("mount-a:18080", "", "r1", []string{"3,a", "3,b"}, 60*time.Second, alwaysOwner)
|
||||
if len(res.Rejected) != 0 {
|
||||
t.Errorf("expected 0 rejected, got %v", res.Rejected)
|
||||
}
|
||||
|
||||
lookup := d.Lookup([]string{"3,a", "3,b", "3,c"}, alwaysOwner)
|
||||
if got := len(lookup.PeersByFid["3,a"]); got != 1 {
|
||||
t.Errorf("3,a: expected 1 holder, got %d", got)
|
||||
}
|
||||
if got := len(lookup.PeersByFid["3,b"]); got != 1 {
|
||||
t.Errorf("3,b: expected 1 holder, got %d", got)
|
||||
}
|
||||
if got := len(lookup.PeersByFid["3,c"]); got != 0 {
|
||||
t.Errorf("3,c (unknown): expected 0 holders, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerDirectory_OwnerRejection(t *testing.T) {
|
||||
d := NewPeerDirectory()
|
||||
|
||||
res := d.Announce("mount-a:18080", "", "", []string{"3,mine", "3,yours"}, time.Minute, ownerSet("3,mine"))
|
||||
if len(res.Rejected) != 1 || res.Rejected[0] != "3,yours" {
|
||||
t.Errorf("expected 3,yours rejected, got %v", res.Rejected)
|
||||
}
|
||||
|
||||
lookup := d.Lookup([]string{"3,mine", "3,yours"}, ownerSet("3,mine"))
|
||||
if got := len(lookup.PeersByFid["3,mine"]); got != 1 {
|
||||
t.Errorf("3,mine should have 1 holder, got %d", got)
|
||||
}
|
||||
if len(lookup.NotOwnerFids) != 1 || lookup.NotOwnerFids[0] != "3,yours" {
|
||||
t.Errorf("expected 3,yours in not-owner list, got %v", lookup.NotOwnerFids)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerDirectory_MultipleHolders(t *testing.T) {
|
||||
d := NewPeerDirectory()
|
||||
d.Announce("mount-a:18080", "", "r1", []string{"3,x"}, time.Minute, alwaysOwner)
|
||||
d.Announce("mount-b:18080", "", "r2", []string{"3,x"}, time.Minute, alwaysOwner)
|
||||
|
||||
holders := d.Lookup([]string{"3,x"}, alwaysOwner).PeersByFid["3,x"]
|
||||
if len(holders) != 2 {
|
||||
t.Fatalf("expected 2 holders, got %d", len(holders))
|
||||
}
|
||||
sort.Slice(holders, func(i, j int) bool { return holders[i].PeerAddr < holders[j].PeerAddr })
|
||||
if holders[0].PeerAddr != "mount-a:18080" || holders[1].PeerAddr != "mount-b:18080" {
|
||||
t.Errorf("unexpected holder addrs: %+v", holders)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerDirectory_RenewExtendsExpiry(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
d := newPeerDirectoryWithClock(func() time.Time { return now })
|
||||
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,a"}, 30*time.Second, alwaysOwner)
|
||||
|
||||
now = now.Add(25 * time.Second)
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,a"}, 30*time.Second, alwaysOwner)
|
||||
|
||||
now = now.Add(10 * time.Second)
|
||||
lookup := d.Lookup([]string{"3,a"}, alwaysOwner)
|
||||
if len(lookup.PeersByFid["3,a"]) != 1 {
|
||||
t.Errorf("renew should keep entry alive, got %d holders", len(lookup.PeersByFid["3,a"]))
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerDirectory_TTLExpiry(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
d := newPeerDirectoryWithClock(func() time.Time { return now })
|
||||
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,a"}, 10*time.Second, alwaysOwner)
|
||||
now = now.Add(15 * time.Second)
|
||||
holders := d.Lookup([]string{"3,a"}, alwaysOwner).PeersByFid["3,a"]
|
||||
if len(holders) != 0 {
|
||||
t.Errorf("expected TTL-expired holder to be gone, got %+v", holders)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerDirectory_Sweep(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
d := newPeerDirectoryWithClock(func() time.Time { return now })
|
||||
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,a"}, 10*time.Second, alwaysOwner)
|
||||
d.Announce("mount-b:18080", "", "", []string{"3,b"}, 60*time.Second, alwaysOwner)
|
||||
|
||||
now = now.Add(30 * time.Second)
|
||||
got := d.Sweep()
|
||||
if got != 1 {
|
||||
t.Errorf("expected 1 swept entry, got %d", got)
|
||||
}
|
||||
_, _, _, _, entries := d.Stats()
|
||||
if entries != 1 {
|
||||
t.Errorf("expected 1 surviving entry, got %d", entries)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeerDirectory_LookupOrdersByRecency guards the LRU contract: the
|
||||
// most recently-announced holder comes first in the lookup response, so
|
||||
// fetchers that try holders in order hit the holder most likely to still
|
||||
// have the chunk cached locally.
|
||||
func TestPeerDirectory_LookupOrdersByRecency(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
d := newPeerDirectoryWithClock(func() time.Time { return now })
|
||||
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner)
|
||||
now = now.Add(10 * time.Second)
|
||||
d.Announce("mount-b:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner)
|
||||
now = now.Add(10 * time.Second)
|
||||
d.Announce("mount-c:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner)
|
||||
|
||||
got := d.Lookup([]string{"3,x"}, alwaysOwner).PeersByFid["3,x"]
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("expected 3 holders, got %d", len(got))
|
||||
}
|
||||
want := []string{"mount-c:18080", "mount-b:18080", "mount-a:18080"}
|
||||
for i, w := range want {
|
||||
if got[i].PeerAddr != w {
|
||||
t.Errorf("holder[%d] = %q, want %q (LRU order)", i, got[i].PeerAddr, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeerDirectory_LookupOrderAfterRenewal ensures renewing an older
|
||||
// holder promotes it to the head — matches LRU touch-on-access semantics
|
||||
// and keeps the freshest liveness signal in front.
|
||||
func TestPeerDirectory_LookupOrderAfterRenewal(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
d := newPeerDirectoryWithClock(func() time.Time { return now })
|
||||
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner)
|
||||
now = now.Add(10 * time.Second)
|
||||
d.Announce("mount-b:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner)
|
||||
now = now.Add(10 * time.Second)
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner)
|
||||
|
||||
got := d.Lookup([]string{"3,x"}, alwaysOwner).PeersByFid["3,x"]
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("expected 2 holders, got %d", len(got))
|
||||
}
|
||||
if got[0].PeerAddr != "mount-a:18080" {
|
||||
t.Errorf("after renewal, head should be mount-a; got %q", got[0].PeerAddr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerDirectory_StatsCountersIncrement(t *testing.T) {
|
||||
d := NewPeerDirectory()
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,a", "3,b"}, time.Minute, alwaysOwner)
|
||||
d.Lookup([]string{"3,a", "3,c"}, alwaysOwner)
|
||||
d.Announce("mount-a:18080", "", "", []string{"3,nope"}, time.Minute, ownerSet("nothing"))
|
||||
|
||||
announces, lookups, rejected, _, _ := d.Stats()
|
||||
if announces != 2 {
|
||||
t.Errorf("announces counter: %d", announces)
|
||||
}
|
||||
if lookups != 2 {
|
||||
t.Errorf("lookups counter: %d", lookups)
|
||||
}
|
||||
if rejected != 1 {
|
||||
t.Errorf("rejected counter: %d", rejected)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeerDirectory_LookupCapsHolderList verifies that a hot fid with
|
||||
// more than maxLookupHolders holders returns only the top-N LRU entries.
|
||||
// The fetcher typically tries only the first 1-3 anyway, so caching
|
||||
// every holder on the wire is overhead.
|
||||
func TestPeerDirectory_LookupCapsHolderList(t *testing.T) {
|
||||
now := time.Unix(1000, 0)
|
||||
d := newPeerDirectoryWithClock(func() time.Time { return now })
|
||||
|
||||
// Announce 32 holders at strictly increasing timestamps so LRU order
|
||||
// is well-defined and deterministic.
|
||||
total := 2 * maxLookupHolders
|
||||
for i := 0; i < total; i++ {
|
||||
addr := fmt.Sprintf("mount-%02d:18080", i)
|
||||
d.Announce(addr, "", "", []string{"3,hot"}, 60*time.Second, alwaysOwner)
|
||||
now = now.Add(time.Millisecond)
|
||||
}
|
||||
|
||||
got := d.Lookup([]string{"3,hot"}, alwaysOwner).PeersByFid["3,hot"]
|
||||
if len(got) != maxLookupHolders {
|
||||
t.Fatalf("expected %d holders in response, got %d", maxLookupHolders, len(got))
|
||||
}
|
||||
// Head should be the most recent (last announced).
|
||||
want := fmt.Sprintf("mount-%02d:18080", total-1)
|
||||
if got[0].PeerAddr != want {
|
||||
t.Errorf("head of list: got %q want %q (LRU order)", got[0].PeerAddr, want)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package mount
|
||||
|
||||
// fakeChunkCache satisfies chunk_cache.ChunkCache with a simple in-memory
|
||||
// map keyed by fid. Just enough to exercise the peer-serve handler paths.
|
||||
type fakeChunkCache struct {
|
||||
chunks map[string][]byte
|
||||
}
|
||||
|
||||
func newFakeChunkCache() *fakeChunkCache {
|
||||
return &fakeChunkCache{chunks: map[string][]byte{}}
|
||||
}
|
||||
|
||||
func (f *fakeChunkCache) Put(fid string, data []byte) {
|
||||
buf := make([]byte, len(data))
|
||||
copy(buf, data)
|
||||
f.chunks[fid] = buf
|
||||
}
|
||||
|
||||
func (f *fakeChunkCache) ReadChunkAt(data []byte, fileId string, offset uint64) (int, error) {
|
||||
b, ok := f.chunks[fileId]
|
||||
if !ok {
|
||||
return 0, nil
|
||||
}
|
||||
if int(offset) >= len(b) {
|
||||
return 0, nil
|
||||
}
|
||||
return copy(data, b[offset:]), nil
|
||||
}
|
||||
|
||||
func (f *fakeChunkCache) SetChunk(fileId string, data []byte) {
|
||||
f.Put(fileId, data)
|
||||
}
|
||||
|
||||
func (f *fakeChunkCache) IsInCache(fileId string, lockNeeded bool) bool {
|
||||
_, ok := f.chunks[fileId]
|
||||
return ok
|
||||
}
|
||||
|
||||
func (f *fakeChunkCache) GetMaxFilePartSizeInCache() uint64 {
|
||||
return 8 * 1024 * 1024
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
package mount
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/mount_peer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util/mem"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
// fetchChunkStreamSize is the frame size used when server-streaming a chunk's
|
||||
// bytes back to a peer. 1 MiB is well above gRPC's 32 KiB preferred frame but
|
||||
// comfortably under the default 4 MiB message cap, so each Recv on the client
|
||||
// returns quickly and the chunk is assembled with ~16 Recv calls for typical
|
||||
// 16 MiB chunks.
|
||||
const fetchChunkStreamSize = 1 * 1024 * 1024
|
||||
|
||||
// maxFetchChunkBytes caps the size of a single FetchChunk read buffer.
|
||||
// The caller's expected_size is untrusted — a misbehaving peer could
|
||||
// claim 1 TiB and OOM the server. 64 MiB is well above any realistic
|
||||
// chunk (SeaweedFS defaults to 16 MiB, filer manifests cap at 32 MiB);
|
||||
// anything larger is treated as invalid input.
|
||||
const maxFetchChunkBytes = 64 * 1024 * 1024
|
||||
|
||||
// PeerGrpcServer is the single mount-to-mount gRPC endpoint. It serves:
|
||||
// - ChunkAnnounce / ChunkLookup — the tier-2 directory RPCs, populated
|
||||
// by inbound announces and queried by inbound lookups. Each handler
|
||||
// is HRW-gated on the caller-side seed view.
|
||||
// - FetchChunk (server stream) — serves bytes from the local
|
||||
// chunk_cache to peers. Replaces the earlier HTTP-only peer-serve
|
||||
// endpoint: one port, one authentication path, one connection pool.
|
||||
//
|
||||
// Transport credentials (TLS/mTLS) are injected via serverOpts. When the
|
||||
// caller passes nothing the server runs plaintext — only appropriate on
|
||||
// a trusted single-host test cluster. Production wiring passes
|
||||
// security.LoadServerTLS from security.toml so cross-host traffic is
|
||||
// authenticated + encrypted.
|
||||
type PeerGrpcServer struct {
|
||||
mount_peer_pb.UnimplementedMountPeerServer
|
||||
|
||||
dir *PeerDirectory
|
||||
cache chunk_cache.ChunkCache
|
||||
ownerFor func(fid string) string // HRW owner predicate on current seeds
|
||||
selfAddr string
|
||||
serverOpts []grpc.ServerOption
|
||||
grpcS *grpc.Server
|
||||
listener net.Listener
|
||||
stopped bool
|
||||
}
|
||||
|
||||
// NewPeerGrpcServer constructs the server. cache is the local chunk_cache
|
||||
// (used to serve FetchChunk); dir is the local directory shard (used to
|
||||
// answer ChunkAnnounce / ChunkLookup); ownerFor returns the HRW owner of a
|
||||
// fid on the current seed view; serverOpts are forwarded to the underlying
|
||||
// gRPC server (use grpc.Creds(...) for TLS/mTLS).
|
||||
func NewPeerGrpcServer(cache chunk_cache.ChunkCache, dir *PeerDirectory, ownerFor func(fid string) string, selfAddr string, serverOpts ...grpc.ServerOption) *PeerGrpcServer {
|
||||
return &PeerGrpcServer{
|
||||
cache: cache,
|
||||
dir: dir,
|
||||
ownerFor: ownerFor,
|
||||
selfAddr: selfAddr,
|
||||
serverOpts: serverOpts,
|
||||
}
|
||||
}
|
||||
|
||||
// Start binds a TCP listener at addr and registers the MountPeer service.
|
||||
func (s *PeerGrpcServer) Start(addr string) error {
|
||||
ln, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("peer grpc listen %s: %w", addr, err)
|
||||
}
|
||||
s.listener = ln
|
||||
s.grpcS = pb.NewGrpcServer(s.serverOpts...)
|
||||
mount_peer_pb.RegisterMountPeerServer(s.grpcS, s)
|
||||
go func() {
|
||||
if err := s.grpcS.Serve(ln); err != nil && err != grpc.ErrServerStopped {
|
||||
glog.Warningf("peer-grpc terminated: %v", err)
|
||||
}
|
||||
}()
|
||||
glog.V(0).Infof("peer-grpc listening on %s", ln.Addr())
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop halts the gRPC server without waiting for in-flight streams.
|
||||
func (s *PeerGrpcServer) Stop() {
|
||||
if s.stopped {
|
||||
return
|
||||
}
|
||||
s.stopped = true
|
||||
if s.grpcS != nil {
|
||||
s.grpcS.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
// Addr returns the bound address (useful when the caller used ":0").
|
||||
func (s *PeerGrpcServer) Addr() string {
|
||||
if s.listener == nil {
|
||||
return ""
|
||||
}
|
||||
return s.listener.Addr().String()
|
||||
}
|
||||
|
||||
// ChunkAnnounce accepts holder entries for fids this mount owns; rejects
|
||||
// others so the caller can retry against the correct owner.
|
||||
func (s *PeerGrpcServer) ChunkAnnounce(ctx context.Context, req *mount_peer_pb.ChunkAnnounceRequest) (*mount_peer_pb.ChunkAnnounceResponse, error) {
|
||||
ttl := time.Duration(req.TtlSeconds) * time.Second
|
||||
res := s.dir.Announce(req.PeerAddr, req.DataCenter, req.Rack, req.FileIds, ttl, s.ownerPredicate)
|
||||
return &mount_peer_pb.ChunkAnnounceResponse{
|
||||
RejectedFileIds: res.Rejected,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ChunkLookup returns known holders for each requested fid in LRU order.
|
||||
func (s *PeerGrpcServer) ChunkLookup(ctx context.Context, req *mount_peer_pb.ChunkLookupRequest) (*mount_peer_pb.ChunkLookupResponse, error) {
|
||||
res := s.dir.Lookup(req.FileIds, s.ownerPredicate)
|
||||
resp := &mount_peer_pb.ChunkLookupResponse{
|
||||
PeersByFid: make(map[string]*mount_peer_pb.PeerSet, len(res.PeersByFid)),
|
||||
NotOwnerFileIds: res.NotOwnerFids,
|
||||
}
|
||||
for fid, holders := range res.PeersByFid {
|
||||
peers := &mount_peer_pb.PeerSet{}
|
||||
for _, h := range holders {
|
||||
peers.Peers = append(peers.Peers, &mount_peer_pb.PeerInfo{
|
||||
PeerAddr: h.PeerAddr,
|
||||
DataCenter: h.DataCenter,
|
||||
Rack: h.Rack,
|
||||
})
|
||||
}
|
||||
resp.PeersByFid[fid] = peers
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// FetchChunk streams bytes of a cached chunk back to the caller. Missing
|
||||
// fid → gRPC NOT_FOUND. Bytes are framed at fetchChunkStreamSize so gRPC's
|
||||
// default 4 MiB message cap does not constrain chunk size.
|
||||
func (s *PeerGrpcServer) FetchChunk(req *mount_peer_pb.FetchChunkRequest, stream mount_peer_pb.MountPeer_FetchChunkServer) error {
|
||||
if s.cache == nil {
|
||||
return status.Error(codes.Unavailable, "chunk cache not configured")
|
||||
}
|
||||
fid := req.FileId
|
||||
if fid == "" {
|
||||
return status.Error(codes.InvalidArgument, "missing file_id")
|
||||
}
|
||||
|
||||
// Size the read buffer to the caller-reported chunk length. The
|
||||
// TieredChunkCache.ReadChunkAt wrapper only returns success when n
|
||||
// equals len(data), so a buffer larger than the actual stored chunk
|
||||
// (e.g. an 8 MiB buffer for a 2 MiB chunk) makes every read look
|
||||
// like a miss. Client sends expected_size from FileChunk.Size so we
|
||||
// allocate exactly the right amount. Fall back to the max-part-size
|
||||
// when the caller left it zero (older clients) and hope the chunk
|
||||
// is that big.
|
||||
//
|
||||
// expected_size is untrusted — cap at maxFetchChunkBytes to prevent
|
||||
// a misbehaving peer from requesting an OOM-sized allocation.
|
||||
if req.ExpectedSize > maxFetchChunkBytes {
|
||||
return status.Errorf(codes.InvalidArgument,
|
||||
"expected_size %d exceeds max %d", req.ExpectedSize, maxFetchChunkBytes)
|
||||
}
|
||||
if req.Length > maxFetchChunkBytes {
|
||||
return status.Errorf(codes.InvalidArgument,
|
||||
"length %d exceeds max %d", req.Length, maxFetchChunkBytes)
|
||||
}
|
||||
readSize := int(req.ExpectedSize)
|
||||
if readSize <= 0 {
|
||||
max := s.cache.GetMaxFilePartSizeInCache()
|
||||
if max == 0 {
|
||||
max = 8 * 1024 * 1024
|
||||
}
|
||||
readSize = int(max)
|
||||
}
|
||||
// mem.Allocate rounds up to the nearest power-of-2 slot backed by a
|
||||
// shared sync.Pool; avoids an allocation per FetchChunk call.
|
||||
buf := mem.Allocate(readSize)
|
||||
defer mem.Free(buf)
|
||||
|
||||
n, err := s.cache.ReadChunkAt(buf, fid, 0)
|
||||
if err != nil || n <= 0 {
|
||||
return status.Errorf(codes.NotFound, "fid %s not cached", fid)
|
||||
}
|
||||
|
||||
// Apply optional offset / length range to the whole-chunk buffer.
|
||||
// Default (both 0) is a full-chunk transfer starting at byte 0,
|
||||
// which is what the fetcher uses today. Non-default is reserved
|
||||
// for future range-read callers.
|
||||
lo := int(req.Offset)
|
||||
if lo < 0 || lo > n {
|
||||
return status.Errorf(codes.OutOfRange, "offset %d outside chunk length %d", lo, n)
|
||||
}
|
||||
hi := n
|
||||
if req.Length > 0 {
|
||||
hi = lo + int(req.Length)
|
||||
if hi > n {
|
||||
hi = n
|
||||
}
|
||||
}
|
||||
|
||||
for off := lo; off < hi; off += fetchChunkStreamSize {
|
||||
end := off + fetchChunkStreamSize
|
||||
if end > hi {
|
||||
end = hi
|
||||
}
|
||||
if sendErr := stream.Send(&mount_peer_pb.FetchChunkResponse{
|
||||
Data: buf[off:end],
|
||||
}); sendErr != nil {
|
||||
return sendErr
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *PeerGrpcServer) ownerPredicate(fid string) bool {
|
||||
if s.ownerFor == nil {
|
||||
return true // no HRW configured → accept all (single-mount mode)
|
||||
}
|
||||
return s.ownerFor(fid) == s.selfAddr
|
||||
}
|
||||
|
||||
// peerDirectorySweepInterval is how often the mount evicts expired
|
||||
// directory entries. Lookup no longer deletes inline (it takes only an
|
||||
// RLock), so this sweeper is the sole memory reclamation path.
|
||||
const peerDirectorySweepInterval = 60 * time.Second
|
||||
|
||||
// runPeerDirectorySweeper runs until stopCh closes. Stopping it on
|
||||
// unmount prevents the goroutine from outliving the WFS, which matters
|
||||
// for tests and for any embedded use of NewSeaweedFileSystem where the
|
||||
// filesystem object is recreated without a process exit.
|
||||
func (wfs *WFS) runPeerDirectorySweeper(stopCh <-chan struct{}) {
|
||||
ticker := time.NewTicker(peerDirectorySweepInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-stopCh:
|
||||
return
|
||||
case <-ticker.C:
|
||||
dir := wfs.peerDirectory
|
||||
if dir == nil {
|
||||
return
|
||||
}
|
||||
if evicted := dir.Sweep(); evicted > 0 {
|
||||
glog.V(2).Infof("peer directory: evicted %d expired entries", evicted)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
package mount
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/mount_peer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
func newTestPeerGrpc(t *testing.T, cache chunk_cache.ChunkCache, ownerFor func(fid string) string, selfAddr string) (mount_peer_pb.MountPeerClient, func()) {
|
||||
t.Helper()
|
||||
dir := NewPeerDirectory()
|
||||
srv := NewPeerGrpcServer(cache, dir, ownerFor, selfAddr)
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
srv.listener = ln
|
||||
srv.grpcS = grpc.NewServer()
|
||||
mount_peer_pb.RegisterMountPeerServer(srv.grpcS, srv)
|
||||
go srv.grpcS.Serve(ln)
|
||||
|
||||
conn, err := grpc.NewClient(ln.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
if err != nil {
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
client := mount_peer_pb.NewMountPeerClient(conn)
|
||||
cleanup := func() {
|
||||
conn.Close()
|
||||
srv.Stop()
|
||||
}
|
||||
return client, cleanup
|
||||
}
|
||||
|
||||
func TestPeerGrpcServer_AnnounceAndLookup(t *testing.T) {
|
||||
self := "self:18080"
|
||||
client, cleanup := newTestPeerGrpc(t, nil, func(fid string) string {
|
||||
return self
|
||||
}, self)
|
||||
defer cleanup()
|
||||
|
||||
_, err := client.ChunkAnnounce(context.Background(), &mount_peer_pb.ChunkAnnounceRequest{
|
||||
FileIds: []string{"3,a", "3,b"},
|
||||
PeerAddr: "holder:18080",
|
||||
TtlSeconds: 60,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("announce: %v", err)
|
||||
}
|
||||
|
||||
resp, err := client.ChunkLookup(context.Background(), &mount_peer_pb.ChunkLookupRequest{
|
||||
FileIds: []string{"3,a", "3,missing"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("lookup: %v", err)
|
||||
}
|
||||
if got := len(resp.PeersByFid["3,a"].Peers); got != 1 {
|
||||
t.Errorf("3,a: expected 1 holder, got %d", got)
|
||||
}
|
||||
if _, ok := resp.PeersByFid["3,missing"]; !ok {
|
||||
t.Errorf("3,missing should have a (nil/empty) peer set entry")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeerGrpcServer_OwnerMismatch(t *testing.T) {
|
||||
client, cleanup := newTestPeerGrpc(t, nil, func(fid string) string {
|
||||
return "some-other:18080"
|
||||
}, "self:18080")
|
||||
defer cleanup()
|
||||
|
||||
ann, err := client.ChunkAnnounce(context.Background(), &mount_peer_pb.ChunkAnnounceRequest{
|
||||
FileIds: []string{"3,x"},
|
||||
PeerAddr: "holder:18080",
|
||||
TtlSeconds: 60,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("announce: %v", err)
|
||||
}
|
||||
if len(ann.RejectedFileIds) != 1 || ann.RejectedFileIds[0] != "3,x" {
|
||||
t.Errorf("expected 3,x rejected, got %v", ann.RejectedFileIds)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeerGrpcServer_FetchChunk_StreamsHit exercises the new byte-transfer
|
||||
// path: a cached chunk returned as multiple gRPC frames, concatenated by
|
||||
// the caller into the original payload.
|
||||
func TestPeerGrpcServer_FetchChunk_StreamsHit(t *testing.T) {
|
||||
cache := newFakeChunkCache()
|
||||
// A payload deliberately larger than fetchChunkStreamSize to force
|
||||
// multiple Send() calls.
|
||||
payload := make([]byte, fetchChunkStreamSize*2+123)
|
||||
for i := range payload {
|
||||
payload[i] = byte(i & 0xff)
|
||||
}
|
||||
cache.Put("3,stream", payload)
|
||||
|
||||
client, cleanup := newTestPeerGrpc(t, cache, nil, "self:18080")
|
||||
defer cleanup()
|
||||
|
||||
stream, err := client.FetchChunk(context.Background(), &mount_peer_pb.FetchChunkRequest{FileId: "3,stream"})
|
||||
if err != nil {
|
||||
t.Fatalf("FetchChunk: %v", err)
|
||||
}
|
||||
var got []byte
|
||||
for {
|
||||
resp, err := stream.Recv()
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("Recv: %v", err)
|
||||
}
|
||||
got = append(got, resp.Data...)
|
||||
}
|
||||
if len(got) != len(payload) {
|
||||
t.Fatalf("got %d bytes, want %d", len(got), len(payload))
|
||||
}
|
||||
for i, b := range got {
|
||||
if b != payload[i] {
|
||||
t.Fatalf("byte mismatch at offset %d: got %02x want %02x", i, b, payload[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestPeerGrpcServer_FetchChunk_NotFound verifies that a miss returns
|
||||
// gRPC NOT_FOUND so the client can distinguish miss from transport error.
|
||||
func TestPeerGrpcServer_FetchChunk_NotFound(t *testing.T) {
|
||||
cache := newFakeChunkCache()
|
||||
client, cleanup := newTestPeerGrpc(t, cache, nil, "self:18080")
|
||||
defer cleanup()
|
||||
|
||||
stream, err := client.FetchChunk(context.Background(), &mount_peer_pb.FetchChunkRequest{FileId: "3,missing"})
|
||||
if err != nil {
|
||||
t.Fatalf("FetchChunk dial: %v", err)
|
||||
}
|
||||
_, err = stream.Recv()
|
||||
if err == nil {
|
||||
t.Fatalf("expected error on missing fid, got nil")
|
||||
}
|
||||
if status.Code(err) != codes.NotFound {
|
||||
t.Errorf("expected NotFound code, got %v", status.Code(err))
|
||||
}
|
||||
}
|
||||
+49
-7
@@ -140,6 +140,9 @@ type WFS struct {
|
||||
posixLocks *PosixLockTable
|
||||
rdmaClient *RDMAMountClient
|
||||
peerRegistrar *PeerRegistrar
|
||||
peerDirectory *PeerDirectory
|
||||
peerGrpcServer *PeerGrpcServer
|
||||
peerDirectoryStop chan struct{} // closed on unmount to stop the sweeper goroutine
|
||||
FilerConf *filer.FilerConf
|
||||
filerClient *wdclient.FilerClient // Cached volume location client
|
||||
refreshMu sync.Mutex
|
||||
@@ -337,6 +340,17 @@ func NewSeaweedFileSystem(option *Option) *WFS {
|
||||
if wfs.rdmaClient != nil {
|
||||
wfs.rdmaClient.Close()
|
||||
}
|
||||
if wfs.peerGrpcServer != nil {
|
||||
wfs.peerGrpcServer.Stop()
|
||||
}
|
||||
if wfs.peerDirectoryStop != nil {
|
||||
select {
|
||||
case <-wfs.peerDirectoryStop:
|
||||
// already closed
|
||||
default:
|
||||
close(wfs.peerDirectoryStop)
|
||||
}
|
||||
}
|
||||
if wfs.peerRegistrar != nil {
|
||||
wfs.peerRegistrar.Stop()
|
||||
}
|
||||
@@ -360,13 +374,13 @@ func NewSeaweedFileSystem(option *Option) *WFS {
|
||||
}
|
||||
|
||||
// Peer chunk sharing: register with every configured filer's mount
|
||||
// registry. PR 3 resolved the advertise address; pass it through so
|
||||
// the registrar heartbeats a reachable identity rather than a wildcard
|
||||
// bind. Broadcasting to the full filer set is what lets mounts pointing
|
||||
// at different filers see each other — each filer's registry is
|
||||
// in-memory and there is no filer-to-filer sync. The gRPC server that
|
||||
// serves ChunkAnnounce/Lookup/FetchChunk is started later (PR 5); until
|
||||
// then the registrar is a no-op beyond heartbeats.
|
||||
// registry + start the single gRPC server that handles ChunkAnnounce /
|
||||
// ChunkLookup / FetchChunk. Broadcasting registration to the full
|
||||
// filer set is what lets mounts pointing at different filers see
|
||||
// each other — each filer's registry is in-memory with no
|
||||
// filer-to-filer sync, so the registrar reconstructs the union
|
||||
// client-side. One port, one identity — the advertise address
|
||||
// resolved in PR #3 is used for everything.
|
||||
if option.PeerEnabled {
|
||||
selfAddr, err := ResolvePeerAdvertiseAddr(option.PeerListen, option.PeerAdvertise)
|
||||
if err != nil {
|
||||
@@ -386,6 +400,34 @@ func NewSeaweedFileSystem(option *Option) *WFS {
|
||||
if err := wfs.peerRegistrar.Start(context.Background()); err != nil {
|
||||
glog.Warningf("peer registrar start: %v", err)
|
||||
}
|
||||
|
||||
wfs.peerDirectory = NewPeerDirectory()
|
||||
// Wire TLS/mTLS from security.toml's grpc.mount section so
|
||||
// cross-host peer RPCs are authenticated + encrypted. When
|
||||
// the section is empty both options come back nil and the
|
||||
// server runs plaintext — intentional for dev/test.
|
||||
peerTLSCreds, peerTLSVerify := security.LoadServerTLS(util.GetViper(), "grpc.mount")
|
||||
var peerServerOpts []grpc.ServerOption
|
||||
if peerTLSCreds != nil {
|
||||
peerServerOpts = append(peerServerOpts, peerTLSCreds)
|
||||
}
|
||||
if peerTLSVerify != nil {
|
||||
peerServerOpts = append(peerServerOpts, peerTLSVerify)
|
||||
}
|
||||
wfs.peerGrpcServer = NewPeerGrpcServer(
|
||||
wfs.chunkCache,
|
||||
wfs.peerDirectory,
|
||||
wfs.peerRegistrar.OwnerFor,
|
||||
selfAddr,
|
||||
peerServerOpts...,
|
||||
)
|
||||
if err := wfs.peerGrpcServer.Start(option.PeerListen); err != nil {
|
||||
glog.Warningf("peer grpc start: %v", err)
|
||||
wfs.peerGrpcServer = nil
|
||||
} else {
|
||||
wfs.peerDirectoryStop = make(chan struct{})
|
||||
go wfs.runPeerDirectorySweeper(wfs.peerDirectoryStop)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user