volume: count EC intervals served by reconstruction (#12392)

* volume: count EC reads that fall back to reconstruction

When a needle read can't read an EC shard interval directly (the shard isn't
mounted here and no holder answers), both volume servers rebuild it from the
other shards and the read still succeeds, with nothing to show for it.

Count those fallbacks in SeaweedFS_volumeServer_ec_degraded_read_total{result}
(success, failure), one per interval, at the one place each server falls back:
Go readOneEcShardInterval, Rust fetch_one_interval. Corruption repair and scrub
aren't counted. A line binding each one to the needle and the volume's encode
generation is logged at V(3) in Go and debug in Rust.

* volume: name the EC counter for what it counts

Reed-Solomon reconstruction can't tell corrupt inputs from good ones, and the
needle is validated later, so "success" only ever meant the reconstruction
produced data. Rename the counter to
SeaweedFS_volumeServer_ec_reconstructed_interval_total and the log line to
"ec reconstructed interval", say so in the help text, and pin the behaviour
with a test: an interval rebuilt from corrupt parity counts success while the
read itself fails.

* storage: restore stderr when the log-line test stops early

A fixture failure in readEcNeedle ends the test through t.Fatalf, which
skipped restoring os.Stderr and closing the pipe, so later tests wrote
into a pipe nobody read.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

seaweed-volume/src/metrics.rs

@@ -252,6 +252,21 @@ pub static SCRUB_SHARD_FAILURES: LazyLock<IntCounterVec> = LazyLock::new(|| {
     .expect("metric can be created")
 });
 
