From d6c65cd9248b368e53ce64dcf58292ce74dac3fc Mon Sep 17 00:00:00 2001 From: Ping Qiu Date: Wed, 4 Nov 2026 08:30:41 -0800 Subject: [PATCH] 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 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- seaweed-volume/src/metrics.rs | 16 ++ seaweed-volume/src/server/handlers.rs | 165 ++++++++++++++++++ seaweed-volume/src/server/store_ec.rs | 34 +++- weed/stats/metrics.go | 13 ++ weed/storage/store_ec.go | 15 ++ .../store_ec_reconstructed_interval_test.go | 133 ++++++++++++++ 6 files changed, 375 insertions(+), 1 deletion(-) create mode 100644 weed/storage/store_ec_reconstructed_interval_test.go diff --git a/seaweed-volume/src/metrics.rs b/seaweed-volume/src/metrics.rs index 45a52a9957..bb5dddfd45 100644 --- a/seaweed-volume/src/metrics.rs +++ b/seaweed-volume/src/metrics.rs @@ -252,6 +252,21 @@ pub static SCRUB_SHARD_FAILURES: LazyLock = 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 = 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 = 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()), diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index e7a3b122ab..846e42e95f 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/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) { + 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>>); + + impl std::io::Write for Captured { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + 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="))), + "{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="))), + "{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, diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index d11486fd47..e26920efa0 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/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, 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() => "".to_string(), + Ok(_) => "a holder reported the needle deleted".to_string(), + Err(e) => e.to_string(), + } + ); } async fn read_remote_ec_shard_interval( diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index 4678c227d5..a7d2e75b96 100644 --- a/weed/stats/metrics.go +++ b/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) diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 2da3327e74..7df9bf090a 100644 --- a/weed/storage/store_ec.go +++ b/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() diff --git a/weed/storage/store_ec_reconstructed_interval_test.go b/weed/storage/store_ec_reconstructed_interval_test.go new file mode 100644 index 0000000000..a2864f61c9 --- /dev/null +++ b/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=") { + 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=") { + 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) + } +}