fix(ecstore): split internal get metadata metrics (#5687)

Classify expected metadata-missing errors separately from unknown get pipeline failures and attribute internal meta-bucket reader failures to an internal_meta path instead of legacy_duplex.

This keeps scanner/data-usage metadata probes from polluting user GET/mixed failure attribution while preserving the existing read error behavior.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/diagnostics/get.rs

@@ -24,6 +24,7 @@ pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty";
 pub(crate) const GET_OBJECT_PATH_DIRECT_MEMORY: &str = "direct_memory";
 pub(crate) const GET_OBJECT_PATH_BODY_CACHE: &str = "body_cache";
 pub(crate) const GET_OBJECT_PATH_INLINE_DIRECT: &str = "inline_direct";
+pub(crate) const GET_OBJECT_PATH_INTERNAL_META: &str = "internal_meta";
 pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex";
 pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";
 pub(crate) const GET_OBJECT_PATH_SET_DISK: &str = "set_disk";
@@ -163,6 +164,7 @@ pub(crate) enum GetObjectFailureReason {
     DecodeError,
     DownstreamClosed,
     Io,
+    MetadataMissing,
     RangeOrLengthInvalid,
     ReadQuorum,
     ShortRead,
@@ -177,6 +179,7 @@ impl GetObjectFailureReason {
             Self::DecodeError => "decode_error",
             Self::DownstreamClosed => "downstream_closed",
             Self::Io => "io",
+            Self::MetadataMissing => "metadata_missing",
             Self::RangeOrLengthInvalid => "range_or_length_invalid",
             Self::ReadQuorum => "read_quorum",
             Self::ShortRead => "short_read",
@@ -190,6 +193,13 @@ pub(crate) fn classify_storage_error(err: &StorageError) -> GetObjectFailureReas
     match err {
         StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) => GetObjectFailureReason::ReadQuorum,
         StorageError::FileCorrupt => GetObjectFailureReason::BitrotMismatch,
+        StorageError::FileNotFound
+        | StorageError::FileVersionNotFound
+        | StorageError::VolumeNotFound
+        | StorageError::BucketNotFound(_)
+        | StorageError::ObjectNotFound(_, _)
+        | StorageError::VersionNotFound(_, _, _)
+        | StorageError::ConfigNotFound => GetObjectFailureReason::MetadataMissing,
         StorageError::InvalidRangeSpec(_) => GetObjectFailureReason::RangeOrLengthInvalid,
         StorageError::Io(io_err) => classify_io_error(io_err),
         _ => GetObjectFailureReason::Unknown,
@@ -293,6 +303,34 @@ mod tests {
             classify_storage_error(&StorageError::InvalidRangeSpec("bad range".to_string())),
             GetObjectFailureReason::RangeOrLengthInvalid
         );
+        assert_eq!(
+            classify_storage_error(&StorageError::FileNotFound),
+            GetObjectFailureReason::MetadataMissing
+        );
+        assert_eq!(
+            classify_storage_error(&StorageError::VolumeNotFound),
+            GetObjectFailureReason::MetadataMissing
+        );
+        assert_eq!(
+            classify_storage_error(&StorageError::ObjectNotFound("bucket".to_string(), "object".to_string())),
+            GetObjectFailureReason::MetadataMissing
+        );
+        assert_eq!(
+            classify_storage_error(&StorageError::BucketNotFound("bucket".to_string())),
+            GetObjectFailureReason::MetadataMissing
+        );
+        assert_eq!(
+            classify_storage_error(&StorageError::VersionNotFound(
+                "bucket".to_string(),
+                "object".to_string(),
+                "version".to_string()
+            )),
+            GetObjectFailureReason::MetadataMissing
+        );
+        assert_eq!(
+            classify_storage_error(&StorageError::ConfigNotFound),
+            GetObjectFailureReason::MetadataMissing
+        );
 
         let internal_broken_pipe = StorageError::Io(io::Error::from(io::ErrorKind::BrokenPipe));
         assert_eq!(classify_storage_error(&internal_broken_pipe), GetObjectFailureReason::Io);
@@ -354,10 +392,12 @@ mod tests {
         assert_eq!(GetObjectFailureReason::DownstreamClosed.as_str(), "downstream_closed");
         assert_eq!(GetObjectFailureReason::BitrotMismatch.as_str(), "bitrot_mismatch");
         assert_eq!(GetObjectFailureReason::DecodeError.as_str(), "decode_error");
+        assert_eq!(GetObjectFailureReason::MetadataMissing.as_str(), "metadata_missing");
         assert_eq!(GET_READER_BUFFER_OUTPUT, "output");
         assert_eq!(GET_READER_BUFFER_PREFETCH, "prefetch");
         assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, "codec_streaming_legacy_engine");
         assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, "codec_streaming_rustfs_engine");
+        assert_eq!(GET_OBJECT_PATH_INTERNAL_META, "internal_meta");
         assert_eq!(GET_DIRECT_MEMORY_DECISION_USE, "use");
         assert_eq!(GET_DIRECT_MEMORY_DECISION_FALLBACK, "fallback");
         assert_eq!(GET_DIRECT_MEMORY_REASON_NONE, "none");

crates/ecstore/src/set_disk/mod.rs

@@ -61,10 +61,11 @@ use crate::data_usage::record_compression_total_memory;
 use crate::diagnostics::get::{
     GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
     GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_DIRECT_MEMORY,
-    GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION,
+    GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX,
-    GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE, GET_STAGE_LOCK_ACQUIRE,
+    GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE,
-    GET_STAGE_METADATA, GET_STAGE_OBJECT_INFO, GET_STAGE_PATH_DECISION, GET_STAGE_READER_SETUP, classify_storage_error,
+    GET_STAGE_LOCK_ACQUIRE, GET_STAGE_METADATA, GET_STAGE_OBJECT_INFO, GET_STAGE_PATH_DECISION, GET_STAGE_READER_SETUP,
-    get_stage_timer_if_enabled, record_get_object_pipeline_failure, record_get_stage_duration_if_enabled,
+    classify_storage_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
+    record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
 };
 use crate::disk::error_reduce::{
     BUCKET_OP_IGNORED_ERRS, OBJECT_OP_IGNORED_ERRS, build_write_quorum_failure_summary, count_errs, reduce_read_quorum_errs,

crates/ecstore/src/set_disk/ops/object.rs

@@ -383,7 +383,12 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 Ok((fi, files, disks)) => (fi, files, disks, None),
                 Err(err) => {
                     rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64());
-                    record_get_object_pipeline_failure(GET_STAGE_METADATA, classify_storage_error(&err));
+                    let failure_path = if is_meta_bucketname(bucket) {
+                        GET_OBJECT_PATH_INTERNAL_META
+                    } else {
+                        GET_OBJECT_PATH_LEGACY_DUPLEX
+                    };
+                    record_get_object_pipeline_failure_for_path(failure_path, GET_STAGE_METADATA, classify_storage_error(&err));
                     return Err(to_object_err(err, vec![bucket, object]));
                 }
             }
@@ -5488,7 +5493,8 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support {
 mod get_object_downstream_close_accounting_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks;
     use super::*;
-    use crate::diagnostics::get::{GET_STAGE_DECODE, GET_STAGE_EMIT, GetObjectFailureReason};
+    use crate::diagnostics::get::{GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT, GetObjectFailureReason};
+    use crate::disk::RUSTFS_META_BUCKET;
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
     use crate::test_metrics::CapturingRecorder;
@@ -5581,6 +5587,65 @@ mod get_object_downstream_close_accounting_tests {
         assert!(decode_failures > 0, "the producer must expose the downstream close at decode");
         assert_eq!(emit_failures, 0, "downstream closure must not be counted as an emit failure");
     }
+
+    #[test]
+    #[serial_test::serial]
+    fn missing_internal_meta_reader_records_internal_meta_path() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (internal_missing, legacy_unknown) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
+                let options = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                let result = set_disks
+                    .get_object_reader(
+                        RUSTFS_META_BUCKET,
+                        "buckets/.usage-cache/nonexistent.bin",
+                        None,
+                        HeaderMap::new(),
+                        &options,
+                    )
+                    .await;
+                assert!(result.is_err(), "missing internal metadata must still return an error");
+
+                (
+                    recorder.counter_value(
+                        "rustfs_io_get_object_pipeline_failures_total",
+                        &[
+                            ("path", GET_OBJECT_PATH_INTERNAL_META),
+                            ("stage", GET_STAGE_METADATA),
+                            ("reason", GetObjectFailureReason::MetadataMissing.as_str()),
+                        ],
+                    ),
+                    recorder.counter_value(
+                        "rustfs_io_get_object_pipeline_failures_total",
+                        &[
+                            ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
+                            ("stage", GET_STAGE_METADATA),
+                            ("reason", GetObjectFailureReason::Unknown.as_str()),
+                        ],
+                    ),
+                )
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert!(internal_missing > 0, "internal metadata miss must use the internal_meta path");
+        assert_eq!(
+            legacy_unknown, 0,
+            "internal metadata miss must not be attributed to legacy_duplex/unknown"
+        );
+    }
 }
 
 #[cfg(test)]

rustfs/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 //! RustFS — high-performance S3-compatible object storage.
 //!
 //! This library exposes the [`embedded`] module which lets you start a