telemetry: count erasure-coded volumes in the reported totals (#10632)

collectVolumeStats walked only DataNode.GetVolumes(), which returns the
regular volumes on each disk. An encoded volume leaves that set and is
reported through GetEcShards instead, so total_disk_bytes and
total_volume_count silently excluded every erasure-coded volume: a cluster
that encoded everything reported zero bytes and zero volumes while still
counting as a volume server.

Sum each holder's shard sizes into the byte total, parity and extra copies
included, matching how a replicated volume's used size counts every replica
and how CollectionEcVolumeStats already reports EC footprint. Count volume
ids rather than shard entries, since one volume's shards are spread over
many nodes and would otherwise multiply the volume count by the number of
holders.

weed/telemetry/collector.go

@@ -6,6 +6,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/telemetry/proto"
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/topology"
 )
 
@@ -152,6 +153,7 @@ func (c *Collector) countVolumeServers() int {
 func (c *Collector) collectVolumeStats() (uint64, int) {
 	var totalDiskBytes uint64
 	var totalVolumeCount int
+	ecVolumeIds := make(map[needle.VolumeId]struct{})
 
 	for _, dcNode := range c.topo.Children() {
 		dc := dcNode.(*topology.DataCenter)
@@ -164,11 +166,21 @@ func (c *Collector) collectVolumeStats() (uint64, int) {
 					totalVolumeCount++
 					totalDiskBytes += volumeInfo.Size
 				}
+				// An encoded volume leaves GetVolumes and is reported as
+				// shards, so without this a cluster reports none of the
+				// bytes it erasure-coded. Every shard copy counts, parity
+				// included, the way a replicated volume counts every replica.
+				for _, ecInfo := range dn.GetEcShards() {
+					totalDiskBytes += uint64(ecInfo.ShardsInfo.TotalSize())
+					// One volume's shards are spread over many nodes, so
+					// count the volume once rather than once per holder.
+					ecVolumeIds[ecInfo.VolumeId] = struct{}{}
+				}
 			}
 		}
 	}
 
-	return totalDiskBytes, totalVolumeCount
+	return totalDiskBytes, totalVolumeCount + len(ecVolumeIds)
 }
 
 // countFilers counts the number of active filer servers across all groups

weed/telemetry/collector_test.go

@@ -0,0 +1,66 @@
+package telemetry
+
+import (
+	"testing"
+
+	"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"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+// ecShardMessage builds a heartbeat message for the given shard ids, each
+// sized (id+1)*sizeUnit bytes.
+func ecShardMessage(vid uint32, sizeUnit int, 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,
+		EcIndexBits: shards.Bitmap(),
+		ShardSizes:  shards.SizesInt64(),
+	}
+}
+
+func TestCollectVolumeStatsCountsEcShards(t *testing.T) {
+	topo := topology.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)
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Size: 1000, FileCount: 10, Version: uint32(needle.GetCurrentVersion())},
+		{Id: 2, Size: 2000, FileCount: 20, Version: uint32(needle.GetCurrentVersion())},
+	}, dn1)
+
+	// Volume 10's shards span both nodes, with a second copy of shard 0 on
+	// dn2; volume 20 sits entirely on dn1.
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		ecShardMessage(10, 100, 0, 1, 2, 3, 4, 5, 6),
+		ecShardMessage(20, 10, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13),
+	}, dn1)
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		ecShardMessage(10, 100, 0, 7, 8, 9, 10, 11, 12, 13),
+	}, dn2)
+
+	collector := NewCollector(nil, topo, nil)
+	diskBytes, volumeCount := collector.collectVolumeStats()
+
+	// 3000 from the two regular volumes, plus every shard copy of volume 10
+	// sized (id+1)*100 — 100..700 on dn1, and 800..1400 plus the second copy
+	// of shard 0 on dn2 — and volume 20's 14 shards at (id+1)*10.
+	const expectedDiskBytes = 3000 + 2800 + (7700 + 100) + 1050
+	if diskBytes != expectedDiskBytes {
+		t.Errorf("Expected %d total disk bytes, got %d", expectedDiskBytes, diskBytes)
+	}
+
+	// 2 regular volumes plus EC volumes 10 and 20 — volume 10 counts once
+	// even though two nodes hold its shards.
+	if volumeCount != 4 {
+		t.Errorf("Expected 4 volumes, got %d", volumeCount)
+	}
+}