mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-07-19 14:32:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 's3-lock-ring-view' into s3-route-by-key

Browse Source 
# Conflicts:
#	weed/s3api/s3api_server.go
This commit is contained in:
Chris Lu
2026-05-22 23:09:37 -07:00
parent 20934cc854 71de73bcd4
commit ee56d1a5cd
1 changed files with 21 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/s3api/s3api_server.go
+21 -13
View File
@@ -159,6 +159,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
// 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)
@@ -170,6 +171,21 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
clientHost = util.DetectedHostAddress()
}
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())
@@ -270,20 +286,12 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
}
if len(option.Filers) > 0 {
objectWriteLockClient := cluster.NewLockClient(option.GrpcDialOption, option.Filers[0])
s3ApiServer.objectWriteLockClient = objectWriteLockClient
// Mirror the master's lock-ring view so each object lock dials the key's
// primary filer directly instead of forwarding through the seed filer.
// The masterClient already filters updates to this server's filer group.
if masterClient != nil {
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)
})
// 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)