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