From 81ed37988425ccfea3127d3dff8dd5b13e741641 Mon Sep 17 00:00:00 2001 From: jk2lx <54355029+jk2lx@users.noreply.github.com> Date: Fri, 26 Jun 2026 22:22:57 +0100 Subject: [PATCH] volume server: route VolumeMarkReadonly to raft leader (#10120) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * volume server: route VolumeMarkReadonly to raft leader After a master raft election, volume servers may still heartbeat a follower while admin paths such as weed shell volume.mark call notifyMasterVolumeReadonly via vs.GetMaster(). Followers reject VolumeMarkReadonly with NotLeader, which breaks tiering and other mark-readonly workflows until the heartbeat loop reconnects. Resolve the leader through GetMasterConfiguration on configured -master peers (same Leader field filer/master clients already use) before calling VolumeMarkReadonly. When the leader differs from the heartbeat peer, update currentMaster so the heartbeat loop converges faster. Adds operation.LookupRaftLeaderMaster with unit tests. Co-authored-by: Cursor * fix: address review feedback on volume.mark raft leader routing Do not update currentMaster during leader lookup — heartbeat owns that field and uses stream GetLeader() to reconnect. Try the heartbeat peer first and only resolve the raft leader after a NotLeader rejection. Add ctx.Err() early exit and quieter logging for context cancellation. Co-authored-by: Cursor * fix(operation): thread the lookup timeout ctx into connection invalidation The 5s timeout drove only the RPC; WithMasterServerClient saw the unbounded outer ctx, so a self-inflicted timeout (slow GetMasterConfiguration during an election) was treated as a stale channel and tore down the shared master connection. Pass the timeout ctx into the helper so its own expiry leaves ctx.Err() set and spares the connection. --------- Co-authored-by: Cursor Co-authored-by: Chris Lu --- weed/operation/lookup_raft_leader.go | 54 ++++++++++ weed/operation/lookup_raft_leader_test.go | 99 +++++++++++++++++++ weed/server/volume_grpc_admin.go | 27 ++++- weed/server/volume_grpc_client_to_master.go | 56 +++++++++-- .../volume_grpc_client_to_master_test.go | 62 ++++++++++++ 5 files changed, 284 insertions(+), 14 deletions(-) create mode 100644 weed/operation/lookup_raft_leader.go create mode 100644 weed/operation/lookup_raft_leader_test.go create mode 100644 weed/server/volume_grpc_client_to_master_test.go diff --git a/weed/operation/lookup_raft_leader.go b/weed/operation/lookup_raft_leader.go new file mode 100644 index 0000000000..4fc8c4958f --- /dev/null +++ b/weed/operation/lookup_raft_leader.go @@ -0,0 +1,54 @@ +package operation + +import ( + "context" + "fmt" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/grpc" +) + +// LookupRaftLeaderMaster returns the current raft leader by querying configured +// master peers. Followers answer GetMasterConfiguration with the leader address; +// topology mutations such as VolumeMarkReadonly must run on that node. +// +// The volume server's heartbeat peer (GetMaster) may still be a follower after an +// election until the heartbeat loop reconnects. +func LookupRaftLeaderMaster(ctx context.Context, masters []pb.ServerAddress, grpcDialOption grpc.DialOption) (pb.ServerAddress, error) { + if len(masters) == 0 { + return "", fmt.Errorf("no master peers configured") + } + var lastErr error + for _, peer := range masters { + if err := ctx.Err(); err != nil { + return "", err + } + // The same timeout ctx drives both the RPC and WithMasterServerClient's + // connection-invalidation decision, so a self-inflicted 5s timeout leaves + // callCtx.Err() != nil and does not poison the shared master connection. + callCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + var leader pb.ServerAddress + err := WithMasterServerClient(callCtx, false, peer, grpcDialOption, func(client master_pb.SeaweedClient) error { + resp, err := client.GetMasterConfiguration(callCtx, &master_pb.GetMasterConfigurationRequest{}) + if err != nil { + return err + } + if resp.Leader == "" { + return fmt.Errorf("master %s returned empty leader", peer) + } + leader = pb.ServerAddress(resp.Leader) + return nil + }) + cancel() + if err == nil && leader != "" { + return leader, nil + } + lastErr = err + } + if lastErr != nil { + return "", fmt.Errorf("lookup raft leader from %d master peer(s): %w", len(masters), lastErr) + } + return "", fmt.Errorf("lookup raft leader: no leader from %d master peer(s)", len(masters)) +} diff --git a/weed/operation/lookup_raft_leader_test.go b/weed/operation/lookup_raft_leader_test.go new file mode 100644 index 0000000000..f765002a01 --- /dev/null +++ b/weed/operation/lookup_raft_leader_test.go @@ -0,0 +1,99 @@ +package operation + +import ( + "context" + "errors" + "fmt" + "net" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +type fakeLeaderConfigServer struct { + master_pb.UnimplementedSeaweedServer + leader string +} + +func (s *fakeLeaderConfigServer) GetMasterConfiguration(_ context.Context, _ *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) { + return &master_pb.GetMasterConfigurationResponse{Leader: s.leader}, nil +} + +func startFakeMasterServer(t *testing.T, srv master_pb.SeaweedServer) pb.ServerAddress { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + grpcServer := grpc.NewServer() + master_pb.RegisterSeaweedServer(grpcServer, srv) + go func() { _ = grpcServer.Serve(lis) }() + t.Cleanup(grpcServer.GracefulStop) + + _, port, _ := net.SplitHostPort(lis.Addr().String()) + return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port)) +} + +func TestLookupRaftLeaderMasterFromFollower(t *testing.T) { + follower := startFakeMasterServer(t, &fakeLeaderConfigServer{ + leader: "leader.example.com:9333.19333", + }) + + leader, err := LookupRaftLeaderMaster( + context.Background(), + []pb.ServerAddress{follower}, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + t.Fatalf("lookup: %v", err) + } + if got, want := leader.ToHttpAddress(), "leader.example.com:9333"; got != want { + t.Fatalf("leader = %q, want %q", got, want) + } +} + +func TestLookupRaftLeaderMasterTriesNextPeer(t *testing.T) { + down := pb.ServerAddress("127.0.0.1:1") + peer := startFakeMasterServer(t, &fakeLeaderConfigServer{ + leader: "leader.example.com:9333.19333", + }) + + leader, err := LookupRaftLeaderMaster( + context.Background(), + []pb.ServerAddress{down, peer}, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + t.Fatalf("lookup: %v", err) + } + if got, want := leader.ToHttpAddress(), "leader.example.com:9333"; got != want { + t.Fatalf("leader = %q, want %q", got, want) + } +} + +func TestLookupRaftLeaderMasterNoPeers(t *testing.T) { + _, err := LookupRaftLeaderMaster(context.Background(), nil, nil) + if err == nil { + t.Fatal("expected error for empty peers") + } +} + +func TestLookupRaftLeaderMasterCanceledContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := LookupRaftLeaderMaster( + ctx, + []pb.ServerAddress{"127.0.0.1:1"}, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err == nil { + t.Fatal("expected error for canceled context") + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } +} diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index 34e45e90dd..7963a4c5ad 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -5,6 +5,7 @@ import ( "fmt" "net" "path/filepath" + "strings" "time" "google.golang.org/grpc/codes" @@ -279,9 +280,27 @@ func (vs *VolumeServer) makeVolumeWritable(ctx context.Context, v *storage.Volum return nil } +func isNotLeaderErr(err error) bool { + return err != nil && strings.Contains(err.Error(), "Not current leader") +} + func (vs *VolumeServer) notifyMasterVolumeReadonly(ctx context.Context, v *storage.Volume, isReadOnly bool) error { - if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(ctx), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { - _, err := client.VolumeMarkReadonly(context.Background(), &master_pb.VolumeMarkReadonlyRequest{ + master := vs.GetMaster(ctx) + err := vs.volumeMarkReadonlyOnMaster(ctx, master, v, isReadOnly) + if err != nil && isNotLeaderErr(err) { + leader, lookupErr := vs.lookupRaftLeaderMaster(ctx) + if lookupErr != nil { + return fmt.Errorf("heartbeat master %s rejected mark-readonly and leader lookup failed: %w", master, lookupErr) + } + master = leader + err = vs.volumeMarkReadonlyOnMaster(ctx, master, v, isReadOnly) + } + return err +} + +func (vs *VolumeServer) volumeMarkReadonlyOnMaster(ctx context.Context, master pb.ServerAddress, v *storage.Volume, isReadOnly bool) error { + if grpcErr := pb.WithMasterClient(ctx, false, master, vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + _, err := client.VolumeMarkReadonly(ctx, &master_pb.VolumeMarkReadonlyRequest{ Ip: vs.store.Ip, Port: uint32(vs.store.Port), VolumeId: uint32(v.Id), @@ -296,8 +315,8 @@ func (vs *VolumeServer) notifyMasterVolumeReadonly(ctx context.Context, v *stora } return nil }); grpcErr != nil { - glog.V(0).Infof("connect to %s: %v", vs.GetMaster(context.Background()), grpcErr) - return fmt.Errorf("grpc VolumeMarkReadonly with master %s: %v", vs.GetMaster(context.Background()), grpcErr) + glog.V(0).Infof("connect to %s: %v", master, grpcErr) + return fmt.Errorf("grpc VolumeMarkReadonly with master %s: %v", master, grpcErr) } return nil } diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index 82b5cceadb..e55a6ca21a 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -1,6 +1,8 @@ package weed_server import ( + "context" + "errors" "fmt" "os" "strings" @@ -10,14 +12,14 @@ import ( "github.com/seaweedfs/seaweedfs/weed/stats" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" - "context" - "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/util" @@ -27,6 +29,40 @@ func (vs *VolumeServer) GetMaster(ctx context.Context) pb.ServerAddress { return vs.getCurrentMaster() } +// lookupRaftLeaderMaster resolves the raft leader for topology mutations via +// GetMasterConfiguration on configured peers. It does not update currentMaster; +// the heartbeat loop owns that field and reconnects when the stream reports a +// different leader than the connected peer. +func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAddress, error) { + if ctx == nil { + ctx = context.Background() + } + leader, err := operation.LookupRaftLeaderMaster(ctx, vs.SeedMasterNodes, vs.grpcDialOption) + if err != nil { + if isContextDoneErr(err) { + glog.V(1).Infof("volume server %s:%d: raft leader lookup: %v", vs.store.Ip, vs.store.Port, err) + } else { + glog.V(0).Infof("volume server %s:%d: raft leader lookup: %v", vs.store.Ip, vs.store.Port, err) + } + return "", err + } + current := vs.getCurrentMaster() + if !leader.Equals(current) { + glog.V(1).Infof("volume server %s:%d: raft leader %v (heartbeat peer %v)", vs.store.Ip, vs.store.Port, leader, current) + } + return leader, nil +} + +func isContextDoneErr(err error) bool { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return true + } + if st, ok := status.FromError(err); ok { + return st.Code() == codes.Canceled || st.Code() == codes.DeadlineExceeded + } + return false +} + // getCurrentMaster returns vs.currentMaster under a read lock so callers // (e.g. Ping admission) do not race with the heartbeat goroutine that // rewrites it on leader changes. @@ -262,10 +298,10 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp case first := <-vs.store.NewEcShardsChan: shards := util.DrainChannel(vs.store.NewEcShardsChan, first) deltaBeat := &master_pb.Heartbeat{ - Ip: ip, - Port: port, - DataCenter: dataCenter, - Rack: rack, + Ip: ip, + Port: port, + DataCenter: dataCenter, + Rack: rack, NewEcShards: shards, } for _, s := range shards { @@ -295,10 +331,10 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp case first := <-vs.store.DeletedEcShardsChan: shards := util.DrainChannel(vs.store.DeletedEcShardsChan, first) deltaBeat := &master_pb.Heartbeat{ - Ip: ip, - Port: port, - DataCenter: dataCenter, - Rack: rack, + Ip: ip, + Port: port, + DataCenter: dataCenter, + Rack: rack, DeletedEcShards: shards, } for _, s := range shards { diff --git a/weed/server/volume_grpc_client_to_master_test.go b/weed/server/volume_grpc_client_to_master_test.go new file mode 100644 index 0000000000..90d1ba18f4 --- /dev/null +++ b/weed/server/volume_grpc_client_to_master_test.go @@ -0,0 +1,62 @@ +package weed_server + +import ( + "context" + "fmt" + "net" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +func TestLookupRaftLeaderMasterDoesNotUpdateCurrentMaster(t *testing.T) { + follower := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{ + leader: "leader.example.com:9333.19333", + }) + heartbeatPeer := pb.ServerAddress("10.0.0.1:9333") + vs := &VolumeServer{ + SeedMasterNodes: []pb.ServerAddress{follower}, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: &storage.Store{Ip: "127.0.0.1", Port: 8080}, + } + vs.setCurrentMaster(heartbeatPeer) + + leader, err := vs.lookupRaftLeaderMaster(context.Background()) + if err != nil { + t.Fatalf("lookup: %v", err) + } + if got, want := leader.ToHttpAddress(), "leader.example.com:9333"; got != want { + t.Fatalf("leader = %q, want %q", got, want) + } + if got := vs.getCurrentMaster(); got != heartbeatPeer { + t.Fatalf("currentMaster = %v, want unchanged %v", got, heartbeatPeer) + } +} + +type fakeLeaderConfigServer struct { + master_pb.UnimplementedSeaweedServer + leader string +} + +func (s *fakeLeaderConfigServer) GetMasterConfiguration(_ context.Context, _ *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) { + return &master_pb.GetMasterConfigurationResponse{Leader: s.leader}, nil +} + +func startFakeMasterServerForLeaderLookup(t *testing.T, srv master_pb.SeaweedServer) pb.ServerAddress { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + grpcServer := grpc.NewServer() + master_pb.RegisterSeaweedServer(grpcServer, srv) + go func() { _ = grpcServer.Serve(lis) }() + t.Cleanup(grpcServer.GracefulStop) + + _, port, _ := net.SplitHostPort(lis.Addr().String()) + return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port)) +}