Skip to content

Commit 9869ef3

Browse files
committed
Add xattr-based attribute support to LocalFileSystem
Store object attributes as extended attributes on the local filesystem, enabling put_opts and put_multipart_opts to persist attributes. Attributes are set on the staging file before atomic rename to preserve consistency. - Add xattr dependency (gated by fs feature) - Map standard attributes to user.* xattr namespace - Read attributes back in get_opts
1 parent 8ef1aaa commit 9869ef3

2 files changed

Lines changed: 180 additions & 22 deletions

File tree

Cargo.toml

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ thiserror = "2.0.2"
4444
tracing = { version = "0.1" }
4545
url = "2.2"
4646
walkdir = { version = "2", optional = true }
47+
xattr = { version = "1", optional = true }
4748

4849
# Cloud storage support
4950
base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
@@ -73,7 +74,7 @@ wasm-bindgen-futures = "0.4.18"
7374
default = ["fs"]
7475
cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/stream", "chrono/serde", "base64", "rand", "ring", "http-body-util", "form_urlencoded", "serde_urlencoded"]
7576
azure = ["cloud", "httparse"]
76-
fs = ["walkdir"]
77+
fs = ["walkdir", "xattr"]
7778
gcp = ["cloud", "rustls-pki-types"]
7879
aws = ["cloud", "md-5"]
7980
http = ["cloud"]

src/local.rs

Lines changed: 178 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
// under the License.
1717

1818
//! An object store implementation for a local filesystem
19+
use std::borrow::Cow;
1920
use std::fs::{File, Metadata, OpenOptions, metadata, symlink_metadata};
2021
use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
2122
use std::ops::Range;
@@ -34,9 +35,9 @@ use url::Url;
3435
use walkdir::{DirEntry, WalkDir};
3536

3637
use crate::{
37-
Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta,
38-
ObjectStore, PutMode, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
39-
UploadPart, maybe_spawn_blocking,
38+
Attribute, Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload,
39+
ObjectMeta, ObjectStore, PutMode, PutMultipartOptions, PutOptions, PutPayload, PutResult,
40+
Result, UploadPart, maybe_spawn_blocking,
4041
path::{Path, absolute_path_to_url},
4142
util::InvalidGetRange,
4243
};
@@ -112,6 +113,18 @@ pub(crate) enum Error {
112113

113114
#[error("Upload aborted")]
114115
Aborted,
116+
117+
#[error("Unable to set extended attribute on {}: {source}", path.display())]
118+
UnableToSetXattr {
119+
source: std::io::Error,
120+
path: PathBuf,
121+
},
122+
123+
#[error("Unable to read extended attribute on {}: {source}", path.display())]
124+
UnableToReadXattr {
125+
source: std::io::Error,
126+
path: PathBuf,
127+
},
115128
}
116129

117130
impl From<Error> for super::Error {
@@ -321,6 +334,79 @@ fn is_valid_file_path(path: &Path) -> bool {
321334
}
322335
}
323336

