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) + } +}