master: let VolumeList ask for the volumes it wants (#10674)

* master: let VolumeList ask for the volumes it wants

The request carried nothing, so every caller was answered with the whole
cluster. A dashboard opening one volume's page, or a capacity probe adding up
one bucket, was served all 800k of them and threw away the rest -- and the
master built every one of those messages first.

The topology, its disks and their counters are still reported in full: a caller
reading free space or replica placement needs the cluster whichever volumes it
asked about. Only what is listed under a disk is selected, ec shards included.

An empty collection and a zero volume id take everything, the way volume.list
already reads its own -collectionPattern and -volumeId, so a caller that
forgets to narrow is answered too much rather than answered wrongly. That
leaves the default collection unnameable, since it is the one the empty string
names, so it gets a field of its own.

An older client sends none of it and is answered exactly as before.

* admin: ask the master for the volume the page is showing

A volume's detail page was pulling every volume in the cluster to find one and
its replicas, and discarding the rest.

* admin: ask the master for the ec volume the page is showing

Same as the volume detail page: one volume's shards were found by pulling every
ec shard in the cluster.

* s3: ask the master for the bucket's own collection

The SOSAPI capacity probe summed one collection's volumes out of a listing of
every volume in the cluster. Cluster capacity still comes out the same: it is
read from the disk counters, which a filtered listing reports in full.

* topology: read the disk usage counters atomically

They are written with atomic.AddInt64 from heartbeats but were read plainly by
the two listings and by FreeSpace, and the map they sit in was iterated without
the lock its neighbour takes. Under -race a listing concurrent with a heartbeat
trips on both.

weed/admin/dash/ec_shard_management.go

@@ -559,7 +559,7 @@ func (s *AdminServer) GetEcVolumeDetails(volumeID uint32, sortBy string, sortOrd
 
 	// Get detailed EC shard information for the specific volume via gRPC
 	err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
-		resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
+		resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{VolumeId: volumeID})
 		if err != nil {
 			return err
 		}
@@ -571,6 +571,7 @@ func (s *AdminServer) GetEcVolumeDetails(volumeID uint32, sortBy string, sortOrd
 						for _, diskInfo := range node.DiskInfos {
 							// Process EC shard information for this specific volume
 							for _, ecShardInfo := range diskInfo.EcShardInfos {
+								// An older master ignores the filter.
 								if ecShardInfo.Id == volumeID {
 									collection = ecShardInfo.Collection
 									dataCenters[dc.Id] = true

weed/admin/dash/volume_management.go

@@ -309,7 +309,7 @@ func (s *AdminServer) sortVolumes(volumes []VolumeWithTopology, sortBy string, s
 }
 
 // GetVolumeDetails retrieves detailed information about a specific volume
-func (s *AdminServer) GetVolumeDetails(volumeID int, server string) (*VolumeDetailsData, error) {
+func (s *AdminServer) GetVolumeDetails(volumeID uint32, server string) (*VolumeDetailsData, error) {
 	var primaryVolume VolumeWithTopology
 	var replicas []VolumeWithTopology
 	var volumeSizeLimit uint64
@@ -317,7 +317,7 @@ func (s *AdminServer) GetVolumeDetails(volumeID int, server string) (*VolumeDeta
 
 	// Find the volume and all its replicas in the cluster
 	err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
-		resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
+		resp, err := client.VolumeList(context.Background(), &master_pb.VolumeListRequest{VolumeId: volumeID})
 		if err != nil {
 			return err
 		}
@@ -328,7 +328,8 @@ func (s *AdminServer) GetVolumeDetails(volumeID int, server string) (*VolumeDeta
 					for _, node := range rack.DataNodeInfos {
 						for _, diskInfo := range node.DiskInfos {
 							for _, volInfo := range diskInfo.VolumeInfos {
-								if int(volInfo.Id) == volumeID {
+								// An older master ignores the filter.
+								if volInfo.Id == volumeID {
 									diskType := volInfo.DiskType
 									if diskType == "" {
 										diskType = "hdd"

weed/admin/handlers/cluster_handlers.go

@@ -155,14 +155,14 @@ func (h *ClusterHandlers) ShowVolumeDetails(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
-	volumeID, err := strconv.Atoi(volumeIDStr)
+	volumeID, err := strconv.ParseUint(volumeIDStr, 10, 32)
 	if err != nil {
 		writeJSONError(w, http.StatusBadRequest, "Invalid volume ID")
 		return
 	}
 
 	// Get volume details
-	volumeDetails, err := h.adminServer.GetVolumeDetails(volumeID, server)
+	volumeDetails, err := h.adminServer.GetVolumeDetails(uint32(volumeID), server)
 	if err != nil {
 		writeJSONError(w, http.StatusInternalServerError, "Failed to get volume details: "+err.Error())
 		return

weed/pb/master.proto

@@ -406,6 +406,13 @@ message TopologyInfo {
   map<string, DiskInfo> diskInfos = 3;
 }
 message VolumeListRequest {
+  // Empty and zero take everything. Only the volumes and ec shards listed
+  // under a disk are selected; the topology and its disk counters are always
+  // reported in full.
+  string collection = 1;
+  uint32 volume_id = 2;
+  // The one collection the empty string cannot name. A named collection wins.
+  bool default_collection_only = 3;
 }
 message VolumeListResponse {
   TopologyInfo topology_info = 1;

weed/pb/master_pb/master.pb.go

@@ -2780,9 +2780,16 @@ func (x *TopologyInfo) GetDiskInfos() map[string]*DiskInfo {
 }
 
 type VolumeListRequest struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Empty and zero take everything. Only the volumes and ec shards listed
+	// under a disk are selected; the topology and its disk counters are always
+	// reported in full.
+	Collection string `protobuf:"bytes,1,opt,name=collection,proto3" json:"collection,omitempty"`
+	VolumeId   uint32 `protobuf:"varint,2,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"`
+	// The one collection the empty string cannot name. A named collection wins.
+	DefaultCollectionOnly bool `protobuf:"varint,3,opt,name=default_collection_only,json=defaultCollectionOnly,proto3" json:"default_collection_only,omitempty"`
+	unknownFields         protoimpl.UnknownFields
+	sizeCache             protoimpl.SizeCache
 }
 
 func (x *VolumeListRequest) Reset() {
@@ -2815,6 +2822,27 @@ func (*VolumeListRequest) Descriptor() ([]byte, []int) {
 	return file_master_proto_rawDescGZIP(), []int{35}
 }
 
+func (x *VolumeListRequest) GetCollection() string {
+	if x != nil {
+		return x.Collection
+	}
+	return ""
+}
+
+func (x *VolumeListRequest) GetVolumeId() uint32 {
+	if x != nil {
+		return x.VolumeId
+	}
+	return 0
+}
+
+func (x *VolumeListRequest) GetDefaultCollectionOnly() bool {
+	if x != nil {
+		return x.DefaultCollectionOnly
+	}
+	return false
+}
+
 type VolumeListResponse struct {
 	state             protoimpl.MessageState `protogen:"open.v1"`
 	TopologyInfo      *TopologyInfo          `protobuf:"bytes,1,opt,name=topology_info,json=topologyInfo,proto3" json:"topology_info,omitempty"`
@@ -5087,8 +5115,13 @@ const file_master_proto_rawDesc = "" +
 	"\tdiskInfos\x18\x03 \x03(\v2&.master_pb.TopologyInfo.DiskInfosEntryR\tdiskInfos\x1aQ\n" +
 	"\x0eDiskInfosEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12)\n" +
-	"\x05value\x18\x02 \x01(\v2\x13.master_pb.DiskInfoR\x05value:\x028\x01\"\x13\n" +
-	"\x11VolumeListRequest\"\x83\x01\n" +
+	"\x05value\x18\x02 \x01(\v2\x13.master_pb.DiskInfoR\x05value:\x028\x01\"\x88\x01\n" +
+	"\x11VolumeListRequest\x12\x1e\n" +
+	"\n" +
+	"collection\x18\x01 \x01(\tR\n" +
+	"collection\x12\x1b\n" +
+	"\tvolume_id\x18\x02 \x01(\rR\bvolumeId\x126\n" +
+	"\x17default_collection_only\x18\x03 \x01(\bR\x15defaultCollectionOnly\"\x83\x01\n" +
 	"\x12VolumeListResponse\x12<\n" +
 	"\rtopology_info\x18\x01 \x01(\v2\x17.master_pb.TopologyInfoR\ftopologyInfo\x12/\n" +
 	"\x14volume_size_limit_mb\x18\x02 \x01(\x04R\x11volumeSizeLimitMb\"4\n" +

weed/s3api/s3api_sosapi.go

@@ -156,9 +156,13 @@ func (s3a *S3ApiServer) getCapacityInfo(ctx context.Context, bucket string) (cap
 		masterMap[string(master)] = master
 	}
 
+	// Cluster capacity below is read from disk counters, which a filtered
+	// listing still reports in full.
+	collectionName := s3a.getCollectionName(bucket)
+
 	// Connect to any available master and get volume list (topology)
 	err = pb.WithOneOfGrpcMasterClients(false, masterMap, s3a.option.GrpcDialOption, func(client master_pb.SeaweedClient) error {
-		resp, vErr := client.VolumeList(ctx, &master_pb.VolumeListRequest{})
+		resp, vErr := client.VolumeList(ctx, &master_pb.VolumeListRequest{Collection: collectionName})
 		if vErr != nil {
 			return vErr
 		}
@@ -168,7 +172,7 @@ func (s3a *S3ApiServer) getCapacityInfo(ctx context.Context, bucket string) (cap
 		}
 
 		// Calculate used size for the bucket by summing up volumes in the collection
-		used = collectBucketUsageFromTopology(resp.TopologyInfo, s3a.getCollectionName(bucket))
+		used = collectBucketUsageFromTopology(resp.TopologyInfo, collectionName)
 
 		// Calculate cluster capacity if no quota
 		if quota > 0 {

weed/server/master_grpc_server_volume.go

@@ -280,7 +280,7 @@ func (ms *MasterServer) VolumeList(ctx context.Context, req *master_pb.VolumeLis
 	}
 
 	resp := &master_pb.VolumeListResponse{
-		TopologyInfo:      ms.Topo.ToTopologyInfo(),
+		TopologyInfo:      ms.Topo.ToTopologyInfo(topology.NewVolumeFilter(req)),
 		VolumeSizeLimitMb: uint64(ms.option.VolumeSizeLimitMB),
 	}
 

weed/topology/collection_statistics_test.go

@@ -117,7 +117,7 @@ func statsTopology(t *testing.T) *Topology {
 func TestCollectionStatisticsMatchesAFullListing(t *testing.T) {
 	topo := statsTopology(t)
 
-	want := referenceCollectionStatistics(topo.ToTopologyInfo())
+	want := referenceCollectionStatistics(topo.ToTopologyInfo(VolumeFilter{}))
 	got := map[string]*CollectionStatistics{}
 	for _, s := range topo.CollectionStatistics() {
 		got[s.Collection] = s

weed/topology/data_center.go

@@ -56,14 +56,14 @@ func (dc *DataCenter) ToInfo() (info DataCenterInfo) {
 	return
 }
 
-func (dc *DataCenter) ToDataCenterInfo() *master_pb.DataCenterInfo {
+func (dc *DataCenter) ToDataCenterInfo(filter VolumeFilter) *master_pb.DataCenterInfo {
 	m := &master_pb.DataCenterInfo{
 		Id:        string(dc.Id()),
 		DiskInfos: dc.diskUsages.ToDiskInfo(),
 	}
 	for _, c := range dc.Children() {
 		rack := c.(*Rack)
-		m.RackInfos = append(m.RackInfos, rack.ToRackInfo())
+		m.RackInfos = append(m.RackInfos, rack.ToRackInfo(filter))
 	}
 	return m
 }

weed/topology/data_node.go

@@ -351,7 +351,7 @@ func (dn *DataNode) ToInfo() (info DataNodeInfo) {
 	return
 }
 
-func (dn *DataNode) ToDataNodeInfo() *master_pb.DataNodeInfo {
+func (dn *DataNode) ToDataNodeInfo(filter VolumeFilter) *master_pb.DataNodeInfo {
 	m := &master_pb.DataNodeInfo{
 		Id: string(dn.Id()),
 		// Start from disk usage counters so empty disks are still represented
@@ -373,7 +373,7 @@ func (dn *DataNode) ToDataNodeInfo() *master_pb.DataNodeInfo {
 
 	for _, c := range dn.Children() {
 		disk := c.(*Disk)
-		m.DiskInfos[string(disk.Id())] = disk.ToDiskInfo()
+		m.DiskInfos[string(disk.Id())] = disk.ToDiskInfo(filter)
 	}
 
 	dn.RLock()

weed/topology/data_node_ec_multi_disk_test.go

@@ -86,7 +86,7 @@ func TestEcShardsAcrossMultipleDisksOnSameNode(t *testing.T) {
 		1: {1, 4, 9},
 		3: {8, 12},
 	}
-	dnInfo := dn.ToDataNodeInfo()
+	dnInfo := dn.ToDataNodeInfo(VolumeFilter{})
 	gotPerDisk := map[uint32][]erasure_coding.ShardId{}
 	for _, diskInfo := range dnInfo.DiskInfos {
 		for _, eci := range diskInfo.EcShardInfos {

weed/topology/data_node_empty_disk_test.go

@@ -21,7 +21,7 @@ func TestToDataNodeInfoReportsEmptyPhysicalDisks(t *testing.T) {
 		{DiskId: 2, Type: "", MaxVolumeCount: 300}, // empty disk, never held a volume
 	})
 
-	info := dn.ToDataNodeInfo()
+	info := dn.ToDataNodeInfo(VolumeFilter{})
 	di, ok := info.DiskInfos[""]
 	if !ok {
 		t.Fatalf("missing HDD disk info")
@@ -61,7 +61,7 @@ func TestToDataNodeInfoKeepsZeroCapacityDisk(t *testing.T) {
 		{DiskId: 2, Type: "", MaxVolumeCount: 0}, // unavailable disk
 	})
 
-	di := dn.ToDataNodeInfo().DiskInfos[""]
+	di := dn.ToDataNodeInfo(VolumeFilter{}).DiskInfos[""]
 	if len(di.MaxVolumeCountByDisk) != 3 {
 		t.Fatalf("want 3 physical disks (incl the zero-capacity one), got %d", len(di.MaxVolumeCountByDisk))
 	}
@@ -81,7 +81,7 @@ func TestToDataNodeInfoFallsBackWhenNoCapacityReported(t *testing.T) {
 		{DiskId: 1},
 	})
 
-	di := dn.ToDataNodeInfo().DiskInfos[""]
+	di := dn.ToDataNodeInfo(VolumeFilter{}).DiskInfos[""]
 	if len(di.MaxVolumeCountByDisk) != 0 {
 		t.Fatalf("want no per-disk max (fallback), got %d", len(di.MaxVolumeCountByDisk))
 	}

weed/topology/disk.go

@@ -98,16 +98,19 @@ func (d *DiskUsages) negative() *DiskUsages {
 }
 
 func (d *DiskUsages) ToDiskInfo() map[string]*master_pb.DiskInfo {
+	d.RLock()
+	defer d.RUnlock()
 	ret := make(map[string]*master_pb.DiskInfo)
 	for diskType, diskUsageCounts := range d.usages {
+		usage := diskUsageCounts.snapshot()
 		m := &master_pb.DiskInfo{
-			VolumeCount:       diskUsageCounts.volumeCount,
-			MaxVolumeCount:    diskUsageCounts.maxVolumeCount,
-			FreeVolumeCount:   diskUsageCounts.maxVolumeCount - (diskUsageCounts.volumeCount - diskUsageCounts.remoteVolumeCount) - ecShardSlots(diskUsageCounts.ecShardCount),
-			ActiveVolumeCount: diskUsageCounts.activeVolumeCount,
-			RemoteVolumeCount: diskUsageCounts.remoteVolumeCount,
-			DiskTotalBytes:    uint64(max(0, diskUsageCounts.diskTotalBytes)),
-			DiskFreeBytes:     uint64(max(0, diskUsageCounts.diskFreeBytes)),
+			VolumeCount:       usage.volumeCount,
+			MaxVolumeCount:    usage.maxVolumeCount,
+			FreeVolumeCount:   usage.maxVolumeCount - (usage.volumeCount - usage.remoteVolumeCount) - ecShardSlots(usage.ecShardCount),
+			ActiveVolumeCount: usage.activeVolumeCount,
+			RemoteVolumeCount: usage.remoteVolumeCount,
+			DiskTotalBytes:    uint64(max(0, usage.diskTotalBytes)),
+			DiskFreeBytes:     uint64(max(0, usage.diskFreeBytes)),
 		}
 		ret[string(diskType)] = m
 	}
@@ -154,8 +157,24 @@ func (a *DiskUsageCounts) addDiskUsageCounts(b *DiskUsageCounts) {
 	atomic.AddInt64(&a.diskFreeBytes, b.diskFreeBytes)
 }
 
+// snapshot reads each counter atomically, so a reader sees whole values rather
+// than ones a concurrent heartbeat is halfway through writing. They are still
+// read one at a time, so they need not all describe the same instant.
+func (a *DiskUsageCounts) snapshot() DiskUsageCounts {
+	return DiskUsageCounts{
+		volumeCount:       atomic.LoadInt64(&a.volumeCount),
+		remoteVolumeCount: atomic.LoadInt64(&a.remoteVolumeCount),
+		activeVolumeCount: atomic.LoadInt64(&a.activeVolumeCount),
+		ecShardCount:      atomic.LoadInt64(&a.ecShardCount),
+		maxVolumeCount:    atomic.LoadInt64(&a.maxVolumeCount),
+		diskTotalBytes:    atomic.LoadInt64(&a.diskTotalBytes),
+		diskFreeBytes:     atomic.LoadInt64(&a.diskFreeBytes),
+	}
+}
+
 func (a *DiskUsageCounts) FreeSpace() int64 {
-	return a.maxVolumeCount + a.remoteVolumeCount - a.volumeCount - ecShardSlots(a.ecShardCount)
+	u := a.snapshot()
+	return u.maxVolumeCount + u.remoteVolumeCount - u.volumeCount - ecShardSlots(u.ecShardCount)
 }
 
 func (du *DiskUsages) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts {
@@ -368,24 +387,35 @@ func (d *Disk) FreeSpace() int64 {
 	return t.FreeSpace()
 }
 
-func (d *Disk) ToDiskInfo() *master_pb.DiskInfo {
-	diskUsage := d.diskUsages.getOrCreateDisk(types.ToDiskType(string(d.Id())))
+func (d *Disk) ToDiskInfo(filter VolumeFilter) *master_pb.DiskInfo {
+	diskUsage := d.diskUsages.getOrCreateDisk(types.ToDiskType(string(d.Id()))).snapshot()
 
 	// Built under the read lock rather than from a copy as large as the
 	// messages it fed. Nothing here re-enters the topology, so the hold is safe.
 	d.RLock()
-	volumeInfos := make([]*master_pb.VolumeInformationMessage, 0, len(d.volumes))
+	// Reserving room for every volume would keep what a filter set out not to
+	// build.
+	capacity := 0
+	if filter.SelectsEverything() {
+		capacity = len(d.volumes)
+	}
+	volumeInfos := make([]*master_pb.VolumeInformationMessage, 0, capacity)
 	var diskId uint32
+	var haveDiskId bool
 	for _, v := range d.volumes {
-		if len(volumeInfos) == 0 {
-			diskId = v.DiskId
+		// Any volume names the disk, including one filtered out.
+		if !haveDiskId {
+			diskId, haveDiskId = v.DiskId, true
+		}
+		if !filter.matches(v.Collection, v.Id) {
+			continue
 		}
 		volumeInfos = append(volumeInfos, v.ToVolumeInformationMessage())
 	}
 	d.RUnlock()
 
 	ecShards := d.GetEcShards()
-	if len(volumeInfos) == 0 && len(ecShards) > 0 {
+	if !haveDiskId && len(ecShards) > 0 {
 		diskId = ecShards[0].DiskId
 	}
 
@@ -401,8 +431,15 @@ func (d *Disk) ToDiskInfo() *master_pb.DiskInfo {
 		DiskFreeBytes:     uint64(max(0, diskUsage.diskFreeBytes)),
 	}
 	m.VolumeInfos = volumeInfos
-	m.EcShardInfos = make([]*master_pb.VolumeEcShardInformationMessage, 0, len(ecShards))
+	ecCapacity := 0
+	if filter.SelectsEverything() {
+		ecCapacity = len(ecShards)
+	}
+	m.EcShardInfos = make([]*master_pb.VolumeEcShardInformationMessage, 0, ecCapacity)
 	for _, ecv := range ecShards {
+		if !filter.matches(ecv.Collection, ecv.VolumeId) {
+			continue
+		}
 		m.EcShardInfos = append(m.EcShardInfos, ecv.ToVolumeEcShardInformationMessage())
 	}
 	return m

weed/topology/disk_info_test.go

@@ -23,7 +23,7 @@ func TestDiskInfoReportsEveryVolume(t *testing.T) {
 
 	var info *master_pb.DiskInfo
 	for _, c := range dn.Children() {
-		info = c.(*Disk).ToDiskInfo()
+		info = c.(*Disk).ToDiskInfo(VolumeFilter{})
 	}
 	if info == nil {
 		t.Fatal("the node reported no disk")
@@ -58,7 +58,7 @@ func TestDiskInfoReportsTheDiskIdOfEcOnlyDisks(t *testing.T) {
 	}, dn)
 
 	for _, c := range dn.Children() {
-		info := c.(*Disk).ToDiskInfo()
+		info := c.(*Disk).ToDiskInfo(VolumeFilter{})
 		if len(info.VolumeInfos) != 0 {
 			t.Fatalf("expected no regular volumes, got %d", len(info.VolumeInfos))
 		}

weed/topology/rack.go

@@ -148,14 +148,14 @@ func (r *Rack) ToInfo() (info RackInfo) {
 	return
 }
 
-func (r *Rack) ToRackInfo() *master_pb.RackInfo {
+func (r *Rack) ToRackInfo(filter VolumeFilter) *master_pb.RackInfo {
 	m := &master_pb.RackInfo{
 		Id:        string(r.Id()),
 		DiskInfos: r.diskUsages.ToDiskInfo(),
 	}
 	for _, c := range r.Children() {
 		dn := c.(*DataNode)
-		m.DataNodeInfos = append(m.DataNodeInfos, dn.ToDataNodeInfo())
+		m.DataNodeInfos = append(m.DataNodeInfos, dn.ToDataNodeInfo(filter))
 	}
 	return m
 }

weed/topology/topology_info.go

@@ -124,14 +124,14 @@ func (t *Topology) ToVolumeLocations() (volumeLocations []*master_pb.VolumeLocat
 	return
 }
 
-func (t *Topology) ToTopologyInfo() *master_pb.TopologyInfo {
+func (t *Topology) ToTopologyInfo(filter VolumeFilter) *master_pb.TopologyInfo {
 	m := &master_pb.TopologyInfo{
 		Id:        string(t.Id()),
 		DiskInfos: t.diskUsages.ToDiskInfo(),
 	}
 	for _, c := range t.Children() {
 		dc := c.(*DataCenter)
-		m.DataCenterInfos = append(m.DataCenterInfos, dc.ToDataCenterInfo())
+		m.DataCenterInfos = append(m.DataCenterInfos, dc.ToDataCenterInfo(filter))
 	}
 	return m
 }

weed/topology/topology_test.go

@@ -176,7 +176,7 @@ func TestDataNodeToDataNodeInfo_IncludeEmptyDiskFromUsage(t *testing.T) {
 	usage := dn.diskUsages.getOrCreateDisk(types.HardDriveType)
 	usage.maxVolumeCount = 8
 
-	info := dn.ToDataNodeInfo()
+	info := dn.ToDataNodeInfo(VolumeFilter{})
 	diskInfo, found := info.DiskInfos[""]
 	if !found {
 		t.Fatalf("expected default disk entry for empty node")

weed/topology/volume_filter.go

@@ -0,0 +1,49 @@
+package topology
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// VolumeFilter narrows a topology listing to the volumes a caller asked about,
+// selecting only what is listed under a disk. A nil field filters nothing, and
+// only nil does: the empty collection is a real one.
+type VolumeFilter struct {
+	Collection *string
+	VolumeId   *needle.VolumeId
+}
+
+// NewVolumeFilter reads what a VolumeList request asked for, where empty and
+// zero mean everything so a caller that forgets to narrow gets too much rather
+// than the wrong thing.
+func NewVolumeFilter(req *master_pb.VolumeListRequest) VolumeFilter {
+	var filter VolumeFilter
+	switch {
+	case req.Collection != "":
+		collection := req.Collection
+		filter.Collection = &collection
+	case req.DefaultCollectionOnly:
+		defaultCollection := ""
+		filter.Collection = &defaultCollection
+	}
+	if req.VolumeId != 0 {
+		volumeId := needle.VolumeId(req.VolumeId)
+		filter.VolumeId = &volumeId
+	}
+	return filter
+}
+
+// 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
+}
+
+func (f VolumeFilter) matches(collection string, id needle.VolumeId) bool {
+	if f.Collection != nil && *f.Collection != collection {
+		return false
+	}
+	if f.VolumeId != nil && *f.VolumeId != id {
+		return false
+	}
+	return true
+}

weed/topology/volume_filter_test.go

@@ -0,0 +1,209 @@
+package topology
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+func filterTestTopology(t *testing.T) *Topology {
+	t.Helper()
+	topo := NewTopology("filter", nil, 32*1024*1024*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100})
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: "", Size: 1000, Version: 3, DiskId: 2},
+		{Id: 2, Collection: "c", Size: 2000, Version: 3, DiskId: 2},
+		{Id: 3, Collection: "other", Size: 3000, Version: 3, DiskId: 2},
+	}, dn)
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		{Id: 8, Collection: "c", EcIndexBits: 0x3fff, DiskId: 2},
+		{Id: 9, Collection: "other", EcIndexBits: 0x3fff, DiskId: 2},
+	}, dn)
+	return topo
+}
+
+// listed returns the volume and ec ids a filtered listing carries.
+func listed(info *master_pb.TopologyInfo) (volumes []uint32, ecVolumes []uint32) {
+	for _, dc := range info.DataCenterInfos {
+		for _, rack := range dc.RackInfos {
+			for _, node := range rack.DataNodeInfos {
+				for _, disk := range node.DiskInfos {
+					for _, v := range disk.VolumeInfos {
+						volumes = append(volumes, v.Id)
+					}
+					for _, ec := range disk.EcShardInfos {
+						ecVolumes = append(ecVolumes, ec.Id)
+					}
+				}
+			}
+		}
+	}
+	return
+}
+
+func equalIds(got, want []uint32) bool {
+	if len(got) != len(want) {
+		return false
+	}
+	seen := map[uint32]int{}
+	for _, id := range got {
+		seen[id]++
+	}
+	for _, id := range want {
+		seen[id]--
+	}
+	for _, n := range seen {
+		if n != 0 {
+			return false
+		}
+	}
+	return true
+}
+
+func TestVolumeFilterSelects(t *testing.T) {
+	topo := filterTestTopology(t)
+	collection := func(name string) *string { return &name }
+	volume := func(id uint32) *needle.VolumeId { v := needle.VolumeId(id); return &v }
+
+	for _, tc := range []struct {
+		name          string
+		filter        VolumeFilter
+		wantVolumes   []uint32
+		wantEcVolumes []uint32
+	}{
+		{"no filter takes everything", VolumeFilter{}, []uint32{1, 2, 3}, []uint32{8, 9}},
+		{"one collection", VolumeFilter{Collection: collection("c")}, []uint32{2}, []uint32{8}},
+		// Asking for it must not read as asking for everything.
+		{"the default collection", VolumeFilter{Collection: collection("")}, []uint32{1}, nil},
+		{"a collection nothing is in", VolumeFilter{Collection: collection("none")}, nil, nil},
+		{"one volume", VolumeFilter{VolumeId: volume(3)}, []uint32{3}, nil},
+		{"one ec volume", VolumeFilter{VolumeId: volume(9)}, nil, []uint32{9}},
+		{"both, agreeing", VolumeFilter{Collection: collection("other"), VolumeId: volume(3)}, []uint32{3}, nil},
+		{"both, disagreeing", VolumeFilter{Collection: collection("c"), VolumeId: volume(3)}, nil, nil},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			volumes, ecVolumes := listed(topo.ToTopologyInfo(tc.filter))
+			if !equalIds(volumes, tc.wantVolumes) {
+				t.Errorf("listed volumes %v, want %v", volumes, tc.wantVolumes)
+			}
+			if !equalIds(ecVolumes, tc.wantEcVolumes) {
+				t.Errorf("listed ec volumes %v, want %v", ecVolumes, tc.wantEcVolumes)
+			}
+		})
+	}
+}
+
+// A filter never changes which disks are reported or what they say about
+// themselves.
+func TestVolumeFilterKeepsTheTopology(t *testing.T) {
+	topo := filterTestTopology(t)
+	full := topo.ToTopologyInfo(VolumeFilter{})
+	name := "c"
+	narrowed := topo.ToTopologyInfo(VolumeFilter{Collection: &name})
+
+	for _, info := range []*master_pb.TopologyInfo{full, narrowed} {
+		if len(info.DataCenterInfos) != 1 || len(info.DataCenterInfos[0].RackInfos) != 1 {
+			t.Fatalf("listing lost its topology: %v", info)
+		}
+	}
+
+	fullDisk := full.DataCenterInfos[0].RackInfos[0].DataNodeInfos[0].DiskInfos[""]
+	narrowedDisk := narrowed.DataCenterInfos[0].RackInfos[0].DataNodeInfos[0].DiskInfos[""]
+	if narrowedDisk == nil {
+		t.Fatal("the filtered listing dropped the disk")
+	}
+	if narrowedDisk.VolumeCount != fullDisk.VolumeCount {
+		t.Errorf("volume count %d, want the disk's own %d", narrowedDisk.VolumeCount, fullDisk.VolumeCount)
+	}
+	if narrowedDisk.MaxVolumeCount != fullDisk.MaxVolumeCount {
+		t.Errorf("max volume count %d, want %d", narrowedDisk.MaxVolumeCount, fullDisk.MaxVolumeCount)
+	}
+	if narrowedDisk.DiskId != fullDisk.DiskId {
+		t.Errorf("disk id %d, want %d", narrowedDisk.DiskId, fullDisk.DiskId)
+	}
+}
+
+// So a caller that finds nothing can tell an empty answer from a missing disk.
+func TestVolumeFilterKeepsTheDiskIdWhenNothingMatches(t *testing.T) {
+	topo := filterTestTopology(t)
+	name := "none"
+	info := topo.ToTopologyInfo(VolumeFilter{Collection: &name})
+	disk := info.DataCenterInfos[0].RackInfos[0].DataNodeInfos[0].DiskInfos[""]
+	if len(disk.VolumeInfos) != 0 {
+		t.Fatalf("expected nothing listed, got %d volumes", len(disk.VolumeInfos))
+	}
+	if disk.DiskId != 2 {
+		t.Errorf("reported disk id %d, want the one its volumes are on", disk.DiskId)
+	}
+}
+
+func TestNewVolumeFilterReadsTheRequest(t *testing.T) {
+	collectionOf := func(f VolumeFilter) string {
+		if f.Collection == nil {
+			return "<every>"
+		}
+		return *f.Collection
+	}
+
+	for _, tc := range []struct {
+		name    string
+		request *master_pb.VolumeListRequest
+		want    string
+	}{
+		// A caller passing through its own "" is answered too much, not wrongly.
+		{"an empty request", &master_pb.VolumeListRequest{}, "<every>"},
+		{"an empty collection", &master_pb.VolumeListRequest{Collection: ""}, "<every>"},
+		{"a named collection", &master_pb.VolumeListRequest{Collection: "c"}, "c"},
+		{"the default collection", &master_pb.VolumeListRequest{DefaultCollectionOnly: true}, ""},
+		// Contradictory, so the more specific of the two wins.
+		{"both", &master_pb.VolumeListRequest{Collection: "c", DefaultCollectionOnly: true}, "c"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := collectionOf(NewVolumeFilter(tc.request)); got != tc.want {
+				t.Errorf("selected collection %q, want %q", got, tc.want)
+			}
+		})
+	}
+
+	if f := NewVolumeFilter(&master_pb.VolumeListRequest{}); f.VolumeId != nil {
+		t.Error("a zero volume id must not filter")
+	}
+	f := NewVolumeFilter(&master_pb.VolumeListRequest{VolumeId: 7})
+	if f.VolumeId == nil || uint32(*f.VolumeId) != 7 {
+		t.Errorf("volume id not carried across: %v", f.VolumeId)
+	}
+}
+
+// Everything a listing reads off a disk must be read under its lock, including
+// how much room to reserve, or a heartbeat writing the map races it.
+func TestVolumeFilterListsWhileVolumesChange(t *testing.T) {
+	topo := filterTestTopology(t)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("10.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100})
+
+	stop := make(chan struct{})
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := 0; ; i++ {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+				{Id: uint32(100 + i%50), Collection: "c", Size: 1, Version: 3, DiskId: 2},
+			}, dn)
+		}
+	}()
+
+	name := "c"
+	for i := 0; i < 200; i++ {
+		topo.ToTopologyInfo(VolumeFilter{})
+		topo.ToTopologyInfo(VolumeFilter{Collection: &name})
+	}
+	close(stop)
+	<-done
+}