Files
seaweedfs/weed/telemetry/collector_test.go
T
Chris Lu 0973634fd4 telemetry: keep only clusters that store at least 10 GiB (#11138)
* telemetry: tidy the server module after the protobuf bump

Claude-Session: https://claude.ai/code/session_01VGiphDxpsKMwu9XpUFhybQ

* telemetry: keep only clusters that store at least 10 GiB

Fresh weed server runs, CI jobs and throwaway containers each mint their
own cluster id. They came in at tens of thousands a day, were most of
the counted clusters and held almost none of the bytes, and the state
file and the metrics page grew with every one of them. Reports under
the floor are counted and dropped, and a state file written before the
floor sheds them on the first restart.

Claude-Session: https://claude.ai/code/session_01VGiphDxpsKMwu9XpUFhybQ

* master: report telemetry only once the cluster stores 10 GiB

A throwaway cluster no longer registers itself with its first report a
minute after start; a real one begins reporting at the first daily tick
after it crosses the floor.

Claude-Session: https://claude.ai/code/session_01VGiphDxpsKMwu9XpUFhybQ
2026-09-03 12:36:00 -07:00

88 lines
3.7 KiB
Go

package telemetry
import (
"testing"
"github.com/seaweedfs/seaweedfs/telemetry/proto"
"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)
}
}
func TestCollectDataWaitsForMinDiskBytes(t *testing.T) {
topo := topology.NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 25})
collector := NewCollector(nil, topo, nil)
under := &master_pb.VolumeInformationMessage{Id: 1, Size: proto.MinDiskBytes - 1, Version: uint32(needle.GetCurrentVersion())}
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{under}, dn)
if data := collector.collectData(); data != nil {
t.Fatalf("cluster under the floor reported %+v", data)
}
one := &master_pb.VolumeInformationMessage{Id: 2, Size: 1, Version: uint32(needle.GetCurrentVersion())}
topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{under, one}, dn)
data := collector.collectData()
if data == nil || data.TotalDiskBytes != proto.MinDiskBytes {
t.Fatalf("cluster on the floor reported %+v", data)
}
}