337+
/// Sets extended attributes on a file from an Attributes collection
338+
fn set_xattrs(path: &std::path::Path, attributes: &Attributes) -> Result<()> {
339+
for (attr, value) in attributes {
340+
let name: Cow<'static, str> = match attr {
341+
Attribute::CacheControl => Cow::Borrowed("user.cache_control"),
342+
Attribute::ContentDisposition => Cow::Borrowed("user.content_disposition"),
343+
Attribute::ContentEncoding => Cow::Borrowed("user.content_encoding"),
344+
Attribute::ContentLanguage => Cow::Borrowed("user.content_language"),
345+
Attribute::ContentType => Cow::Borrowed("user.content_type"),
346+
Attribute::StorageClass => Cow::Borrowed("user.storage_class"),
347+
Attribute::Metadata(key) => Cow::Owned(format!("user.{key}")),
348+
};
349+
xattr::set(path, name.as_ref(), value.as_ref().as_bytes()).map_err(|source| {
350+
Error::UnableToSetXattr {
351+
source,
352+
path: path.into(),
353+
}
354+
})?;
355+
}
356+
Ok(())
357+
}
358+
359+
/// Reads extended attributes from a file and returns an Attributes collection
360+
fn get_xattrs(path: &std::path::Path) -> Result<Attributes> {
361+
let mut attributes = Attributes::new();
362+
363+
let list = match xattr::list(path) {
364+
Ok(list) => list,
365+
Err(source) => {
366+
return Err(Error::UnableToReadXattr {
367+
source,
368+
path: path.into(),
369+
}
370+
.into());
371+
}
372+
};
373+
374+
for name in list {
375+
let name_str = match name.to_str() {
376+
Some(s) if s.starts_with("user.") => s,
377+
_ => continue,
378+
};
379+
380+
let value = match xattr::get(path, &name) {
381+
Ok(Some(v)) => match String::from_utf8(v) {
382+
Ok(s) => s,
383+
Err(_) => continue,
384+
},
385+
Ok(None) => continue,
386+
Err(source) => {
387+
return Err(Error::UnableToReadXattr {
388+
source,
389+
path: path.into(),
390+
}
391+
.into());
392+
}
393+
};
394+
395+
let key = &name_str["user.".len()..];
396+
let attr = match key {
397+
"cache_control" => Attribute::CacheControl,
398+
"content_disposition" => Attribute::ContentDisposition,
399+
"content_encoding" => Attribute::ContentEncoding,
400+
"content_language" => Attribute::ContentLanguage,
401+
"content_type" => Attribute::ContentType,
402+
"storage_class" => Attribute::StorageClass,
403+
key => Attribute::Metadata(key.to_string().into()),
404+
};
405+
attributes.insert(attr, value.into());
406+
}
407+
Ok(attributes)
408+
}
409+
324410
#[async_trait]
325411
impl ObjectStore for LocalFileSystem {
326412
async fn put_opts(
@@ -336,13 +422,6 @@ impl ObjectStore for LocalFileSystem {
336422
});
337423
}
338424

