volume: stop reporting read-only volumes that are no longer here (#10867)

* volume: clear per-collection metrics when a collection leaves a server

The read-only and disk size gauges are only ever set for collections the
heartbeat still finds here, and nothing zeroes the rest. volume.balance marks a
volume read-only to move it, so the last heartbeat that saw it counts it
read-only - and if it was the collection's last volume on that server, that
count stands until the process restarts. The dashboard then shows read-only
volumes that volume.list -readonly cannot find anywhere.

Remember what each heartbeat set, and drop what is gone on the next one.

* volume: stop the read-only volume count from wrapping at 256

The per-collection counters were uint8, so a server holding 256 read-only
volumes of one collection reported zero of them.

* volume: read the read-only flags once when counting them

The heartbeat asked IsReadOnly for the verdict and then read noWriteOrDelete
and noWriteCanDelete straight off the volume, unlocked, so the reasons could
disagree with the verdict they were explaining. Take them together, under one
lock. The location is now nil-checked rather than skipped by short-circuit
evaluation, so a volume that has not joined a disk location yet stays safe.

* volume: let only a surviving volume keep its collection reported

A volume being deleted for expiry still made an entry in the read-only counts,
which is what the cleanup reads as "this collection is still here". The
collection's last volume could go and its series would stand for one more
heartbeat. Count the survivors only.

* volume: size a collection from the volumes it still has

The size totals are rebuilt from scratch every heartbeat, so subtracting a
volume that is about to be deleted took the surviving volumes' sizes down with
it: a collection keeping a small volume and losing a larger one reported the
difference, or lost its entry and kept the previous heartbeat's number.

* volume: cover the deleted bytes total in the surviving volume test

Deleted bytes are totalled the same way as sizes and were going unchecked, so
the test now leaves deleted needles on both volumes and pins that gauge too.

seaweed-volume/src/metrics.rs

@@ -330,6 +330,16 @@ pub fn delete_collection_metrics(collection: &str) {
     delete_partial_match_collection(&DISK_SIZE_GAUGE, collection);
 }
 
+/// Drop a collection's volume server series once its last volume leaves this
+/// server. These gauges are only ever set for collections still present, so the
+/// values from the heartbeat that saw the last volume would otherwise stand
+/// until the process restarts.
+pub fn delete_volume_server_collection_metrics(collection: &str) {
+    let _ = DISK_SIZE_GAUGE.remove_label_values(&[collection, DISK_SIZE_LABEL_NORMAL]);
+    let _ = DISK_SIZE_GAUGE.remove_label_values(&[collection, DISK_SIZE_LABEL_DELETED_BYTES]);
+    delete_partial_match_collection(&READ_ONLY_VOLUME_GAUGE, collection);
+}
+
 /// Remove all metric entries from a GaugeVec where the "collection" label matches.
 /// This emulates Go's `DeletePartialMatch(prometheus.Labels{"collection": collection})`.
 fn delete_partial_match_collection(gauge: &GaugeVec, collection: &str) {

seaweed-volume/src/server/heartbeat.rs

@@ -956,24 +956,30 @@ fn build_heartbeat_with_ec_status(
                 should_delete_volume = true;
             }
 
-            // Track disk size by collection
-            let entry = disk_sizes.entry(vol.collection.clone()).or_insert((0, 0));
+            // Track disk size by collection. A volume on its way out is left
+            // out: an entry here is also what says the collection is still on
+            // this server.
             if !should_delete_volume {
+                let entry = disk_sizes.entry(vol.collection.clone()).or_insert((0, 0));
                 entry.0 += volume_size;
                 entry.1 += vol.deleted_size();
             }
 
-            let read_only = ro_counts.entry(vol.collection.clone()).or_default();
-            if !should_delete_volume && vol.is_read_only() {
-                read_only.is_read_only += 1;
-                if vol.is_no_write_or_delete() {
-                    read_only.no_write_or_delete += 1;
-                }
-                if vol.is_no_write_can_delete() {
-                    read_only.no_write_can_delete += 1;
-                }
-                if loc.is_disk_space_low.load(Ordering::Relaxed) {
-                    read_only.is_disk_space_low += 1;
+            // An entry here is what says the collection is still on this
+            // server, so a volume on its way out must not make one.
+            if !should_delete_volume {
+                let read_only = ro_counts.entry(vol.collection.clone()).or_default();
+                if vol.is_read_only() {
+                    read_only.is_read_only += 1;
+                    if vol.is_no_write_or_delete() {
+                        read_only.no_write_or_delete += 1;
+                    }
+                    if vol.is_no_write_can_delete() {
+                        read_only.no_write_can_delete += 1;
+                    }
+                    if loc.is_disk_space_low.load(Ordering::Relaxed) {
+                        read_only.is_disk_space_low += 1;
+                    }
                 }
             }
 
@@ -1007,6 +1013,17 @@ fn build_heartbeat_with_ec_status(
             .with_label_values(&[col, crate::metrics::READ_ONLY_LABEL_IS_DISK_SPACE_LOW])
             .set(counts.is_disk_space_low as f64);
     }
+    // ro_counts has an entry for every collection that kept a volume through
+    // this pass, including the ones counting zero read-only volumes.
+    {
+        let mut reported = store.reported_collections.lock().unwrap();
+        for col in reported.iter() {
+            if !ro_counts.contains_key(col) {
+                crate::metrics::delete_volume_server_collection_metrics(col);
+            }
+        }
+        *reported = ro_counts.keys().cloned().collect();
+    }
     // Update max volumes gauge
     let total_max: i64 = max_volume_counts.values().map(|v| *v as i64).sum();
     crate::metrics::MAX_VOLUMES.set(total_max);
@@ -1082,6 +1099,14 @@ fn collect_live_ec_shards(
                 .with_label_values(&[col, crate::metrics::DISK_SIZE_LABEL_EC])
                 .set(*size as f64);
         }
+        let mut reported = store.reported_ec_collections.lock().unwrap();
+        for col in reported.iter() {
+            if !ec_sizes.contains_key(col) {
+                let _ = crate::metrics::DISK_SIZE_GAUGE
+                    .remove_label_values(&[col, crate::metrics::DISK_SIZE_LABEL_EC]);
+            }
+        }
+        *reported = ec_sizes.keys().cloned().collect();
     }
 
     ec_shards
@@ -1601,6 +1626,82 @@ mod tests {
         );
     }
 
+    fn collection_series(gauge: &prometheus::GaugeVec, collection: &str) -> usize {
+        use prometheus::core::Collector;
+        gauge
+            .collect()
+            .iter()
+            .flat_map(|family| family.get_metric().to_vec())
+            .filter(|metric| {
+                metric
+                    .get_label()
+                    .iter()
+                    .any(|label| label.get_name() == "collection" && label.get_value() == collection)
+            })
+            .count()
+    }
+
+    // The per-collection gauges are only ever set for collections the heartbeat
+    // still finds on this server. A volume.balance that moves a collection's
+    // last volume off a server used to leave its read-only count - marked
+    // read-only for the move, moments before it went - standing on that server
+    // until a restart, with nothing in volume.list to match it.
+    #[test]
+    fn test_build_heartbeat_clears_metrics_of_departed_collection() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let dir = temp_dir.path().to_str().unwrap();
+        let collection = "heartbeat_departed_case";
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                dir,
+                dir,
+                8,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+        store
+            .add_volume(
+                VolumeId(21),
+                collection,
+                None,
+                None,
+                0,
+                DiskType::HardDrive,
+                Version::current(),
+            )
+            .unwrap();
+        {
+            let (_, volume) = store.find_volume_mut(VolumeId(21)).unwrap();
+            volume.set_read_only().unwrap();
+        }
+
+        build_heartbeat(&test_config(), &mut store);
+        assert_eq!(
+            READ_ONLY_VOLUME_GAUGE
+                .with_label_values(&[collection, READ_ONLY_LABEL_IS_READ_ONLY])
+                .get(),
+            1.0
+        );
+
+        assert!(store.unmount_volume(VolumeId(21)));
+        build_heartbeat(&test_config(), &mut store);
+
+        assert_eq!(
+            collection_series(&READ_ONLY_VOLUME_GAUGE, collection),
+            0,
+            "read-only series left after the collection left the server"
+        );
+        assert_eq!(
+            collection_series(&DISK_SIZE_GAUGE, collection),
+            0,
+            "disk size series left after the collection left the server"
+        );
+    }
+
     #[test]
     fn test_build_heartbeat_reports_disk_bytes() {
         let temp_dir = tempfile::tempdir().unwrap();

seaweed-volume/src/storage/store.rs

@@ -4,8 +4,10 @@
 //! It coordinates volume placement, lookup, and lifecycle operations.
 //! Matches Go's storage/store.go.
 
+use std::collections::HashSet;
 use std::io;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Mutex;
 
 use crate::config::MinFreeSpace;
 use crate::pb::master_pb;
@@ -32,6 +34,12 @@ pub struct Store {
     pub data_center: String,
     pub rack: String,
     pub volume_report: crate::storage::volume_report::VolumeReportState,
+    /// Collections the last heartbeat set per-collection gauges for. Those
+    /// gauges are only ever set for collections still held here, so one whose
+    /// last volume leaves - moved away by volume.balance, say - would keep
+    /// reporting the heartbeat that saw it.
+    pub reported_collections: Mutex<HashSet<String>>,
+    pub reported_ec_collections: Mutex<HashSet<String>>,
 }
 
 impl Store {
@@ -49,6 +57,8 @@ impl Store {
             volume_report: Default::default(),
             data_center: String::new(),
             rack: String::new(),
+            reported_collections: Mutex::new(HashSet::new()),
+            reported_ec_collections: Mutex::new(HashSet::new()),
         }
     }
 

weed/stats/metrics.go

@@ -1101,6 +1101,16 @@ func DeleteCollectionMetrics(collection string) {
 	glog.V(0).Infof("delete collection metrics, %s: %d", collection, c)
 }
 
+// DeleteVolumeServerCollectionMetrics drops a collection's volume server series
+// once its last volume leaves this server. These gauges are only ever set for
+// collections still present, so the values from the heartbeat that saw the last
+// volume would otherwise stand until the process restarts.
+func DeleteVolumeServerCollectionMetrics(collection string) {
+	VolumeServerDiskSizeGauge.DeleteLabelValues(collection, "normal")
+	VolumeServerDiskSizeGauge.DeleteLabelValues(collection, "deleted_bytes")
+	VolumeServerReadOnlyVolumeGauge.DeletePartialMatch(prometheus.Labels{"collection": collection})
+}
+
 func bucketMetricTTLControl() {
 	ttlNs := bucketAtiveTTL.Nanoseconds()
 	for {

weed/storage/store.go

@@ -86,6 +86,12 @@ type Store struct {
 	// runs, so two overlapping scans would each forget what the other marked
 	// and name every volume it holds as departed.
 	collectHeartbeatLock sync.Mutex
+	// Collections the last heartbeat set per-collection gauges for. Those gauges
+	// are only ever set for collections still held here, so one whose last
+	// volume leaves - moved away by volume.balance, say - would keep reporting
+	// the heartbeat that saw it. Written only from the heartbeat goroutine.
+	reportedCollections   map[string]struct{}
+	reportedEcCollections map[string]struct{}
 }
 
 func (s *Store) String() (str string) {
@@ -449,7 +455,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	var maxFileKey NeedleId
 	collectionVolumeSize := make(map[string]int64)
 	collectionVolumeDeletedBytes := make(map[string]int64)
-	collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
+	collectionVolumeReadOnlyCount := make(map[string]map[string]int)
 	// Filled once per volume and kept only by the heartbeat that carries it, so
 	// a server with nothing to say fills the same message all the way through.
 	scratchMessage := &master_pb.VolumeInformationMessage{}
@@ -531,38 +537,35 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 				}
 			}
 
-			if _, exist := collectionVolumeSize[v.Collection]; !exist {
-				collectionVolumeSize[v.Collection] = 0
-				collectionVolumeDeletedBytes[v.Collection] = 0
-			}
+			// The totals are rebuilt from scratch every heartbeat, so a volume
+			// on its way out is simply not added. Subtracting it took the
+			// surviving volumes' sizes down with it, and an entry here is also
+			// what says the collection is still on this server.
 			if !shouldDeleteVolume {
 				collectionVolumeSize[v.Collection] += int64(volumeMessage.Size)
 				collectionVolumeDeletedBytes[v.Collection] += int64(volumeMessage.DeletedByteCount)
-			} else {
-				collectionVolumeSize[v.Collection] -= int64(volumeMessage.Size)
-				if collectionVolumeSize[v.Collection] <= 0 {
-					delete(collectionVolumeSize, v.Collection)
-				}
-			}
 
-			if _, exist := collectionVolumeReadOnlyCount[v.Collection]; !exist {
-				collectionVolumeReadOnlyCount[v.Collection] = map[string]uint8{
-					stats.IsReadOnly:       0,
-					stats.NoWriteOrDelete:  0,
-					stats.NoWriteCanDelete: 0,
-					stats.IsDiskSpaceLow:   0,
+				counts, exist := collectionVolumeReadOnlyCount[v.Collection]
+				if !exist {
+					counts = map[string]int{
+						stats.IsReadOnly:       0,
+						stats.NoWriteOrDelete:  0,
+						stats.NoWriteCanDelete: 0,
+						stats.IsDiskSpaceLow:   0,
+					}
+					collectionVolumeReadOnlyCount[v.Collection] = counts
 				}
-			}
-			if !shouldDeleteVolume && v.IsReadOnly() {
-				collectionVolumeReadOnlyCount[v.Collection][stats.IsReadOnly] += 1
-				if v.noWriteOrDelete {
-					collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteOrDelete] += 1
-				}
-				if v.noWriteCanDelete {
-					collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteCanDelete] += 1
-				}
-				if v.location.isDiskSpaceLow.Load() {
-					collectionVolumeReadOnlyCount[v.Collection][stats.IsDiskSpaceLow] += 1
+				if readOnly, noWriteOrDelete, noWriteCanDelete, diskSpaceLow := v.ReadOnlyReasons(); readOnly {
+					counts[stats.IsReadOnly] += 1
+					if noWriteOrDelete {
+						counts[stats.NoWriteOrDelete] += 1
+					}
+					if noWriteCanDelete {
+						counts[stats.NoWriteCanDelete] += 1
+					}
+					if diskSpaceLow {
+						counts[stats.IsDiskSpaceLow] += 1
+					}
 				}
 			}
 		}
@@ -616,6 +619,19 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		}
 	}
 
