rc_core/
ops.rs

1//! Typed operational APIs for health probes and storage usage.
2
3use std::time::Duration;
4
5use async_trait::async_trait;
6use jiff::Timestamp;
7use serde::{Deserialize, Serialize};
8
9use crate::Result;
10
11/// RustFS health probe kind.
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
13#[serde(rename_all = "snake_case")]
14pub enum HealthProbe {
15    /// Process liveness through the public `/health` endpoint.
16    Liveness,
17    /// Dependency readiness through the public `/health/ready` endpoint.
18    Readiness,
19}
20
21impl HealthProbe {
22    /// Return the documented RustFS beta.10 endpoint for this probe.
23    pub const fn path(self) -> &'static str {
24        match self {
25            Self::Liveness => "/health",
26            Self::Readiness => "/health/ready",
27        }
28    }
29
30    /// Return the stable machine-readable probe name.
31    pub const fn as_str(self) -> &'static str {
32        match self {
33            Self::Liveness => "liveness",
34            Self::Readiness => "readiness",
35        }
36    }
37}
38
39/// Result returned for a completed HTTP health probe.
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41pub struct HealthReport {
42    /// Probe kind.
43    pub probe: HealthProbe,
44    /// Configured service endpoint, without credentials.
45    pub endpoint: String,
46    /// Probe path used for the request.
47    pub path: String,
48    /// HTTP status code returned by the service.
49    pub status_code: u16,
50    /// Whether the endpoint reported the requested health state.
51    pub healthy: bool,
52    /// Round-trip latency in milliseconds.
53    pub latency_ms: u64,
54    /// Optional server-provided state such as `ok` or `degraded`.
55    pub status: Option<String>,
56    /// Optional server-provided service identity.
57    pub service: Option<String>,
58    /// Optional server-provided version.
59    pub server_version: Option<String>,
60}
61
62/// Adapter boundary for public health endpoints.
63#[async_trait]
64pub trait HealthApi: Send + Sync {
65    /// Run one health probe with an explicit upper time bound.
66    async fn check_health(&self, probe: HealthProbe, timeout: Duration) -> Result<HealthReport>;
67}
68
69/// Origin of a usage report.
70#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
71#[serde(rename_all = "snake_case")]
72pub enum UsageSource {
73    /// RustFS background scanner snapshot.
74    ServerSnapshot,
75    /// Portable client-side S3 listing.
76    ClientScan,
77}
78
79impl UsageSource {
80    /// Return the stable machine-readable source name.
81    pub const fn as_str(self) -> &'static str {
82        match self {
83            Self::ServerSnapshot => "server_snapshot",
84            Self::ClientScan => "client_scan",
85        }
86    }
87}
88
89/// Requested usage scope.
90#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
91#[serde(rename_all = "snake_case")]
92pub enum UsageScope {
93    /// All buckets visible to the configured alias.
94    Cluster,
95    /// A single bucket.
96    Bucket,
97    /// A prefix within a bucket.
98    Prefix,
99}
100
101impl UsageScope {
102    /// Return the stable machine-readable scope name.
103    pub const fn as_str(self) -> &'static str {
104        match self {
105            Self::Cluster => "cluster",
106            Self::Bucket => "bucket",
107            Self::Prefix => "prefix",
108        }
109    }
110}
111
112/// Per-bucket usage values.
113#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
114pub struct UsageBucket {
115    /// Bucket name.
116    pub name: String,
117    /// Bytes represented by the selected scan dimensions.
118    pub total_bytes: u64,
119    /// Current object count.
120    pub object_count: u64,
121    /// Non-delete object version count, when collected.
122    pub version_count: Option<u64>,
123    /// Delete marker count, when collected.
124    pub delete_marker_count: Option<u64>,
125    /// Incomplete multipart upload count, when collected.
126    pub incomplete_upload_count: Option<u64>,
127    /// Uploaded part bytes belonging to incomplete uploads, when collected.
128    pub incomplete_upload_bytes: Option<u64>,
129}
130
131/// A bucket that could not be included in a multi-bucket client scan.
132#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
133pub struct UsageFailure {
134    /// Bucket name.
135    pub bucket: String,
136    /// Sanitizable diagnostic that never contains configured credentials.
137    pub message: String,
138}
139
140/// Complete or partial usage result.
141#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
142pub struct UsageReport {
143    /// Data source used for the report.
144    pub source: UsageSource,
145    /// Scope represented by the report.
146    pub scope: UsageScope,
147    /// Alias-relative path for bucket and prefix scopes.
148    pub path: Option<String>,
149    /// Server snapshot timestamp; absent for client scans.
150    pub snapshot_at: Option<Timestamp>,
151    /// Total bytes represented by the report.
152    pub total_bytes: u64,
153    /// Current object count.
154    pub object_count: u64,
155    /// Version count, when supplied by the selected source.
156    pub version_count: Option<u64>,
157    /// Delete marker count, when supplied by the selected source.
158    pub delete_marker_count: Option<u64>,
159    /// Incomplete upload count, when explicitly scanned.
160    pub incomplete_upload_count: Option<u64>,
161    /// Incomplete uploaded-part bytes, when explicitly scanned.
162    pub incomplete_upload_bytes: Option<u64>,
163    /// Deterministically sorted bucket rows.
164    pub buckets: Vec<UsageBucket>,
165    /// Whether one or more requested buckets could not be scanned.
166    pub partial: bool,
167    /// Per-bucket failures for partial multi-bucket scans.
168    pub failures: Vec<UsageFailure>,
169}
170
171impl UsageReport {
172    /// Create an empty report that can be populated with bucket rows.
173    pub fn empty(source: UsageSource, scope: UsageScope, path: Option<String>) -> Self {
174        Self {
175            source,
176            scope,
177            path,
178            snapshot_at: None,
179            total_bytes: 0,
180            object_count: 0,
181            version_count: None,
182            delete_marker_count: None,
183            incomplete_upload_count: None,
184            incomplete_upload_bytes: None,
185            buckets: Vec::new(),
186            partial: false,
187            failures: Vec::new(),
188        }
189    }
190
191    /// Add one bucket row before final aggregation.
192    pub fn push_bucket(&mut self, bucket: UsageBucket) {
193        self.buckets.push(bucket);
194    }
195
196    /// Add one recoverable bucket failure.
197    pub fn push_failure(&mut self, failure: UsageFailure) {
198        self.partial = true;
199        self.failures.push(failure);
200    }
201
202    /// Sort rows and recompute report totals using saturating arithmetic.
203    pub fn finish(&mut self) {
204        let versions_were_requested = self.version_count.is_some();
205        let delete_markers_were_requested = self.delete_marker_count.is_some();
206        let incomplete_uploads_were_requested = self.incomplete_upload_count.is_some();
207        let incomplete_bytes_were_requested = self.incomplete_upload_bytes.is_some();
208        self.buckets
209            .sort_by(|left, right| left.name.cmp(&right.name));
210        self.failures
211            .sort_by(|left, right| left.bucket.cmp(&right.bucket));
212        self.total_bytes = self.buckets.iter().fold(0_u64, |total, bucket| {
213            total.saturating_add(bucket.total_bytes)
214        });
215        self.object_count = self.buckets.iter().fold(0_u64, |total, bucket| {
216            total.saturating_add(bucket.object_count)
217        });
218        self.version_count = sum_optional(self.buckets.iter().map(|bucket| bucket.version_count))
219            .or_else(|| versions_were_requested.then_some(0));
220        self.delete_marker_count =
221            sum_optional(self.buckets.iter().map(|bucket| bucket.delete_marker_count))
222                .or_else(|| delete_markers_were_requested.then_some(0));
223        self.incomplete_upload_count = sum_optional(
224            self.buckets
225                .iter()
226                .map(|bucket| bucket.incomplete_upload_count),
227        )
228        .or_else(|| incomplete_uploads_were_requested.then_some(0));
229        self.incomplete_upload_bytes = sum_optional(
230            self.buckets
231                .iter()
232                .map(|bucket| bucket.incomplete_upload_bytes),
233        )
234        .or_else(|| incomplete_bytes_were_requested.then_some(0));
235    }
236}
237
238fn sum_optional(values: impl Iterator<Item = Option<u64>>) -> Option<u64> {
239    let mut values = values;
240    let first = values.next()??;
241    values.try_fold(first, |total, value| Some(total.saturating_add(value?)))
242}
243
244/// Options for an explicitly authorized client-side usage scan.
245#[derive(Debug, Clone, PartialEq, Eq)]
246pub struct UsageScanRequest {
247    /// Optional bucket; absent means all visible buckets.
248    pub bucket: Option<String>,
249    /// Optional prefix within `bucket`.
250    pub prefix: Option<String>,
251    /// Include all object versions and delete markers.
252    pub include_versions: bool,
253    /// Include incomplete multipart uploads and uploaded part bytes.
254    pub include_incomplete_uploads: bool,
255}
256
257impl UsageScanRequest {
258    /// Resolve the request scope.
259    pub const fn scope(&self) -> UsageScope {
260        if self.prefix.is_some() {
261            UsageScope::Prefix
262        } else if self.bucket.is_some() {
263            UsageScope::Bucket
264        } else {
265            UsageScope::Cluster
266        }
267    }
268
269    /// Whether the request asks for data unavailable in the cluster snapshot.
270    pub const fn requires_client_scan(&self) -> bool {
271        self.prefix.is_some() || self.include_incomplete_uploads
272    }
273
274    /// Build the canonical alias-relative output path.
275    pub fn path(&self, alias: &str) -> Option<String> {
276        let bucket = self.bucket.as_deref()?;
277        Some(match self.prefix.as_deref() {
278            Some(prefix) => format!("{alias}/{bucket}/{prefix}"),
279            None => format!("{alias}/{bucket}"),
280        })
281    }
282}
283
284/// Adapter boundary for the RustFS background-scanner snapshot.
285#[async_trait]
286pub trait UsageSnapshotApi: Send + Sync {
287    /// Fetch the cluster-wide server snapshot.
288    async fn usage_snapshot(&self) -> Result<UsageReport>;
289}
290
291/// Adapter boundary for a portable S3 usage scan.
292#[async_trait]
293pub trait UsageScanApi: Send + Sync {
294    /// Scan usage using paginated S3 APIs.
295    async fn scan_usage(&self, request: &UsageScanRequest) -> Result<UsageReport>;
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301
302    #[test]
303    fn optional_totals_remain_unknown_when_any_bucket_is_unknown() {
304        assert_eq!(sum_optional(std::iter::empty()), None);
305        assert_eq!(sum_optional([Some(1), None].into_iter()), None);
306        assert_eq!(sum_optional([Some(1), Some(2)].into_iter()), Some(3));
307    }
308}
rc_core/ops.rs

rc_core/
ops.rs
