master: count EC volumes in statistics used size (#10457)

Statistics aggregates the volume layouts of a collection, but EC volumes
are tracked outside collectionMap, so they were reported as nothing. A
mount over a cluster whose volumes have mostly been encoded showed a df
used size of a few GiB against terabytes of EC data.

Walk the data nodes and add the EC volumes of the requested collection.
Every shard copy counts, parity included, the way a regular volume's used
size counts every replica, so used size stays the space the cluster
actually occupies.

File count comes from the volume-wide .ecx and .ecj counts, taking the
largest a holder reports rather than summing them: both files travel with
the shards on a move, so several nodes can report the same tombstones.

weed/server/master_grpc_server_volume.go

@@ -205,7 +205,8 @@ func (ms *MasterServer) Statistics(ctx context.Context, req *master_pb.Statistic
 	}
 
 	// an empty collection means all collections, and a named collection covers
-	// all its layouts, so used size matches the topology-wide total size below
+	// all its layouts and EC volumes, so used size matches the topology-wide
+	// total size below
 	stats := ms.Topo.CollectionVolumeStats(req.Collection)
 	totalSize := ms.Topo.GetDiskUsages().GetMaxVolumeCount() * int64(ms.option.VolumeSizeLimitMB) * 1024 * 1024
 	resp := &master_pb.StatisticsResponse{

weed/topology/topology.go

@@ -431,8 +431,9 @@ func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.Replic
 	}).(*Collection).GetOrCreateVolumeLayout(rp, ttl, diskType)
 }
 
-// CollectionVolumeStats aggregates stats across all volume layouts of one
-// collection, or across every collection when collectionName is empty.
+// CollectionVolumeStats aggregates stats across all volume layouts and EC
+// volumes of one collection, or across every collection when collectionName is
+// empty.
 func (t *Topology) CollectionVolumeStats(collectionName string) *VolumeLayoutStats {
 	ret := &VolumeLayoutStats{}
 	var collections []*Collection
@@ -451,6 +452,12 @@ func (t *Topology) CollectionVolumeStats(collectionName string) *VolumeLayoutSta
 			ret.FileCount += stats.FileCount
 		}
 	}
+	// EC volumes live outside collectionMap, so a collection whose volumes are
+	// all encoded has no layout left to report them
+	ecStats := t.CollectionEcVolumeStats(collectionName)
+	ret.TotalSize += ecStats.TotalSize
+	ret.UsedSize += ecStats.UsedSize
+	ret.FileCount += ecStats.FileCount
 	return ret
 }
 

weed/topology/topology_ec.go

@@ -172,6 +172,64 @@ func (t *Topology) LookupEcShards(vid needle.VolumeId) (locations *EcShardLocati
 	return
 }
 
+// ecVolumeCounts accumulates one EC volume's needle counts while they are
+// collected from every node reporting its shards.
+type ecVolumeCounts struct {
+	fileCount   uint64
+	deleteCount uint64
+}
+
+// CollectionEcVolumeStats sums the disk footprint and live needle count of the
+// EC volumes in one collection, or in every collection when collectionName is
+// empty. Every shard copy counts, parity included, the way a regular volume's
+// used size counts every replica; needle counts are per volume, again as a
+// regular volume reports them.
+func (t *Topology) CollectionEcVolumeStats(collectionName string) *VolumeLayoutStats {
+	ret := &VolumeLayoutStats{}
+	perVolume := make(map[needle.VolumeId]*ecVolumeCounts)
+
+	for _, c := range t.Children() {
+		for _, r := range c.(*DataCenter).Children() {
+			for _, n := range r.(*Rack).Children() {
+				for _, ecInfo := range n.(*DataNode).GetEcShards() {
+					if collectionName != "" && ecInfo.Collection != collectionName {
+						continue
+					}
+					ret.UsedSize += uint64(ecInfo.ShardsInfo.TotalSize())
+					counts, found := perVolume[ecInfo.VolumeId]
+					if !found {
+						counts = &ecVolumeCounts{}
+						perVolume[ecInfo.VolumeId] = counts
+					}
+					// .ecx and .ecj are both volume-wide files that travel with
+					// the shards, so take the largest count any holder reports
+					// rather than summing: a node still loading .ecx reports 0
+					// and must not pin the total down, and a shard move copies
+					// the journal, so several holders can report the same
+					// tombstones. Deletes recorded only on another holder since
+					// then are missed, which errs toward reporting files that
+					// are gone rather than losing a whole volume's count.
+					if ecInfo.FileCount > counts.fileCount {
+						counts.fileCount = ecInfo.FileCount
+					}
+					if ecInfo.DeleteCount > counts.deleteCount {
+						counts.deleteCount = ecInfo.DeleteCount
+					}
+				}
+			}
+		}
+	}
+
+	// an EC volume is sealed, so it offers no room beyond what it holds
+	ret.TotalSize = ret.UsedSize
+	for _, counts := range perVolume {
+		if counts.fileCount > counts.deleteCount {
+			ret.FileCount += counts.fileCount - counts.deleteCount
+		}
+	}
+	return ret
+}
+
 func (t *Topology) ListEcServersByCollection(collection string) (dataNodes []pb.ServerAddress) {
 	t.ecShardMapLock.RLock()
 	defer t.ecShardMapLock.RUnlock()

weed/topology/topology_stats_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 )
 
