From 0973634fd4ea0a183913ef752fef5eb7ea418485 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 3 Sep 2026 12:36:00 -0700 Subject: [PATCH] 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 --- telemetry/README.md | 3 + telemetry/proto/floor.go | 7 ++ telemetry/server/api/handlers_test.go | 2 +- telemetry/server/go.mod | 2 +- telemetry/server/go.sum | 4 +- telemetry/server/storage/confirmed_test.go | 2 +- telemetry/server/storage/floor_test.go | 90 ++++++++++++++++++++ telemetry/server/storage/history_test.go | 8 +- telemetry/server/storage/persistence.go | 9 +- telemetry/server/storage/persistence_test.go | 4 +- telemetry/server/storage/prometheus.go | 14 ++- telemetry/server/storage/sizes_test.go | 10 +-- weed/telemetry/collector.go | 15 +++- weed/telemetry/collector_test.go | 21 +++++ 14 files changed, 168 insertions(+), 23 deletions(-) create mode 100644 telemetry/proto/floor.go create mode 100644 telemetry/server/storage/floor_test.go diff --git a/telemetry/README.md b/telemetry/README.md index 4edd58e27..85fc48598 100644 --- a/telemetry/README.md +++ b/telemetry/README.md @@ -68,6 +68,8 @@ message TelemetryData { | `broker_count` | Number of broker servers | `1` | | `timestamp` | When data was collected | `1640995200` | +A master starts reporting once its cluster stores at least 10 GiB, and the server drops reports under that floor. Every fresh `weed server`, CI job and throwaway container mints its own cluster id, and at tens of thousands a day they were nearly all of the counted clusters and almost none of the bytes. + ## Quick Start ### 1. Deploy Telemetry Server @@ -156,6 +158,7 @@ series across upgrades. To slice values by version or OS, join with ### Server Metrics - `seaweedfs_telemetry_reports_received_total`: Total telemetry reports received +- `seaweedfs_telemetry_reports_skipped_total`: Reports dropped because the cluster stores less than 10 GiB ## API Endpoints diff --git a/telemetry/proto/floor.go b/telemetry/proto/floor.go new file mode 100644 index 000000000..46c9b559f --- /dev/null +++ b/telemetry/proto/floor.go @@ -0,0 +1,7 @@ +package proto + +// MinDiskBytes is what a cluster must store before its master reports it and +// the server keeps the report: fresh `weed server` runs, CI jobs and throwaway +// containers each mint their own cluster id, and they were most of the counted +// clusters while holding almost none of the bytes. +const MinDiskBytes = 10 << 30 diff --git a/telemetry/server/api/handlers_test.go b/telemetry/server/api/handlers_test.go index 80debb53c..ec7eef971 100644 --- a/telemetry/server/api/handlers_test.go +++ b/telemetry/server/api/handlers_test.go @@ -21,7 +21,7 @@ func validReport() *proto.TelemetryData { Version: "4.40", Os: "linux/amd64", VolumeServerCount: 5, - TotalDiskBytes: 123456789, + TotalDiskBytes: 123 << 30, TotalVolumeCount: 42, FilerCount: 2, BrokerCount: 1, diff --git a/telemetry/server/go.mod b/telemetry/server/go.mod index 4f0778b3d..b952361c4 100644 --- a/telemetry/server/go.mod +++ b/telemetry/server/go.mod @@ -5,7 +5,7 @@ go 1.26 require ( github.com/prometheus/client_golang v1.24.1 github.com/seaweedfs/seaweedfs v0.0.0-00010101000000-000000000000 - google.golang.org/protobuf v1.36.11 + google.golang.org/protobuf v1.36.12 ) require ( diff --git a/telemetry/server/go.sum b/telemetry/server/go.sum index 9e8f705c0..40859a1b8 100644 --- a/telemetry/server/go.sum +++ b/telemetry/server/go.sum @@ -30,7 +30,7 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= -google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/telemetry/server/storage/confirmed_test.go b/telemetry/server/storage/confirmed_test.go index e06d04b21..2e70309d9 100644 --- a/telemetry/server/storage/confirmed_test.go +++ b/telemetry/server/storage/confirmed_test.go @@ -14,7 +14,7 @@ func report(id, version string) *proto.TelemetryData { Version: version, Os: "linux/amd64", VolumeServerCount: 1, - TotalDiskBytes: 100, + TotalDiskBytes: proto.MinDiskBytes, TotalVolumeCount: 1, } } diff --git a/telemetry/server/storage/floor_test.go b/telemetry/server/storage/floor_test.go new file mode 100644 index 000000000..f32034f69 --- /dev/null +++ b/telemetry/server/storage/floor_test.go @@ -0,0 +1,90 @@ +package storage + +import ( + "path/filepath" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/seaweedfs/seaweedfs/telemetry/proto" +) + +func counterValue(t *testing.T, c prometheus.Counter) float64 { + t.Helper() + var m dto.Metric + if err := c.Write(&m); err != nil { + t.Fatal(err) + } + return m.GetCounter().GetValue() +} + +func TestClustersUnderTheFloorAreNotKept(t *testing.T) { + s := newPrometheusStorage(prometheus.NewRegistry()) + + // report() sits exactly on the floor, which is enough. + if err := s.StoreTelemetry(report("aaaaaaaa-0000-0000-0000-000000000001", "4.40")); err != nil { + t.Fatal(err) + } + small := report("bbbbbbbb-0000-0000-0000-000000000002", "4.40") + small.TotalDiskBytes = proto.MinDiskBytes - 1 + if err := s.StoreTelemetry(small); err != nil { + t.Fatal(err) + } + + if _, ok := s.instances[small.TopologyId]; ok { + t.Fatal("cluster under the floor was kept") + } + if _, ok := s.GetHistory(small.TopologyId, 90); ok { + t.Fatal("history kept for a cluster under the floor") + } + if stats := statsOf(t, s); stats["active_instances"] != 1 { + t.Errorf("active = %v, want the one cluster on the floor", stats["active_instances"]) + } + if got := counterValue(t, s.telemetryReceived); got != 2 { + t.Errorf("received = %v, want 2", got) + } + if got := counterValue(t, s.reportsSkipped); got != 1 { + t.Errorf("skipped = %v, want 1", got) + } +} + +// State written before the floor existed carries clusters under it; loading +// drops them and marks the state dirty so the next save sheds them from disk. +func TestLoadStateDropsClustersUnderTheFloor(t *testing.T) { + path := filepath.Join(t.TempDir(), "telemetry-state.json") + + s := newPrometheusStorage(prometheus.NewRegistry()) + kept := report("aaaaaaaa-0000-0000-0000-000000000001", "4.40") + if err := s.StoreTelemetry(kept); err != nil { + t.Fatal(err) + } + small := report("bbbbbbbb-0000-0000-0000-000000000002", "4.40") + small.TotalDiskBytes = proto.MinDiskBytes - 1 + s.instances[small.TopologyId] = &telemetryData{TelemetryData: small, ReceivedAt: time.Now()} + s.histories[small.TopologyId] = []HistorySample{{Ts: time.Now().Unix(), TotalDiskBytes: small.TotalDiskBytes}} + if err := s.SaveStateIfDirty(path); err != nil { + t.Fatal(err) + } + + s = newPrometheusStorage(prometheus.NewRegistry()) + n, err := s.LoadState(path) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Errorf("loaded %d instances, want 1", n) + } + if _, ok := s.instances[small.TopologyId]; ok { + t.Error("cluster under the floor survived the load") + } + if _, ok := s.histories[small.TopologyId]; ok { + t.Error("history of a cluster under the floor survived the load") + } + if _, ok := s.instances[kept.TopologyId]; !ok { + t.Error("cluster on the floor was dropped") + } + if !s.dirty { + t.Error("dropping a cluster left the state clean, so it would stay on disk") + } +} diff --git a/telemetry/server/storage/history_test.go b/telemetry/server/storage/history_test.go index 0f3775eed..e1f8bf3c7 100644 --- a/telemetry/server/storage/history_test.go +++ b/telemetry/server/storage/history_test.go @@ -17,7 +17,7 @@ func TestClusterHistory(t *testing.T) { Version: "4.40", Os: "linux/amd64", VolumeServerCount: 3, - TotalDiskBytes: 1000, + TotalDiskBytes: proto.MinDiskBytes + 1000, TotalVolumeCount: 10, } if err := s.StoreTelemetry(report); err != nil { @@ -25,7 +25,7 @@ func TestClusterHistory(t *testing.T) { } // A second report on the same UTC day replaces the day's sample. - report.TotalDiskBytes = 2000 + report.TotalDiskBytes = proto.MinDiskBytes + 2000 if err := s.StoreTelemetry(report); err != nil { t.Fatalf("store: %v", err) } @@ -36,7 +36,7 @@ func TestClusterHistory(t *testing.T) { if len(samples) != 1 { t.Fatalf("got %d samples, want 1 (same-day replace)", len(samples)) } - if samples[0].TotalDiskBytes != 2000 { + if samples[0].TotalDiskBytes != proto.MinDiskBytes+2000 { t.Errorf("same-day sample not replaced: got %d", samples[0].TotalDiskBytes) } @@ -62,7 +62,7 @@ func TestClusterHistory(t *testing.T) { if !ok || len(samples) != 2 { t.Fatalf("after round trip: ok=%v samples=%d, want 2", ok, len(samples)) } - if samples[0].TotalDiskBytes != 500 || samples[1].TotalDiskBytes != 2000 { + if samples[0].TotalDiskBytes != 500 || samples[1].TotalDiskBytes != proto.MinDiskBytes+2000 { t.Errorf("samples corrupted after round trip: %+v", samples) } diff --git a/telemetry/server/storage/persistence.go b/telemetry/server/storage/persistence.go index a61740ec7..194c848ac 100644 --- a/telemetry/server/storage/persistence.go +++ b/telemetry/server/storage/persistence.go @@ -5,6 +5,8 @@ import ( "fmt" "os" "path/filepath" + + "github.com/seaweedfs/seaweedfs/telemetry/proto" ) // persistedState is the on-disk snapshot of the in-memory instance map. @@ -16,7 +18,8 @@ type persistedState struct { // LoadState restores the instance map and Prometheus gauges from a state file // written by SaveStateIfDirty. A missing file is not an error. Original // ReceivedAt timestamps are preserved so cleanup and the active-cluster -// windows stay correct across restarts. +// windows stay correct across restarts. Clusters under proto.MinDiskBytes are +// dropped, so state written before the floor sheds them on the first restart. func (s *PrometheusStorage) LoadState(path string) (int, error) { b, err := os.ReadFile(path) if err != nil { @@ -39,6 +42,10 @@ func (s *PrometheusStorage) LoadState(path string) (int, error) { if instance == nil || instance.TelemetryData == nil || instance.TelemetryData.TopologyId == "" { continue } + if instance.TelemetryData.TotalDiskBytes < proto.MinDiskBytes { + s.dirty = true // so the next save sheds it + continue + } s.instances[id] = instance s.setClusterMetrics(instance.TelemetryData) loaded++ diff --git a/telemetry/server/storage/persistence_test.go b/telemetry/server/storage/persistence_test.go index df11545f6..6417f7016 100644 --- a/telemetry/server/storage/persistence_test.go +++ b/telemetry/server/storage/persistence_test.go @@ -26,7 +26,7 @@ func TestStateRoundTrip(t *testing.T) { Version: "4.40", Os: "linux/amd64", VolumeServerCount: 5, - TotalDiskBytes: 123456789, + TotalDiskBytes: 123 << 30, TotalVolumeCount: 42, FilerCount: 2, BrokerCount: 1, @@ -79,7 +79,7 @@ func TestLoadStateFillsNewestSampleVersion(t *testing.T) { path := filepath.Join(t.TempDir(), "telemetry-state.json") s := newPrometheusStorage(prometheus.NewRegistry()) - report := &proto.TelemetryData{TopologyId: "test-cluster-1", Version: "4.40", Os: "linux/amd64"} + report := &proto.TelemetryData{TopologyId: "test-cluster-1", Version: "4.40", Os: "linux/amd64", TotalDiskBytes: proto.MinDiskBytes} if err := s.StoreTelemetry(report); err != nil { t.Fatalf("store: %v", err) } diff --git a/telemetry/server/storage/prometheus.go b/telemetry/server/storage/prometheus.go index 7641a6e0c..b9756a8b3 100644 --- a/telemetry/server/storage/prometheus.go +++ b/telemetry/server/storage/prometheus.go @@ -1,6 +1,7 @@ package storage import ( + "fmt" "sync" "time" @@ -21,6 +22,7 @@ type PrometheusStorage struct { brokerCount *prometheus.GaugeVec clusterInfo *prometheus.GaugeVec telemetryReceived prometheus.Counter + reportsSkipped prometheus.Counter // In-memory storage for API endpoints (if needed) mu sync.RWMutex @@ -83,6 +85,10 @@ func newPrometheusStorage(reg prometheus.Registerer) *PrometheusStorage { Name: "seaweedfs_telemetry_reports_received_total", Help: "Total number of telemetry reports received", }), + reportsSkipped: promauto.NewCounter(prometheus.CounterOpts{ + Name: "seaweedfs_telemetry_reports_skipped_total", + Help: fmt.Sprintf("Reports not kept because the cluster stores less than %d GiB", proto.MinDiskBytes>>30), + }), instances: make(map[string]*telemetryData), histories: make(map[string][]HistorySample), stats: make(map[string]interface{}), @@ -93,6 +99,12 @@ func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error { s.mu.Lock() defer s.mu.Unlock() + s.telemetryReceived.Inc() + if data.TotalDiskBytes < proto.MinDiskBytes { + s.reportsSkipped.Inc() + return nil + } + // Drop the cluster_info series recorded under the previous label set when // a cluster reports back with a different version or OS, so it is not // counted under two versions at once. @@ -102,8 +114,6 @@ func (s *PrometheusStorage) StoreTelemetry(data *proto.TelemetryData) error { } s.setClusterMetrics(data) - s.telemetryReceived.Inc() - // Store in memory for API endpoints receivedAt := time.Now().UTC() s.instances[data.TopologyId] = &telemetryData{ diff --git a/telemetry/server/storage/sizes_test.go b/telemetry/server/storage/sizes_test.go index c4bec1663..6217a5378 100644 --- a/telemetry/server/storage/sizes_test.go +++ b/telemetry/server/storage/sizes_test.go @@ -112,13 +112,13 @@ func TestClusterSizeSeriesUsesLatestDailySample(t *testing.T) { TopologyId: "aaaaaaaa-0000-0000-0000-000000000001", Version: "4.40", Os: "linux/amd64", - TotalDiskBytes: 100, + TotalDiskBytes: proto.MinDiskBytes + 100, VolumeServerCount: 4, } if err := s.StoreTelemetry(data); err != nil { t.Fatal(err) } - data.TotalDiskBytes = 700 + data.TotalDiskBytes = proto.MinDiskBytes + 700 data.VolumeServerCount = 6 if err := s.StoreTelemetry(data); err != nil { t.Fatal(err) @@ -133,14 +133,14 @@ func TestClusterSizeSeriesUsesLatestDailySample(t *testing.T) { if today := time.Now().UTC().Format("2006-01-02"); len(series.Dates) != 1 || series.Dates[0] != today { t.Errorf("dates = %v, want %s only", series.Dates, today) } - if got := series.Clusters[0].Disk; !equal(got, []uint64{700}) { + if got := series.Clusters[0].Disk; !equal(got, []uint64{proto.MinDiskBytes + 700}) { t.Errorf("disk = %v, want today's latest sample only", got) } if got := series.Clusters[0].Servers; !equal(got, []uint64{6}) { t.Errorf("servers = %v, want today's latest sample only", got) } - if series.TotalDisk != 700 || series.TotalServers != 6 { - t.Errorf("totals = %d disk / %d servers, want 700/6", series.TotalDisk, series.TotalServers) + if series.TotalDisk != proto.MinDiskBytes+700 || series.TotalServers != 6 { + t.Errorf("totals = %d disk / %d servers, want %d/6", series.TotalDisk, series.TotalServers, proto.MinDiskBytes+700) } } diff --git a/weed/telemetry/collector.go b/weed/telemetry/collector.go index ea1745d68..8cf392ee4 100644 --- a/weed/telemetry/collector.go +++ b/weed/telemetry/collector.go @@ -68,8 +68,9 @@ func (c *Collector) CollectAndSendAsync() { } go func() { - data := c.collectData() - c.client.SendTelemetryAsync(data) + if data := c.collectData(); data != nil { + c.client.SendTelemetryAsync(data) + } }() } @@ -80,7 +81,7 @@ func (c *Collector) StartPeriodicCollection(interval time.Duration) { return } - glog.V(0).Infof("Reporting anonymous cluster statistics to %s every %v, use -telemetry=false to opt out", c.client.url, interval) + glog.V(0).Infof("Reporting anonymous cluster statistics to %s every %v once %d GiB are stored, use -telemetry=false to opt out", c.client.url, interval, proto.MinDiskBytes>>30) // Send initial telemetry after a short delay go func() { @@ -107,7 +108,9 @@ func (c *Collector) StartPeriodicCollection(interval time.Duration) { }() } -// collectData gathers telemetry data from the topology +// collectData gathers telemetry data from the topology. It returns nil while +// the cluster stores less than proto.MinDiskBytes, so throwaway clusters never +// report. func (c *Collector) collectData() *proto.TelemetryData { data := &proto.TelemetryData{ Version: c.version, @@ -131,6 +134,10 @@ func (c *Collector) collectData() *proto.TelemetryData { data.BrokerCount = int32(c.countBrokers()) } + if data.TotalDiskBytes < proto.MinDiskBytes { + glog.V(2).Infof("Skipping telemetry: %d bytes stored, reporting starts at %d", data.TotalDiskBytes, proto.MinDiskBytes) + return nil + } return data } diff --git a/weed/telemetry/collector_test.go b/weed/telemetry/collector_test.go index b7127d93f..e4db1e304 100644 --- a/weed/telemetry/collector_test.go +++ b/weed/telemetry/collector_test.go @@ -3,6 +3,7 @@ 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" @@ -64,3 +65,23 @@ func TestCollectVolumeStatsCountsEcShards(t *testing.T) { 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) + } +}