339-
if !opts.attributes.is_empty() {
340-
return Err(crate::Error::NotImplemented {
341-
operation: "`put_opts` with `opts.attributes` specified".into(),
342-
implementer: self.to_string(),
343-
});
344-
}
345-
346425
let path = self.path_to_filesystem(location)?;
347426
maybe_spawn_blocking(move || {
348427
let (mut file, staging_path) = new_staged_upload(&path)?;
@@ -355,6 +434,11 @@ impl ObjectStore for LocalFileSystem {
355434
path: path.to_string_lossy().to_string(),
356435
})?;
357436
e_tag = Some(get_etag(&metadata));
437+
438+
if !opts.attributes.is_empty() {
439+
set_xattrs(&staging_path, &opts.attributes)?;
440+
}
441+
358442
match opts.mode {
359443
PutMode::Overwrite => {
360444
// For some fuse types of file systems, the file must be closed first
@@ -402,16 +486,9 @@ impl ObjectStore for LocalFileSystem {
402486
location: &Path,
403487
opts: PutMultipartOptions,
404488
) -> Result<Box<dyn MultipartUpload>> {
405-
if !opts.attributes.is_empty() {
406-
return Err(crate::Error::NotImplemented {
407-
operation: "`put_multipart_opts` with `opts.attributes` specified".into(),
408-
implementer: self.to_string(),
409-
});
410-
}
411-
412489
let dest = self.path_to_filesystem(location)?;
413490
let (file, src) = new_staged_upload(&dest)?;
414-
Ok(Box::new(LocalUpload::new(src, dest, file)))
491+
Ok(Box::new(LocalUpload::new(src, dest, file, opts.attributes)))
415492
}
416493

417494
async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
@@ -429,9 +506,11 @@ impl ObjectStore for LocalFileSystem {
429506
None => 0..meta.size,
430507
};
431508

509+
let attributes = get_xattrs(&path)?;
510+
432511
Ok(GetResult {
433512
payload: GetResultPayload::File(file, path),
434-
attributes: Attributes::default(),
513+
attributes,
435514
range,
436515
meta,
437516
})
@@ -829,6 +908,8 @@ struct LocalUpload {
829908
src: Option<PathBuf>,
830909
/// The next offset to write into the file
831910
offset: u64,
911+
/// Attributes to set on the file
912+
attributes: Attributes,
832913
}
833914

834915
#[derive(Debug)]
@@ -838,14 +919,15 @@ struct UploadState {
838919
}
839920

840921
impl LocalUpload {
841-
pub(crate) fn new(src: PathBuf, dest: PathBuf, file: File) -> Self {
922+
pub(crate) fn new(src: PathBuf, dest: PathBuf, file: File, attributes: Attributes) -> Self {
842923
Self {
843924
state: Arc::new(UploadState {
844925
dest,
845926
file: Mutex::new(file),
846927
}),
847928
src: Some(src),
848929
offset: 0,
930+
attributes,
849931
}
850932
}
851933
}
@@ -876,9 +958,15 @@ impl MultipartUpload for LocalUpload {
876958
async fn complete(&mut self) -> Result<PutResult> {
877959
let src = self.src.take().ok_or(Error::Aborted)?;
878960
let s = Arc::clone(&self.state);
961+
let attributes = std::mem::take(&mut self.attributes);
879962
maybe_spawn_blocking(move || {
880963
// Ensure no inflight writes
881964
let file = s.file.lock();
965+
966+
if !attributes.is_empty() {
967+
set_xattrs(&src, &attributes)?;
968+
}
969+
882970
std::fs::rename(&src, &s.dest)
883971
.map_err(|source| Error::UnableToRenameFile { source })?;
884972
let metadata = file.metadata().map_err(|e| Error::Metadata {
@@ -1177,6 +1265,7 @@ mod tests {
11771265
copy_rename_nonexistent_object(&integration).await;
11781266
stream_get(&integration).await;
11791267
put_opts(&integration, false).await;
1268+
put_get_attributes(&integration).await;
11801269
}
11811270

11821271
#[test]
@@ -1790,7 +1879,7 @@ mod unix_test {
17901879
use tempfile::TempDir;
17911880

17921881
use crate::local::LocalFileSystem;
1793-
use crate::{ObjectStoreExt, Path};
1882+
use crate::{Attribute, Attributes, ObjectStore, ObjectStoreExt, Path, PutOptions};
17941883

17951884
#[tokio::test]
17961885
async fn test_fifo() {
@@ -1810,4 +1899,72 @@ mod unix_test {
18101899

18111900
spawned.await.unwrap();
18121901
}
1902+
1903+
#[tokio::test]
1904+
async fn test_put_get_attributes() {
1905+
let root = TempDir::new().unwrap();
1906+
let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
1907+
1908+
let location = Path::from("test_file");
1909+
let data = "test data";
1910+
1911+
let mut attributes = Attributes::new();
1912+
attributes.insert(Attribute::ContentType, "text/plain".into());
1913+
attributes.insert(Attribute::CacheControl, "max-age=3600".into());
1914+
attributes.insert(Attribute::ContentDisposition, "inline".into());
1915+
attributes.insert(Attribute::ContentEncoding, "gzip".into());
1916+
attributes.insert(Attribute::ContentLanguage, "en-US".into());
1917+
attributes.insert(Attribute::StorageClass, "STANDARD".into());
1918+
attributes.insert(
1919+
Attribute::Metadata("custom_key".into()),
1920+
"custom_value".into(),
1921+
);
1922+
1923+
let opts = PutOptions {
1924+
attributes: attributes.clone(),
1925+
..Default::default()
1926+
};
1927+
1928+
integration
1929+
.put_opts(&location, data.into(), opts)
1930+
.await
1931+
.unwrap();
1932+
1933+
let result = integration.get(&location).await.unwrap();
1934+
assert_eq!(result.attributes, attributes);
1935+
1936+
let bytes = result.bytes().await.unwrap();
1937+
assert_eq!(bytes.as_ref(), data.as_bytes());
1938+
}
1939+
1940+
#[tokio::test]
1941+
async fn test_multipart_attributes() {
1942+
let root = TempDir::new().unwrap();
1943+
let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
1944+
1945+
let location = Path::from("multipart_file");
1946+
1947+
let mut attributes = Attributes::new();
1948+
attributes.insert(Attribute::ContentType, "application/octet-stream".into());
1949+
attributes.insert(Attribute::Metadata("part_count".into()), "2".into());
1950+
1951+
let opts = crate::PutMultipartOptions {
1952+
attributes: attributes.clone(),
1953+
..Default::default()
1954+
};
1955+
1956+
let mut upload = integration
1957+
.put_multipart_opts(&location, opts)
1958+
.await
1959+
.unwrap();
1960+
upload.put_part("part1".into()).await.unwrap();
1961+
upload.put_part("part2".into()).await.unwrap();
1962+
upload.complete().await.unwrap();
1963+
1964+
let result = integration.get(&location).await.unwrap();
1965+
assert_eq!(result.attributes, attributes);
1966+
1967+
let bytes = result.bytes().await.unwrap();
1968+
assert_eq!(bytes.as_ref(), b"part1part2");
1969+
}
18131970
}

0 commit comments

Comments
 (0)