+	// collectionVolumeReadOnlyCount has an entry for every collection that kept
+	// a volume through this pass, including the ones counting zero read-only
+	// volumes.
+	for col := range s.reportedCollections {
+		if _, stillHere := collectionVolumeReadOnlyCount[col]; !stillHere {
+			stats.DeleteVolumeServerCollectionMetrics(col)
+		}
+	}
+	s.reportedCollections = make(map[string]struct{}, len(collectionVolumeReadOnlyCount))
+	for col := range collectionVolumeReadOnlyCount {
+		s.reportedCollections[col] = struct{}{}
+	}
+
 	departedVolumes := s.volumeReport.commit(reportPass, reportGeneration, sendFullList)
 
 	// has_no_volumes says the server holds nothing, so it may only be derived

weed/storage/store_collection_metrics_test.go

@@ -0,0 +1,182 @@
+package storage
+
+import (
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// fillTestVolume writes one needle and dates the volume now. A volume with no
+// content never expires, and the needle carries no append time of its own,
+// which would leave the volume dated to the epoch.
+func fillTestVolume(t *testing.T, v *Volume) {
+	t.Helper()
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(v.Id)), true, false, false); err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	v.lastModifiedTsSeconds = uint64(time.Now().Unix())
+}
+
+// The per-collection gauges are only ever set for collections the heartbeat
+// still finds on this server. A volume.balance that moves a collection's last
+// volume off a server used to leave its read-only count - marked read-only for
+// the move, moments before it went - standing on that server until a restart,
+// with nothing in volume.list to match it.
+func TestCollectHeartbeatClearsMetricsOfDepartedCollection(t *testing.T) {
+	stats.VolumeServerReadOnlyVolumeGauge.Reset()
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerReadOnlyVolumeGauge.Reset)
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store := newTestStore(t, 1)
+	mountTestVolume(t, store.Locations[0], 1, "pics").noWriteOrDelete = true
+
+	store.CollectHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues("pics", stats.IsReadOnly)); got != 1 {
+		t.Fatalf("read-only volumes of pics = %v, want 1", got)
+	}
+
+	if err := store.UnmountVolume(1); err != nil {
+		t.Fatalf("UnmountVolume: %v", err)
+	}
+	store.CollectHeartbeat()
+
+	if n := testutil.CollectAndCount(stats.VolumeServerReadOnlyVolumeGauge); n != 0 {
+		t.Errorf("%d read-only series left after the collection left the server", n)
+	}
+	if n := testutil.CollectAndCount(stats.VolumeServerDiskSizeGauge); n != 0 {
+		t.Errorf("%d disk size series left after the collection left the server", n)
+	}
+}
+
+// A volume being deleted for expiry is already gone as far as the gauges are
+// concerned, so the heartbeat that drops the collection's last one has to take
+// its series along rather than leave them standing for another pass.
+func TestCollectHeartbeatClearsMetricsWhenTheLastVolumeExpires(t *testing.T) {
+	stats.VolumeServerReadOnlyVolumeGauge.Reset()
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerReadOnlyVolumeGauge.Reset)
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store := newTestStore(t, 1)
+	store.SetVolumeSizeLimit(30 << 30)
+	v := mountTestVolume(t, store.Locations[0], 1, "pics")
+	v.Ttl = &needle.TTL{Count: 1, Unit: needle.Minute}
+	fillTestVolume(t, v)
+
+	store.CollectHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues("pics", "normal")); got == 0 {
+		t.Fatal("disk size of pics = 0, want the written needle")
+	}
+
+	v.lastModifiedTsSeconds = uint64(time.Now().Add(-time.Hour).Unix())
+	store.CollectHeartbeat()
+
+	if store.findVolume(1) != nil {
+		t.Fatal("the expired volume outlived the heartbeat")
+	}
+	if n := testutil.CollectAndCount(stats.VolumeServerReadOnlyVolumeGauge); n != 0 {
+		t.Errorf("%d read-only series left after the last volume expired", n)
+	}
+	if n := testutil.CollectAndCount(stats.VolumeServerDiskSizeGauge); n != 0 {
+		t.Errorf("%d disk size series left after the last volume expired", n)
+	}
+}
+
+// A collection that loses one volume to expiry and keeps another must report
+// what is left, not what is left minus what went.
+func TestCollectHeartbeatSizesOnlySurvivingVolumes(t *testing.T) {
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store := newTestStore(t, 2)
+	store.SetVolumeSizeLimit(30 << 30)
+	// One volume per location, so the surviving one is always scanned first.
+	surviving := mountTestVolume(t, store.Locations[0], 1, "pics")
+	fillTestVolume(t, surviving)
+	expiring := mountTestVolume(t, store.Locations[1], 2, "pics")
+	expiring.Ttl = &needle.TTL{Count: 1, Unit: needle.Minute}
+	fillTestVolume(t, expiring)
+	// Both volumes carry deleted bytes, which are totalled the same way.
+	for _, v := range []*Volume{surviving, expiring} {
+		n := newRandomNeedle(uint64(v.Id) + 100)
+		if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
+			t.Fatalf("write needle: %v", err)
+		}
+		if _, err := v.deleteNeedle2(n); err != nil {
+			t.Fatalf("delete needle: %v", err)
+		}
+	}
+
+	heartbeat := store.CollectHeartbeat()
+	var survivingSize, survivingDeleted uint64
+	for _, m := range heartbeat.Volumes {
+		if m.Id == 1 {
+			survivingSize, survivingDeleted = m.Size, m.DeletedByteCount
+		}
+	}
+	if survivingSize == 0 || survivingDeleted == 0 {
+		t.Fatalf("the surviving volume reported size %d and %d deleted bytes, want both", survivingSize, survivingDeleted)
+	}
+
+	expiring.lastModifiedTsSeconds = uint64(time.Now().Add(-time.Hour).Unix())
+	store.CollectHeartbeat()
+
+	if store.findVolume(1) == nil {
+		t.Fatal("the volume without a ttl was deleted")
+	}
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues("pics", "normal")); got != float64(survivingSize) {
+		t.Errorf("disk size of pics = %v, want %d, the volume still here", got, survivingSize)
+	}
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues("pics", "deleted_bytes")); got != float64(survivingDeleted) {
+		t.Errorf("deleted bytes of pics = %v, want %d, the volume still here", got, survivingDeleted)
+	}
+}
+
+// The counts are per collection and per server, and a server holds far more
+// than 255 volumes of one collection.
+func TestCollectHeartbeatCountsPast255ReadOnlyVolumes(t *testing.T) {
+	stats.VolumeServerReadOnlyVolumeGauge.Reset()
+	t.Cleanup(stats.VolumeServerReadOnlyVolumeGauge.Reset)
+
+	store := newTestStore(t, 1)
+	const readOnlyVolumes = 300
+	for i := 1; i <= readOnlyVolumes; i++ {
+		mountTestVolume(t, store.Locations[0], needle.VolumeId(i), "pics")
+		store.findVolume(needle.VolumeId(i)).noWriteOrDelete = true
+	}
+
+	store.CollectHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues("pics", stats.IsReadOnly)); got != readOnlyVolumes {
+		t.Errorf("read-only volumes of pics = %v, want %d", got, readOnlyVolumes)
+	}
+}
+
+func TestCollectErasureCodingHeartbeatClearsMetricsOfDepartedCollection(t *testing.T) {
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store, _, vid, collection, plant := setupECStoreWithMixedDisks(t)
+	plant()
+	if err := store.MountEcShards(collection, vid, 0, ""); err != nil {
+		t.Fatalf("MountEcShards: %v", err)
+	}
+
+	store.CollectErasureCodingHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues(collection, "ec")); got == 0 {
+		t.Fatalf("ec disk size of %s = 0, want the mounted shard's size", collection)
+	}
+
+	if err := store.UnmountEcShards(vid, erasure_coding.ShardId(0), 0); err != nil {
+		t.Fatalf("UnmountEcShards: %v", err)
+	}
+	store.CollectErasureCodingHeartbeat()
+
+	if n := testutil.CollectAndCount(stats.VolumeServerDiskSizeGauge); n != 0 {
+		t.Errorf("%d disk size series left after the collection's shards left the server", n)
+	}
+}

