master: stream volume listings (#10676)

* 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.

weed/admin/maintenance/maintenance_scanner.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types"
 )
@@ -101,7 +102,8 @@ func (ms *MaintenanceScanner) getVolumeHealthMetrics() ([]*types.VolumeHealthMet
 	glog.V(1).Infof("Collecting volume health metrics from master")
 	err := ms.adminClient.WithMasterClient(func(client master_pb.SeaweedClient) error {
 
-		resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
+		// Streamed, so the master never builds the whole listing to send it.
+		resp, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{})
 		if err != nil {
 			return err
 		}

weed/pb/master.proto

@@ -27,6 +27,8 @@ service Seaweed {
   }
   rpc VolumeList (VolumeListRequest) returns (VolumeListResponse) {
   }
+  rpc VolumeListStream (VolumeListRequest) returns (stream VolumeListStreamResponse) {
+  }
   rpc LookupEcVolume (LookupEcVolumeRequest) returns (LookupEcVolumeResponse) {
   }
   rpc VacuumVolume (VacuumVolumeRequest) returns (VacuumVolumeResponse) {
@@ -419,6 +421,24 @@ message VolumeListResponse {
   uint64 volume_size_limit_mb = 2;
 }
 
+// VolumeListStream answers the same request as VolumeList without either end
+// holding every volume in the cluster at once. At 800k volumes the reply is
+// 36MB on the wire but 305MB as messages, which the master built in full
+// before sending any of it.
+message VolumeListStreamResponse {
+  // Sent once, first, listing no volumes: the topology, its disks and their
+  // counters. Every message after carries volumes for one of those disks.
+  VolumeListResponse header = 1;
+  // Which disk this batch is from. A disk arrives over as many batches as it
+  // takes, so append rather than assign.
+  string data_center = 2;
+  string rack = 3;
+  string data_node = 4;
+  string disk_type = 5;
+  repeated VolumeInformationMessage volume_infos = 6;
+  repeated VolumeEcShardInformationMessage ec_shard_infos = 7;
+}
+
 message LookupEcVolumeRequest {
   uint32 volume_id = 1;
 }

weed/pb/master_pb/master_grpc.pb.go

@@ -28,6 +28,7 @@ const (
 	Seaweed_CollectionList_FullMethodName         = "/master_pb.Seaweed/CollectionList"
 	Seaweed_CollectionDelete_FullMethodName       = "/master_pb.Seaweed/CollectionDelete"
 	Seaweed_VolumeList_FullMethodName             = "/master_pb.Seaweed/VolumeList"
+	Seaweed_VolumeListStream_FullMethodName       = "/master_pb.Seaweed/VolumeListStream"
 	Seaweed_LookupEcVolume_FullMethodName         = "/master_pb.Seaweed/LookupEcVolume"
 	Seaweed_VacuumVolume_FullMethodName           = "/master_pb.Seaweed/VacuumVolume"
 	Seaweed_DisableVacuum_FullMethodName          = "/master_pb.Seaweed/DisableVacuum"
@@ -60,6 +61,7 @@ type SeaweedClient interface {
 	CollectionList(ctx context.Context, in *CollectionListRequest, opts ...grpc.CallOption) (*CollectionListResponse, error)
 	CollectionDelete(ctx context.Context, in *CollectionDeleteRequest, opts ...grpc.CallOption) (*CollectionDeleteResponse, error)
 	VolumeList(ctx context.Context, in *VolumeListRequest, opts ...grpc.CallOption) (*VolumeListResponse, error)
+	VolumeListStream(ctx context.Context, in *VolumeListRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[VolumeListStreamResponse], error)
 	LookupEcVolume(ctx context.Context, in *LookupEcVolumeRequest, opts ...grpc.CallOption) (*LookupEcVolumeResponse, error)
 	VacuumVolume(ctx context.Context, in *VacuumVolumeRequest, opts ...grpc.CallOption) (*VacuumVolumeResponse, error)
 	DisableVacuum(ctx context.Context, in *DisableVacuumRequest, opts ...grpc.CallOption) (*DisableVacuumResponse, error)
@@ -186,6 +188,25 @@ func (c *seaweedClient) VolumeList(ctx context.Context, in *VolumeListRequest, o
 	return out, nil
 }
 
+func (c *seaweedClient) VolumeListStream(ctx context.Context, in *VolumeListRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[VolumeListStreamResponse], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &Seaweed_ServiceDesc.Streams[3], Seaweed_VolumeListStream_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[VolumeListRequest, VolumeListStreamResponse]{ClientStream: stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Seaweed_VolumeListStreamClient = grpc.ServerStreamingClient[VolumeListStreamResponse]
+
 func (c *seaweedClient) LookupEcVolume(ctx context.Context, in *LookupEcVolumeRequest, opts ...grpc.CallOption) (*LookupEcVolumeResponse, error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
 	out := new(LookupEcVolumeResponse)
@@ -369,6 +390,7 @@ type SeaweedServer interface {
 	CollectionList(context.Context, *CollectionListRequest) (*CollectionListResponse, error)
 	CollectionDelete(context.Context, *CollectionDeleteRequest) (*CollectionDeleteResponse, error)
 	VolumeList(context.Context, *VolumeListRequest) (*VolumeListResponse, error)
+	VolumeListStream(*VolumeListRequest, grpc.ServerStreamingServer[VolumeListStreamResponse]) error
 	LookupEcVolume(context.Context, *LookupEcVolumeRequest) (*LookupEcVolumeResponse, error)
 	VacuumVolume(context.Context, *VacuumVolumeRequest) (*VacuumVolumeResponse, error)
 	DisableVacuum(context.Context, *DisableVacuumRequest) (*DisableVacuumResponse, error)
@@ -423,6 +445,9 @@ func (UnimplementedSeaweedServer) CollectionDelete(context.Context, *CollectionD
 func (UnimplementedSeaweedServer) VolumeList(context.Context, *VolumeListRequest) (*VolumeListResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method VolumeList not implemented")
 }
+func (UnimplementedSeaweedServer) VolumeListStream(*VolumeListRequest, grpc.ServerStreamingServer[VolumeListStreamResponse]) error {
+	return status.Error(codes.Unimplemented, "method VolumeListStream not implemented")
+}
 func (UnimplementedSeaweedServer) LookupEcVolume(context.Context, *LookupEcVolumeRequest) (*LookupEcVolumeResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method LookupEcVolume not implemented")
 }
@@ -624,6 +649,17 @@ func _Seaweed_VolumeList_Handler(srv interface{}, ctx context.Context, dec func(
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Seaweed_VolumeListStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(VolumeListRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(SeaweedServer).VolumeListStream(m, &grpc.GenericServerStream[VolumeListRequest, VolumeListStreamResponse]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Seaweed_VolumeListStreamServer = grpc.ServerStreamingServer[VolumeListStreamResponse]
+
 func _Seaweed_LookupEcVolume_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(LookupEcVolumeRequest)
 	if err := dec(in); err != nil {
@@ -1049,6 +1085,11 @@ var Seaweed_ServiceDesc = grpc.ServiceDesc{
 			ServerStreams: true,
 			ClientStreams: true,
 		},
+		{
+			StreamName:    "VolumeListStream",
+			Handler:       _Seaweed_VolumeListStream_Handler,
+			ServerStreams: true,
+		},
 	},
 	Metadata: "master.proto",
 }

weed/pb/volume_list_stream.go

@@ -0,0 +1,212 @@
+package pb
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// ReceiveVolumeList reads a streamed volume listing: the topology first, then
+// its volumes in batches as they arrive. A caller that works a volume at a
+// time never holds the cluster; one that needs it whole can use
+// CollectVolumeList.
+//
+// onTopology is given a listing whose disks name themselves but list nothing,
+// because the volumes come through onVolumes instead. That holds however the
+// master answered: one too old for the stream is asked the old way and its
+// reply cut into the same batches, so a caller cannot tell the difference and
+// must not read volumes off the topology either way.
+func ReceiveVolumeList(ctx context.Context, client master_pb.SeaweedClient, request *master_pb.VolumeListRequest,
+	onTopology func(*master_pb.VolumeListResponse) error,
+	onVolumes func(*master_pb.VolumeListStreamResponse) error) error {
+
+	stream, err := client.VolumeListStream(ctx, request)
+	if err == nil {
+		started, streamErr := receiveVolumeListStream(stream, onTopology, onVolumes)
+		// Only a stream that said nothing can be asked again the old way. Past
+		// its first message the master plainly does have the method, and the
+		// error may even be the caller's own, so starting over would hand back
+		// what has already been handed over.
+		if started || status.Code(streamErr) != codes.Unimplemented {
+			return streamErr
+		}
+	} else if status.Code(err) != codes.Unimplemented {
+		return err
+	}
+
+	response, err := client.VolumeList(ctx, request)
+	if err != nil {
+		return err
+	}
+	return replayVolumeList(response, onTopology, onVolumes)
+}
+
+// receiveVolumeListStream reports whether the stream said anything at all,
+// which decides whether it can be started over as an unstreamed listing.
+func receiveVolumeListStream(stream master_pb.Seaweed_VolumeListStreamClient,
+	onTopology func(*master_pb.VolumeListResponse) error,
+	onVolumes func(*master_pb.VolumeListStreamResponse) error) (started bool, err error) {
+
+	told := false
+	for {
+		batch, err := stream.Recv()
+		if err == io.EOF {
+			if !told {
+				return started, fmt.Errorf("volume list stream ended before its topology")
+			}
+			return started, nil
+		}
+		if err != nil {
+			return started, err
+		}
+		started = true
+		if batch.Header != nil {
+			if told {
+				return started, fmt.Errorf("volume list stream sent its topology twice")
+			}
+			told = true
+			if onTopology != nil {
+				if err := onTopology(batch.Header); err != nil {
+					return started, err
+				}
+			}
+			continue
+		}
+		if !told {
+			return started, fmt.Errorf("volume list stream sent volumes before its topology")
+		}
+		if onVolumes != nil {
+			if err := onVolumes(batch); err != nil {
+				return started, err
+			}
+		}
+	}
+}
+
+// replayVolumeList cuts an unstreamed reply into the batches a caller expects.
+// Each disk's volumes are moved out of the topology rather than shared with it,
+// so the topology handed over lists nothing, exactly as a streamed one does.
+func replayVolumeList(response *master_pb.VolumeListResponse,
+	onTopology func(*master_pb.VolumeListResponse) error,
+	onVolumes func(*master_pb.VolumeListStreamResponse) error) error {
+
+	type batch struct {
+		key    [4]string
+		volume []*master_pb.VolumeInformationMessage
+		ec     []*master_pb.VolumeEcShardInformationMessage
+	}
+	var batches []batch
+
+	if response.TopologyInfo != nil {
+		for _, dc := range response.TopologyInfo.DataCenterInfos {
+			for _, rack := range dc.RackInfos {
+				for _, node := range rack.DataNodeInfos {
+					for diskType, disk := range node.DiskInfos {
+						if len(disk.VolumeInfos) == 0 && len(disk.EcShardInfos) == 0 {
+							continue
+						}
+						batches = append(batches, batch{
+							key:    [4]string{dc.Id, rack.Id, node.Id, diskType},
+							volume: disk.VolumeInfos,
+							ec:     disk.EcShardInfos,
+						})
+						disk.VolumeInfos, disk.EcShardInfos = nil, nil
+					}
+				}
+			}
+		}
+	}
+
+	if onTopology != nil {
+		if err := onTopology(response); err != nil {
+			return err
+		}
+	}
+	if onVolumes == nil {
+		return nil
+	}
+	for _, b := range batches {
+		err := onVolumes(&master_pb.VolumeListStreamResponse{
+			DataCenter:   b.key[0],
+			Rack:         b.key[1],
+			DataNode:     b.key[2],
+			DiskType:     b.key[3],
+			VolumeInfos:  b.volume,
+			EcShardInfos: b.ec,
+		})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// CollectVolumeList streams a listing and puts it back together, for callers
+// that need the whole topology. The master still never holds it all, which is
+// the point; this only moves that cost to the caller.
+func CollectVolumeList(ctx context.Context, client master_pb.SeaweedClient, request *master_pb.VolumeListRequest) (*master_pb.VolumeListResponse, error) {
+	var response *master_pb.VolumeListResponse
+	var skipped int
+	disks := make(map[[4]string]*master_pb.DiskInfo)
+
+	err := ReceiveVolumeList(ctx, client, request,
+		func(topology *master_pb.VolumeListResponse) error {
+			response = topology
+			return nil
+		},
+		func(batch *master_pb.VolumeListStreamResponse) error {
+			key := [4]string{batch.DataCenter, batch.Rack, batch.DataNode, batch.DiskType}
+			disk, known := disks[key]
+			if !known {
+				disk = findDisk(response, key)
+				disks[key] = disk
+			}
+			if disk == nil {
+				// A disk registered after the topology went out. It is not in
+				// the listing being rebuilt and has nowhere to go, so leave it
+				// to the next one rather than failing this one -- an unstreamed
+				// listing would not have shown it either, having read each
+				// node's disks once.
+				skipped++
+				return nil
+			}
+			disk.VolumeInfos = append(disk.VolumeInfos, batch.VolumeInfos...)
+			disk.EcShardInfos = append(disk.EcShardInfos, batch.EcShardInfos...)
+			return nil
+		})
+	if err != nil {
+		return nil, err
+	}
+	if skipped > 0 {
+		glog.V(1).Infof("volume list: %d batches were for disks added after the topology was sent", skipped)
+	}
+	return response, nil
+}
+
+func findDisk(response *master_pb.VolumeListResponse, key [4]string) *master_pb.DiskInfo {
+	if response == nil || response.TopologyInfo == nil {
+		return nil
+	}
+	for _, dc := range response.TopologyInfo.DataCenterInfos {
+		if dc.Id != key[0] {
+			continue
+		}
+		for _, rack := range dc.RackInfos {
+			if rack.Id != key[1] {
+				continue
+			}
+			for _, node := range rack.DataNodeInfos {
+				if node.Id != key[2] {
+					continue
+				}
+				return node.DiskInfos[key[3]]
+			}
+		}
+	}
+	return nil
+}

weed/pb/volume_list_stream_test.go

@@ -0,0 +1,306 @@
+package pb
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+	"google.golang.org/protobuf/proto"
+)
+
+// fakeMaster answers a listing either way, so the same assertions can be made
+// of a master that streams and one too old to.
+type fakeMaster struct {
+	master_pb.UnimplementedSeaweedServer
+	response *master_pb.VolumeListResponse
+	streams  bool
+	batch    int
+	// extraBatch is sent after the rest, standing for a disk that registered
+	// once the topology had gone out.
+	extraBatch *master_pb.VolumeListStreamResponse
+	// failAfterHeader stands for a master that plainly has the method but
+	// gives up mid-stream, reporting the one code that means "ask the old way".
+	failAfterHeader bool
+}
+
+func (m *fakeMaster) VolumeList(ctx context.Context, req *master_pb.VolumeListRequest) (*master_pb.VolumeListResponse, error) {
+	return cloneListing(m.response), nil
+}
+
+func (m *fakeMaster) VolumeListStream(req *master_pb.VolumeListRequest, stream master_pb.Seaweed_VolumeListStreamServer) error {
+	if !m.streams {
+		return status_Unimplemented()
+	}
+	full := cloneListing(m.response)
+	header := &master_pb.VolumeListResponse{
+		TopologyInfo:      &master_pb.TopologyInfo{Id: full.TopologyInfo.Id},
+		VolumeSizeLimitMb: full.VolumeSizeLimitMb,
+	}
+	// The header names every disk but lists nothing on it.
+	for _, dc := range full.TopologyInfo.DataCenterInfos {
+		headerDc := &master_pb.DataCenterInfo{Id: dc.Id}
+		for _, rack := range dc.RackInfos {
+			headerRack := &master_pb.RackInfo{Id: rack.Id}
+			for _, node := range rack.DataNodeInfos {
+				headerNode := &master_pb.DataNodeInfo{Id: node.Id, DiskInfos: map[string]*master_pb.DiskInfo{}}
+				for diskType, disk := range node.DiskInfos {
+					headerNode.DiskInfos[diskType] = &master_pb.DiskInfo{Type: diskType, DiskId: disk.DiskId}
+				}
+				headerRack.DataNodeInfos = append(headerRack.DataNodeInfos, headerNode)
+			}
+			headerDc.RackInfos = append(headerDc.RackInfos, headerRack)
+		}
+		header.TopologyInfo.DataCenterInfos = append(header.TopologyInfo.DataCenterInfos, headerDc)
+	}
+	if err := stream.Send(&master_pb.VolumeListStreamResponse{Header: header}); err != nil {
+		return err
+	}
+	if m.failAfterHeader {
+		return status_Unimplemented()
+	}
+
+	for _, dc := range full.TopologyInfo.DataCenterInfos {
+		for _, rack := range dc.RackInfos {
+			for _, node := range rack.DataNodeInfos {
+				for diskType, disk := range node.DiskInfos {
+					for start := 0; start < len(disk.VolumeInfos); start += m.batch {
+						end := min(start+m.batch, len(disk.VolumeInfos))
+						err := stream.Send(&master_pb.VolumeListStreamResponse{
+							DataCenter: dc.Id, Rack: rack.Id, DataNode: node.Id, DiskType: diskType,
+							VolumeInfos: disk.VolumeInfos[start:end],
+						})
+						if err != nil {
+							return err
+						}
+					}
+					if len(disk.EcShardInfos) > 0 {
+						err := stream.Send(&master_pb.VolumeListStreamResponse{
+							DataCenter: dc.Id, Rack: rack.Id, DataNode: node.Id, DiskType: diskType,
+							EcShardInfos: disk.EcShardInfos,
+						})
+						if err != nil {
+							return err
+						}
+					}
+				}
+			}
+		}
+	}
+	if m.extraBatch != nil {
+		if err := stream.Send(m.extraBatch); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func testListing(volumes int) *master_pb.VolumeListResponse {
+	disk := &master_pb.DiskInfo{Type: "", DiskId: 2}
+	for i := 1; i <= volumes; i++ {
+		disk.VolumeInfos = append(disk.VolumeInfos, &master_pb.VolumeInformationMessage{
+			Id: uint32(i), Size: uint64(i) * 100, Collection: "c",
+		})
+	}
+	disk.EcShardInfos = append(disk.EcShardInfos, &master_pb.VolumeEcShardInformationMessage{
+		Id: 900, Collection: "c", EcIndexBits: 0x3fff,
+	})
+	return &master_pb.VolumeListResponse{
+		VolumeSizeLimitMb: 30000,
+		TopologyInfo: &master_pb.TopologyInfo{
+			Id: "topo",
+			DataCenterInfos: []*master_pb.DataCenterInfo{{
+				Id: "dc1",
+				RackInfos: []*master_pb.RackInfo{{
+					Id: "rack1",
+					DataNodeInfos: []*master_pb.DataNodeInfo{{
+						Id:        "10.0.0.1:8080",
+						DiskInfos: map[string]*master_pb.DiskInfo{"": disk},
+					}},
+				}},
+			}},
+		},
+	}
+}
+
+func dial(t *testing.T, master *fakeMaster) master_pb.SeaweedClient {
+	t.Helper()
+	listener := bufconn.Listen(1 << 20)
+	server := grpc.NewServer()
+	master_pb.RegisterSeaweedServer(server, master)
+	go server.Serve(listener)
+	t.Cleanup(server.Stop)
+
+	conn, err := grpc.NewClient("passthrough://bufnet",
+		grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return listener.DialContext(ctx) }),
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return master_pb.NewSeaweedClient(conn)
+}
+
+// Whichever way the master answers, a caller sees the same volumes, and sees
+// none of them on the topology it is handed.
+func TestReceiveVolumeListIsTheSameEitherWay(t *testing.T) {
+	const volumes = 250
+	for _, streams := range []bool{true, false} {
+		t.Run(fmt.Sprintf("streaming=%v", streams), func(t *testing.T) {
+			client := dial(t, &fakeMaster{response: testListing(volumes), streams: streams, batch: 32})
+
+			var got []uint32
+			var ec []uint32
+			var topology *master_pb.VolumeListResponse
+			err := ReceiveVolumeList(context.Background(), client, &master_pb.VolumeListRequest{},
+				func(header *master_pb.VolumeListResponse) error {
+					topology = header
+					return nil
+				},
+				func(batch *master_pb.VolumeListStreamResponse) error {
+					for _, v := range batch.VolumeInfos {
+						got = append(got, v.Id)
+					}
+					for _, s := range batch.EcShardInfos {
+						ec = append(ec, s.Id)
+					}
+					return nil
+				})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(got) != volumes {
+				t.Errorf("received %d volumes, want %d", len(got), volumes)
+			}
+			if len(ec) != 1 {
+				t.Errorf("received %d ec shards, want 1", len(ec))
+			}
+			if topology == nil {
+				t.Fatal("never told the topology")
+			}
+			if topology.VolumeSizeLimitMb != 30000 {
+				t.Errorf("volume size limit %d, want 30000", topology.VolumeSizeLimitMb)
+			}
+			disk := topology.TopologyInfo.DataCenterInfos[0].RackInfos[0].DataNodeInfos[0].DiskInfos[""]
+			if len(disk.VolumeInfos) != 0 || len(disk.EcShardInfos) != 0 {
+				t.Errorf("the topology handed over listed %d volumes and %d ec shards, want none",
+					len(disk.VolumeInfos), len(disk.EcShardInfos))
+			}
+		})
+	}
+}
+
+// Reassembly must put back exactly what an unstreamed listing holds -- in
+// particular it must not double the volumes when the master did not stream.
+func TestCollectVolumeListRebuildsTheListing(t *testing.T) {
+	const volumes = 250
+	for _, streams := range []bool{true, false} {
+		t.Run(fmt.Sprintf("streaming=%v", streams), func(t *testing.T) {
+			client := dial(t, &fakeMaster{response: testListing(volumes), streams: streams, batch: 32})
+
+			response, err := CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			disk := response.TopologyInfo.DataCenterInfos[0].RackInfos[0].DataNodeInfos[0].DiskInfos[""]
+			if len(disk.VolumeInfos) != volumes {
+				t.Fatalf("rebuilt %d volumes, want %d", len(disk.VolumeInfos), volumes)
+			}
+			if len(disk.EcShardInfos) != 1 {
+				t.Fatalf("rebuilt %d ec shards, want 1", len(disk.EcShardInfos))
+			}
+			if disk.DiskId != 2 {
+				t.Errorf("rebuilt disk id %d, want 2", disk.DiskId)
+			}
+			seen := make(map[uint32]int, volumes)
+			for _, v := range disk.VolumeInfos {
+				seen[v.Id]++
+			}
+			for id, n := range seen {
+				if n != 1 {
+					t.Fatalf("volume %d rebuilt %d times", id, n)
+				}
+			}
+		})
+	}
+}
+
+func status_Unimplemented() error {
+	return status.Error(codes.Unimplemented, "this master does not stream volume listings")
+}
+
+func cloneListing(r *master_pb.VolumeListResponse) *master_pb.VolumeListResponse {
+	return proto.Clone(r).(*master_pb.VolumeListResponse)
+}
+
+// A heartbeat can register a disk between the topology going out and the
+// batches following it. Those volumes have nowhere to go in the listing being
+// rebuilt, but they must not fail it: the scan that reads it runs every 30
+// minutes and would lose the whole cluster over one new disk.
+func TestCollectVolumeListSurvivesADiskAddedMidStream(t *testing.T) {
+	const volumes = 100
+	master := &fakeMaster{response: testListing(volumes), streams: true, batch: 32}
+	master.extraBatch = &master_pb.VolumeListStreamResponse{
+		DataCenter: "dc1", Rack: "rack1", DataNode: "10.0.0.1:8080", DiskType: "ssd",
+		VolumeInfos: []*master_pb.VolumeInformationMessage{{Id: 5000, Collection: "c"}},
+	}
+	client := dial(t, master)
+
+	response, err := CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{})
+	if err != nil {
+		t.Fatalf("a disk arriving mid-stream failed the listing: %v", err)
+	}
+	node := response.TopologyInfo.DataCenterInfos[0].RackInfos[0].DataNodeInfos[0]
+	if len(node.DiskInfos[""].VolumeInfos) != volumes {
+		t.Errorf("rebuilt %d volumes, want %d", len(node.DiskInfos[""].VolumeInfos), volumes)
+	}
+	if _, appeared := node.DiskInfos["ssd"]; appeared {
+		t.Error("the listing grew a disk its topology never named")
+	}
+}
+
+// A stream that has already spoken cannot be started over as an unstreamed
+// listing: the caller would be handed the same volumes twice.
+func TestReceiveVolumeListDoesNotRestartAStreamThatBegan(t *testing.T) {
+	client := dial(t, &fakeMaster{response: testListing(100), streams: true, batch: 32, failAfterHeader: true})
+
+	topologies, batches := 0, 0
+	err := ReceiveVolumeList(context.Background(), client, &master_pb.VolumeListRequest{},
+		func(*master_pb.VolumeListResponse) error { topologies++; return nil },
+		func(*master_pb.VolumeListStreamResponse) error { batches++; return nil })
+	if err == nil {
+		t.Fatal("a stream that failed after its topology was quietly restarted")
+	}
+	if topologies != 1 {
+		t.Errorf("handed the topology %d times, want 1", topologies)
+	}
+	if batches != 0 {
+		t.Errorf("handed %d volume batches, want none", batches)
+	}
+}
+
+// A caller's own error must reach it, even when it happens to carry the code
+// that means an older master.
+func TestReceiveVolumeListDoesNotRestartOnACallersError(t *testing.T) {
+	client := dial(t, &fakeMaster{response: testListing(100), streams: true, batch: 32})
+
+	batches := 0
+	err := ReceiveVolumeList(context.Background(), client, &master_pb.VolumeListRequest{},
+		nil,
+		func(*master_pb.VolumeListStreamResponse) error {
+			batches++
+			return status.Error(codes.Unimplemented, "the caller cannot handle this")
+		})
+	if err == nil {
+		t.Fatal("the caller's error was swallowed and the listing restarted")
+	}
+	if batches != 1 {
+		t.Errorf("called back %d times, want 1 before giving up", batches)
+	}
+}

weed/server/master_grpc_server_volume.go

@@ -287,6 +287,29 @@ func (ms *MasterServer) VolumeList(ctx context.Context, req *master_pb.VolumeLis
 	return resp, nil
 }
 
+// VolumeListStream answers VolumeList without building the whole reply first.
+// The topology goes out on its own, then the volumes in batches, so the master
+// holds one batch rather than every volume in the cluster.
+func (ms *MasterServer) VolumeListStream(req *master_pb.VolumeListRequest, stream master_pb.Seaweed_VolumeListStreamServer) error {
+
+	if !ms.Topo.IsLeader() {
+		return raft.NotLeaderError
+	}
+
+	listed := ms.Topo.ToTopologyInfo(topology.NoVolumes())
+	err := stream.Send(&master_pb.VolumeListStreamResponse{
+		Header: &master_pb.VolumeListResponse{
+			TopologyInfo:      listed,
+			VolumeSizeLimitMb: uint64(ms.option.VolumeSizeLimitMB),
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	return ms.Topo.StreamVolumes(listed, topology.NewVolumeFilter(req), 0, stream.Send)
+}
+
 func (ms *MasterServer) LookupEcVolume(ctx context.Context, req *master_pb.LookupEcVolumeRequest) (*master_pb.LookupEcVolumeResponse, error) {
 
 	if !ms.Topo.IsLeader() {

weed/topology/volume_filter.go

@@ -11,6 +11,14 @@ import (
 type VolumeFilter struct {
 	Collection *string
 	VolumeId   *needle.VolumeId
+	// nothing selects the topology alone, for a listing whose volumes travel
+	// in messages of their own.
+	nothing bool
+}
+
+// NoVolumes selects the topology and no volume in it.
+func NoVolumes() VolumeFilter {
+	return VolumeFilter{nothing: true}
 }
 
 // NewVolumeFilter reads what a VolumeList request asked for, where empty and
@@ -35,10 +43,13 @@ func NewVolumeFilter(req *master_pb.VolumeListRequest) VolumeFilter {
 
 // SelectsEverything lets a caller size its result for the whole disk up front.
 func (f VolumeFilter) SelectsEverything() bool {
-	return f.Collection == nil && f.VolumeId == nil
+	return !f.nothing && f.Collection == nil && f.VolumeId == nil
 }
 
 func (f VolumeFilter) matches(collection string, id needle.VolumeId) bool {
+	if f.nothing {
+		return false
+	}
 	if f.Collection != nil && *f.Collection != collection {
 		return false
 	}

weed/topology/volume_stream.go

@@ -0,0 +1,137 @@
+package topology
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// StreamVolumes hands every volume the filter selects to send, in batches, so
+// that neither end holds the whole cluster to move it. Batches are built under
+// their disk's lock and handed over outside it, so a slow reader stalls the
+// stream rather than the topology.
+//
+// Only the disks `listed` names are streamed. That listing went out first, and
+// the topology is walked again here, so without it a disk registering in
+// between would have its volumes sent to a client with nowhere to put them.
+// Bounding the walk by what was already announced makes the two agree by
+// construction: a disk arriving mid-listing is in neither, and is reported by
+// the next one.
+//
+// The batches still do not share one instant. Neither did a single listing,
+// which takes each disk's lock in turn, so a volume that moves while either is
+// running can be seen twice or not at all.
+func (t *Topology) StreamVolumes(listed *master_pb.TopologyInfo, filter VolumeFilter, batchSize int, send func(*master_pb.VolumeListStreamResponse) error) error {
+	if batchSize <= 0 {
+		batchSize = defaultVolumeStreamBatch
+	}
+	announced := announcedDisks(listed)
+	for _, dcNode := range t.Children() {
+		dc := dcNode.(*DataCenter)
+		for _, rackNode := range dc.Children() {
+			rack := rackNode.(*Rack)
+			for _, dnNode := range rack.Children() {
+				dn := dnNode.(*DataNode)
+				for _, diskNode := range dn.Children() {
+					disk := diskNode.(*Disk)
+					if !announced[[4]string{string(dc.Id()), string(rack.Id()), string(dn.Id()), string(disk.Id())}] {
+						continue
+					}
+					batch := func() *master_pb.VolumeListStreamResponse {
+						return &master_pb.VolumeListStreamResponse{
+							DataCenter: string(dc.Id()),
+							Rack:       string(rack.Id()),
+							DataNode:   string(dn.Id()),
+							DiskType:   string(disk.Id()),
+						}
+					}
+					if err := disk.streamVolumes(filter, batchSize, batch, send); err != nil {
+						return err
+					}
+					if err := disk.streamEcShards(filter, batchSize, batch, send); err != nil {
+						return err
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
+const defaultVolumeStreamBatch = 10000
+
+// announcedDisks names the disks a listing carried.
+func announcedDisks(listed *master_pb.TopologyInfo) map[[4]string]bool {
+	announced := make(map[[4]string]bool)
+	if listed == nil {
+		return announced
+	}
+	for _, dc := range listed.DataCenterInfos {
+		for _, rack := range dc.RackInfos {
+			for _, node := range rack.DataNodeInfos {
+				for diskType := range node.DiskInfos {
+					announced[[4]string{dc.Id, rack.Id, node.Id, diskType}] = true
+				}
+			}
+		}
+	}
+	return announced
+}
+
+// streamVolumes sends this disk's volumes a batch at a time. The ids are taken
+// in one pass and the messages built in later ones, so the lock is held for a
+// batch rather than for the disk, and only 4 bytes per volume are carried
+// between passes. A volume that leaves in between is simply not sent.
+func (d *Disk) streamVolumes(filter VolumeFilter, batchSize int, newBatch func() *master_pb.VolumeListStreamResponse, send func(*master_pb.VolumeListStreamResponse) error) error {
+	d.RLock()
+	ids := make([]needle.VolumeId, 0, len(d.volumes))
+	for id := range d.volumes {
+		ids = append(ids, id)
+	}
+	d.RUnlock()
+
+	for start := 0; start < len(ids); start += batchSize {
+		end := min(start+batchSize, len(ids))
+		batch := newBatch()
+
+		d.RLock()
+		for _, id := range ids[start:end] {
+			v, found := d.volumes[id]
+			if !found || !filter.matches(v.Collection, v.Id) {
+				continue
+			}
+			batch.VolumeInfos = append(batch.VolumeInfos, v.ToVolumeInformationMessage())
+		}
+		d.RUnlock()
+
+		if len(batch.VolumeInfos) == 0 {
+			continue
+		}
+		if err := send(batch); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (d *Disk) streamEcShards(filter VolumeFilter, batchSize int, newBatch func() *master_pb.VolumeListStreamResponse, send func(*master_pb.VolumeListStreamResponse) error) error {
+	// GetEcShards already copies under the lock, and a cluster holds far fewer
+	// ec shards than volumes, so these only need cutting into batches.
+	shards := d.GetEcShards()
+	for start := 0; start < len(shards); start += batchSize {
+		end := min(start+batchSize, len(shards))
+		batch := newBatch()
+		for _, ecv := range shards[start:end] {
+			if !filter.matches(ecv.Collection, ecv.VolumeId) {
+				continue
+			}
+			batch.EcShardInfos = append(batch.EcShardInfos, ecv.ToVolumeEcShardInformationMessage())
+		}
+		if len(batch.EcShardInfos) == 0 {
+			continue
+		}
+		if err := send(batch); err != nil {
+			return err
+		}
+	}
+	return nil
+}

weed/topology/volume_stream_bounds_test.go

@@ -0,0 +1,45 @@
+package topology
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+// A disk that registers after the listing went out has nowhere to go in it, so
+// its volumes must not be streamed into a client that cannot place them. It is
+// reported by the next listing instead.
+func TestStreamVolumesStaysInsideTheAnnouncedTopology(t *testing.T) {
+	topo := NewTopology("bounds", nil, 32*1024*1024*1024, 5, false)
+	rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
+	known := rack.GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100})
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: "c", Size: 100, Version: 3},
+		{Id: 2, Collection: "c", Size: 200, Version: 3},
+	}, known)
+
+	listed := topo.ToTopologyInfo(NoVolumes())
+
+	// The heartbeat that lands between the listing and the walk below.
+	late := rack.GetOrCreateDataNode("10.0.0.2", 8080, 18080, "", "", map[string]uint32{"": 100})
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 3, Collection: "c", Size: 300, Version: 3},
+	}, late)
+
+	var streamed []uint32
+	err := topo.StreamVolumes(listed, VolumeFilter{}, 10, func(b *master_pb.VolumeListStreamResponse) error {
+		if b.DataNode == string(late.Id()) {
+			t.Errorf("streamed a batch for %s, which the listing never named", b.DataNode)
+		}
+		for _, v := range b.VolumeInfos {
+			streamed = append(streamed, v.Id)
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !equalIds(streamed, []uint32{1, 2}) {
+		t.Errorf("streamed %v, want the two volumes the listing named", streamed)
+	}
+}