rustfs_filemeta/filemeta/
inline_data.rs

1// Copyright 2024 RustFS Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use super::*;
16use rustfs_utils::http::{SUFFIX_TRANSITION_STATUS, get_bytes};
17use std::collections::HashSet;
18
19fn physical_data_dir(version: &FileMetaShallowVersion) -> Result<Option<Uuid>> {
20    if version.header.version_type != VersionType::Object {
21        return Ok(None);
22    }
23
24    let version = FileMetaVersion::try_from(version.meta.as_slice())?;
25    let Some(obj) = version.object else {
26        return Ok(None);
27    };
28
29    if obj.inlinedata() {
30        return Ok(None);
31    }
32
33    if let Some(status) = get_bytes(&obj.meta_sys, SUFFIX_TRANSITION_STATUS)
34        && status.as_slice() == TRANSITION_COMPLETE.as_bytes()
35        && !is_restored_object_on_disk(&obj.meta_user)
36    {
37        return Ok(None);
38    }
39
40    Ok(obj.data_dir.filter(|dir| !dir.is_nil()))
41}
42
43impl FileMeta {
44    pub fn find_unshared_data_dir_for_version(&self, version_id: Option<Uuid>) -> Option<Uuid> {
45        let vid = version_id.unwrap_or_default();
46        let mut target_data_dir = None;
47        let mut target_selected = false;
48        let mut other_data_dirs = HashSet::new();
49
50        for version in self.versions.iter().filter(|v| v.header.version_type == VersionType::Object) {
51            let is_target_version = version.header.version_id.unwrap_or_default() == vid;
52            let dir = match physical_data_dir(version) {
53                Ok(dir) => dir,
54                Err(_) => return None,
55            };
56
57            if is_target_version {
58                if target_selected {
59                    continue;
60                }
61
62                target_selected = true;
63                target_data_dir = dir;
64                if let Some(dir) = target_data_dir
65                    && other_data_dirs.contains(&dir)
66                {
67                    return None;
68                }
69                continue;
70            }
71
72            if let Some(dir) = dir {
73                if target_data_dir == Some(dir) {
74                    return None;
75                }
76                other_data_dirs.insert(dir);
77            }
78        }
79
80        target_data_dir
81    }
82
83    pub fn shard_data_dir_count(&self, vid: &Option<Uuid>, data_dir: &Option<Uuid>) -> usize {
84        let vid = vid.unwrap_or_default();
85        let Some(data_dir) = data_dir else {
86            return 0;
87        };
88
89        let mut count = 0;
90        for version in self
91            .versions
92            .iter()
93            .filter(|v| v.header.version_type == VersionType::Object && v.header.version_id != Some(vid))
94        {
95            match physical_data_dir(version) {
96                Ok(Some(dir)) if dir == *data_dir => count += 1,
97                Ok(_) => {}
98                Err(_) => return 1,
99            }
100        }
101        count
102    }
103
104    pub fn get_data_dirs(&self) -> Result<Vec<Option<Uuid>>> {
105        let mut data_dirs = Vec::new();
106        for version in &self.versions {
107            if version.header.version_type == VersionType::Object {
108                let ver = FileMetaVersion::try_from(version.meta.as_slice())?;
109                data_dirs.push(ver.get_data_dir());
110            }
111        }
112        Ok(data_dirs)
113    }
114
115    /// Count shared data directories
116    pub fn shared_data_dir_count(&self, version_id: Option<Uuid>, data_dir: Option<Uuid>) -> usize {
117        let vid = version_id.unwrap_or_default();
118        let Some(data_dir) = data_dir else {
119            return 0;
120        };
121
122        if self.data.entries().unwrap_or_default() > 0
123            && self.find_inline_data_for_version(version_id).unwrap_or_default().is_some()
124        {
125            return 0;
126        }
127
128        let mut count = 0;
129        for version in self
130            .versions
131            .iter()
132            .filter(|v| v.header.version_type == VersionType::Object && v.header.version_id != Some(vid))
133        {
134            match physical_data_dir(version) {
135                Ok(Some(dir)) if dir == data_dir => count += 1,
136                Ok(_) => {}
137                Err(_) => return 1,
138            }
139        }
140        count
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147    use time::format_description::well_known::Rfc3339;
148    use time::{Duration, OffsetDateTime};
149
150    fn make_plain_file_info(version_id: Uuid, data_dir: Uuid) -> FileInfo {
151        make_file_info_with_metadata(version_id, data_dir, HashMap::from([("etag".to_string(), format!("etag-{version_id}"))]))
152    }
153
154    fn make_plain_null_version_file_info(data_dir: Uuid) -> FileInfo {
155        let mut fi = make_plain_file_info(Uuid::nil(), data_dir);
156        fi.version_id = None;
157        fi
158    }
159
160    fn make_file_info(version_id: Uuid, data_dir: Uuid) -> FileInfo {
161        let restore_header = format!(
162            "ongoing-request=\"false\", expiry-date=\"{}\"",
163            (OffsetDateTime::now_utc() + Duration::days(1))
164                .format(&Rfc3339)
165                .expect("format restore expiry"),
166        );
167        make_file_info_with_metadata(
168            version_id,
169            data_dir,
170            HashMap::from([
171                ("etag".to_string(), format!("etag-{version_id}")),
172                (metadata_keys::RESTORE.to_string(), restore_header),
173            ]),
174        )
175    }
176
177    fn make_file_info_with_metadata(version_id: Uuid, data_dir: Uuid, metadata: HashMap<String, String>) -> FileInfo {
178        FileInfo {
179            version_id: Some(version_id),
180            data_dir: Some(data_dir),
181            size: 64 * 1024,
182            mod_time: Some(OffsetDateTime::now_utc()),
183            metadata,
184            erasure: ErasureInfo {
185                algorithm: ErasureAlgo::ReedSolomon.to_string(),
186                data_blocks: 4,
187                parity_blocks: 2,
188                block_size: 1024 * 1024,
189                index: 1,
190                distribution: vec![1, 2, 3, 4, 5, 6],
191                ..Default::default()
192            },
193            ..Default::default()
194        }
195    }
196
197    #[test]
198    fn find_unshared_data_dir_for_version_returns_data_dir_when_unique() {
199        let target_version = Uuid::new_v4();
200        let target_data_dir = Uuid::new_v4();
201        let mut meta = FileMeta::new();
202        meta.add_version(make_file_info(target_version, target_data_dir))
203            .expect("seed target version");
204        meta.add_version(make_file_info(Uuid::new_v4(), Uuid::new_v4()))
205            .expect("seed non-shared version");
206
207        let got = meta.find_unshared_data_dir_for_version(Some(target_version));
208        assert_eq!(got, Some(target_data_dir));
209    }
210
211    #[test]
212    fn find_unshared_data_dir_for_version_returns_none_when_shared() {
213        let target_version = Uuid::new_v4();
214        let shared_data_dir = Uuid::new_v4();
215        let mut meta = FileMeta::new();
216        meta.add_version(make_file_info(target_version, shared_data_dir))
217            .expect("seed target version");
218        meta.add_version(make_file_info(Uuid::new_v4(), shared_data_dir))
219            .expect("seed shared version");
220
221        let got = meta.find_unshared_data_dir_for_version(Some(target_version));
222        assert_eq!(got, None);
223    }
224
225    #[test]
226    fn find_unshared_data_dir_for_version_uses_data_dir_when_header_flag_is_false() {
227        let target_version = Uuid::new_v4();
228        let target_data_dir = Uuid::new_v4();
229        let mut meta = FileMeta::new();
230        meta.add_version(make_plain_file_info(target_version, target_data_dir))
231            .expect("seed target version");
232        meta.add_version(make_plain_file_info(Uuid::new_v4(), Uuid::new_v4()))
233            .expect("seed non-shared version");
234
235        let target = meta
236            .versions
237            .iter()
238            .find(|version| version.header.version_id == Some(target_version))
239            .expect("target version header");
240        assert!(!target.header.uses_data_dir());
241
242        let got = meta.find_unshared_data_dir_for_version(Some(target_version));
243        assert_eq!(got, Some(target_data_dir));
244    }
245
246    #[test]
247    fn find_unshared_data_dir_for_null_version_uses_data_dir_when_header_flag_is_false() {
248        let target_data_dir = Uuid::new_v4();
249        let mut meta = FileMeta::new();
250        meta.add_version(make_plain_null_version_file_info(target_data_dir))
251            .expect("seed null version");
252
253        let target = meta
254            .versions
255            .iter()
256            .find(|version| version.header.version_id == Some(Uuid::nil()))
257            .expect("null version header");
258        assert!(!target.header.uses_data_dir());
259
260        let got = meta.find_unshared_data_dir_for_version(None);
261        assert_eq!(got, Some(target_data_dir));
262    }
263
264    #[test]
265    fn shared_data_dir_count_uses_data_dir_when_header_flag_is_false() {
266        let target_version = Uuid::new_v4();
267        let other_version = Uuid::new_v4();
268        let shared_data_dir = Uuid::new_v4();
269        let mut meta = FileMeta::new();
270        meta.add_version(make_plain_file_info(target_version, shared_data_dir))
271            .expect("seed target version");
272        meta.add_version(make_plain_file_info(other_version, shared_data_dir))
273            .expect("seed shared version");
274
275        assert!(meta.versions.iter().all(|version| !version.header.uses_data_dir()));
276        assert_eq!(meta.shared_data_dir_count(Some(target_version), Some(shared_data_dir)), 1);
277
278        let old_dir = meta
279            .delete_version(&FileInfo {
280                version_id: Some(target_version),
281                ..Default::default()
282            })
283            .expect("delete target version");
284        assert_eq!(old_dir, None);
285    }
286
287    #[test]
288    fn find_unshared_data_dir_for_version_skips_remote_transitioned_data_dir() {
289        let target_version = Uuid::new_v4();
290        let target_data_dir = Uuid::new_v4();
291        let mut fi = make_plain_file_info(target_version, target_data_dir);
292        fi.transition_status = TRANSITION_COMPLETE.to_string();
293
294        let mut meta = FileMeta::new();
295        meta.add_version(fi).expect("seed remote transitioned version");
296
297        let got = meta.find_unshared_data_dir_for_version(Some(target_version));
298        assert_eq!(got, None);
299    }
300}
rustfs_filemeta/filemeta/inline_data.rs

rustfs_filemeta/filemeta/
inline_data.rs