@@ -45,3 +46,56 @@ func TestCollectionVolumeStats(t *testing.T) {
 		t.Errorf("stats query should not create a phantom collection")
 	}
 }
+
+// ecShardMessage builds a heartbeat message for the given shard ids, sized
+// (id+1)*sizeUnit bytes each.
+func ecShardMessage(vid uint32, collection string, sizeUnit int, fileCount, deleteCount uint64, shardIds ...erasure_coding.ShardId) *master_pb.VolumeEcShardInformationMessage {
+	shards := erasure_coding.NewShardsInfo()
+	for _, id := range shardIds {
+		shards.Set(erasure_coding.NewShardInfo(id, erasure_coding.ShardSize((int(id)+1)*sizeUnit)))
+	}
+	return &master_pb.VolumeEcShardInformationMessage{
+		Id:          vid,
+		Collection:  collection,
+		EcIndexBits: shards.Bitmap(),
+		ShardSizes:  shards.SizesInt64(),
+		FileCount:   fileCount,
+		DeleteCount: deleteCount,
+	}
+}
+
+func TestCollectionVolumeStatsWithEcVolumes(t *testing.T) {
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+
+	rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
+	maxVolumeCounts := map[string]uint32{"": 25}
+	dn1 := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", maxVolumeCounts)
+	dn2 := rack.GetOrCreateDataNode("127.0.0.2", 34534, 0, "127.0.0.2", "", maxVolumeCounts)
+
+	// volume 10 spans both nodes, and shard 0 has a second copy on dn2. dn2 has
+	// not finished loading its .ecx yet and reports a zero file count, and both
+	// nodes report the 5 tombstones of the journal they share.
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		ecShardMessage(10, "c1", 100, 50, 5, 0, 1, 2, 3, 4, 5, 6),
+		ecShardMessage(20, "", 10, 7, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13),
+	}, dn1)
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		ecShardMessage(10, "c1", 100, 0, 5, 0, 7, 8, 9, 10, 11, 12, 13),
+	}, dn2)
+
+	// every shard copy of volume 10, parity included, each sized (id+1)*100:
+	// 100..700 on dn1, plus a second copy of shard 0 and 800..1400 on dn2
+	c1Stats := topo.CollectionVolumeStats("c1")
+	assert(t, "c1 ec used size", int(c1Stats.UsedSize), 2800+7800)
+	// the shared journal counts once, not once per holder: 50 - 5
+	assert(t, "c1 ec file count", int(c1Stats.FileCount), 45)
+
+	// volume 20 adds all 14 shards, each sized (id+1)*10
+	allStats := topo.CollectionVolumeStats("")
+	assert(t, "all collections ec used size", int(allStats.UsedSize), 10600+1050)
+	assert(t, "all collections ec file count", int(allStats.FileCount), 45+7)
+
+	if _, found := topo.FindCollection("c1"); found {
+		t.Errorf("ec stats query should not create a phantom collection")
+	}
+}