From d35c4b3d2df1e23ad69396eb6739c2b4f04b51a3 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 7 Jul 2026 12:42:46 -0700 Subject: [PATCH] s3: fail over routed object writes when the owner filer is unreachable (#10251) * s3: fail over routed object writes when the owner filer is unreachable A routed object write (multipart completion, PUT, delete, versioned finalize, metadata replace) dialed the ring-selected owner filer directly with no failover. After a filer restarts onto a new address the lock ring can still name the old one, so every routed write hangs on the dead address until the gateway is restarted; CompleteMultipartUpload in particular exceeds client timeouts. Route the transaction through withFilerClientFailover, skipping an owner that recently failed, so a live filer forwards it to the real owner by route_key. Mirrors the read path's getObjectEntryRoutedByKey. * s3: fail over bucket-config writes when the owner filer is unreachable patchBucketEntry dialed the bucket's ring owner directly, so a restarted filer's stale ring address hung every bucket-config write (versioning, lifecycle, object lock, ownership, ACL, policy, CORS) - the same failure as routed object writes. Route it through objectTxnOnFiler so it skips an unreachable owner and a live filer forwards by route_key. --- weed/s3api/s3api_bucket_config.go | 12 +-- weed/s3api/s3api_object_routed_write.go | 20 ++++- weed/s3api/s3api_object_routed_write_test.go | 89 ++++++++++++++++++++ 3 files changed, 113 insertions(+), 8 deletions(-) diff --git a/weed/s3api/s3api_bucket_config.go b/weed/s3api/s3api_bucket_config.go index 3e8fe3706..6a8bdd09b 100644 --- a/weed/s3api/s3api_bucket_config.go +++ b/weed/s3api/s3api_bucket_config.go @@ -17,7 +17,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/kms" - "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/cors" @@ -587,8 +586,7 @@ func (s3a *S3ApiServer) patchBucketEntry(bucket string, m *filer_pb.ObjectMutati RouteKey: objectWriteRouteKeyPrefix + bucketPath, Mutations: []*filer_pb.ObjectMutation{m}, } - txn := func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.ObjectTransaction(context.Background(), req) + respErr := func(resp *filer_pb.ObjectTransactionResponse, err error) error { if err != nil { return err } @@ -598,11 +596,13 @@ func (s3a *S3ApiServer) patchBucketEntry(bucket string, m *filer_pb.ObjectMutati return nil } if s3a.objectWriteLockClient != nil { - if owner := s3a.objectWriteLockClient.PrimaryForKey(objectWriteRouteKeyPrefix + bucketPath); owner != "" { - return pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, txn) + if owner := s3a.objectWriteLockClient.PrimaryForKey(req.RouteKey); owner != "" { + return respErr(s3a.objectTxnOnFiler(owner, req)) } } - return s3a.WithFilerClient(false, txn) + return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return respErr(client.ObjectTransaction(context.Background(), req)) + }) } // isVersioningEnabled checks if versioning is enabled for a bucket (with caching) diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index af0b0b19f..5adf732b4 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -148,13 +148,29 @@ func singleStrongETag(v string) (string, bool) { return strings.Trim(v, `"`), true } +// objectTxnOnFiler applies a routed transaction, preferring the object's owner +// filer but failing over to a live filer when the owner is unreachable, so a +// restarted filer's stale ring address (e.g. a rolled K8s pod's old IP) can't hang +// the write. The reached filer forwards to the real owner by route_key, keeping its +// per-path lock authoritative. Mirrors getObjectEntryRoutedByKey on the read path. func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { var resp *filer_pb.ObjectTransactionResponse - err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + txn := func(client filer_pb.SeaweedFilerClient) error { var e error resp, e = client.ObjectTransaction(context.Background(), req) return e - }) + } + if s3a.filerClient == nil { + err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, txn) + return resp, err + } + // Skip an owner whose recent write hit a transport error so a dead/stale + // address isn't re-dialed every request. + preferred := owner + if s3a.ownerRecentlyUnreachable(owner) { + preferred = "" + } + err := s3a.withFilerClientFailover(preferred, false, txn) return resp, err } diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index a3fc71181..0e47b1559 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -1,13 +1,21 @@ package s3api import ( + "context" + "fmt" + "net" "net/http" + "sync/atomic" "testing" "time" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "github.com/seaweedfs/seaweedfs/weed/wdclient" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" ) func reqWith(headers map[string]string) *http.Request { @@ -242,6 +250,87 @@ func TestSingleStrongETag(t *testing.T) { } } +// fakeTxnFiler is a minimal SeaweedFiler gRPC server that records ObjectTransaction +// calls, standing in for a live filer that a routed write fails over to. +type fakeTxnFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + calls int32 +} + +func (f *fakeTxnFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + atomic.AddInt32(&f.calls, 1) + return &filer_pb.ObjectTransactionResponse{}, nil +} + +// startFakeTxnFiler serves impl on a random localhost port and returns the S3-style +// filer address whose ToGrpcAddress resolves back to that port. +func startFakeTxnFiler(t *testing.T, impl *fakeTxnFiler) pb.ServerAddress { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + srv := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(srv, impl) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + port := lis.Addr().(*net.TCPAddr).Port + return pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", port)) +} + +// closedFilerAddress returns an address whose gRPC port has nothing listening, +// modeling a filer whose ring address is stale after a pod restart. +func closedFilerAddress(t *testing.T) pb.ServerAddress { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := lis.Addr().(*net.TCPAddr).Port + lis.Close() + return pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", port)) +} + +// A routed write whose owner address is dead fails over to a live filer (which +// forwards to the real owner by route_key) instead of hanging on the dead owner, +// so an object write survives a filer pod IP change without an S3 gateway restart. +func TestObjectTxnFailsOverStaleOwner(t *testing.T) { + live := &fakeTxnFiler{} + liveAddr := startFakeTxnFiler(t, live) + deadOwner := closedFilerAddress(t) + + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + s3a := &S3ApiServer{ + option: &S3ApiServerOption{GrpcDialOption: dialOption}, + filerClient: wdclient.NewFilerClient([]pb.ServerAddress{liveAddr}, dialOption, ""), + } + req := &filer_pb.ObjectTransactionRequest{LockKey: "/buckets/b/o", RouteKey: "s3.object.write:/buckets/b/o"} + + // Owner not yet flagged: the first attempt dials it, fails fast, then fails + // over to the live filer and flags the owner unreachable. + resp, err := s3a.objectTxnOnFiler(deadOwner, req) + if err != nil { + t.Fatalf("expected failover success, got %v", err) + } + if resp == nil || resp.Error != "" { + t.Fatalf("unexpected response %+v", resp) + } + if got := atomic.LoadInt32(&live.calls); got != 1 { + t.Fatalf("live filer calls = %d, want 1", got) + } + if !s3a.ownerRecentlyUnreachable(deadOwner) { + t.Fatal("dead owner should be flagged unreachable after the failed dial") + } + + // Once flagged, the owner is skipped entirely and the write still lands. + if _, err := s3a.objectTxnOnFiler(deadOwner, req); err != nil { + t.Fatalf("expected success while owner flagged, got %v", err) + } + if got := atomic.LoadInt32(&live.calls); got != 2 { + t.Fatalf("live filer calls = %d, want 2", got) + } +} + func TestRouteWriteCondition(t *testing.T) { // Unconditional routes either way. if c, ok := routeWriteCondition(reqWith(nil), false); !ok || c != nil {