From 52d74df4d1b5770de3a2079c824cf5aa0e7239e0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 10 Aug 2026 09:51:08 -0700 Subject: [PATCH] clients: stream the volume listings that ask for everything (#10679) * master: stream volume listings A listing of 800k volumes is 36MB on the wire but 305MB as messages, and the master built all of it, then held it while grpc encoded it. Two of those at once is most of a small master's heap, and the maintenance scanner asks every 30 minutes. The topology goes out first, listing nothing, then its volumes in batches, so the master holds a batch rather than a cluster: 341MB of live heap for one listing becomes 4.4MB. It allocates much the same either way -- what changes is how much of it has to be live at once, which is what sets the heap ceiling. Batches are built under their disk's lock and sent outside it, so a slow reader stalls the stream rather than the topology. They therefore do not share one instant, which a single listing did not either: it takes each disk's lock in turn, so a volume moving during either can be seen twice or not at all. The client helper hides which kind of master answered: one too old for the stream is asked the old way and its reply cut into the same batches. Either way the topology handed over lists no volumes, so a caller cannot come to depend on finding them there. * admin: stream the listing the maintenance scan reads It asks for every volume in the cluster every 30 minutes. Reassembling it client-side keeps the scan identical -- ActiveTopology splits disks by the disk ids on the volumes, so it needs them in the topology -- while the master no longer builds the whole reply to send it. * topology: report a disk id that does not depend on map order A topology disk that fronts several physical disks took its reported id from whichever volume the map yielded first, so two listings of an unchanged disk could disagree. Take the smallest instead. * topology: test that a streamed listing rebuilds to the whole one The callers that stream now rebuild the listing from a topology sent without volumes plus the batches after it, so that has to come out the same as being sent it whole, at every batch size and under a filter. * clients: stream the volume listings that ask for everything The dashboard's list and export pages, the collection and ec shard pages, the topology view, the worker metrics and two shell commands each asked the master to build all 800k volumes into one reply. They read the same listing as before, rebuilt on their side, so the master no longer holds it. The three that already ask for one volume or one collection stay as they are: their replies are small, and streaming one costs a round trip to say so. --- weed/admin/dash/admin_server.go | 2 +- weed/admin/dash/cluster_topology.go | 3 +- weed/admin/dash/collection_management.go | 3 +- weed/admin/dash/ec_shard_management.go | 5 +- weed/admin/dash/volume_export.go | 3 +- weed/admin/dash/volume_management.go | 5 +- weed/plugin/worker/volume_metrics.go | 2 +- weed/shell/command_ec_common.go | 2 +- weed/shell/command_fs_merge_volumes.go | 3 +- weed/topology/disk.go | 14 ++- weed/topology/volume_stream_test.go | 134 +++++++++++++++++++++++ 11 files changed, 161 insertions(+), 15 deletions(-) create mode 100644 weed/topology/volume_stream_test.go diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index bfe965161..9643d8b01 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -2050,7 +2050,7 @@ func (s *AdminServer) getCollectionStats() (map[string]collectionStats, error) { } err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err } diff --git a/weed/admin/dash/cluster_topology.go b/weed/admin/dash/cluster_topology.go index b91f3dd6c..eac475606 100644 --- a/weed/admin/dash/cluster_topology.go +++ b/weed/admin/dash/cluster_topology.go @@ -9,6 +9,7 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) @@ -116,7 +117,7 @@ func (s *AdminServer) getTopologyViaGRPC(topology *ClusterTopology) error { // Get cluster status from master err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { currentMaster := s.masterClient.GetMaster(context.Background()) glog.Errorf("Failed to get volume list from master %s: %v", currentMaster, err) diff --git a/weed/admin/dash/collection_management.go b/weed/admin/dash/collection_management.go index fe55cf402..f868ca90e 100644 --- a/weed/admin/dash/collection_management.go +++ b/weed/admin/dash/collection_management.go @@ -5,6 +5,7 @@ import ( "sort" "time" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" ) @@ -19,7 +20,7 @@ func (s *AdminServer) GetClusterCollections() (*ClusterCollectionsData, error) { // Get actual collection information from volume data err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err } diff --git a/weed/admin/dash/ec_shard_management.go b/weed/admin/dash/ec_shard_management.go index aed62f4ee..2d8ae0356 100644 --- a/weed/admin/dash/ec_shard_management.go +++ b/weed/admin/dash/ec_shard_management.go @@ -6,6 +6,7 @@ import ( "sort" "time" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" ) @@ -44,7 +45,7 @@ func (s *AdminServer) GetClusterEcShards(page int, pageSize int, sortBy string, // Get detailed EC shard information via gRPC err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err } @@ -261,7 +262,7 @@ func (s *AdminServer) GetClusterEcVolumes(page int, pageSize int, sortBy string, // Get detailed EC shard information via gRPC err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err } diff --git a/weed/admin/dash/volume_export.go b/weed/admin/dash/volume_export.go index c6de3264c..520e96456 100644 --- a/weed/admin/dash/volume_export.go +++ b/weed/admin/dash/volume_export.go @@ -5,6 +5,7 @@ import ( "sort" "time" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" @@ -121,7 +122,7 @@ func (s *AdminServer) ExportClusterVolumeList(ctx context.Context, collection st } err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(ctx, &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(ctx, client, &master_pb.VolumeListRequest{}) if err != nil { return err } diff --git a/weed/admin/dash/volume_management.go b/weed/admin/dash/volume_management.go index ab26ea774..59c2b231d 100644 --- a/weed/admin/dash/volume_management.go +++ b/weed/admin/dash/volume_management.go @@ -7,6 +7,7 @@ import ( "sort" "time" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" ) @@ -32,7 +33,7 @@ func (s *AdminServer) GetClusterVolumes(page int, pageSize int, sortBy string, s // Get detailed volume information via gRPC err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err } @@ -419,7 +420,7 @@ func (s *AdminServer) GetClusterVolumeServers() (*ClusterVolumeServersData, erro // Make only ONE VolumeList call and use it for both topology building AND EC shard processing err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { - resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err } diff --git a/weed/plugin/worker/volume_metrics.go b/weed/plugin/worker/volume_metrics.go index 0a9ff97a6..4988bc715 100644 --- a/weed/plugin/worker/volume_metrics.go +++ b/weed/plugin/worker/volume_metrics.go @@ -108,7 +108,7 @@ func FetchVolumeList(ctx context.Context, address string, grpcDialOption grpc.Di client := master_pb.NewSeaweedClient(conn) callCtx, cancelCall := context.WithTimeout(ctx, 10*time.Second) - response, callErr := client.VolumeList(callCtx, &master_pb.VolumeListRequest{}) + response, callErr := pb.CollectVolumeList(callCtx, client, &master_pb.VolumeListRequest{}) cancelCall() _ = conn.Close() diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index cc93d7cff..c222b6116 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -173,7 +173,7 @@ func collectTopologyInfo(commandEnv *CommandEnv, delayBeforeCollecting time.Dura var resp *master_pb.VolumeListResponse err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error { - resp, err = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + resp, err = pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) return err }) if err != nil { diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 4bc256a35..88953b285 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -22,6 +22,7 @@ import ( "google.golang.org/protobuf/proto" "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/util" @@ -331,7 +332,7 @@ func (c *commandFsMergeVolumes) reloadVolumesInfo(masterClient *wdclient.MasterC c.volumes = make(map[needle.VolumeId]*master_pb.VolumeInformationMessage) return masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { - volumes, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{}) + volumes, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) if err != nil { return err } diff --git a/weed/topology/disk.go b/weed/topology/disk.go index 7bffdfcf1..6b8081001 100644 --- a/weed/topology/disk.go +++ b/weed/topology/disk.go @@ -411,8 +411,10 @@ func (d *Disk) ToDiskInfo(filter VolumeFilter) *master_pb.DiskInfo { var diskId uint32 var haveDiskId bool for _, v := range d.volumes { - // Any volume names the disk, including one filtered out. - if !haveDiskId { + // Any volume names the disk, including one filtered out. The smallest + // rather than whichever the map yields first, so that two listings of + // an unchanged disk agree when it fronts several physical disks. + if !haveDiskId || v.DiskId < diskId { diskId, haveDiskId = v.DiskId, true } if !filter.matches(v.Collection, v.Id) { @@ -423,8 +425,12 @@ func (d *Disk) ToDiskInfo(filter VolumeFilter) *master_pb.DiskInfo { d.RUnlock() ecShards := d.GetEcShards() - if !haveDiskId && len(ecShards) > 0 { - diskId = ecShards[0].DiskId + if !haveDiskId { + for _, ecv := range ecShards { + if !haveDiskId || ecv.DiskId < diskId { + diskId, haveDiskId = ecv.DiskId, true + } + } } m := &master_pb.DiskInfo{ diff --git a/weed/topology/volume_stream_test.go b/weed/topology/volume_stream_test.go new file mode 100644 index 000000000..37e8e4e6f --- /dev/null +++ b/weed/topology/volume_stream_test.go @@ -0,0 +1,134 @@ +package topology + +import ( + "fmt" + "sort" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" +) + +func streamTestTopology(t *testing.T) *Topology { + t.Helper() + topo := NewTopology("stream", nil, 32*1024*1024*1024, 5, false) + for dc := 1; dc <= 2; dc++ { + rack := topo.GetOrCreateDataCenter(fmt.Sprintf("dc%d", dc)).GetOrCreateRack("rack1") + for n := 1; n <= 2; n++ { + dn := rack.GetOrCreateDataNode(fmt.Sprintf("10.0.%d.%d", dc, n), 8080, 18080, "", "", + map[string]uint32{"": 10000}) + var volumes []*master_pb.VolumeInformationMessage + for i := 0; i < 250; i++ { + volumes = append(volumes, &master_pb.VolumeInformationMessage{ + Id: uint32(dc*10000 + n*1000 + i), Size: uint64(i) * 100, + Collection: fmt.Sprintf("c%d", i%3), Version: 3, DiskId: uint32(i % 2), + }) + } + topo.SyncDataNodeRegistration(volumes, dn) + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + {Id: uint32(dc*100 + n), Collection: "c1", EcIndexBits: 0x3fff, DiskId: 1}, + }, dn) + } + } + return topo +} + +// summarise reduces a listing to what a caller reads off it, so a streamed one +// can be held against an unstreamed one without depending on ordering. +func summarise(info *master_pb.TopologyInfo) []string { + var lines []string + for _, dc := range info.DataCenterInfos { + for _, rack := range dc.RackInfos { + for _, node := range rack.DataNodeInfos { + for diskType, disk := range node.DiskInfos { + where := fmt.Sprintf("%s/%s/%s/%s", dc.Id, rack.Id, node.Id, diskType) + lines = append(lines, + fmt.Sprintf("%s disk id=%d max=%d count=%d", where, disk.DiskId, disk.MaxVolumeCount, disk.VolumeCount)) + for _, v := range disk.VolumeInfos { + lines = append(lines, fmt.Sprintf("%s vol %d size=%d collection=%s disk=%d", + where, v.Id, v.Size, v.Collection, v.DiskId)) + } + for _, ec := range disk.EcShardInfos { + lines = append(lines, fmt.Sprintf("%s ec %d collection=%s bits=%d", + where, ec.Id, ec.Collection, ec.EcIndexBits)) + } + } + } + } + } + sort.Strings(lines) + return lines +} + +// Every caller that moved onto the stream rebuilds the listing from a topology +// sent without volumes plus the batches that follow. That has to come out the +// same as the listing they used to be sent whole. +func TestStreamedListingRebuildsToTheSameThing(t *testing.T) { + topo := streamTestTopology(t) + + for _, batchSize := range []int{1, 7, 250, 100000} { + t.Run(fmt.Sprintf("batch=%d", batchSize), func(t *testing.T) { + rebuilt := topo.ToTopologyInfo(NoVolumes()) + byPlace := map[[4]string]*master_pb.DiskInfo{} + for _, dc := range rebuilt.DataCenterInfos { + for _, rack := range dc.RackInfos { + for _, node := range rack.DataNodeInfos { + for diskType, disk := range node.DiskInfos { + byPlace[[4]string{dc.Id, rack.Id, node.Id, diskType}] = disk + if len(disk.VolumeInfos) != 0 || len(disk.EcShardInfos) != 0 { + t.Fatalf("the header listed volumes on %s", node.Id) + } + } + } + } + } + + err := topo.StreamVolumes(rebuilt, VolumeFilter{}, batchSize, func(b *master_pb.VolumeListStreamResponse) error { + if batchSize < 100000 && len(b.VolumeInfos) > batchSize { + t.Errorf("batch carried %d volumes, more than the %d asked for", len(b.VolumeInfos), batchSize) + } + disk := byPlace[[4]string{b.DataCenter, b.Rack, b.DataNode, b.DiskType}] + if disk == nil { + t.Fatalf("batch named a disk the header did not: %s/%s", b.DataNode, b.DiskType) + } + disk.VolumeInfos = append(disk.VolumeInfos, b.VolumeInfos...) + disk.EcShardInfos = append(disk.EcShardInfos, b.EcShardInfos...) + return nil + }) + if err != nil { + t.Fatal(err) + } + + want := summarise(topo.ToTopologyInfo(VolumeFilter{})) + got := summarise(rebuilt) + if len(got) != len(want) { + t.Fatalf("rebuilt %d lines, want %d", len(got), len(want)) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("line %d:\n got %q\nwant %q", i, got[i], want[i]) + } + } + }) + } +} + +// A filter must select the same volumes streamed as it does whole. +func TestStreamedListingHonoursTheFilter(t *testing.T) { + topo := streamTestTopology(t) + name := "c1" + + var streamed []uint32 + err := topo.StreamVolumes(topo.ToTopologyInfo(NoVolumes()), VolumeFilter{Collection: &name}, 16, func(b *master_pb.VolumeListStreamResponse) error { + for _, v := range b.VolumeInfos { + streamed = append(streamed, v.Id) + } + return nil + }) + if err != nil { + t.Fatal(err) + } + whole, _ := listed(topo.ToTopologyInfo(VolumeFilter{Collection: &name})) + if !equalIds(streamed, whole) { + t.Errorf("streamed %d volumes, whole listing had %d", len(streamed), len(whole)) + } +}