From a2ff9cca27fc0ba114e8f499585eebfd880dcadb Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 9 Aug 2026 21:59:42 -0700 Subject: [PATCH] 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 | 3 +- weed/admin/dash/volume_management.go | 7 +- weed/admin/handlers/cluster_handlers.go | 4 +- weed/pb/master.proto | 7 + weed/pb/master_pb/master.pb.go | 43 +++- weed/s3api/s3api_sosapi.go | 8 +- weed/server/master_grpc_server_volume.go | 2 +- weed/topology/collection_statistics_test.go | 2 +- weed/topology/data_center.go | 4 +- weed/topology/data_node.go | 4 +- weed/topology/data_node_ec_multi_disk_test.go | 2 +- weed/topology/data_node_empty_disk_test.go | 6 +- weed/topology/disk.go | 67 ++++-- weed/topology/disk_info_test.go | 4 +- weed/topology/rack.go | 4 +- weed/topology/topology_info.go | 4 +- weed/topology/topology_test.go | 2 +- weed/topology/volume_filter.go | 49 ++++ weed/topology/volume_filter_test.go | 209 ++++++++++++++++++ 19 files changed, 386 insertions(+), 45 deletions(-) create mode 100644 weed/topology/volume_filter.go create mode 100644 weed/topology/volume_filter_test.go diff --git a/weed/admin/dash/ec_shard_management.go b/weed/admin/dash/ec_shard_management.go index 7dc9277c6..aed62f4ee 100644 --- a/weed/admin/dash/ec_shard_management.go +++ b/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 diff --git a/weed/admin/dash/volume_management.go b/weed/admin/dash/volume_management.go index a0543a647..ab26ea774 100644 --- a/weed/admin/dash/volume_management.go +++ b/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" diff --git a/weed/admin/handlers/cluster_handlers.go b/weed/admin/handlers/cluster_handlers.go index 1d7d490ab..5a453a17c 100644 --- a/weed/admin/handlers/cluster_handlers.go +++ b/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 diff --git a/weed/pb/master.proto b/weed/pb/master.proto index 37ee52a80..7ef6bbc4c 100644 --- a/weed/pb/master.proto +++ b/weed/pb/master.proto @@ -406,6 +406,13 @@ message TopologyInfo { map 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; diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index 45bf1606d..751edb0f4 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/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" + diff --git a/weed/s3api/s3api_sosapi.go b/weed/s3api/s3api_sosapi.go index e0594a6dc..f1c635e7a 100644 --- a/weed/s3api/s3api_sosapi.go +++ b/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 { diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 2b95ad94f..2ec6c407d 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/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), } diff --git a/weed/topology/collection_statistics_test.go b/weed/topology/collection_statistics_test.go index b7d5f5b1c..4c8078277 100644 --- a/weed/topology/collection_statistics_test.go +++ b/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 diff --git a/weed/topology/data_center.go b/weed/topology/data_center.go index e036621b4..b461fb07a 100644 --- a/weed/topology/data_center.go +++ b/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 } diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index 847d30cf8..19c8b2007 100644 --- a/weed/topology/data_node.go +++ b/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() diff --git a/weed/topology/data_node_ec_multi_disk_test.go b/weed/topology/data_node_ec_multi_disk_test.go index 698a4d114..cd294ca05 100644 --- a/weed/topology/data_node_ec_multi_disk_test.go +++ b/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 { diff --git a/weed/topology/data_node_empty_disk_test.go b/weed/topology/data_node_empty_disk_test.go index 74c7e15e3..06c6fd614 100644 --- a/weed/topology/data_node_empty_disk_test.go +++ b/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)) } diff --git a/weed/topology/disk.go b/weed/topology/disk.go index 527d655b6..1a952b481 100644 --- a/weed/topology/disk.go +++ b/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 diff --git a/weed/topology/disk_info_test.go b/weed/topology/disk_info_test.go index 53f2296f5..627807e4b 100644 --- a/weed/topology/disk_info_test.go +++ b/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)) } diff --git a/weed/topology/rack.go b/weed/topology/rack.go index 4facd30f6..876122b81 100644 --- a/weed/topology/rack.go +++ b/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 } diff --git a/weed/topology/topology_info.go b/weed/topology/topology_info.go index 1c9bfa6b9..c60b68329 100644 --- a/weed/topology/topology_info.go +++ b/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 } diff --git a/weed/topology/topology_test.go b/weed/topology/topology_test.go index ec3f4585e..4e4aa665e 100644 --- a/weed/topology/topology_test.go +++ b/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") diff --git a/weed/topology/volume_filter.go b/weed/topology/volume_filter.go new file mode 100644 index 000000000..bec87862b --- /dev/null +++ b/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 +} diff --git a/weed/topology/volume_filter_test.go b/weed/topology/volume_filter_test.go new file mode 100644 index 000000000..0fcbfad5a --- /dev/null +++ b/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 "" + } + 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{}, ""}, + {"an empty collection", &master_pb.VolumeListRequest{Collection: ""}, ""}, + {"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 +}