From 9863f4848d74e28aab222198cb794429ef1575b7 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 4 Sep 2026 09:40:03 +0800 Subject: [PATCH] feat(tier): report cluster tier stats and count tier requests (#7110) `GET /v3/tier-stats` answered from whichever process received the request, returning that node's rolling 24-hour transition counters as if they were cluster totals, and the `TierRequestsSuccess` and `TierRequestsFailure` metric names had no producer at all. The body now separates the two quantities a tier carries. Stored inventory comes from the persisted scanner usage snapshot, which is already cluster-wide; rolling activity is summed over every member through a new read-only `TierDailyStats` peer RPC. Rings are merged rather than added, so an idle node's expired hours age out, and each node counts only its own committed transitions, so a retry is counted once. Coverage travels with the numbers: `activity.status` names the reporting members and the ones that could not be asked, timed out, or answered with a ring this build refuses to merge, and per-tier inventory is absent rather than zero when the snapshot has no accounting. The version 1 body stays reachable at `?format=legacy`. Tier request counters are recorded at the two seams every remote request passes through, so a new provider is counted by construction, with a closed operation/outcome label set that can never grow a tier name, endpoint or object key. Closes rustfs/backlog#2207 Co-authored-by: cxymds --- crates/e2e_test/README.md | 2 +- crates/e2e_test/src/lib.rs | 4 + .../e2e_test/src/reliant/grpc_lock_server.rs | 7 + .../e2e_test/src/tier_stats_cluster_test.rs | 144 +++++ crates/ecstore/src/api/mod.rs | 4 +- .../bucket/lifecycle/tier_last_day_stats.rs | 156 ++++- .../src/cluster/rpc/peer_rest_client.rs | 37 +- .../ecstore/src/services/notification_sys.rs | 175 ++++++ .../ecstore/src/services/tier/tier_config.rs | 2 +- .../ecstore/src/services/tier/warm_backend.rs | 259 +++++++- crates/obs/src/metrics/collectors/mod.rs | 2 + crates/obs/src/metrics/collectors/tier.rs | 96 +++ crates/obs/src/metrics/scheduler.rs | 5 +- .../obs/src/metrics/schema/entry/subsystem.rs | 4 + crates/obs/src/metrics/schema/mod.rs | 1 + crates/obs/src/metrics/schema/tier.rs | 43 ++ crates/obs/src/metrics/stats_collector.rs | 19 +- .../src/generated/proto_gen/node_service.rs | 66 +++ crates/protos/src/node.proto | 17 + crates/scanner-metrics/src/metrics.rs | 184 ++++++ docs/architecture/README.md | 1 + docs/architecture/tier-stats-contract.md | 43 ++ rustfs/src/admin/handlers/tier.rs | 556 +++++++++++++++++- rustfs/src/admin/storage_api.rs | 11 +- rustfs/src/storage/rpc/node_service.rs | 7 + .../src/storage/rpc/node_service/metrics.rs | 36 +- rustfs/src/storage/storage_api.rs | 9 +- 27 files changed, 1844 insertions(+), 46 deletions(-) create mode 100644 crates/e2e_test/src/tier_stats_cluster_test.rs create mode 100644 crates/obs/src/metrics/collectors/tier.rs create mode 100644 crates/obs/src/metrics/schema/tier.rs create mode 100644 docs/architecture/tier-stats-contract.md diff --git a/crates/e2e_test/README.md b/crates/e2e_test/README.md index 2685661b3..ed5a65d97 100644 --- a/crates/e2e_test/README.md +++ b/crates/e2e_test/README.md @@ -25,7 +25,7 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern: | **policy** | [`src/policy/`](src/policy), `existing_object_tag_policy_test`, `bucket_policy_check_test`, `anonymous_access_test`, `security_boundary_test`, `multipart_auth_test` | IAM / bucket-policy / STS session policy, policy variables, anonymous access, DoS/SSRF boundaries. Own guide: [`src/policy/README.md`](src/policy/README.md) | | **protocols** | [`src/protocols/`](src/protocols) | FTPS, WebDAV, SFTP compliance. Fixed ports, own guide: [`src/protocols/README.md`](src/protocols/README.md) | | **reliant** | [`src/reliant/`](src/reliant) | Tests that reuse an **externally started** server (SQL/select, conditional writes, lifecycle, deleted-object reads, node-interact). Run via [`scripts/run_e2e_tests.sh`](../../scripts/run_e2e_tests.sh); see [`src/reliant/README.md`](src/reliant/README.md) | -| **cluster** | `cluster_concurrency_test`, `stale_multipart_cleanup_cluster_test`, `namespace_lock_quorum_test`, `admin_timeout_regression_test`, `object_lambda_test`, `replication_extension_test` | Multi-node scenarios via `RustFSTestClusterEnvironment` | +| **cluster** | `cluster_concurrency_test`, `stale_multipart_cleanup_cluster_test`, `namespace_lock_quorum_test`, `admin_timeout_regression_test`, `object_lambda_test`, `replication_extension_test`, `tier_stats_cluster_test` | Multi-node scenarios via `RustFSTestClusterEnvironment` | | **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup | | **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory | diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index e68ee426a..9279c39c4 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -218,6 +218,10 @@ mod cluster_multidrive_pool_test; #[cfg(test)] mod inline_fast_path_cluster_test; +// backlog#2207: two-node gate for the cluster-authoritative tier stats contract. +#[cfg(test)] +mod tier_stats_cluster_test; + // PutObject / MultipartUpload with checksum (Content-MD5, x-amz-checksum-*) #[cfg(test)] mod checksum_upload_test; diff --git a/crates/e2e_test/src/reliant/grpc_lock_server.rs b/crates/e2e_test/src/reliant/grpc_lock_server.rs index a03a8a11e..d73a44787 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_server.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_server.rs @@ -904,6 +904,13 @@ impl NodeService for MinimalLockNodeService { ) -> Result, Status> { Err(Status::unimplemented("lock-only test server")) } + + async fn tier_daily_stats( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } } /// Spawn a gRPC lock server on a random port diff --git a/crates/e2e_test/src/tier_stats_cluster_test.rs b/crates/e2e_test/src/tier_stats_cluster_test.rs new file mode 100644 index 000000000..b596a03d4 --- /dev/null +++ b/crates/e2e_test/src/tier_stats_cluster_test.rs @@ -0,0 +1,144 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Two-node gate for the tier stats wire contract (rustfs/backlog#2207). +//! +//! Before this contract, `GET /v3/tier-stats` answered from the process that +//! happened to receive the request, so the same query returned different +//! numbers depending on which node a client reached, with nothing in the body +//! saying so. These tests pin the two properties that fix costs: the answer is +//! node-independent, and it states how much of the cluster it covers. + +use crate::common::{RustFSTestClusterEnvironment, admin_request, init_logging}; +use http::Method; +use http::StatusCode; +use serde_json::Value; +use std::time::Duration; +use tokio::time::{Instant, sleep}; + +type TestResult = Result>; + +const TIER_STATS_PATH: &str = "/rustfs/admin/v3/tier-stats"; +const PEER_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(30); + +async fn tier_stats(cluster: &RustFSTestClusterEnvironment, node: usize, query: &str) -> TestResult<(StatusCode, String)> { + admin_request( + &cluster.nodes[node].url, + Method::GET, + &format!("{TIER_STATS_PATH}{query}"), + None, + &cluster.access_key, + &cluster.secret_key, + ) + .await +} + +async fn tier_stats_json(cluster: &RustFSTestClusterEnvironment, node: usize) -> TestResult { + let (status, body) = tier_stats(cluster, node, "").await?; + assert_eq!(status, StatusCode::OK, "node {node} must answer tier-stats: {body}"); + Ok(serde_json::from_str(&body)?) +} + +/// Wait until every node reports the whole cluster. +/// +/// Peer clients are established during startup, so a query issued in the first +/// moments can legitimately see a peer as unavailable. Polling separates that +/// startup window from the failure this test is about: an answer that stays +/// partial because the peer never reports at all. +async fn wait_for_complete_activity(cluster: &RustFSTestClusterEnvironment) -> TestResult> { + let deadline = Instant::now() + PEER_CONVERGENCE_TIMEOUT; + loop { + let mut bodies = Vec::with_capacity(cluster.nodes.len()); + for node in 0..cluster.nodes.len() { + bodies.push(tier_stats_json(cluster, node).await?); + } + + if bodies.iter().all(|body| body["activity"]["status"] == "complete") { + return Ok(bodies); + } + if Instant::now() >= deadline { + return Err(format!("tier-stats activity never became complete on every node: {bodies:?}").into()); + } + sleep(Duration::from_millis(500)).await; + } +} + +#[tokio::test] +async fn tier_stats_answers_the_same_cluster_result_from_either_node() -> TestResult { + init_logging(); + + let mut cluster = RustFSTestClusterEnvironment::new(2).await?; + cluster.start().await?; + + let bodies = wait_for_complete_activity(&cluster).await?; + let (first, second) = (&bodies[0], &bodies[1]); + + for (node, body) in bodies.iter().enumerate() { + assert_eq!(body["contractVersion"], 2, "node {node} must name the current contract version"); + assert_eq!( + body["activity"]["nodesExpected"], 2, + "node {node} must expect both cluster members, not only itself" + ); + assert_eq!( + body["activity"]["nodesReporting"], 2, + "node {node} must include its peer's rolling window in the sum" + ); + assert_eq!( + body["activity"]["unavailableNodes"], + serde_json::json!([]), + "node {node} reported a complete result while naming an unavailable member" + ); + } + + // A fixture cluster has no remote tier, so this pair is equal over an + // empty list; `nodesReporting` above is what proves the peer answered. + // The per-tier summing itself is pinned by the aggregator unit tests. + assert_eq!( + first["tiers"], second["tiers"], + "the same query must not depend on which node received it" + ); + assert_eq!( + first["inventory"]["status"], second["inventory"]["status"], + "both nodes read the same persisted usage snapshot" + ); + + cluster.stop(); + Ok(()) +} + +#[tokio::test] +async fn tier_stats_keeps_the_legacy_body_reachable_and_rejects_unknown_formats() -> TestResult { + init_logging(); + + let mut cluster = RustFSTestClusterEnvironment::new(2).await?; + cluster.start().await?; + + let (status, body) = tier_stats(&cluster, 0, "?format=legacy").await?; + assert_eq!(status, StatusCode::OK, "the pinned legacy body must stay reachable: {body}"); + let legacy: Value = serde_json::from_str(&body)?; + assert!( + legacy.is_object() && legacy.get("contractVersion").is_none(), + "the legacy body is the bare tier map, not the current envelope: {legacy}" + ); + + let (status, body) = tier_stats(&cluster, 0, "?format=v3").await?; + assert_eq!( + status, + StatusCode::BAD_REQUEST, + "an unknown format must be rejected rather than answered in another shape: {body}" + ); + + cluster.stop(); + Ok(()) +} diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 110a617b4..6ad57fb21 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -118,7 +118,7 @@ pub mod bucket { } pub mod tier_last_day_stats { - pub use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats}; + pub use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats, TierDailyStatsWire}; } pub mod tier_sweeper { @@ -472,7 +472,7 @@ pub mod notification { #[cfg(any(test, feature = "test-util"))] pub use crate::services::notification_sys::rotate_cross_pool_fence_fleet_proof_for_test; pub use crate::services::notification_sys::{ - CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant, + ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys, new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe, }; diff --git a/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs b/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs index 2d8d6c2ea..ec4b5b5cc 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs @@ -19,6 +19,7 @@ #![allow(clippy::all)] use rustfs_data_usage::TierStats; +use serde::{Deserialize, Serialize}; use sha2::Sha256; use std::collections::HashMap; use std::ops::Sub; @@ -27,7 +28,24 @@ use tracing::{error, warn}; pub type DailyAllTierStats = HashMap; -#[derive(Clone)] +/// One bin per hour of the rolling day. The bin index is the UTC hour, so the +/// array is a ring the writer ages forward rather than a queue. +pub const TIER_DAILY_STATS_BINS: usize = 24; + +/// Interchange form of [`LastDayTierStats`] for the internode tier-stats RPC. +/// +/// The in-memory type keeps its bins private because the ring is only +/// meaningful together with `updated_at`; this type carries both across the +/// wire and is validated back into the ring by [`LastDayTierStats::from_wire`]. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct TierDailyStatsWire { + pub bins: Vec, + /// Seconds since the Unix epoch. Bins are hour-resolution, so a coarser + /// timestamp than the in-memory `OffsetDateTime` loses nothing. + pub updated_at_unix_secs: i64, +} + +#[derive(Clone, Debug)] pub struct LastDayTierStats { bins: [TierStats; 24], updated_at: OffsetDateTime, @@ -80,11 +98,57 @@ impl LastDayTierStats { } } - #[allow( - dead_code, - reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)" - )] - fn merge(&self, m: LastDayTierStats) -> LastDayTierStats { + /// The rolling ring as observed, without aging it forward. + /// + /// Only meaningful together with [`LastDayTierStats::updated_at`]: a bin + /// belongs to the hour of its index within the day that ends at + /// `updated_at`. + pub fn bins(&self) -> &[TierStats; TIER_DAILY_STATS_BINS] { + &self.bins + } + + pub fn updated_at(&self) -> OffsetDateTime { + self.updated_at + } + + pub fn to_wire(&self) -> TierDailyStatsWire { + TierDailyStatsWire { + bins: self.bins.to_vec(), + updated_at_unix_secs: self.updated_at.unix_timestamp(), + } + } + + /// Rebuild the ring from a peer's response. + /// + /// A ring of the wrong width or an unrepresentable timestamp is corrupt + /// peer input, not a zero sample: it returns an error so the caller can + /// report the node as non-reporting instead of merging a plausible but + /// wrong day into a cluster total. + pub fn from_wire(wire: TierDailyStatsWire) -> Result { + let bins: [TierStats; TIER_DAILY_STATS_BINS] = wire.bins.try_into().map_err(|bins: Vec| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("tier daily stats must carry {TIER_DAILY_STATS_BINS} bins, got {}", bins.len()), + ) + })?; + let updated_at = OffsetDateTime::from_unix_timestamp(wire.updated_at_unix_secs).map_err(|err| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("tier daily stats carry an unrepresentable timestamp: {err}"), + ) + })?; + + Ok(Self { bins, updated_at }) + } + + /// Combine two independently observed rings. + /// + /// Each node counts only the transitions it completed itself, so summing + /// bins across nodes is a cluster total rather than a double count. The + /// older ring is aged forward to the newer one's clock first, so a node + /// that stopped transitioning hours ago contributes its still-current + /// bins and not its expired ones. + pub fn merge(&self, m: LastDayTierStats) -> LastDayTierStats { let mut cl = self.clone(); let mut cm = m; let mut merged = LastDayTierStats::default(); @@ -108,6 +172,7 @@ impl LastDayTierStats { #[cfg(test)] mod test { use super::*; + use time::Duration; #[test] fn total_sums_all_recorded_stats() { @@ -132,4 +197,83 @@ mod test { } ); } + + fn sample(total_size: u64) -> TierStats { + TierStats { + total_size, + num_versions: 1, + num_objects: 1, + } + } + + #[test] + fn wire_round_trip_preserves_the_ring_and_its_clock() { + let mut stats = LastDayTierStats::default(); + stats.add_stats(sample(10)); + + let restored = LastDayTierStats::from_wire(stats.to_wire()).expect("a ring this node produced must decode"); + + assert_eq!(restored.bins(), stats.bins(), "every bin must survive the wire"); + assert_eq!( + restored.updated_at().unix_timestamp(), + stats.updated_at().unix_timestamp(), + "the ring's clock must survive the wire" + ); + } + + #[test] + fn a_ring_of_the_wrong_width_is_rejected() { + let mut wire = LastDayTierStats::default().to_wire(); + wire.bins.pop(); + + let err = LastDayTierStats::from_wire(wire).expect_err("a short ring must not decode as a zero day"); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + } + + #[test] + fn an_unrepresentable_clock_is_rejected() { + let mut wire = LastDayTierStats::default().to_wire(); + wire.updated_at_unix_secs = i64::MIN; + + let err = LastDayTierStats::from_wire(wire).expect_err("an unrepresentable clock must not decode"); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + } + + #[test] + fn merge_sums_two_nodes_that_transitioned_in_the_same_hour() { + let mut left = LastDayTierStats::default(); + left.add_stats(sample(10)); + let mut right = LastDayTierStats::default(); + right.add_stats(sample(20)); + + assert_eq!( + left.merge(right).total(), + TierStats { + total_size: 30, + num_versions: 2, + num_objects: 2, + }, + "each node counts only its own completions, so a merge is a cluster total" + ); + } + + #[test] + fn merge_ages_out_a_peer_ring_older_than_a_day() { + let mut stale = LastDayTierStats::default(); + stale.add_stats(sample(10)); + stale.updated_at -= Duration::days(2); + + let mut fresh = LastDayTierStats::default(); + fresh.add_stats(sample(20)); + + assert_eq!( + fresh.merge(stale).total(), + TierStats { + total_size: 20, + num_versions: 1, + num_objects: 1, + }, + "a node that stopped transitioning more than a day ago must not keep contributing" + ); + } } diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 3b1ae83fc..03648d8d6 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -23,6 +23,7 @@ use crate::storage_api_contracts::internode::{ SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, }; use crate::{ + bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats, TierDailyStatsWire}, bucket::replication::BucketStats, disk::disk_store::{get_drive_active_check_interval, get_drive_active_check_timeout}, layout::endpoints::EndpointServerPools, @@ -48,7 +49,7 @@ use rustfs_protos::proto_gen::node_service::{ LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ReplacementRecoveryStatusRequest, ScannerActivityRequest, ScannerActivityResponse, ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ScannerPublicationLeaseResponse, ServerInfoRequest, SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest, - StartProfilingRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest, + StartProfilingRequest, StopRebalanceRequest, TierDailyStatsRequest, TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse, TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient, tier_mutation_control_service_client::TierMutationControlServiceClient, }; @@ -986,6 +987,40 @@ impl PeerRestClient { Ok(cpus) } + /// This peer's own rolling-day transition counters, per remote tier. + /// + /// The response is untrusted peer input: a ring of the wrong width or an + /// unrepresentable clock is rejected here rather than merged, so a corrupt + /// answer makes the node non-reporting instead of silently shifting a + /// cluster total. + pub async fn tier_daily_stats(&self) -> Result { + self.finalize_result(self.tier_daily_stats_inner().await).await + } + + async fn tier_daily_stats_inner(&self) -> Result { + let mut client = self.get_client().await?; + let request = Request::new(TierDailyStatsRequest {}); + + let response = client.tier_daily_stats(request).await?.into_inner(); + if !response.success { + if let Some(msg) = response.error_info { + return Err(Error::other(msg)); + } + return Err(peer_failure_without_details("tier_daily_stats", None)); + } + + let mut buf = Deserializer::new(Cursor::new(response.tier_daily_stats)); + let wire: HashMap = Deserialize::deserialize(&mut buf)?; + + wire.into_iter() + .map(|(tier, stats)| { + LastDayTierStats::from_wire(stats) + .map(|stats| (tier, stats)) + .map_err(Error::from) + }) + .collect() + } + pub async fn get_net_info(&self) -> Result { self.finalize_result(self.get_net_info_inner().await).await } diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 01d402704..a7e399031 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::bucket::lifecycle::tier_last_day_stats::DailyAllTierStats; use crate::cluster::rpc::{PeerRestClient, ScannerPeerActivity, ScannerPublicationLease, TierConfigReloadOutcome}; use crate::diagnostics::admin_server_info::get_commit_id; use crate::disk::DiskAPI; @@ -50,6 +51,7 @@ const LOG_SUBSYSTEM_NOTIFICATION: &str = "notification"; const EVENT_NOTIFICATION_PEER_PROPAGATION: &str = "notification_peer_propagation"; const EVENT_NOTIFICATION_CAPABILITY_PROBE: &str = "notification_capability_probe"; const SCANNER_ACTIVITY_PROBE_TIMEOUT: Duration = Duration::from_secs(5); +const TIER_DAILY_STATS_PROBE_TIMEOUT: Duration = Duration::from_secs(5); const TIER_CONFIG_RELOAD_RETRY_BASE: Duration = Duration::from_millis(100); const TIER_CONFIG_RELOAD_RETRY_CAP: Duration = Duration::from_secs(5); const REMOTE_VERSION_STATE_PROBE_INTERVAL: Duration = Duration::from_secs(10); @@ -973,6 +975,120 @@ impl NotificationSys { } } +/// Rolling tier activity summed over every cluster member that answered, with +/// the reporting coverage behind the sum. +/// +/// The coverage is part of the result rather than a log line: a sum over a +/// subset of the cluster is not a cluster total, and a caller that renders it +/// as one is the defect this type exists to prevent. +pub struct ClusterTierDailyStats { + pub stats: DailyAllTierStats, + /// Members whose rolling day is included, always at least this node. + pub nodes_reporting: usize, + /// Members this deployment expects to hear from, including this node. + pub nodes_expected: usize, + /// Members that could not be asked, could not answer, or answered with a + /// ring this build refuses to merge. Sorted, and named by grid host. + pub unavailable_nodes: Vec, +} + +impl ClusterTierDailyStats { + pub fn is_complete(&self) -> bool { + self.unavailable_nodes.is_empty() && self.nodes_reporting == self.nodes_expected + } +} + +/// Fold one member's rolling day into the running cluster total. +/// +/// Merging (rather than adding totals) ages each member's ring to the newer +/// clock first, so a member that stopped transitioning yesterday contributes +/// only the hours still inside the rolling day. +fn merge_tier_daily_stats(into: &mut DailyAllTierStats, from: DailyAllTierStats) { + for (tier, stats) in from { + match into.remove(&tier) { + Some(existing) => { + into.insert(tier, existing.merge(stats)); + } + None => { + into.insert(tier, stats); + } + } + } +} + +impl NotificationSys { + /// Sum this node's rolling tier activity with every reachable peer's. + /// + /// Each node records only the transitions it completed itself, so the sum + /// is a cluster total and a retried transition is counted once, by the + /// node that finally committed it. Peers are probed concurrently under a + /// per-peer deadline so one black-holed member cannot hold the admin + /// request open. + pub async fn tier_daily_stats(&self, local: DailyAllTierStats) -> ClusterTierDailyStats { + let mut stats = local; + let nodes_expected = self.peer_clients.len() + 1; + let mut nodes_reporting = 1; + let mut unavailable_nodes = Vec::new(); + + let mut probes = Vec::with_capacity(self.peer_clients.len()); + for (idx, client) in self.peer_clients.iter().enumerate() { + let host = self.tier_daily_stats_peer_host(idx, client.as_ref()); + probes.push(async move { + let Some(client) = client.as_ref() else { + return (host, Err(Error::other("peer is not reachable"))); + }; + // The peer is already named by the caller's `peer` log field + // and by `unavailable_nodes`, so the deadline is reported as + // the typed variant rather than a formatted fragment. + let result = timeout(TIER_DAILY_STATS_PROBE_TIMEOUT, client.tier_daily_stats()) + .await + .unwrap_or(Err(Error::Timeout)); + (host, result) + }); + } + + for (host, result) in join_all(probes).await { + match result { + Ok(peer_stats) => { + nodes_reporting += 1; + merge_tier_daily_stats(&mut stats, peer_stats); + } + Err(err) => { + warn!( + event = EVENT_NOTIFICATION_CAPABILITY_PROBE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + peer = host, + error = %err, + "tier daily stats peer did not report" + ); + unavailable_nodes.push(host); + } + } + } + + unavailable_nodes.sort(); + ClusterTierDailyStats { + stats, + nodes_reporting, + nodes_expected, + unavailable_nodes, + } + } + + /// Name a peer slot even when no client was ever built for it, so an + /// unreachable member is reported by host instead of disappearing. + fn tier_daily_stats_peer_host(&self, idx: usize, client: Option<&PeerRestClient>) -> String { + if let Some(client) = client { + return client.grid_host.clone(); + } + self.peer_topology_hosts + .get(idx) + .cloned() + .unwrap_or_else(|| format!("peer[{idx}]")) + } +} + pub struct NotificationPeerErr { pub host: String, pub err: Option, @@ -2935,6 +3051,65 @@ fn aggregate_scanner_dirty_usage_acknowledgement_results( #[cfg(test)] mod tests { use super::*; + use crate::bucket::lifecycle::tier_last_day_stats::LastDayTierStats; + use rustfs_data_usage::TierStats; + + fn ring(total_size: u64) -> LastDayTierStats { + let mut stats = LastDayTierStats::default(); + stats.add_stats(TierStats { + total_size, + num_versions: 1, + num_objects: 1, + }); + stats + } + + #[test] + fn merging_peer_rings_sums_each_tier_without_double_counting() { + let mut cluster = DailyAllTierStats::from([("WARM".to_string(), ring(10))]); + + merge_tier_daily_stats( + &mut cluster, + DailyAllTierStats::from([("WARM".to_string(), ring(20)), ("COLD".to_string(), ring(5))]), + ); + + assert_eq!( + cluster.get("WARM").expect("the shared tier must survive the merge").total(), + TierStats { + total_size: 30, + num_versions: 2, + num_objects: 2, + }, + "both nodes' completions belong in the cluster total" + ); + assert_eq!( + cluster.get("COLD").expect("a tier only one node saw must be kept").total(), + TierStats { + total_size: 5, + num_versions: 1, + num_objects: 1, + } + ); + } + + #[test] + fn a_single_member_result_is_complete_and_a_missing_peer_is_not() { + let complete = ClusterTierDailyStats { + stats: DailyAllTierStats::new(), + nodes_reporting: 1, + nodes_expected: 1, + unavailable_nodes: Vec::new(), + }; + assert!(complete.is_complete(), "a single-member deployment reports its whole cluster"); + + let partial = ClusterTierDailyStats { + stats: DailyAllTierStats::new(), + nodes_reporting: 1, + nodes_expected: 2, + unavailable_nodes: vec!["node-b:9000".to_string()], + }; + assert!(!partial.is_complete(), "a silent member must make the sum partial"); + } #[test] fn cross_pool_policy_versions_authorize_only_their_supported_protocols() { diff --git a/crates/ecstore/src/services/tier/tier_config.rs b/crates/ecstore/src/services/tier/tier_config.rs index 01d8444cc..94fe4e408 100644 --- a/crates/ecstore/src/services/tier/tier_config.rs +++ b/crates/ecstore/src/services/tier/tier_config.rs @@ -38,7 +38,7 @@ const WASABI_ALTERNATIVE_ENDPOINTS: &[(&str, &str)] = &[ ("eu-south-1", "https://s3.it-1.wasabisys.com"), ]; -#[derive(Serialize, Deserialize, Default, Debug, Clone)] +#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)] pub enum TierType { #[default] Unsupported, diff --git a/crates/ecstore/src/services/tier/warm_backend.rs b/crates/ecstore/src/services/tier/warm_backend.rs index cc1392b27..8eb6e4058 100644 --- a/crates/ecstore/src/services/tier/warm_backend.rs +++ b/crates/ecstore/src/services/tier/warm_backend.rs @@ -43,6 +43,7 @@ use rustfs_s3_client::{ api_put_object::{AdvancedPutOptions, PutObjectOptions}, transition_api::{ReadCloser, ReaderImpl}, }; +use rustfs_scanner_metrics::metrics::{TierRequestOperation, TierRequestOutcome, global_metrics}; use rustfs_utils::egress::validate_outbound_url; use rustfs_utils::http::headers::{ CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _, @@ -346,6 +347,110 @@ pub(crate) fn optimal_part_size(object_size: i64, min_part_size: i64) -> Result< Ok(part_size) } +/// Counts every remote-tier request exactly once, whatever backend performs it. +/// +/// The counters are the only production update path for the `tier` request +/// metrics, and they must not grow with tier names, endpoints or object keys, +/// so the wrapper deliberately records nothing but the fixed operation and +/// outcome labels. Wrapping here rather than inside each provider keeps a new +/// backend counted by construction, and keeps one call per request: the +/// overridden `remove_exact` delegates to the inner backend, so the inner +/// default's own `remove` cannot count the same request a second time. +struct MeteredWarmBackend { + inner: WarmBackendImpl, +} + +impl MeteredWarmBackend { + fn record(operation: TierRequestOperation, result: Result) -> Result { + let outcome = match &result { + Ok(_) => TierRequestOutcome::Success, + Err(err) => TierRequestOutcome::from_error(err), + }; + global_metrics().record_tier_request(operation, outcome); + result + } +} + +#[async_trait::async_trait] +impl WarmBackend for MeteredWarmBackend { + /// Delegated without a counter: only one backend issues a remote request + /// here, and every other one takes the trait default, so a `validate` + /// counter would mostly record requests that never happened. + async fn validate(&self) -> Result<(), std::io::Error> { + self.inner.validate().await + } + + /// A local check of a value this process already holds; it issues no + /// remote request, so it is delegated without a counter. + fn validate_remote_version_id(&self, remote_version_id: &str) -> Result<(), std::io::Error> { + self.inner.validate_remote_version_id(remote_version_id) + } + + async fn put(&self, object: &str, r: ReaderImpl, length: i64) -> Result { + Self::record(TierRequestOperation::Put, self.inner.put(object, r, length).await) + } + + async fn put_with_meta( + &self, + object: &str, + r: ReaderImpl, + length: i64, + meta: HashMap, + ) -> Result { + Self::record(TierRequestOperation::Put, self.inner.put_with_meta(object, r, length, meta).await) + } + + async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result { + Self::record(TierRequestOperation::Get, self.inner.get(object, rv, opts).await) + } + + async fn remove(&self, object: &str, rv: &str) -> Result<(), std::io::Error> { + Self::record(TierRequestOperation::Remove, self.inner.remove(object, rv).await) + } + + async fn remove_exact(&self, object: &str, rv: &str) -> Result<(), std::io::Error> { + Self::record(TierRequestOperation::Remove, self.inner.remove_exact(object, rv).await) + } + + async fn probe_transition_candidate(&self, object: &str) -> Result { + let result = self.inner.probe_transition_candidate(object).await; + // `Unsupported` is the trait default: the backend issued no request, + // so counting it would inflate the probe counter on every backend that + // does not implement probing. + if matches!(result, Ok(TransitionCandidateProbe::Unsupported)) { + return result; + } + Self::record(TierRequestOperation::Probe, result) + } + + async fn in_use(&self) -> Result { + Self::record(TierRequestOperation::InUse, self.inner.in_use().await) + } +} + +/// The reconciler is a second remote probe path, reached from recovery rather +/// than from the `WarmBackend` handle, so it needs its own counter: a `probe` +/// counter that saw only one of the two paths would read as a complete count +/// while missing every recovery probe. +struct MeteredTransitionCandidateReconciler { + inner: Box, +} + +#[async_trait::async_trait] +impl TransitionCandidateReconciler for MeteredTransitionCandidateReconciler { + async fn probe_transition_candidate_for( + &self, + object: &str, + identity: TransitionCandidateIdentity, + ) -> Result { + let result = self.inner.probe_transition_candidate_for(object, identity).await; + if matches!(result, Ok(TransitionCandidateProbe::Unsupported)) { + return result; + } + MeteredWarmBackend::record(TierRequestOperation::Probe, result) + } +} + pub async fn check_warm_backend(w: Option<&WarmBackendImpl>) -> Result<(), AdminError> { let w = w.ok_or_else(|| ERR_TIER_NOT_FOUND.clone())?; w.validate().await.map_err(|_| ERR_TIER_INVALID_CONFIG.clone())?; @@ -593,6 +698,8 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result return Ok(None), }; - Ok(Some(reconciler)) + Ok(Some(Box::new(MeteredTransitionCandidateReconciler { inner: reconciler }))) } #[cfg(test)] @@ -655,6 +762,156 @@ mod tests { const PROBE_VERSION: &str = "remote-v2"; + struct CountingBackend { + put_result: fn() -> Result, + removes: Arc, + } + + #[async_trait::async_trait] + impl WarmBackend for CountingBackend { + async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result { + (self.put_result)() + } + + async fn put_with_meta( + &self, + object: &str, + r: ReaderImpl, + length: i64, + _meta: HashMap, + ) -> Result { + self.put(object, r, length).await + } + + async fn get(&self, _object: &str, _rv: &str, _opts: WarmBackendGetOpts) -> Result { + Err(std::io::Error::other("unused")) + } + + async fn remove(&self, _object: &str, _rv: &str) -> Result<(), std::io::Error> { + self.removes.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + + async fn in_use(&self) -> Result { + Ok(false) + } + } + + fn tier_cell(operation: TierRequestOperation, outcome: TierRequestOutcome) -> u64 { + global_metrics() + .tier_request_counts() + .into_iter() + .find(|count| count.operation == operation && count.outcome == outcome) + .map(|count| count.count) + .expect("every operation/outcome cell must be reported") + } + + #[tokio::test] + async fn a_metered_put_counts_its_outcome_once() { + let before_success = tier_cell(TierRequestOperation::Put, TierRequestOutcome::Success); + let before_failure = tier_cell(TierRequestOperation::Put, TierRequestOutcome::BackendError); + let backend = MeteredWarmBackend { + inner: Box::new(CountingBackend { + put_result: || Ok("remote-v1".to_string()), + removes: Arc::new(AtomicUsize::new(0)), + }), + }; + + backend + .put("object", ReaderImpl::Body(Bytes::from_static(b"payload")), 7) + .await + .expect("the fake backend accepts the put"); + + assert_eq!( + tier_cell(TierRequestOperation::Put, TierRequestOutcome::Success), + before_success + 1, + "an acknowledged put must be counted once" + ); + assert_eq!( + tier_cell(TierRequestOperation::Put, TierRequestOutcome::BackendError), + before_failure, + "a success must not also increment a failure cell" + ); + } + + #[tokio::test] + async fn a_metered_put_failure_is_classified_by_error_kind() { + let before_timeout = tier_cell(TierRequestOperation::Put, TierRequestOutcome::Timeout); + let backend = MeteredWarmBackend { + inner: Box::new(CountingBackend { + put_result: || Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "endpoint stalled")), + removes: Arc::new(AtomicUsize::new(0)), + }), + }; + + backend + .put("object", ReaderImpl::Body(Bytes::from_static(b"payload")), 7) + .await + .expect_err("the fake backend rejects the put"); + + assert_eq!( + tier_cell(TierRequestOperation::Put, TierRequestOutcome::Timeout), + before_timeout + 1, + "a timed-out request must land in the timeout cell" + ); + } + + fn tier_probe_total() -> u64 { + global_metrics() + .tier_request_counts() + .into_iter() + .filter(|count| count.operation == TierRequestOperation::Probe) + .map(|count| count.count) + .sum() + } + + #[tokio::test] + async fn an_unsupported_probe_is_not_counted_as_a_request() { + let before = tier_probe_total(); + let backend = MeteredWarmBackend { + inner: Box::new(CountingBackend { + put_result: || Ok(String::new()), + removes: Arc::new(AtomicUsize::new(0)), + }), + }; + + let probe = backend + .probe_transition_candidate("object") + .await + .expect("the trait default answers without a remote request"); + + assert_eq!(probe, TransitionCandidateProbe::Unsupported); + assert_eq!( + tier_probe_total(), + before, + "a backend that issues no probe request must not appear in the probe counters" + ); + } + + #[tokio::test] + async fn an_exact_remove_is_counted_once_not_twice() { + let before = tier_cell(TierRequestOperation::Remove, TierRequestOutcome::Success); + let removes = Arc::new(AtomicUsize::new(0)); + let backend = MeteredWarmBackend { + inner: Box::new(CountingBackend { + put_result: || Ok(String::new()), + removes: Arc::clone(&removes), + }), + }; + + backend + .remove_exact("object", "remote-v1") + .await + .expect("the fake backend accepts the remove"); + + assert_eq!(removes.load(Ordering::SeqCst), 1, "the inner backend performs one request"); + assert_eq!( + tier_cell(TierRequestOperation::Remove, TierRequestOutcome::Success), + before + 1, + "the default remove_exact must not count its own inner remove a second time" + ); + } + struct RejectingValidationBackend { validations: Arc, puts: Arc, diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs index e641540de..cb4078669 100644 --- a/crates/obs/src/metrics/collectors/mod.rs +++ b/crates/obs/src/metrics/collectors/mod.rs @@ -40,6 +40,7 @@ pub mod system_memory; pub mod system_network; pub mod system_network_host; pub mod system_process; +pub mod tier; pub(crate) use audit::{AuditTargetRuntimeStats, collect_audit_runtime_metrics}; pub use audit::{AuditTargetStats, collect_audit_metrics}; @@ -94,3 +95,4 @@ pub use system_process::{ ProcessAttributeError, ProcessAttributes, ProcessStats, ProcessStatusType, collect_process_attributes, collect_process_metrics, }; +pub use tier::{TierRequestStats, collect_tier_request_metrics}; diff --git a/crates/obs/src/metrics/collectors/tier.rs b/crates/obs/src/metrics/collectors/tier.rs new file mode 100644 index 000000000..4007c0b28 --- /dev/null +++ b/crates/obs/src/metrics/collectors/tier.rs @@ -0,0 +1,96 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Remote tier request metrics collector. + +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::tier::*; + +/// One operation/outcome cell of the tier request counters. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TierRequestStats { + pub operation: &'static str, + pub outcome: &'static str, + pub count: u64, +} + +/// Split the fixed operation/outcome cells over the success and failure +/// counters. +/// +/// The success counter carries no outcome label: `success` is the only outcome +/// it can report, and repeating it would make the two counters look like they +/// share a label set they do not. +pub fn collect_tier_request_metrics(stats: &[TierRequestStats]) -> Vec { + stats + .iter() + .map(|stat| { + if stat.outcome == "success" { + PrometheusMetric::from_descriptor(&TIER_REQUESTS_SUCCESS_MD, stat.count as f64) + .with_label(OPERATION_LABEL, stat.operation) + } else { + PrometheusMetric::from_descriptor(&TIER_REQUESTS_FAILURE_MD, stat.count as f64) + .with_label(OPERATION_LABEL, stat.operation) + .with_label(OUTCOME_LABEL, stat.outcome) + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn stats() -> Vec { + vec![ + TierRequestStats { + operation: "put", + outcome: "success", + count: 7, + }, + TierRequestStats { + operation: "put", + outcome: "timeout", + count: 2, + }, + ] + } + + #[test] + fn success_and_failure_land_on_their_own_counters() { + let metrics = collect_tier_request_metrics(&stats()); + + let success = metrics + .iter() + .find(|metric| metric.name == TIER_REQUESTS_SUCCESS_MD.get_full_metric_name()) + .expect("a success cell must produce the success counter"); + assert_eq!(success.value, 7.0); + assert!( + success.labels.iter().all(|(name, _)| *name != OUTCOME_LABEL), + "the success counter must not carry an outcome label" + ); + + let failure = metrics + .iter() + .find(|metric| metric.name == TIER_REQUESTS_FAILURE_MD.get_full_metric_name()) + .expect("a non-success cell must produce the failure counter"); + assert_eq!(failure.value, 2.0); + assert!( + failure + .labels + .iter() + .any(|(name, value)| *name == OUTCOME_LABEL && value.as_ref() == "timeout"), + "the failure counter must keep the outcome that produced it" + ); + } +} diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 9ac05e79e..8d527c8df 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -77,6 +77,7 @@ use crate::metrics::collectors::{ collect_request_metrics, collect_resource_metrics, collect_scanner_runtime_metrics, + collect_tier_request_metrics, }; use crate::metrics::config::{ DEFAULT_AUDIT_METRICS_INTERVAL, DEFAULT_BUCKET_METRICS_INTERVAL, DEFAULT_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, @@ -151,7 +152,7 @@ use crate::metrics::stats_collector::{ collect_disk_and_system_drive_runtime_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats, collect_ilm_runtime_metric_stats, collect_internode_network_stats, collect_on_demand_migration_backfill_stats, collect_on_demand_migration_stats, collect_process_metric_bundle_with, collect_replication_stats, - collect_scanner_runtime_metric_stats, collect_system_cpu_and_memory_stats_with, + collect_scanner_runtime_metric_stats, collect_system_cpu_and_memory_stats_with, collect_tier_request_metric_stats, }; use crate::node_identity::{SERVER_LABEL, current_local_node_identity}; use crate::telemetry::retire_metric_series; @@ -2397,6 +2398,8 @@ pub fn init_metrics_runtime(token: CancellationToken) { metrics.extend(collect_ilm_runtime_metrics(&stats)); } + metrics.extend(collect_tier_request_metrics(&collect_tier_request_metric_stats())); + let mut retire_scanner_cycle_bucket_drive_result_keys = Vec::new(); let mut retire_scanner_bucket_drive_result_keys = Vec::new(); let mut retire_scanner_active_bucket_drive_keys = Vec::new(); diff --git a/crates/obs/src/metrics/schema/entry/subsystem.rs b/crates/obs/src/metrics/schema/entry/subsystem.rs index 255fe6892..40e5a9ffd 100644 --- a/crates/obs/src/metrics/schema/entry/subsystem.rs +++ b/crates/obs/src/metrics/schema/entry/subsystem.rs @@ -47,6 +47,7 @@ pub enum MetricSubsystem { // other service related subsystems Ilm, + Tier, Audit, Replication, Notification, @@ -91,6 +92,7 @@ impl MetricSubsystem { // other service related subsystems Self::Ilm => "/ilm", + Self::Tier => "/tier", Self::Audit => "/audit", Self::Replication => "/replication", Self::Notification => "/notification", @@ -140,6 +142,7 @@ impl MetricSubsystem { // Other service-related subsystems "/ilm" => Self::Ilm, + "/tier" => Self::Tier, "/audit" => Self::Audit, "/replication" => Self::Replication, "/notification" => Self::Notification, @@ -202,6 +205,7 @@ pub mod subsystems { pub const CLUSTER_IAM: MetricSubsystem = MetricSubsystem::ClusterIam; pub const CLUSTER_CONFIG: MetricSubsystem = MetricSubsystem::ClusterConfig; pub const ILM: MetricSubsystem = MetricSubsystem::Ilm; + pub const TIER: MetricSubsystem = MetricSubsystem::Tier; pub const AUDIT: MetricSubsystem = MetricSubsystem::Audit; pub const REPLICATION: MetricSubsystem = MetricSubsystem::Replication; pub const NOTIFICATION: MetricSubsystem = MetricSubsystem::Notification; diff --git a/crates/obs/src/metrics/schema/mod.rs b/crates/obs/src/metrics/schema/mod.rs index 878fad62a..2d90828ac 100644 --- a/crates/obs/src/metrics/schema/mod.rs +++ b/crates/obs/src/metrics/schema/mod.rs @@ -40,6 +40,7 @@ pub mod system_memory; pub mod system_network; pub mod system_network_host; pub mod system_process; +pub mod tier; pub use entry::descriptor::MetricDescriptor; pub use entry::metric_name::MetricName; diff --git a/crates/obs/src/metrics/schema/tier.rs b/crates/obs/src/metrics/schema/tier.rs new file mode 100644 index 000000000..f6992dd3b --- /dev/null +++ b/crates/obs/src/metrics/schema/tier.rs @@ -0,0 +1,43 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Remote tier request metric descriptors. +//! +//! The label set is fixed by the operation and outcome enums the recording +//! site uses, so the series count is bounded by construction and cannot grow +//! with tier names, endpoints or object keys. + +use crate::{MetricDescriptor, MetricName, new_counter_md, subsystems}; +use std::sync::LazyLock; + +pub const OPERATION_LABEL: &str = "operation"; +pub const OUTCOME_LABEL: &str = "outcome"; + +pub static TIER_REQUESTS_SUCCESS_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::TierRequestsSuccess, + "Remote tier requests the backend acknowledged, by operation", + &[OPERATION_LABEL], + subsystems::TIER, + ) +}); + +pub static TIER_REQUESTS_FAILURE_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::TierRequestsFailure, + "Remote tier requests that did not complete, by operation and outcome", + &[OPERATION_LABEL, OUTCOME_LABEL], + subsystems::TIER, + ) +}); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 36b4641d7..1a2021b26 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -27,7 +27,7 @@ use crate::metrics::collectors::{ DriveDetailedStats, DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmBackpressureStats, IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats, OdmBackfillRuntimeStats, OnDemandMigrationBucketStats, ProcessStats, ProcessStatusType, ReplicationMetricsSnapshot, - ResourceStats, ScannerRuntimeStats, ScannerStats, + ResourceStats, ScannerRuntimeStats, ScannerStats, TierRequestStats, }; use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot}; use crate::metrics::{ @@ -1396,6 +1396,23 @@ fn ilm_backpressure_stats(metrics: &ScannerMetricsReport) -> Vec Vec { + global_metrics() + .tier_request_counts() + .into_iter() + .map(|count| TierRequestStats { + operation: count.operation.as_label(), + outcome: count.outcome.as_label(), + count: count.count, + }) + .collect() +} + /// Collect ILM metrics from the current lifecycle runtime state. pub async fn collect_ilm_metric_stats() -> Option { collect_ilm_runtime_metric_stats().await.map(|stats| stats.stats) diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 7a5058db0..037e9e597 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -1482,6 +1482,25 @@ pub struct LoadTransitionTierConfigResponse { #[prost(enumeration = "ControlPlaneErrorCode", optional, tag = "3")] pub error_code: ::core::option::Option, } +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct TierDailyStatsRequest {} +/// One node's own rolling-day transition counters, per remote tier. +/// +/// `tier_daily_stats` is a msgpack map of tier name to the responder's 24-bin +/// ring plus the clock that ring was last aged to. A node counts only the +/// transitions it completed itself, so a caller sums the rings of every node to +/// obtain a cluster total. A peer that does not implement this RPC answers +/// UNIMPLEMENTED, which the caller reports as a non-reporting node rather than +/// as zero activity. +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct TierDailyStatsResponse { + #[prost(bool, tag = "1")] + pub success: bool, + #[prost(bytes = "bytes", tag = "2")] + pub tier_daily_stats: ::prost::bytes::Bytes, + #[prost(string, optional, tag = "3")] + pub error_info: ::core::option::Option<::prost::alloc::string::String>, +} #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct TierMutationPrepareRequest { #[prost(uint32, tag = "1")] @@ -3075,6 +3094,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "LoadTransitionTierConfig")); self.inner.unary(req, path, codec).await } + pub async fn tier_daily_stats( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/TierDailyStats"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "TierDailyStats")); + self.inner.unary(req, path, codec).await + } pub async fn get_live_events( &mut self, request: impl tonic::IntoRequest, @@ -3475,6 +3509,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn tier_daily_stats( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn get_live_events( &self, request: tonic::Request, @@ -6080,6 +6118,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/TierDailyStats" => { + #[allow(non_camel_case_types)] + struct TierDailyStatsSvc(pub Arc); + impl tonic::server::UnaryService for TierDailyStatsSvc { + type Response = super::TierDailyStatsResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::tier_daily_stats(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = TierDailyStatsSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/GetLiveEvents" => { #[allow(non_camel_case_types)] struct GetLiveEventsSvc(pub Arc); diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index d1e1b0ebc..477b7a120 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -1043,6 +1043,22 @@ message LoadTransitionTierConfigResponse { optional ControlPlaneErrorCode error_code = 3; } +message TierDailyStatsRequest {} + +// One node's own rolling-day transition counters, per remote tier. +// +// `tier_daily_stats` is a msgpack map of tier name to the responder's 24-bin +// ring plus the clock that ring was last aged to. A node counts only the +// transitions it completed itself, so a caller sums the rings of every node to +// obtain a cluster total. A peer that does not implement this RPC answers +// UNIMPLEMENTED, which the caller reports as a non-reporting node rather than +// as zero activity. +message TierDailyStatsResponse { + bool success = 1; + bytes tier_daily_stats = 2; + optional string error_info = 3; +} + message TierMutationPrepareRequest { uint32 version = 1; string mutation_id = 2; @@ -1203,6 +1219,7 @@ service NodeService { rpc CancelDecommission(CancelDecommissionRequest) returns (CancelDecommissionResponse) {}; // auth-policy: body-bound rpc ClearDecommission(ClearDecommissionRequest) returns (ClearDecommissionResponse) {}; // auth-policy: body-bound rpc LoadTransitionTierConfig(LoadTransitionTierConfigRequest) returns (LoadTransitionTierConfigResponse) {}; // auth-policy: body-bound + rpc TierDailyStats(TierDailyStatsRequest) returns (TierDailyStatsResponse) {}; // auth-policy: read-only rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {}; // auth-policy: read-only } diff --git a/crates/scanner-metrics/src/metrics.rs b/crates/scanner-metrics/src/metrics.rs index b0b9d32fb..d91d08497 100644 --- a/crates/scanner-metrics/src/metrics.rs +++ b/crates/scanner-metrics/src/metrics.rs @@ -798,6 +798,88 @@ struct ScannerActiveBucketDriveValue { // Metrics // --------------------------------------------------------------------------- +/// A remote-tier request a warm backend issues on the cluster's behalf. +/// +/// The set is closed on purpose: these values become metric labels, so a new +/// operation is a deliberate schema change rather than something a call site +/// can invent. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum TierRequestOperation { + Put, + Get, + Remove, + Probe, + InUse, +} + +impl TierRequestOperation { + pub const ALL: [Self; 5] = [Self::Put, Self::Get, Self::Remove, Self::Probe, Self::InUse]; + + pub const fn as_label(self) -> &'static str { + match self { + Self::Put => "put", + Self::Get => "get", + Self::Remove => "remove", + Self::Probe => "probe", + Self::InUse => "in_use", + } + } + + const fn index(self) -> usize { + self as usize + } +} + +/// How a remote-tier request ended. +/// +/// This classifies the request only. A transition whose remote PUT succeeded +/// and whose local commit then failed is a `Success` here: the remote service +/// did perform the request, and reporting it as a tier failure would hide a +/// leaked remote object behind an apparent backend outage. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum TierRequestOutcome { + Success, + BackendError, + Timeout, + Cancelled, +} + +impl TierRequestOutcome { + pub const ALL: [Self; 4] = [Self::Success, Self::BackendError, Self::Timeout, Self::Cancelled]; + + pub const fn as_label(self) -> &'static str { + match self { + Self::Success => "success", + Self::BackendError => "backend_error", + Self::Timeout => "timeout", + Self::Cancelled => "cancelled", + } + } + + const fn index(self) -> usize { + self as usize + } + + /// Classify a warm backend error without letting its message into a label. + pub fn from_error(err: &std::io::Error) -> Self { + match err.kind() { + std::io::ErrorKind::TimedOut => Self::Timeout, + std::io::ErrorKind::Interrupted => Self::Cancelled, + _ => Self::BackendError, + } + } +} + +/// One fixed operation/outcome cell of the tier request counters. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct TierRequestCount { + pub operation: TierRequestOperation, + pub outcome: TierRequestOutcome, + pub count: u64, +} + +const TIER_REQUEST_COUNTER_SLOTS: usize = TierRequestOperation::ALL.len() * TierRequestOutcome::ALL.len(); + pub struct Metrics { operations: Vec, latency: Vec, @@ -892,6 +974,10 @@ pub struct Metrics { scanner_transition_queued_total: AtomicU64, scanner_transition_missed_total: AtomicU64, scanner_transition_completed: AtomicU64, + /// Remote tier requests by operation and outcome. The width is fixed by + /// the two label enums, so the series count cannot grow with tier names, + /// endpoints or object keys. + tier_requests: [AtomicU64; TIER_REQUEST_COUNTER_SLOTS], scanner_transition_failed: AtomicU64, scanner_throttle_idle_mode_enabled: AtomicBool, scanner_throttle_sleep_factor_micros: AtomicU64, @@ -1943,6 +2029,7 @@ impl Metrics { scanner_transition_queued_total: AtomicU64::new(0), scanner_transition_missed_total: AtomicU64::new(0), scanner_transition_completed: AtomicU64::new(0), + tier_requests: std::array::from_fn(|_| AtomicU64::new(0)), scanner_transition_failed: AtomicU64::new(0), scanner_throttle_idle_mode_enabled: AtomicBool::new(false), scanner_throttle_sleep_factor_micros: AtomicU64::new(0), @@ -2214,6 +2301,33 @@ impl Metrics { .store(state.compensation_running, Ordering::Relaxed); } + /// Count one completed remote tier request. + /// + /// Called once per request at the single point where every warm backend + /// call returns, so a retried request is counted once per attempt and a + /// successful attempt is never also counted as a failure. + pub fn record_tier_request(&self, operation: TierRequestOperation, outcome: TierRequestOutcome) { + let slot = operation.index() * TierRequestOutcome::ALL.len() + outcome.index(); + self.tier_requests[slot].fetch_add(1, Ordering::Relaxed); + } + + /// Every operation/outcome cell, including the zero ones, so a scrape has + /// a stable series set from the first request onward. + pub fn tier_request_counts(&self) -> Vec { + let mut counts = Vec::with_capacity(TIER_REQUEST_COUNTER_SLOTS); + for operation in TierRequestOperation::ALL { + for outcome in TierRequestOutcome::ALL { + let slot = operation.index() * TierRequestOutcome::ALL.len() + outcome.index(); + counts.push(TierRequestCount { + operation, + outcome, + count: self.tier_requests[slot].load(Ordering::Relaxed), + }); + } + } + counts + } + pub fn record_scanner_transition_completed(&self, count: u64) { self.scanner_transition_completed.fetch_add(count, Ordering::Relaxed); } @@ -3549,6 +3663,76 @@ impl Drop for CloseDiskGuard { mod tests { use super::*; + fn tier_cell(metrics: &Metrics, operation: TierRequestOperation, outcome: TierRequestOutcome) -> u64 { + metrics + .tier_request_counts() + .into_iter() + .find(|count| count.operation == operation && count.outcome == outcome) + .map(|count| count.count) + .expect("every operation/outcome cell must be reported") + } + + #[test] + fn a_tier_request_lands_in_exactly_one_operation_outcome_cell() { + let metrics = Metrics::default(); + metrics.record_tier_request(TierRequestOperation::Put, TierRequestOutcome::Success); + + assert_eq!(tier_cell(&metrics, TierRequestOperation::Put, TierRequestOutcome::Success), 1); + assert_eq!( + metrics + .tier_request_counts() + .into_iter() + .map(|count| count.count) + .sum::(), + 1, + "one request must not increment a second cell" + ); + } + + #[test] + fn every_operation_outcome_pair_has_its_own_cell() { + let metrics = Metrics::default(); + for operation in TierRequestOperation::ALL { + for outcome in TierRequestOutcome::ALL { + metrics.record_tier_request(operation, outcome); + } + } + + let counts = metrics.tier_request_counts(); + assert_eq!( + counts.len(), + TierRequestOperation::ALL.len() * TierRequestOutcome::ALL.len(), + "the series set is fixed by the two label enums" + ); + assert!( + counts.iter().all(|count| count.count == 1), + "two different pairs must not share a counter slot" + ); + } + + #[test] + fn a_backend_error_is_classified_without_reading_its_message() { + assert_eq!( + TierRequestOutcome::from_error(&std::io::Error::new(std::io::ErrorKind::TimedOut, "s3.example.com:9000 timed out")), + TierRequestOutcome::Timeout + ); + assert_eq!( + TierRequestOutcome::from_error(&std::io::Error::new(std::io::ErrorKind::Interrupted, "cancelled")), + TierRequestOutcome::Cancelled + ); + assert_eq!( + TierRequestOutcome::from_error(&std::io::Error::other("AccessDenied")), + TierRequestOutcome::BackendError + ); + } + + #[test] + fn tier_request_labels_are_stable_identifiers() { + assert_eq!(TierRequestOperation::Put.as_label(), "put"); + assert_eq!(TierRequestOperation::InUse.as_label(), "in_use"); + assert_eq!(TierRequestOutcome::BackendError.as_label(), "backend_error"); + } + #[test] fn scanner_metrics_report_timestamps_serialize_as_rfc3339_utc() { let report = ScannerMetricsReport { diff --git a/docs/architecture/README.md b/docs/architecture/README.md index b3a819622..6cbbf569b 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -48,6 +48,7 @@ Required headings and strings in these files are asserted by `scripts/check_arch | [config-model-boundary-adr.md](config-model-boundary-adr.md) | touching the server-config model (`Config`, `KV`, `KVS`) or its persistence, or asking which crate owns which part of server configuration | | [admin-route-action-snapshot.md](admin-route-action-snapshot.md) | adding, moving, or re-authorizing an admin route and needing to know where the route → handler → `AdminAction` contract is enforced | | [kms-bulk-rekey-contract.md](kms-bulk-rekey-contract.md) | changing the bulk envelope re-wrap sweep, its admin endpoints, the re-wrap primitive, or which objects a rekey may touch | +| [tier-stats-contract.md](tier-stats-contract.md) | changing what `GET /rustfs/admin/v3/tier-stats` returns, adding a tier accounting source, or wiring a metric to a remote tier request | ## Support and compatibility matrices (release-facing, keep current) diff --git a/docs/architecture/tier-stats-contract.md b/docs/architecture/tier-stats-contract.md new file mode 100644 index 000000000..5b8bc90ab --- /dev/null +++ b/docs/architecture/tier-stats-contract.md @@ -0,0 +1,43 @@ +# Tier Stats Contract + +**Use this when:** changing what `GET /rustfs/admin/v3/tier-stats` returns, adding a tier accounting source, or wiring a metric to a remote tier request. +**Source of truth:** `rustfs/src/admin/handlers/tier.rs` (`GetTierInfo`, `tier_stats_body`), `crates/ecstore/src/services/notification_sys.rs` (`ClusterTierDailyStats`), `crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs` (`LastDayTierStats`), `crates/ecstore/src/services/tier/warm_backend.rs` (`MeteredWarmBackend`), `crates/obs/src/metrics/schema/tier.rs`. + +## Two quantities, never one + +A tier carries two unrelated numbers, and reporting either as the other is the defect this contract exists to prevent. + +- **Stored inventory** — how many bytes, objects and versions currently live in the tier. It is produced by the scanner, persisted in the data usage snapshot (`DataUsageInfo::tier_stats`), and is already cluster-wide. It is a level, not a rate. +- **Rolling activity** — how many transitions the cluster completed into the tier during the last 24 hours. Each node keeps its own 24-bin ring in memory (`TransitionState::add_lastday_stats`) and counts only the transitions it completed itself. It is a rate window, not a level, and it is lost on restart. + +Neither substitutes for the other. An empty rolling window does not mean an empty tier, and a populated tier does not imply recent activity. + +## Response contract + +`contractVersion` names the body shape. + +Version 1 was a bare map of tier name to the answering process's rolling counters, with nothing in the body distinguishing a node from the cluster or a rate from a level. It remains reachable at `?format=legacy` for callers pinned to it; it is not extended. + +Version 2 is the default body: + +- `inventory.status` is `accounted`, `not-accounted`, or `unavailable`. Per-tier `inventory` values are present only under `accounted`. An absent per-tier accounting means "not accounted", never "zero" — the scanner classifies objects by tier only once a tier exists, so a zero would be indistinguishable from an unscanned cluster. +- `activity.status` is `complete` or `partial`, with `nodesReporting`, `nodesExpected`, and the `unavailableNodes` that could not be asked, timed out, or answered with a ring this build refuses to merge. A peer that predates the `TierDailyStats` RPC answers `UNIMPLEMENTED` and is reported as unavailable rather than as zero activity. +- Each tier entry carries `type` only when the name is a configured remote tier. A name that carries stats without a configuration is a local storage class the scanner accounts for, or a tier removed since the snapshot. + +Field names inside the counter objects are `totalSize`, `numVersions`, `numObjects` — the camelCase spelling admin clients expect from the `madmin` tier stats shape, rather than the Rust field names version 1 leaked. This aligns the spelling; it is not a claim that the envelope is byte-compatible with a specific `madmin` release, and a client-side check belongs with whatever client a release wants to support. + +## Why merging rings is not summing totals + +Nodes are asked concurrently under a per-peer deadline, and each answer is merged with `LastDayTierStats::merge` rather than added. Merging ages the older ring forward to the newer ring's clock first, so a node that stopped transitioning yesterday contributes only the hours still inside the rolling day. Adding raw totals would keep expired hours alive for as long as the node stays up. + +Double counting is prevented at the source, not at the aggregator: `add_lastday_stats` runs once per committed transition on the node that committed it, so a transition retried on another node is counted once, by whichever node finally committed it. + +## Tier request metrics + +`rustfs_tier_requests_success` and `rustfs_tier_requests_failure` are updated at the two seams every remote tier request passes through: `MeteredWarmBackend`, which wraps every backend `new_warm_backend` builds, and `MeteredTransitionCandidateReconciler`, which wraps the separate recovery probe handle `new_transition_candidate_reconciler` builds. A new provider is therefore counted by construction. Two seams are deliberately delegated without a counter: `validate`, whose trait default performs no remote request on every backend but one, and a `probe_transition_candidate` that answers `Unsupported`, which is the same default. Counting either would report requests that were never issued. + +The label set is closed by two enums in `crates/scanner-metrics/src/metrics.rs`: `TierRequestOperation` (`put`, `get`, `remove`, `probe`, `in_use`) and `TierRequestOutcome` (`success`, `backend_error`, `timeout`, `cancelled`). Tier names, endpoints and object keys must never become labels — the endpoint carries credentials in its userinfo form and the key is unbounded. + +`timeout` and `cancelled` are recognised from `std::io::ErrorKind`, never from an error message, so a message that mentions an endpoint cannot reach a label. Until the transition client grows bounded deadlines (rustfs/backlog#2204), few failures actually carry `TimedOut`, so most land in `backend_error`; the classification is the seam that work extends, not a claim that timeouts are already distinguishable. + +The outcome classifies the request only. A transition whose remote PUT succeeded and whose local commit then failed is a `success` here: the remote service did perform the request, and recording it as a tier failure would hide a leaked remote object behind an apparent backend outage. Local commit failure is observable through the ILM task-event metrics instead. diff --git a/rustfs/src/admin/handlers/tier.rs b/rustfs/src/admin/handlers/tier.rs index f42eccc7b..535554a46 100644 --- a/rustfs/src/admin/handlers/tier.rs +++ b/rustfs/src/admin/handlers/tier.rs @@ -16,10 +16,10 @@ use crate::admin::runtime_sources::object_store_from_extensions; use crate::admin::storage_api::runtime_sources::TierConfigMgr; use crate::admin::storage_api::tier::{ - AdminError, DailyAllTierStats, ERR_TIER_ALREADY_EXISTS, ERR_TIER_BACKEND_IN_USE, ERR_TIER_BACKEND_NOT_EMPTY, - ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CONFIG, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_MISSING_CREDENTIALS, - ERR_TIER_NAME_NOT_UPPERCASE, ERR_TIER_NOT_FOUND, ERR_TIER_RESERVED_NAME, TierConfig, TierConfigUpdateError, TierCreds, - TierType, + AdminError, ClusterTierDailyStats, DailyAllTierStats, ECStore, ERR_TIER_ALREADY_EXISTS, ERR_TIER_BACKEND_IN_USE, + ERR_TIER_BACKEND_NOT_EMPTY, ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CONFIG, ERR_TIER_INVALID_CREDENTIALS, + ERR_TIER_MISSING_CREDENTIALS, ERR_TIER_NAME_NOT_UPPERCASE, ERR_TIER_NOT_FOUND, ERR_TIER_RESERVED_NAME, TierConfig, + TierConfigUpdateError, TierCreds, TierType, }; use crate::{ admin::runtime_sources::{current_daily_tier_stats, current_notification_system, current_tier_config_handle}, @@ -42,7 +42,8 @@ use s3s::{ s3_error, }; use serde_urlencoded::from_bytes; -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeSet, HashMap, HashSet}; +use std::sync::Arc; use time::OffsetDateTime; use tracing::{debug, warn}; @@ -174,7 +175,7 @@ fn resolve_tier_name(uri: &Uri, params: &Params<'_, '_>) -> S3Result { AddTierQuery::default() }; - Ok(require_tier_name(&query)?.to_string()) + Ok(require_tier_name(query.tier.as_deref())?.to_string()) } pub fn register_tier_route(r: &mut S3Router) -> std::io::Result<()> { @@ -654,6 +655,107 @@ impl Operation for VerifyTier { } } +/// Version of the `GET /v3/tier-stats` response body. +/// +/// Version 1 was a bare `{"": {"total_size": ...}}` map holding the +/// answering process's rolling 24-hour transition counters, with nothing in +/// the body saying that it was neither a cluster total nor an inventory. +/// Version 2 separates the two quantities and carries the reporting coverage +/// behind each. Callers pinned to version 1 request it with `?format=legacy`. +const TIER_STATS_CONTRACT_VERSION: u32 = 2; + +const TIER_STATS_FORMAT_LEGACY: &str = "legacy"; + +#[derive(Debug, Clone, serde::Deserialize, Default)] +pub struct TierStatsQuery { + pub tier: Option, + /// `legacy` returns the version 1 body. Any other value is rejected rather + /// than silently answered in the current format. + pub format: Option, +} + +/// Counters shared by the inventory and rolling-activity views. +/// +/// The field names are the camelCase spelling admin clients expect from the +/// `madmin` tier stats shape (`totalSize`, `numVersions`, `numObjects`), not +/// the Rust field names the version 1 body leaked. +#[derive(Debug, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +struct TierStatsBody { + total_size: u64, + num_versions: u64, + num_objects: u64, +} + +impl From for TierStatsBody { + fn from(stats: TierStats) -> Self { + Self { + total_size: stats.total_size, + num_versions: stats.num_versions, + num_objects: stats.num_objects, + } + } +} + +/// Where the stored-inventory numbers came from. +/// +/// `accounted` is the only status whose per-tier `inventory` values are +/// present; the other two say why they are absent instead of reporting a +/// plausible zero (`crates/data-usage/src/data_usage.rs` documents that an +/// absent per-tier accounting means "not accounted", never "zero"). +#[derive(Debug, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +struct TierInventoryStatusBody { + status: &'static str, + /// Scanner snapshot time the inventory was taken from, RFC 3339. + #[serde(skip_serializing_if = "Option::is_none")] + updated_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + detail: Option, +} + +/// How much of the cluster the rolling activity counters cover. +#[derive(Debug, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +struct TierActivityStatusBody { + status: &'static str, + nodes_reporting: usize, + nodes_expected: usize, + unavailable_nodes: Vec, +} + +#[derive(Debug, PartialEq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +struct TierInfoBody { + name: String, + /// The configured remote tier type. Absent when the name carries stats but + /// is not a configured remote tier: a local storage class the scanner + /// accounts for, or a tier removed since the snapshot was taken. + #[serde(rename = "type", skip_serializing_if = "Option::is_none")] + tier_type: Option, + /// Cluster-wide stored inventory for this tier. Absent whenever + /// `inventory.status` is not `accounted`. + #[serde(skip_serializing_if = "Option::is_none")] + inventory: Option, + /// Transitions this cluster completed into the tier during the rolling + /// 24 hours, summed over the reporting nodes. Each node counts only its + /// own completions, so a transition retried across nodes is counted once, + /// by the node that committed it. + transitions_last24h: TierStatsBody, + /// Newest hour boundary the merged rolling ring has been aged to, RFC 3339. + #[serde(skip_serializing_if = "Option::is_none")] + transitions_updated_at: Option, +} + +#[derive(Debug, PartialEq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +struct TierStatsBodyV2 { + contract_version: u32, + inventory: TierInventoryStatusBody, + activity: TierActivityStatusBody, + tiers: Vec, +} + pub struct GetTierInfo {} #[async_trait::async_trait] impl Operation for GetTierInfo { @@ -666,23 +768,27 @@ impl Operation for GetTierInfo { let query = { if let Some(query) = req.uri.query() { - let input: AddTierQuery = + let input: TierStatsQuery = from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "failed to decode query"))?; input } else { - AddTierQuery::default() + TierStatsQuery::default() } }; let tier_name = if query.tier.is_some() { - Some(require_tier_name(&query)?) + Some(require_tier_name(query.tier.as_deref())?) } else { None }; - let info = filter_tier_stats(current_daily_tier_stats(), tier_name); - let data = serde_json::to_vec(&info) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal tier err {e}")))?; + let data = if tier_stats_wants_legacy_format(&query)? { + serde_json::to_vec(&filter_tier_stats(current_daily_tier_stats(), tier_name)) + } else { + let store = object_store_from_extensions(&req.extensions); + serde_json::to_vec(&tier_stats_body(store, tier_name).await) + } + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal tier err {e}")))?; let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value")); @@ -691,12 +797,171 @@ impl Operation for GetTierInfo { } } -fn optional_tier_name(query: &AddTierQuery) -> Option<&str> { - query.tier.as_deref().map(str::trim).filter(|tier| !tier.is_empty()) +/// Assemble the version 2 body from the two independent sources. +/// +/// The stored inventory comes from the persisted scanner snapshot, which is +/// already cluster-wide; the rolling activity is per node and is summed over +/// the members that answer. Neither source can stand in for the other, so a +/// failure in one leaves the other's fields populated and says so in its own +/// status. +async fn tier_stats_body(store: Option>, tier_name: Option<&str>) -> TierStatsBodyV2 { + let (inventory_status, inventory) = tier_inventory(store).await; + let activity = cluster_tier_daily_stats().await; + let tier_types = { + let tier_config_mgr_handle = current_tier_config_handle(); + let tier_config_mgr = tier_config_mgr_handle.read().await; + tier_config_mgr + .list_tiers() + .into_iter() + .map(|tier| (tier.name, tier.tier_type)) + .collect() + }; + + assemble_tier_stats_body(tier_types, inventory_status, inventory, activity, tier_name) } -fn require_tier_name(query: &AddTierQuery) -> S3Result<&str> { - optional_tier_name(query).ok_or_else(|| s3_error!(InvalidArgument, "tier is required")) +/// Join the configured tiers, the stored inventory and the rolling activity +/// into one body. +/// +/// A tier is listed when any of the three sources knows it: a configured tier +/// with no data yet must still appear, and a tier that carries data but no +/// configuration must not be dropped just because it cannot be typed. +fn assemble_tier_stats_body( + tier_types: HashMap, + inventory_status: TierInventoryStatusBody, + inventory: HashMap, + activity: ClusterTierDailyStats, + tier_name: Option<&str>, +) -> TierStatsBodyV2 { + let mut names: BTreeSet<&str> = tier_types.keys().map(String::as_str).collect(); + names.extend(inventory.keys().map(String::as_str)); + names.extend(activity.stats.keys().map(String::as_str)); + + let tiers = names + .into_iter() + .filter(|name| tier_name.is_none_or(|requested| name.eq_ignore_ascii_case(requested))) + .map(|name| { + let daily = activity.stats.get(name); + TierInfoBody { + name: name.to_string(), + tier_type: tier_types.get(name).cloned(), + inventory: inventory.get(name).copied().map(TierStatsBody::from), + transitions_last24h: daily.map(|stats| stats.total()).unwrap_or_default().into(), + transitions_updated_at: daily.and_then(|stats| format_rfc3339(stats.updated_at())), + } + }) + .collect(); + + TierStatsBodyV2 { + contract_version: TIER_STATS_CONTRACT_VERSION, + inventory: inventory_status, + activity: TierActivityStatusBody { + status: if activity.is_complete() { "complete" } else { "partial" }, + nodes_reporting: activity.nodes_reporting, + nodes_expected: activity.nodes_expected, + unavailable_nodes: activity.unavailable_nodes, + }, + tiers, + } +} + +/// Resolve the requested body version. +/// +/// An unrecognized value is rejected rather than answered in the current +/// format: a caller that asked for a shape it can parse must not receive a +/// different one with a 200. +fn tier_stats_wants_legacy_format(query: &TierStatsQuery) -> S3Result { + match query.format.as_deref().map(str::trim).filter(|format| !format.is_empty()) { + None => Ok(false), + Some(TIER_STATS_FORMAT_LEGACY) => Ok(true), + Some(_) => Err(invalid_tier_query("unsupported tier-stats format")), + } +} + +/// The rolling ring of every cluster member that answers. +/// +/// Without a notification system this process is the whole cluster it can +/// speak for, so its own ring is reported as a complete single-member result +/// rather than as a cluster total it cannot prove. +async fn cluster_tier_daily_stats() -> ClusterTierDailyStats { + let local = current_daily_tier_stats(); + match current_notification_system() { + Some(notification_sys) => notification_sys.tier_daily_stats(local).await, + None => ClusterTierDailyStats { + stats: local, + nodes_reporting: 1, + nodes_expected: 1, + unavailable_nodes: Vec::new(), + }, + } +} + +async fn tier_inventory(store: Option>) -> (TierInventoryStatusBody, HashMap) { + let Some(store) = store else { + return ( + TierInventoryStatusBody { + status: "unavailable", + updated_at: None, + detail: Some("object store is not initialized".to_string()), + }, + HashMap::new(), + ); + }; + + match crate::admin::storage_api::data_usage::load_admin_data_usage_from_backend_cached(store).await { + Err(err) => ( + TierInventoryStatusBody { + status: "unavailable", + updated_at: None, + detail: Some(format!("usage snapshot could not be read: {err}")), + }, + HashMap::new(), + ), + Ok(usage) => { + let updated_at = usage.last_update.and_then(|updated| { + OffsetDateTime::from(updated) + .format(&time::format_description::well_known::Rfc3339) + .ok() + }); + match usage.tier_stats { + Some(tier_stats) => ( + TierInventoryStatusBody { + status: "accounted", + updated_at, + detail: None, + }, + tier_stats.tiers, + ), + None => ( + TierInventoryStatusBody { + status: "not-accounted", + updated_at, + detail: Some("the persisted usage snapshot carries no per-tier accounting".to_string()), + }, + HashMap::new(), + ), + } + } + } +} + +fn format_rfc3339(at: OffsetDateTime) -> Option { + at.format(&time::format_description::well_known::Rfc3339).ok() +} + +/// One constructor for every tier query rejection, so the mutation handlers +/// and the stats handler cannot drift into different faults for the same +/// class of bad request. +fn invalid_tier_query(message: &str) -> S3Error { + s3_error!(InvalidArgument, "{message}") +} + +fn optional_tier_name(tier: Option<&str>) -> Option<&str> { + tier.map(str::trim).filter(|tier| !tier.is_empty()) +} + +fn require_tier_name(tier: Option<&str>) -> S3Result<&str> { + optional_tier_name(tier).ok_or_else(|| invalid_tier_query("tier is required")) } fn filter_tier_stats(daily_stats: DailyAllTierStats, tier_name: Option<&str>) -> HashMap { @@ -951,7 +1216,7 @@ mod tests { #[test] fn require_tier_name_rejects_missing_value() { - let err = require_tier_name(&AddTierQuery::default()).expect_err("missing tier should return an error"); + let err = require_tier_name(None).expect_err("missing tier should return an error"); assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); assert_eq!(err.message(), Some("tier is required")); @@ -959,11 +1224,7 @@ mod tests { #[test] fn require_tier_name_rejects_empty_value() { - let err = require_tier_name(&AddTierQuery { - tier: Some(" ".to_string()), - ..Default::default() - }) - .expect_err("empty tier should return an error"); + let err = require_tier_name(Some(" ")).expect_err("empty tier should return an error"); assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); assert_eq!(err.message(), Some("tier is required")); @@ -1127,6 +1388,257 @@ mod tests { assert_eq!(query.force, "true"); } + fn accounted_inventory() -> TierInventoryStatusBody { + TierInventoryStatusBody { + status: "accounted", + updated_at: None, + detail: None, + } + } + + fn complete_activity(stats: DailyAllTierStats) -> ClusterTierDailyStats { + ClusterTierDailyStats { + stats, + nodes_reporting: 2, + nodes_expected: 2, + unavailable_nodes: Vec::new(), + } + } + + #[test] + fn tier_stats_body_separates_inventory_from_rolling_activity() { + let inventory = HashMap::from([( + "WARM".to_string(), + TierStats { + total_size: 4096, + num_versions: 8, + num_objects: 8, + }, + )]); + + let body = assemble_tier_stats_body( + HashMap::from([("WARM".to_string(), TierType::S3)]), + accounted_inventory(), + inventory, + complete_activity(sample_daily_stats()), + Some("WARM"), + ); + + let warm = body.tiers.first().expect("the requested tier must be present"); + assert_eq!( + warm.inventory, + Some(TierStatsBody { + total_size: 4096, + num_versions: 8, + num_objects: 8, + }), + "inventory must come from the stored accounting, not the rolling window" + ); + assert_eq!( + warm.transitions_last24h, + TierStatsBody { + total_size: 15, + num_versions: 3, + num_objects: 1, + }, + "rolling activity must come from the transition ring, not the inventory" + ); + assert_eq!(warm.tier_type, Some(TierType::S3)); + } + + #[test] + fn a_configured_tier_without_data_is_still_listed() { + let body = assemble_tier_stats_body( + HashMap::from([("COLD".to_string(), TierType::S3)]), + accounted_inventory(), + HashMap::new(), + complete_activity(DailyAllTierStats::new()), + None, + ); + + let cold = body + .tiers + .first() + .expect("a configured tier must be listed before it has data"); + assert_eq!(cold.name, "COLD"); + assert_eq!(cold.inventory, None, "an unaccounted tier must not report a zero inventory"); + assert_eq!( + cold.transitions_last24h, + TierStatsBody { + total_size: 0, + num_versions: 0, + num_objects: 0, + } + ); + assert_eq!(cold.transitions_updated_at, None); + } + + #[test] + fn a_tier_with_activity_but_no_configuration_keeps_its_counters_untyped() { + let body = assemble_tier_stats_body( + HashMap::new(), + accounted_inventory(), + HashMap::new(), + complete_activity(sample_daily_stats()), + Some("ARCHIVE"), + ); + + let archive = body + .tiers + .first() + .expect("a tier with data must be listed without a configuration"); + assert_eq!(archive.tier_type, None, "an unconfigured name must not be given a type"); + assert_eq!( + archive.transitions_last24h, + TierStatsBody { + total_size: 9, + num_versions: 1, + num_objects: 1, + } + ); + } + + #[test] + fn a_restarted_cluster_keeps_its_inventory_and_loses_only_the_rolling_window() { + // The rolling window lives in each process's memory and the inventory + // lives in the persisted usage snapshot, so a restart empties one and + // leaves the other intact. Reporting the emptied window as the tier's + // contents is the confusion the two fields exist to prevent. + let inventory = HashMap::from([( + "WARM".to_string(), + TierStats { + total_size: 4096, + num_versions: 8, + num_objects: 8, + }, + )]); + + let body = assemble_tier_stats_body( + HashMap::from([("WARM".to_string(), TierType::S3)]), + accounted_inventory(), + inventory, + complete_activity(DailyAllTierStats::new()), + Some("WARM"), + ); + + let warm = body.tiers.first().expect("the tier must survive a restart"); + assert_eq!( + warm.inventory, + Some(TierStatsBody { + total_size: 4096, + num_versions: 8, + num_objects: 8, + }), + "a restart must not empty the stored inventory" + ); + assert_eq!( + warm.transitions_last24h, + TierStatsBody { + total_size: 0, + num_versions: 0, + num_objects: 0, + }, + "a restart empties the rolling window, and that must stay visible as zero activity" + ); + } + + #[test] + fn an_unreachable_node_makes_the_activity_partial() { + let body = assemble_tier_stats_body( + HashMap::new(), + accounted_inventory(), + HashMap::new(), + ClusterTierDailyStats { + stats: sample_daily_stats(), + nodes_reporting: 1, + nodes_expected: 2, + unavailable_nodes: vec!["10.0.0.2:9000".to_string()], + }, + None, + ); + + assert_eq!( + body.activity.status, "partial", + "a sum over part of the cluster must not be presented as a cluster total" + ); + assert_eq!(body.activity.nodes_reporting, 1); + assert_eq!(body.activity.nodes_expected, 2); + assert_eq!(body.activity.unavailable_nodes, vec!["10.0.0.2:9000".to_string()]); + } + + #[test] + fn an_unavailable_inventory_omits_the_per_tier_values() { + let body = assemble_tier_stats_body( + HashMap::from([("WARM".to_string(), TierType::S3)]), + TierInventoryStatusBody { + status: "unavailable", + updated_at: None, + detail: Some("usage snapshot could not be read".to_string()), + }, + HashMap::new(), + complete_activity(sample_daily_stats()), + Some("WARM"), + ); + + assert_eq!(body.inventory.status, "unavailable"); + assert_eq!( + body.tiers.first().expect("the tier must still be listed").inventory, + None, + "an unreadable snapshot must not be rendered as an empty tier" + ); + } + + #[test] + fn the_body_names_its_own_contract_version() { + let body = assemble_tier_stats_body( + HashMap::new(), + accounted_inventory(), + HashMap::new(), + complete_activity(DailyAllTierStats::new()), + None, + ); + let encoded = serde_json::to_value(&body).expect("the body must serialize"); + + assert_eq!(encoded["contractVersion"], 2); + assert!( + encoded["activity"]["nodesExpected"].is_number(), + "the activity coverage must reach the wire" + ); + } + + #[test] + fn tier_stats_counters_use_the_madmin_field_spelling() { + let encoded = serde_json::to_value(TierStatsBody::from(TierStats { + total_size: 1, + num_versions: 2, + num_objects: 3, + })) + .expect("counters must serialize"); + + assert_eq!(encoded["totalSize"], 1); + assert_eq!(encoded["numVersions"], 2); + assert_eq!(encoded["numObjects"], 3); + } + + #[test] + fn the_legacy_format_is_opt_in_and_unknown_formats_are_rejected() { + assert!(!tier_stats_wants_legacy_format(&TierStatsQuery::default()).expect("no format is the current contract")); + assert!( + tier_stats_wants_legacy_format(&TierStatsQuery { + tier: None, + format: Some("legacy".to_string()), + }) + .expect("legacy must stay reachable") + ); + + let err = tier_stats_wants_legacy_format(&TierStatsQuery { + tier: None, + format: Some("v3".to_string()), + }) + .expect_err("an unknown format must not be answered in another shape"); + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + } + fn sample_daily_stats() -> DailyAllTierStats { let mut warm = LastDayTierStats::default(); warm.add_stats(TierStats { diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 8e23fa0e0..03adac373 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -65,7 +65,7 @@ mod ecstore_metrics { mod ecstore_notification { pub(crate) use crate::storage::storage_api::ecstore_notification::{ - CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof, + ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof, }; } @@ -102,6 +102,7 @@ pub(crate) type ECStore = ecstore_storage::ECStore; pub(crate) type EndpointServerPools = ecstore_layout::EndpointServerPools; pub(crate) type MetricType = ecstore_metrics::MetricType; pub(crate) type NotificationSys = ecstore_notification::NotificationSys; +pub(crate) type ClusterTierDailyStats = ecstore_notification::ClusterTierDailyStats; pub(crate) type PeerRestClient = ecstore_rpc::PeerRestClient; pub(crate) type RebalSaveOpt = ecstore_rebalance::RebalSaveOpt; pub(crate) type RebalanceCleanupWarnings = ecstore_rebalance::RebalanceCleanupWarnings; @@ -1031,9 +1032,9 @@ pub(crate) mod s3 { pub(crate) mod tier { pub(crate) use super::{ - AdminError, DailyAllTierStats, ERR_TIER_ALREADY_EXISTS, ERR_TIER_BACKEND_IN_USE, ERR_TIER_BACKEND_NOT_EMPTY, - ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CONFIG, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_MISSING_CREDENTIALS, - ERR_TIER_NAME_NOT_UPPERCASE, ERR_TIER_NOT_FOUND, ERR_TIER_RESERVED_NAME, TierConfig, TierConfigUpdateError, TierCreds, - TierType, + AdminError, ClusterTierDailyStats, DailyAllTierStats, ECStore, ERR_TIER_ALREADY_EXISTS, ERR_TIER_BACKEND_IN_USE, + ERR_TIER_BACKEND_NOT_EMPTY, ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CONFIG, ERR_TIER_INVALID_CREDENTIALS, + ERR_TIER_MISSING_CREDENTIALS, ERR_TIER_NAME_NOT_UPPERCASE, ERR_TIER_NOT_FOUND, ERR_TIER_RESERVED_NAME, TierConfig, + TierConfigUpdateError, TierCreds, TierType, }; } diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 7b5258a98..44578265c 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -2497,6 +2497,13 @@ impl Node for NodeService { } } + async fn tier_daily_stats( + &self, + request: Request, + ) -> Result, Status> { + self.handle_tier_daily_stats(request).await + } + async fn load_transition_tier_config( &self, request: Request, diff --git a/rustfs/src/storage/rpc/node_service/metrics.rs b/rustfs/src/storage/rpc/node_service/metrics.rs index 79e78e076..559a1409b 100644 --- a/rustfs/src/storage/rpc/node_service/metrics.rs +++ b/rustfs/src/storage/rpc/node_service/metrics.rs @@ -14,11 +14,14 @@ use super::NodeService; use crate::storage::rpc::encode_msgpack_map; -use crate::storage::storage_api::rpc_consumer::node_service::{CollectMetricsOpts, MetricType, collect_local_metrics}; +use crate::storage::storage_api::rpc_consumer::node_service::{ + CollectMetricsOpts, MetricType, TierDailyStatsWire, collect_local_metrics, get_global_transition_state, +}; use bytes::Bytes; use rmp_serde::Deserializer; use rustfs_protos::proto_gen::node_service::*; use serde::Deserialize; +use std::collections::HashMap; use std::io::Cursor; use tonic::{Request, Response, Status}; use tracing::error; @@ -72,4 +75,35 @@ impl NodeService { })), } } + + /// This node's own rolling-day transition counters. + /// + /// Only this node's completions are reported; the caller sums the rings of + /// every member, so answering with anything wider would double count. + pub(super) async fn handle_tier_daily_stats( + &self, + _request: Request, + ) -> Result, Status> { + let stats = get_global_transition_state() + .get_daily_all_tier_stats() + .into_iter() + .map(|(tier, stats)| (tier, stats.to_wire())) + .collect::>(); + + match encode_msgpack_map(&stats) { + Ok(buf) => Ok(Response::new(TierDailyStatsResponse { + success: true, + tier_daily_stats: buf.into(), + error_info: None, + })), + Err(err) => { + error!(error = %err, "failed to serialize tier daily stats"); + Ok(Response::new(TierDailyStatsResponse { + success: false, + tier_daily_stats: Bytes::new(), + error_info: Some(err.to_string()), + })) + } + } + } } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index e503cfbe5..9e1e92dc4 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -261,9 +261,9 @@ pub(crate) mod rpc_consumer { ECStore, Error, FileInfoVersions, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MetricType, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq, ReadMultipleResp, ReadOptions, SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt, StoragePeerS3ClientExt, - UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, find_local_disk_by_ref, get_local_server_property, - reload_bucket_metadata, reload_transition_tier_config, remove_bucket_metadata, - validate_batch_read_version_item_count, + TierDailyStatsWire, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, find_local_disk_by_ref, + get_global_transition_state, get_local_server_property, reload_bucket_metadata, reload_transition_tier_config, + remove_bucket_metadata, validate_batch_read_version_item_count, }; pub(crate) type StorageResult = super::super::Result; @@ -511,7 +511,7 @@ pub(crate) mod ecstore_notification { #[cfg(test)] pub(crate) use rustfs_ecstore::api::notification::rotate_cross_pool_fence_fleet_proof_for_test; pub(crate) use rustfs_ecstore::api::notification::{ - CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof, + ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys, new_global_notification_sys, start_remote_version_state_fleet_probe, }; @@ -662,6 +662,7 @@ pub(crate) type BucketBandwidthMonitor = ecstore_bucket::bandwidth::monitor::Mon pub(crate) type CheckPartsResp = ecstore_disk::CheckPartsResp; pub(crate) type CollectMetricsOpts = ecstore_metrics::CollectMetricsOpts; pub(crate) type DailyAllTierStats = ecstore_bucket::lifecycle::tier_last_day_stats::DailyAllTierStats; +pub(crate) type TierDailyStatsWire = ecstore_bucket::lifecycle::tier_last_day_stats::TierDailyStatsWire; pub(crate) type DeleteOptions = ecstore_disk::DeleteOptions; pub(crate) type DiskError = ecstore_disk::error::DiskError; pub(crate) type DiskInfo = ecstore_disk::DiskInfo;