From f037fc4dce66fc27c675e8d7f6958b7ec67ce0e9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 24 May 2026 00:40:43 -0700 Subject: [PATCH] s3: dial the object lock's primary filer directly (#9626) * s3: dial the object lock's primary filer directly The S3 object write lock builds a fresh short-lived lock per write, each starting at the seed filer. When the seed isn't the key's hash-ring primary the filer forwards the request to the primary, and in multi-cluster setups that forward crosses clusters on every write. Give the lock client a view of the filer lock ring, fed by the master's LockRingUpdate broadcasts the gateway already receives, so it dials the primary directly. The view tracks filer membership by version; a stale view stays correct because the filer still forwards as a fallback. Also send the initial ring snapshot to S3 clients, not just filers. * s3: subscribe to lock-ring updates before starting the master loop The master delivers the initial LockRingUpdate once, on connect. Registering the callback after KeepConnectedToMaster started left a window where that first update could arrive before the handler was set and be dropped, delaying the ring view until the next membership change. Build the lock client and register the callback in the masters block before launching the loop; the filers block reuses that client (or creates a plain one when no masters are configured). * lock_manager: build the hash ring in a deterministic server order rebuildRing ranged over the server set (a map), whose iteration order is randomized per process. On a vnode hash collision the last writer into vnodeToServer wins, so two nodes holding the same server set could resolve the collision to different servers and disagree on the primary for keys near that slot. Now that the S3 gateway also computes PrimaryForKey, such a disagreement would route the same key to different filers and defeat per-path serialization. Iterate the servers in sorted order so the ring is identical on every node with the same set, regardless of discovery order. * lock_manager: skip redundant ring rebuilds, trim comments SetRing now ignores a non-zero version at or below the current one once a ring exists, so repeated LockRingUpdate broadcasts on reconnect no longer rebuild the ring. * s3: hold the lock-ring client on the server for route-by-key Store the object-write lock client on S3ApiServer so handlers can resolve a key's owner filer via PrimaryForKey. --- weed/cluster/lock_client.go | 58 ++++++++++++- weed/cluster/lock_client_test.go | 92 +++++++++++++++++++++ weed/cluster/lock_manager/hash_ring.go | 8 ++ weed/cluster/lock_manager/hash_ring_test.go | 35 ++++++++ weed/s3api/s3api_server.go | 29 ++++++- weed/server/master_grpc_server.go | 8 +- 6 files changed, 224 insertions(+), 6 deletions(-) create mode 100644 weed/cluster/lock_client_test.go diff --git a/weed/cluster/lock_client.go b/weed/cluster/lock_client.go index d84f51bef..fa4eea934 100644 --- a/weed/cluster/lock_client.go +++ b/weed/cluster/lock_client.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync" "sync/atomic" "time" @@ -19,6 +20,14 @@ type LockClient struct { maxLockDuration time.Duration sleepDuration time.Duration seedFiler pb.ServerAddress + + // ring is an optional client-side view of the filer lock hash ring. When + // populated, a new lock starts at the key's primary filer instead of the + // seed filer, avoiding the seed->primary forward hop. A stale view stays + // correct: the filer forwards to the real primary as a fallback. + ringMu sync.RWMutex + ring *lock_manager.HashRing + ringVersion int64 } func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) *LockClient { @@ -30,6 +39,49 @@ func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) * } } +// SetRing mirrors the master's LockRingUpdate so the client computes the same +// primary the filers do. A non-zero version at or below the current one is +// ignored once a ring exists, dropping reordered and redundant broadcasts; +// version 0 always applies (bootstrap). +func (lc *LockClient) SetRing(servers []pb.ServerAddress, version int64) { + lc.ringMu.Lock() + defer lc.ringMu.Unlock() + if version != 0 && version <= lc.ringVersion && lc.ring != nil { + return + } + lc.ringVersion = version + if lc.ring == nil { + lc.ring = lock_manager.NewHashRing(lock_manager.DefaultVnodeCount) + } + lc.ring.SetServers(servers) +} + +// hostForKey returns the filer that should own key per the current ring view, +// falling back to the seed filer when no view has been received yet. +func (lc *LockClient) hostForKey(key string) pb.ServerAddress { + lc.ringMu.RLock() + defer lc.ringMu.RUnlock() + if lc.ring == nil { + return lc.seedFiler + } + if primary := lc.ring.GetPrimary(key); primary != "" { + return primary + } + return lc.seedFiler +} + +// PrimaryForKey returns the ring owner for key, or "" before any ring arrives. +// Unlike hostForKey it does not fall back to the seed, so a route-by-key caller +// stays on the distributed lock until the ring is known. +func (lc *LockClient) PrimaryForKey(key string) pb.ServerAddress { + lc.ringMu.RLock() + defer lc.ringMu.RUnlock() + if lc.ring == nil { + return "" + } + return lc.ring.GetPrimary(key) +} + type LiveLock struct { key string renewToken string @@ -51,7 +103,7 @@ type LiveLock struct { func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLock) { lock = &LiveLock{ key: key, - hostFiler: lc.seedFiler, + hostFiler: lc.hostForKey(key), cancelCh: make(chan struct{}), expireAtNs: time.Now().Add(5 * time.Second).UnixNano(), grpcDialOption: lc.grpcDialOption, @@ -72,7 +124,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D } lock := &LiveLock{ key: key, - hostFiler: lc.seedFiler, + hostFiler: lc.hostForKey(key), cancelCh: make(chan struct{}), expireAtNs: time.Now().Add(lockTTL).UnixNano(), grpcDialOption: lc.grpcDialOption, @@ -110,7 +162,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerChange func(newLockOwner string), lockTTL time.Duration) (lock *LiveLock) { lock = &LiveLock{ key: key, - hostFiler: lc.seedFiler, + hostFiler: lc.hostForKey(key), cancelCh: make(chan struct{}), expireAtNs: time.Now().Add(lockTTL).UnixNano(), grpcDialOption: lc.grpcDialOption, diff --git a/weed/cluster/lock_client_test.go b/weed/cluster/lock_client_test.go new file mode 100644 index 000000000..76ba02157 --- /dev/null +++ b/weed/cluster/lock_client_test.go @@ -0,0 +1,92 @@ +package cluster + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" + "github.com/seaweedfs/seaweedfs/weed/pb" +) + +// The gateway must resolve a lock key to the same primary the filers do, +// otherwise it dials the wrong filer and the lock still gets forwarded. Both +// sides use the same HashRing over the same server set, so for every key the +// client's hostForKey must equal the filer ring's GetPrimary. +func TestLockClientHostMatchesFilerRing(t *testing.T) { + servers := []pb.ServerAddress{ + "filer-a:8888", "filer-b:8888", "filer-c:8888", "filer-d:8888", + } + + filerRing := lock_manager.NewHashRing(lock_manager.DefaultVnodeCount) + filerRing.SetServers(servers) + + lc := NewLockClient(nil, "seed:8888") + lc.SetRing(servers, 1) + + for _, key := range []string{ + "s3.object.write:/buckets/b/obj-0", + "s3.object.write:/buckets/b/obj-1", + "s3.object.write:/buckets/b/obj-2", + "s3.object.write:/buckets/gosbench-0/w0obj-kilo-0877", + "some/other/key", + } { + if got, want := lc.hostForKey(key), filerRing.GetPrimary(key); got != want { + t.Errorf("key %q: client host %q != filer primary %q", key, got, want) + } + } +} + +// Without a ring view, the client falls back to the seed filer (which the filer +// forwards from), preserving the pre-optimization behavior. +func TestLockClientHostFallsBackToSeed(t *testing.T) { + lc := NewLockClient(nil, "seed:8888") + if got := lc.hostForKey("any-key"); got != "seed:8888" { + t.Errorf("expected seed fallback, got %q", got) + } + + // An empty ring (no members yet) also falls back to the seed. + lc.SetRing(nil, 1) + if got := lc.hostForKey("any-key"); got != "seed:8888" { + t.Errorf("expected seed fallback on empty ring, got %q", got) + } +} + +// A stale (older-version) update must not regress a newer ring view, while +// version 0 always applies as a bootstrap. +func TestLockClientSetRingVersionGuard(t *testing.T) { + lc := NewLockClient(nil, "seed:8888") + + newer := []pb.ServerAddress{"filer-a:8888", "filer-b:8888"} + lc.SetRing(newer, 10) + primaryAt10 := lc.hostForKey("k") + + // Older version is ignored. + lc.SetRing([]pb.ServerAddress{"filer-z:8888"}, 5) + if got := lc.hostForKey("k"); got != primaryAt10 { + t.Errorf("stale update applied: host changed to %q", got) + } + + // version 0 is always accepted. + lc.SetRing([]pb.ServerAddress{"filer-z:8888"}, 0) + if got := lc.hostForKey("k"); got != "filer-z:8888" { + t.Errorf("bootstrap update not applied, got %q", got) + } +} + +// PrimaryForKey returns "" before any ring is received (so a route-by-key +// caller falls back to the distributed lock) and the ring owner afterwards, +// unlike hostForKey which falls back to the seed. +func TestLockClientPrimaryForKey(t *testing.T) { + lc := NewLockClient(nil, "seed:8888") + if got := lc.PrimaryForKey("k"); got != "" { + t.Errorf("expected empty before ring, got %q", got) + } + + lc.SetRing([]pb.ServerAddress{"filer-a:8888", "filer-b:8888"}, 1) + got := lc.PrimaryForKey("k") + if got == "" { + t.Fatal("expected an owner after ring set") + } + if got != lc.hostForKey("k") { + t.Errorf("PrimaryForKey %q disagrees with hostForKey %q", got, lc.hostForKey("k")) + } +} diff --git a/weed/cluster/lock_manager/hash_ring.go b/weed/cluster/lock_manager/hash_ring.go index 6e543a681..4f3855503 100644 --- a/weed/cluster/lock_manager/hash_ring.go +++ b/weed/cluster/lock_manager/hash_ring.go @@ -145,7 +145,15 @@ func (hr *HashRing) rebuildRing() { hr.vnodeToServer = make(map[uint32]pb.ServerAddress, len(hr.servers)*hr.vnodeCount) hr.sortedHashes = make([]uint32, 0, len(hr.servers)*hr.vnodeCount) + // Sort so a vnode-hash collision resolves to the same server on every node; + // map iteration order alone is randomized per process. + servers := make([]pb.ServerAddress, 0, len(hr.servers)) for server := range hr.servers { + servers = append(servers, server) + } + sort.Slice(servers, func(i, j int) bool { return servers[i] < servers[j] }) + + for _, server := range servers { for i := 0; i < hr.vnodeCount; i++ { vnodeKey := vnodeKeyFor(server, i) hash := hashKey(vnodeKey) diff --git a/weed/cluster/lock_manager/hash_ring_test.go b/weed/cluster/lock_manager/hash_ring_test.go index 6a7dabe2e..3bca6d1a2 100644 --- a/weed/cluster/lock_manager/hash_ring_test.go +++ b/weed/cluster/lock_manager/hash_ring_test.go @@ -171,3 +171,38 @@ func TestHashRing_GetPrimary(t *testing.T) { primary, _ := hr.GetPrimaryAndBackup("mykey") assert.Equal(t, primary, hr.GetPrimary("mykey")) } + +// The ring must be identical on every node holding the same server set, +// regardless of the order servers were added or supplied. Build rings several +// ways and assert they agree on the primary for a wide range of keys. +func TestHashRing_OrderIndependent(t *testing.T) { + servers := []pb.ServerAddress{ + "filer-a:8888", "filer-b:8888", "filer-c:8888", "filer-d:8888", "filer-e:8888", + } + + bySet := NewHashRing(50) + bySet.SetServers(servers) + + byReverse := NewHashRing(50) + rev := append([]pb.ServerAddress(nil), servers...) + for i, j := 0, len(rev)-1; i < j; i, j = i+1, j-1 { + rev[i], rev[j] = rev[j], rev[i] + } + byReverse.SetServers(rev) + + byAdd := NewHashRing(50) + for _, s := range []pb.ServerAddress{"filer-c:8888", "filer-e:8888", "filer-a:8888", "filer-d:8888", "filer-b:8888"} { + byAdd.AddServer(s) + } + + for i := 0; i < 5000; i++ { + key := fmt.Sprintf("s3.object.write:/buckets/b/obj-%d", i) + p := bySet.GetPrimary(key) + if got := byReverse.GetPrimary(key); got != p { + t.Fatalf("reverse-order ring disagrees on %q: %s vs %s", key, got, p) + } + if got := byAdd.GetPrimary(key); got != p { + t.Fatalf("add-order ring disagrees on %q: %s vs %s", key, got, p) + } + } +} diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index b3e6d9d25..2d18f76d5 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -25,6 +25,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/iam/policy" "github.com/seaweedfs/seaweedfs/weed/iam/sts" "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" @@ -95,6 +96,8 @@ type S3ApiServer struct { stsHandlers *STSHandlers // STS HTTP handlers for AssumeRoleWithWebIdentity cipher bool // encrypt data on volume servers newObjectWriteLock func(bucket, object string) objectWriteLock + // objectWriteLockClient resolves a key's owner filer for route-by-key. + objectWriteLockClient *cluster.LockClient // Shared ReaderCache used by the S3 GET streaming path. It lives for the // lifetime of the server so that concurrent and repeat reads share a // single in-flight download per chunk, and so that no per-request @@ -152,6 +155,8 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl // Uses the battle-tested vidMap with filer-based lookups // Supports multiple filer addresses with automatic failover for high availability var filerClient *wdclient.FilerClient + var masterClient *wdclient.MasterClient + var objectWriteLockClient *cluster.LockClient if len(option.Masters) > 0 { // Enable filer discovery via master masterMap := make(map[string]pb.ServerAddress) @@ -162,7 +167,22 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl if clientHost == "0.0.0.0" || clientHost == "" { clientHost = util.DetectedHostAddress() } - masterClient := wdclient.NewMasterClient(option.GrpcDialOption, option.FilerGroup, cluster.S3Type, pb.ServerAddress(util.JoinHostPort(clientHost, option.GrpcPort)), "", "", *pb.NewServiceDiscoveryFromMap(masterMap)) + masterClient = wdclient.NewMasterClient(option.GrpcDialOption, option.FilerGroup, cluster.S3Type, pb.ServerAddress(util.JoinHostPort(clientHost, option.GrpcPort)), "", "", *pb.NewServiceDiscoveryFromMap(masterMap)) + // Build the object-write lock client and subscribe to the master's + // lock-ring updates BEFORE starting the master loop, so the initial + // LockRingUpdate sent on connect isn't dropped (the master only delivers + // it once per connect). The masterClient already filters updates to this + // server's filer group. + if len(option.Filers) > 0 { + objectWriteLockClient = cluster.NewLockClient(option.GrpcDialOption, option.Filers[0]) + masterClient.SetOnLockRingUpdateFn(func(update *master_pb.LockRingUpdate) { + servers := make([]pb.ServerAddress, 0, len(update.Servers)) + for _, s := range update.Servers { + servers = append(servers, pb.ServerAddress(s)) + } + objectWriteLockClient.SetRing(servers, update.Version) + }) + } // Start the master client connection loop - required for GetMaster() to work go masterClient.KeepConnectedToMaster(context.Background()) @@ -263,7 +283,12 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl } if len(option.Filers) > 0 { - objectWriteLockClient := cluster.NewLockClient(option.GrpcDialOption, option.Filers[0]) + // Reuse the lock client built in the masters block (already subscribed to + // ring updates); create a plain one when no masters are configured. + if objectWriteLockClient == nil { + objectWriteLockClient = cluster.NewLockClient(option.GrpcDialOption, option.Filers[0]) + } + s3ApiServer.objectWriteLockClient = objectWriteLockClient s3ApiServer.newObjectWriteLock = func(bucket, object string) objectWriteLock { lockKey := fmt.Sprintf("s3.object.write:%s", s3ApiServer.toFilerPath(bucket, object)) owner := fmt.Sprintf("s3api-%d", s3ApiServer.randomClientId) diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index fd7709d67..aef686959 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -393,7 +393,13 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ } func (ms *MasterServer) initialLockRingUpdate(clientType string, filerGroup string) *master_pb.KeepConnectedResponse { - if clientType != cluster.FilerType || ms.LockRingManager == nil { + if ms.LockRingManager == nil { + return nil + } + // Filers are ring members; S3 gateways are lock clients that need the same + // view to dial a key's primary directly. Both get the initial snapshot; + // later membership changes already broadcast to every connected client. + if clientType != cluster.FilerType && clientType != cluster.S3Type { return nil }