+/// Reads of an EC shard interval that couldn't be read directly and were
+/// served by reconstructing it from the other shards, by result. `success`
+/// means the reconstruction produced data, not that the needle then passed
+/// validation. Mirrors Go's VolumeServerECReconstructedIntervalCounter.
+pub static EC_RECONSTRUCTED_INTERVAL_COUNTER: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "SeaweedFS_volumeServer_ec_reconstructed_interval_total",
+            "Reads of an EC interval served by reconstructing from other shards; success means reconstruction produced data, not that the needle passed validation.",
+        ),
+        &["result"],
+    )
+    .expect("metric can be created")
+});
+
 /// Counter of storage read/write EIO errors on volumes and EC shards.
 /// Mirrors Go's VolumeServerStorageIoErrorCounter.
 pub static STORAGE_IO_ERROR_COUNTER: LazyLock<IntCounter> = LazyLock::new(|| {
@@ -402,6 +417,7 @@ pub fn register_metrics() {
             Box::new(SCRUB_VOLUME_FAILURES.clone()),
             Box::new(SCRUB_SHARD_FAILURES.clone()),
             Box::new(STORAGE_IO_ERROR_COUNTER.clone()),
+            Box::new(EC_RECONSTRUCTED_INTERVAL_COUNTER.clone()),
             Box::new(IO_QUARANTINE_GAUGE.clone()),
             // Legacy metrics
             Box::new(VOLUMES_TOTAL.clone()),

seaweed-volume/src/server/handlers.rs

@@ -6227,6 +6227,171 @@ mod tests {
         })
     }
 
+    /// EC reconstructed-interval counter: one fresh volume per case, the needle (on
+    /// shard 0) EC-encoded under a known generation, the `missing` shards
+    /// unmounted and the shards in `corrupt` overwritten where the needle lies.
+    /// Returns the read's bytes and the marker lines logged at debug.
+    async fn read_ec_reconstructed(
+        vid: u32,
+        missing: &[crate::storage::erasure_coding::ec_shard::ShardId],
+        corrupt: &[crate::storage::erasure_coding::ec_shard::ShardId],
+    ) -> (Option<Vec<u8>>, Vec<String>) {
+        use crate::storage::erasure_coding::ec_encoder::write_ec_files;
+        use crate::storage::needle_map::NeedleMapKind;
+        use crate::storage::volume::{VifEcShardConfig, VifVolumeInfo, Volume, VolumeSpec};
+        use std::io::{Seek, SeekFrom, Write};
+        let tmp = tempfile::TempDir::new().unwrap();
+        let state = volume_test_state(&tmp);
+        let dir = tmp.path().to_str().unwrap();
+        let v = Volume::new(
+            dir,
+            dir,
+            VolumeId(vid),
+            NeedleMapKind::InMemory,
+            &VolumeSpec::default(),
+        )
+        .unwrap();
+        let mut n = Needle {
+            id: NeedleId(7),
+            cookie: Cookie(TEST_COOKIE),
+            data: b"ec degraded read".to_vec(),
+            data_size: 16,
+            ..Needle::default()
+        };
+        v.write_needle(&mut n, true, false).unwrap();
+        v.sync_to_disk().unwrap();
+        v.close();
+        let block_size = write_ec_files(dir, dir, "", VolumeId(vid), 10, 4).unwrap();
+        let vif = format!("{dir}/{vid}.vif");
+        let mut info: VifVolumeInfo =
+            serde_json::from_str(&std::fs::read_to_string(&vif).unwrap()).unwrap();
+        info.ec_shard_config = Some(VifEcShardConfig {
+            data_shards: 10,
+            parity_shards: 4,
+            encode_ts_ns: EC_GENERATION,
+            block_size,
+        });
+        crate::storage::store::save_vif_volume_info(&vif, &info).unwrap();
+        for shard in corrupt {
+            let mut f = std::fs::OpenOptions::new()
+                .write(true)
+                .open(format!("{dir}/{vid}.ec{shard:02}"))
+                .unwrap();
+            f.seek(SeekFrom::Start(0)).unwrap();
+            f.write_all(&[0xA5; 256]).unwrap();
+        }
+        let mounted: Vec<_> = (0..14).filter(|s| !missing.contains(s)).collect();
+        state
+            .store
+            .write()
+            .unwrap()
+            .mount_ec_shards(VolumeId(vid), "", &mounted)
+            .unwrap();
+
+        let out = Captured::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::DEBUG)
+            .with_ansi(false)
+            .with_writer(out.clone())
+            .finish();
+        let guard = tracing::subscriber::set_default(subscriber);
+        // Another test may have reached the marker's callsite first, with no
+        // subscriber, and cached it as never enabled.
+        tracing::callsite::rebuild_interest_cache();
+        let read = crate::server::store_ec::read_ec_shard_needle_or_miss(
+            &state,
+            VolumeId(vid),
+            NeedleId(7),
+        )
+        .await;
+        drop(guard);
+        let text = String::from_utf8(out.0.lock().unwrap().clone()).unwrap();
+        let lines = text
+            .lines()
+            .filter(|l| l.contains("ec reconstructed interval:"))
+            .map(|l| l.trim_end().to_string())
+            .collect();
+        (read.ok().and_then(Result::ok).map(|n| n.data), lines)
+    }
+
+    const EC_GENERATION: i64 = 1_700_000_000_123_456_789;
+
+    #[derive(Clone, Default)]
+    struct Captured(Arc<std::sync::Mutex<Vec<u8>>>);
+
+    impl std::io::Write for Captured {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().unwrap().extend_from_slice(buf);
+            Ok(buf.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Captured {
+        type Writer = Captured;
+        fn make_writer(&'a self) -> Self::Writer {
+            self.clone()
+        }
+    }
+
+    /// What the counter and line mean, case by case. A direct read logs nothing;
+    /// a read that rebuilds shard 0 logs and counts one success; one with 5
+    /// shards left fails and logs one failure; a corrupt interval read directly
+    /// goes to the repair path and isn't marked. `success` is not needle
+    /// validation: rebuilt from corrupt parity, the interval logs success while
+    /// the read fails.
+    #[tokio::test]
+    async fn test_ec_reconstructed_interval() {
+        use crate::metrics::EC_RECONSTRUCTED_INTERVAL_COUNTER as C;
+        let prefix = |vid: u32| {
+            format!(
+                "ec reconstructed interval: volume={vid} generation={EC_GENERATION} needle=7 shard=0 "
+            )
+        };
+        let data = Some(b"ec degraded read".to_vec());
+
+        assert_eq!(
+            read_ec_reconstructed(61, &[], &[]).await,
+            (data.clone(), vec![])
+        );
+
+        let success = C.with_label_values(&["success"]).get();
+        let (read, lines) = read_ec_reconstructed(62, &[0], &[]).await;
+        assert_eq!(read, data);
+        assert!(
+            matches!(&lines[..], [l] if l.ends_with(&(prefix(62) + "result=success error=<nil>"))),
+            "{lines:?}"
+        );
+        assert!(C.with_label_values(&["success"]).get() > success);
+
+        let failure = C.with_label_values(&["failure"]).get();
+        let (read, lines) = read_ec_reconstructed(63, &[0, 6, 7, 8, 9, 10, 11, 12, 13], &[]).await;
+        assert_eq!(read, None);
+        assert!(
+            matches!(&lines[..], [l] if l.contains(&(prefix(63) + "result=failure error=")) && !l.ends_with("error=")),
+            "{lines:?}"
+        );
+        assert!(C.with_label_values(&["failure"]).get() > failure);
+
+        let (read, lines) = read_ec_reconstructed(64, &[], &[0, 10, 11, 12, 13]).await;
+        assert_eq!(
+            (read, lines),
+            (None, vec![]),
+            "an unrepairable corrupt read was marked"
+        );
+
+        let success = C.with_label_values(&["success"]).get();
+        let (read, lines) = read_ec_reconstructed(65, &[0], &[10, 11, 12, 13]).await;
+        assert_eq!(read, None, "a read rebuilt from corrupt parity was served");
+        assert!(
+            matches!(&lines[..], [l] if l.ends_with(&(prefix(65) + "result=success error=<nil>"))),
+            "{lines:?}"
+        );
+        assert!(C.with_label_values(&["success"]).get() > success);
+    }
+
     /// Write `data` as a chunk manifest on a fresh EC volume 2. Chunks stay on volume 1.
     fn put_ec_manifest(
         state: &VolumeServerState,

seaweed-volume/src/server/store_ec.rs

@@ -1356,7 +1356,39 @@ async fn fetch_one_interval(
 
     // Reconstruct: fan-out reads to every other shard at the same
     // (shard_offset, size). Mirrors `recoverOneRemoteEcShardInterval`.
-    recover_one_remote_ec_shard_interval(state, iv, map, &HashSet::from([iv.shard_id])).await
+    let result =
+        recover_one_remote_ec_shard_interval(state, iv, map, &HashSet::from([iv.shard_id])).await;
+    note_ec_reconstructed_interval(iv, &result);
+    result
+}
+
+/// Counts one interval that couldn't be read directly and was reconstructed,
+/// and at debug logs it with the needle and the encode generation, which the
+/// counter can't carry. `success` means the reconstruction produced data; the
+/// needle is validated later. A holder's "deleted" answer with too few shards
+/// rebuilt nothing, so it is a failure, as Go's reconstruct error makes it there.
+fn note_ec_reconstructed_interval(iv: EcInterval, result: &io::Result<(Vec<u8>, bool)>) {
+    let outcome = match result {
+        Ok((buf, _)) if !buf.is_empty() => "success",
+        _ => "failure",
+    };
+    crate::metrics::EC_RECONSTRUCTED_INTERVAL_COUNTER
+        .with_label_values(&[outcome])
+        .inc();
+    // tracing evaluates these arguments only when debug is enabled.
+    tracing::debug!(
+        "ec reconstructed interval: volume={} generation={} needle={} shard={} result={} error={}",
+        iv.vid.0,
+        iv.expected_encode_ts_ns,
+        iv.needle_id,
+        iv.shard_id,
+        outcome,
+        match result {
+            Ok((buf, _)) if !buf.is_empty() => "<nil>".to_string(),
+            Ok(_) => "a holder reported the needle deleted".to_string(),
+            Err(e) => e.to_string(),
+        }
+    );
 }
 
 async fn read_remote_ec_shard_interval(

weed/stats/metrics.go

@@ -668,6 +668,18 @@ var (
 			Help:      "Counter of EC shard rebuild operations by result.",
 		}, []string{"result"})
 
+	// VolumeServerECReconstructedIntervalCounter counts reads of an EC shard
+	// interval that could not be read directly and were served by reconstructing
+	// it from the other shards. success means the reconstruction produced data,
+	// not that the needle then passed validation.
+	VolumeServerECReconstructedIntervalCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: subsystemVolumeServer,
+			Name:      "ec_reconstructed_interval_total",
+			Help:      "Reads of an EC interval served by reconstructing from other shards; success means reconstruction produced data, not that the needle passed validation.",
+		}, []string{"result"})
+
 	S3RequestCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: Namespace,
@@ -1077,6 +1089,7 @@ func init() {
 	Gather.MustRegister(VolumeServerReplicationFailures)
 	Gather.MustRegister(VolumeServerECRebuildHistogram)
 	Gather.MustRegister(VolumeServerECRebuildCounter)
+	Gather.MustRegister(VolumeServerECReconstructedIntervalCounter)
 	Gather.MustRegister(MasterUnderReplicatedVolumes)
 	Gather.MustRegister(MasterVolumeCreationCounter)
 

weed/storage/store_ec.go

@@ -682,6 +682,7 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur
 
 	// try reading by recovering from other shards
 	_, is_deleted, err = s.recoverOneRemoteEcShardInterval(needleId, ecVolume, shardId, data, actualOffset)
+	noteEcReconstructedInterval(ecVolume, needleId, shardId, err)
 	if err == nil {
 		return
 	}
@@ -1264,6 +1265,20 @@ func (s *Store) recoverEcShardInterval(needleId types.NeedleId, ecVolume *erasur
 	return len(buf), is_deleted, nil
 }
 
+// noteEcReconstructedInterval counts one shard interval that could not be read
+// directly and was reconstructed from the other shards, and at V(3) logs it with
+// the needle and the volume's encode generation, which the counter can't carry.
+// success means the reconstruction produced data; the needle is validated later.
+func noteEcReconstructedInterval(ecVolume *erasure_coding.EcVolume, needleId types.NeedleId, shardId erasure_coding.ShardId, err error) {
+	result := "success"
+	if err != nil {
+		result = "failure"
+	}
+	stats.VolumeServerECReconstructedIntervalCounter.WithLabelValues(result).Inc()
+	glog.V(3).Infof("ec reconstructed interval: volume=%d generation=%d needle=%s shard=%d result=%s error=%v",
+		ecVolume.VolumeId, ecVolume.EncodeTsNs, needleId, shardId, result, err)
+}
+
 func (s *Store) EcVolumes() (ecVolumes []*erasure_coding.EcVolume) {
 	for _, location := range s.Locations {
 		location.ecVolumesLock.RLock()

weed/storage/store_ec_reconstructed_interval_test.go

@@ -0,0 +1,133 @@
+package storage
+
+import (
+	"io"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	flag "github.com/seaweedfs/seaweedfs/weed/util/fla9"
+)
+
+type shards = []erasure_coding.ShardId
+
+// reconstructedIntervals reads ec_reconstructed_interval_total; this package
+// has no parallel tests, so a test's before/after difference is its own.
+func reconstructedIntervals() (success, failure float64) {
+	c := stats.VolumeServerECReconstructedIntervalCounter
+	return testutil.ToFloat64(c.WithLabelValues("success")), testutil.ToFloat64(c.WithLabelValues("failure"))
+}
+
+// readEcNeedle writes a one-needle EC volume, overwrites the needle's range on
+// the corrupt shards, mounts every shard but missing (with no location for
+// them, so the read reconstructs), reads the needle, and returns how far each
+// counter moved.
+func readEcNeedle(t *testing.T, vid needle.VolumeId, missing, corrupt shards) (n *needle.Needle, err error, dSuccess, dFailure float64) {
+	t.Helper()
+	store := newTestStore(t, 1)
+	_, n = writeEcVolumeFiles(t, store.Locations[0].Directory, vid)
+	skip := map[erasure_coding.ShardId]bool{}
+	for _, sid := range missing {
+		skip[sid] = true
+	}
+	for sid := erasure_coding.ShardId(0); sid < erasure_coding.TotalShardsCount; sid++ {
+		if skip[sid] {
+			continue
+		}
+		if err := store.MountEcShards("", vid, sid, ""); err != nil {
+			t.Fatalf("mount shard %d: %v", sid, err)
+		}
+	}
+	ecVolume, _ := store.Locations[0].FindEcVolume(vid)
+	ecVolume.EncodeTsNs = 1700000000123456789
+	_, _, intervals, err := ecVolume.LocateEcShardNeedle(n.Id, ecVolume.Version)
+	if err != nil {
+		t.Fatalf("locate needle: %v", err)
+	}
+	_, off := ecVolume.IntervalToShardIdAndOffset(intervals[0])
+	for _, sid := range corrupt {
+		corruptEcShardRange(t, ecVolume, sid, off, int64(intervals[0].Size))
+	}
+	seedShardLocations(ecVolume, pb.NewServerAddressWithGrpcPort("127.0.0.1:1", 1))
+	ecVolume.ShardLocationsLock.Lock()
+	for sid := range skip {
+		delete(ecVolume.ShardLocations, sid)
+	}
+	ecVolume.ShardLocationsLock.Unlock()
+
+	s0, f0 := reconstructedIntervals()
+	got := &needle.Needle{Id: n.Id}
+	if _, err = store.ReadEcShardNeedle(vid, got, nil); err == nil && string(got.Data) != string(n.Data) {
+		t.Fatal("read returned different bytes")
+	}
+	s1, f1 := reconstructedIntervals()
+	return n, err, s1 - s0, f1 - f0
+}
+
+// What the counter means, case by case. A direct read isn't counted, nor is a
+// corrupt interval read directly (that goes to the repair path). A missing
+// shard counts one success when the reconstruction produced data and one
+// failure when it couldn't (5 shards left). success is not needle validation:
+// rebuilt from corrupt parity, the interval counts success while the read fails.
+func TestEcReconstructedIntervalCounter(t *testing.T) {
+	parity := shards{10, 11, 12, 13}
+	for _, c := range []struct {
+		vid              needle.VolumeId
+		missing, corrupt shards
+		fails            bool
+		s, f             float64
+	}{
+		{17, nil, nil, false, 0, 0},
+		{18, shards{0}, nil, false, 1, 0},
+		{19, shards{0, 6, 7, 8, 9, 10, 11, 12, 13}, nil, true, 0, 1},
+		{20, nil, append(shards{0}, parity...), true, 0, 0},
+		{21, shards{0}, parity, true, 1, 0},
+	} {
+		if _, err, s, f := readEcNeedle(t, c.vid, c.missing, c.corrupt); (err != nil) != c.fails || s != c.s || f != c.f {
+			t.Errorf("missing %v, corrupt %v: err %v, success +%v, failure +%v; want failed=%v, +%v, +%v", c.missing, c.corrupt, err, s, f, c.fails, c.s, c.f)
+		}
+	}
+}
+
+// The line binds a reconstruction to the volume's generation and the needle,
+// at V(3) and not below.
+func TestEcReconstructedIntervalLogLine(t *testing.T) {
+	flag.Set("logtostderr", "true")
+	t.Cleanup(func() { flag.Set("vmodule", ""); flag.Set("logtostderr", "false") })
+	markers := func(vmodule string, vid needle.VolumeId, missing shards) (n *needle.Needle, lines []string) {
+		flag.Set("vmodule", vmodule)
+		r, w, _ := os.Pipe()
+		stderr, out := os.Stderr, make(chan []byte, 1)
+		restore := sync.OnceFunc(func() { os.Stderr = stderr; w.Close() })
+		t.Cleanup(restore)
+		go func() { b, _ := io.ReadAll(r); r.Close(); out <- b }()
+		os.Stderr = w
+		n, _, _, _ = readEcNeedle(t, vid, missing, nil)
+		restore()
+		for _, line := range strings.Split(string(<-out), "\n") {
+			if strings.Contains(line, "ec reconstructed interval:") {
+				lines = append(lines, line)
+			}
+		}
+		return n, lines
+	}
+	prefix := func(vid int, n *needle.Needle) string {
+		return "ec reconstructed interval: volume=" + needle.VolumeId(vid).String() + " generation=1700000000123456789 needle=" + n.Id.String() + " shard=0 "
+	}
+	if n, lines := markers("store_ec=3", 22, shards{0}); len(lines) != 1 || !strings.HasSuffix(lines[0], prefix(22, n)+"result=success error=<nil>") {
+		t.Errorf("success line %q", lines)
+	}
+	if n, lines := markers("store_ec=3", 23, shards{0, 6, 7, 8, 9, 10, 11, 12, 13}); len(lines) != 1 ||
+		!strings.Contains(lines[0], prefix(23, n)+"result=failure error=") || strings.HasSuffix(lines[0], "error=<nil>") {
+		t.Errorf("failure line %q", lines)
+	}
+	if _, lines := markers("store_ec=2", 24, shards{0}); len(lines) != 0 {
+		t.Errorf("at V(2) the line was logged: %q", lines)
+	}
+}