weed/storage/store_ec.go

@@ -153,6 +153,16 @@ func (s *Store) CollectErasureCodingHeartbeat() *master_pb.Heartbeat {
 		stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "ec").Set(float64(size))
 	}
 
+	for col := range s.reportedEcCollections {
+		if _, stillHere := collectionEcShardSize[col]; !stillHere {
+			stats.VolumeServerDiskSizeGauge.DeleteLabelValues(col, "ec")
+		}
+	}
+	s.reportedEcCollections = make(map[string]struct{}, len(collectionEcShardSize))
+	for col := range collectionEcShardSize {
+		s.reportedEcCollections[col] = struct{}{}
+	}
+
 	return &master_pb.Heartbeat{
 		EcShards:      ecShardMessages,
 		HasNoEcShards: len(ecShardMessages) == 0,

weed/storage/store_heartbeat_bench_test.go

@@ -14,7 +14,7 @@ func benchCollectHeartbeat(b *testing.B, count int) {
 	store := newTestStore(b, 1)
 	location := store.Locations[0]
 	for i := 1; i <= count; i++ {
-		mountTestVolume(b, location, needle.VolumeId(i))
+		mountTestVolume(b, location, needle.VolumeId(i), "")
 	}
 	store.ResetVolumeReporting()
 	store.AcceptVolumeChanges()

weed/storage/store_heartbeat_digest_test.go

@@ -7,14 +7,15 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 )
 
-func mountTestVolume(t testing.TB, loc *DiskLocation, vid needle.VolumeId) {
+func mountTestVolume(t testing.TB, loc *DiskLocation, vid needle.VolumeId, collection string) *Volume {
 	t.Helper()
-	v, err := NewVolume(loc.Directory, loc.IdxDirectory, "", vid, NeedleMapInMemory,
+	v, err := NewVolume(loc.Directory, loc.IdxDirectory, collection, vid, NeedleMapInMemory,
 		&super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
 	loc.SetVolume(vid, v)
+	return v
 }
 
 // The digest has to cover exactly the volumes the heartbeat carries. A volume
@@ -22,9 +23,9 @@ func mountTestVolume(t testing.TB, loc *DiskLocation, vid needle.VolumeId) {
 // comparison disagree forever.
 func TestCollectHeartbeatDigestsExactlyWhatItReports(t *testing.T) {
 	store := newTestStore(t, 2)
-	mountTestVolume(t, store.Locations[0], 1)
-	mountTestVolume(t, store.Locations[0], 2)
-	mountTestVolume(t, store.Locations[1], 3)
+	mountTestVolume(t, store.Locations[0], 1, "")
+	mountTestVolume(t, store.Locations[0], 2, "")
+	mountTestVolume(t, store.Locations[1], 3, "")
 
 	heartbeat := store.CollectHeartbeat()
 	if heartbeat.VolumeDigest == nil {
@@ -67,14 +68,14 @@ func TestCollectHeartbeatDigestsAnEmptyStore(t *testing.T) {
 
 func TestCollectHeartbeatDigestFollowsVolumeChanges(t *testing.T) {
 	store := newTestStore(t, 1)
-	mountTestVolume(t, store.Locations[0], 1)
+	mountTestVolume(t, store.Locations[0], 1, "")
 	first := store.CollectHeartbeat().GetVolumeDigest()
 
 	if second := store.CollectHeartbeat().GetVolumeDigest(); second != first {
 		t.Errorf("an unchanged store reported a different digest: %d then %d", first, second)
 	}
 
-	mountTestVolume(t, store.Locations[0], 2)
+	mountTestVolume(t, store.Locations[0], 2, "")
 	if grown := store.CollectHeartbeat().GetVolumeDigest(); grown == first {
 		t.Error("mounting a volume left the digest unchanged")
 	}

weed/storage/store_volume_report_test.go

@@ -12,7 +12,7 @@ func reportingStore(t *testing.T, vids ...needle.VolumeId) *Store {
 	t.Helper()
 	store := newTestStore(t, 1)
 	for _, vid := range vids {
-		mountTestVolume(t, store.Locations[0], vid)
+		mountTestVolume(t, store.Locations[0], vid, "")
 	}
 	return store
 }
@@ -73,7 +73,7 @@ func TestHeartbeatReportsOnlyWhatChanged(t *testing.T) {
 	store.AcceptVolumeChanges()
 	store.CollectHeartbeat()
 
-	mountTestVolume(t, store.Locations[0], 3)
+	mountTestVolume(t, store.Locations[0], 3, "")
 	heartbeat := store.CollectHeartbeat()
 
 	if len(heartbeat.Volumes) != 0 {
@@ -129,7 +129,7 @@ func TestRemountedVolumeIsReportedAgain(t *testing.T) {
 	store.Locations[0].UnloadVolume(needle.VolumeId(1))
 	store.CollectHeartbeat()
 
-	mountTestVolume(t, store.Locations[0], 1)
+	mountTestVolume(t, store.Locations[0], 1, "")
 	heartbeat := store.CollectHeartbeat()
 	if len(heartbeat.ChangedVolumes) != 1 {
 		t.Errorf("a remounted volume was not reported: %v", heartbeat.ChangedVolumes)
@@ -201,7 +201,7 @@ func TestFullListCarriesNoDepartures(t *testing.T) {
 // the master unregister a volume the same heartbeat re-adds.
 func TestMovedVolumeIsNotADeparture(t *testing.T) {
 	store := newTestStore(t, 2)
-	mountTestVolume(t, store.Locations[0], 1)
+	mountTestVolume(t, store.Locations[0], 1, "")
 	store.ResetVolumeReporting()
 	store.AcceptVolumeChanges()
 	store.CollectHeartbeat()

weed/storage/volume.go

@@ -540,9 +540,20 @@ func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
 }
 
 func (v *Volume) IsReadOnly() bool {
+	readOnly, _, _, _ := v.ReadOnlyReasons()
+	return readOnly
+}
+
+// ReadOnlyReasons reports whether the volume refuses writes and why, reading the
+// flags once so the reasons cannot disagree with the verdict.
+func (v *Volume) ReadOnlyReasons() (readOnly, noWriteOrDelete, noWriteCanDelete, diskSpaceLow bool) {
 	v.noWriteLock.RLock()
-	defer v.noWriteLock.RUnlock()
-	return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow.Load()
+	noWriteOrDelete, noWriteCanDelete = v.noWriteOrDelete, v.noWriteCanDelete
+	v.noWriteLock.RUnlock()
+	// The location is attached when the volume joins a disk location, which is
+	// after NewVolume hands it back.
+	diskSpaceLow = v.location != nil && v.location.isDiskSpaceLow.Load()
+	return noWriteOrDelete || noWriteCanDelete || diskSpaceLow, noWriteOrDelete, noWriteCanDelete, diskSpaceLow
 }
 
 func (v *Volume) PersistReadOnly(readOnly bool, canDelete bool) {