soma_infra/
zfs.rs

1use std::collections::BTreeMap;
2
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use soma_fleet::{HostId, HostRecord, TopologyRevision};
6use soma_ops::Timestamp;
7use tokio_util::sync::CancellationToken;
8
9use crate::{InfraError, InfraResult};
10
11const MAX_TARGET_CHARS: usize = 256;
12const MAX_ROWS: u32 = 5000;
13
14/// Allowlisted ZFS dataset types.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "snake_case")]
17pub enum ZfsDatasetType {
18    /// Filesystems.
19    Filesystem,
20    /// Block volumes.
21    Volume,
22    /// Snapshots.
23    Snapshot,
24    /// Bookmarks.
25    Bookmark,
26    /// Every supported type.
27    All,
28}
29
30impl ZfsDatasetType {
31    #[cfg(any(feature = "process-driver", test))]
32    pub(crate) const fn as_arg(self) -> &'static str {
33        match self {
34            Self::Filesystem => "filesystem",
35            Self::Volume => "volume",
36            Self::Snapshot => "snapshot",
37            Self::Bookmark => "bookmark",
38            Self::All => "all",
39        }
40    }
41}
42
43/// Request for a ZFS pool listing.
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct ZfsPoolRequest {
46    pool: Option<String>,
47    deadline: Timestamp,
48}
49
50impl ZfsPoolRequest {
51    /// Creates an unfiltered pool request.
52    #[must_use]
53    pub const fn new(deadline: Timestamp) -> Self {
54        Self {
55            pool: None,
56            deadline,
57        }
58    }
59
60    /// Restricts the listing to one pool.
61    pub fn with_pool(mut self, pool: impl Into<String>) -> InfraResult<Self> {
62        self.pool = Some(validate_target("pool", pool.into())?);
63        Ok(self)
64    }
65
66    /// Returns the optional pool filter.
67    #[must_use]
68    pub fn pool(&self) -> Option<&str> {
69        self.pool.as_deref()
70    }
71
72    /// Returns the absolute deadline.
73    #[must_use]
74    pub const fn deadline(&self) -> Timestamp {
75        self.deadline
76    }
77}
78
79/// Request for a ZFS dataset listing.
80#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
81pub struct ZfsDatasetRequest {
82    pool: Option<String>,
83    dataset_type: Option<ZfsDatasetType>,
84    recursive: bool,
85    deadline: Timestamp,
86}
87
88impl ZfsDatasetRequest {
89    /// Creates an unfiltered dataset request.
90    #[must_use]
91    pub const fn new(deadline: Timestamp) -> Self {
92        Self {
93            pool: None,
94            dataset_type: None,
95            recursive: false,
96            deadline,
97        }
98    }
99
100    /// Restricts the listing to one pool or dataset root.
101    pub fn with_pool(mut self, pool: impl Into<String>) -> InfraResult<Self> {
102        self.pool = Some(validate_target("pool", pool.into())?);
103        Ok(self)
104    }
105
106    /// Selects a dataset type.
107    #[must_use]
108    pub const fn with_type(mut self, dataset_type: ZfsDatasetType) -> Self {
109        self.dataset_type = Some(dataset_type);
110        self
111    }
112
113    /// Enables recursive listing.
114    #[must_use]
115    pub const fn recursive(mut self, recursive: bool) -> Self {
116        self.recursive = recursive;
117        self
118    }
119
120    /// Returns the optional pool filter.
121    #[must_use]
122    pub fn pool(&self) -> Option<&str> {
123        self.pool.as_deref()
124    }
125
126    /// Returns the optional dataset type.
127    #[must_use]
128    pub const fn dataset_type(&self) -> Option<ZfsDatasetType> {
129        self.dataset_type
130    }
131
132    /// Returns whether recursive listing is enabled.
133    #[must_use]
134    pub const fn is_recursive(&self) -> bool {
135        self.recursive
136    }
137
138    /// Returns the absolute deadline.
139    #[must_use]
140    pub const fn deadline(&self) -> Timestamp {
141        self.deadline
142    }
143}
144
145/// Request for a bounded ZFS snapshot listing.
146#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
147pub struct ZfsSnapshotRequest {
148    pool: Option<String>,
149    dataset: Option<String>,
150    limit: u32,
151    deadline: Timestamp,
152}
153
154impl ZfsSnapshotRequest {
155    /// Creates an unfiltered request limited to 500 rows.
156    #[must_use]
157    pub const fn new(deadline: Timestamp) -> Self {
158        Self {
159            pool: None,
160            dataset: None,
161            limit: 500,
162            deadline,
163        }
164    }
165
166    /// Sets a pool fallback target.
167    pub fn with_pool(mut self, pool: impl Into<String>) -> InfraResult<Self> {
168        self.pool = Some(validate_target("pool", pool.into())?);
169        Ok(self)
170    }
171
172    /// Sets a dataset target, which takes precedence over the pool.
173    pub fn with_dataset(mut self, dataset: impl Into<String>) -> InfraResult<Self> {
174        self.dataset = Some(validate_target("dataset", dataset.into())?);
175        Ok(self)
176    }
177
178    /// Sets the maximum returned rows.
179    pub fn with_limit(mut self, limit: u32) -> InfraResult<Self> {
180        if limit == 0 || limit > MAX_ROWS {
181            return Err(InfraError::InvalidRequest {
182                domain: "zfs",
183                message: format!("snapshot limit must be 1-{MAX_ROWS}"),
184            });
185        }
186        self.limit = limit;
187        Ok(self)
188    }
189
190    /// Returns the pool fallback.
191    #[must_use]
192    pub fn pool(&self) -> Option<&str> {
193        self.pool.as_deref()
194    }
195
196    /// Returns the dataset target.
197    #[must_use]
198    pub fn dataset(&self) -> Option<&str> {
199        self.dataset.as_deref()
200    }
201
202    /// Returns the row limit.
203    #[must_use]
204    pub const fn limit(&self) -> u32 {
205        self.limit
206    }
207
208    /// Returns the absolute deadline.
209    #[must_use]
210    pub const fn deadline(&self) -> Timestamp {
211        self.deadline
212    }
213}
214
215/// Structured ZFS tabular output.
216#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
217pub struct ZfsTable {
218    /// Target host.
219    pub host: HostId,
220    /// Exact topology revision.
221    pub topology_revision: TopologyRevision,
222    /// Column names in source order.
223    pub columns: Vec<String>,
224    /// Rows keyed by column name.
225    pub rows: Vec<BTreeMap<String, String>>,
226    /// Whether rows were omitted by the request limit.
227    pub truncated: bool,
228}
229
230/// Product-neutral ZFS read engine.
231#[async_trait]
232pub trait ZfsInspector: Send + Sync {
233    /// Lists pools.
234    async fn pools(
235        &self,
236        host: &HostRecord,
237        request: &ZfsPoolRequest,
238        cancellation: &CancellationToken,
239    ) -> InfraResult<ZfsTable>;
240
241    /// Lists datasets.
242    async fn datasets(
243        &self,
244        host: &HostRecord,
245        request: &ZfsDatasetRequest,
246        cancellation: &CancellationToken,
247    ) -> InfraResult<ZfsTable>;
248
249    /// Lists snapshots.
250    async fn snapshots(
251        &self,
252        host: &HostRecord,
253        request: &ZfsSnapshotRequest,
254        cancellation: &CancellationToken,
255    ) -> InfraResult<ZfsTable>;
256}
257
258#[cfg(any(feature = "process-driver", test))]
259pub(crate) fn parse_zfs_table(
260    host: &HostRecord,
261    raw: &str,
262    limit: Option<u32>,
263) -> InfraResult<ZfsTable> {
264    let mut lines = raw.lines().filter(|line| !line.trim().is_empty());
265    let columns = lines
266        .next()
267        .ok_or_else(|| parse_error("ZFS output has no header"))?
268        .split_whitespace()
269        .map(str::to_owned)
270        .collect::<Vec<_>>();
271    if columns.is_empty() {
272        return Err(parse_error("ZFS output has an empty header"));
273    }
274    let mut rows = lines
275        .map(|line| parse_row(&columns, line))
276        .collect::<InfraResult<Vec<_>>>()?;
277    let truncated = limit.is_some_and(|limit| rows.len() > limit as usize);
278    if let Some(limit) = limit {
279        rows.truncate(limit as usize);
280    }
281    Ok(ZfsTable {
282        host: host.id().clone(),
283        topology_revision: host.revision().clone(),
284        columns,
285        rows,
286        truncated,
287    })
288}
289
290#[cfg(any(feature = "process-driver", test))]
291fn parse_row(columns: &[String], line: &str) -> InfraResult<BTreeMap<String, String>> {
292    let values = line.split_whitespace().collect::<Vec<_>>();
293    if values.len() < columns.len() {
294        return Err(parse_error(&format!(
295            "ZFS row has {} values for {} columns",
296            values.len(),
297            columns.len()
298        )));
299    }
300    let mut row = BTreeMap::new();
301    for (index, column) in columns.iter().enumerate() {
302        let value = if index + 1 == columns.len() {
303            values[index..].join(" ")
304        } else {
305            values[index].to_owned()
306        };
307        row.insert(column.clone(), value);
308    }
309    Ok(row)
310}
311
312fn validate_target(kind: &'static str, value: String) -> InfraResult<String> {
313    let count = value.chars().count();
314    if count == 0
315        || count > MAX_TARGET_CHARS
316        || value.starts_with('-')
317        || value.chars().any(|character| {
318            !(character.is_ascii_alphanumeric()
319                || matches!(character, '_' | '-' | '.' | ':' | '/' | '@'))
320        })
321    {
322        Err(InfraError::InvalidRequest {
323            domain: "zfs",
324            message: format!("invalid {kind} target: {value:?}"),
325        })
326    } else {
327        Ok(value)
328    }
329}
330
331#[cfg(any(feature = "process-driver", test))]
332fn parse_error(message: &str) -> InfraError {
333    InfraError::Parse {
334        domain: "zfs",
335        message: message.to_owned(),
336    }
337}
338
339#[cfg(test)]
340#[path = "zfs_tests.rs"]
341mod tests;
soma_infra/zfs.rs

soma_infra/
zfs.rs
