Skip to content

Commit bca231a

Browse files
St4NNiSt4NNi
authored
merge: Merge pull request #198 from arunaengine/feat/list_objects
Backwards compatibility for ListObjects
2 parents 9ec7fd0 + f3441f1 commit bca231a

6 files changed

Lines changed: 153 additions & 17 deletions

File tree

‎components/data_proxy/src/caching/cache.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -894,7 +894,7 @@ impl Cache {
894894
};
895895

896896
self.multi_parts
897-
.retain(|_, v| v.first().map_or(true, |e| e.object_id != id));
897+
.retain(|_, v| v.first().is_none_or(|e| e.object_id != id));
898898

899899
let object = old.1 .0.read().await;
900900
for p in self

‎components/data_proxy/src/data_backends/s3_backend.rs‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@ use http_body_util::StreamBody;
2525
use hyper::body::Frame;
2626
use rand::random;
2727
use tracing::error;
28-
use tracing::trace;
2928

3029
#[allow(dead_code)]
3130
#[derive(Debug, Clone)]

‎components/data_proxy/src/replication/replication_handler.rs‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -516,13 +516,12 @@ impl ReplicationHandler {
516516
continue;
517517
} else {
518518
let parent = {
519-
let mut current_version_id = object_id.clone();
519+
let mut current_version_id = object_id;
520520
if let Some(version) = &object.versions {
521-
version.iter().next().map(|v| match v {
522-
VersionVariant::IsVersion(id) => {
523-
current_version_id = id.clone()
521+
version.iter().next().map(|v| {
522+
if let VersionVariant::IsVersion(id) = v {
523+
current_version_id = *id
524524
}
525-
_ => {}
526525
});
527526
}
528527
cache.get_single_parent(&current_version_id).await.map_err(

‎components/data_proxy/src/s3_frontend/data_handler.rs‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -71,19 +71,20 @@ impl DataHandler {
7171
let parents = if let Some(levels) = path_level {
7272
levels
7373
} else {
74-
let mut object_id = object.id.clone();
74+
let mut object_id = object.id;
7575
if let Some(version) = &object.versions {
76-
version.iter().next().map(|v| match v {
77-
VersionVariant::IsVersion(id) => object_id = id.clone(),
78-
_ => {}
76+
version.iter().next().map(|v| {
77+
if let VersionVariant::IsVersion(id) = v {
78+
object_id = *id
79+
}
7980
});
8081
}
8182
cache.get_single_parent(&object_id).await.map_err(|e| {
8283
error!(error = ?e, msg = e.to_string());
8384
e
8485
})?
8586
};
86-
if parents.get(0).is_none() {
87+
if parents.is_empty() {
8788
error!(?parents, "No parent found");
8889
return Err(anyhow!("No parent found"));
8990
}

‎components/data_proxy/src/s3_frontend/s3service.rs‎

Lines changed: 121 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -181,7 +181,7 @@ impl S3 for ArunaS3Service {
181181
})?;
182182

183183
let response = CompleteMultipartUploadOutput {
184-
e_tag: Some(format!("-{}", object.id.to_string())),
184+
e_tag: Some(format!("-{}", object.id)),
185185
..Default::default()
186186
};
187187

@@ -812,6 +812,125 @@ impl S3 for ArunaS3Service {
812812
Ok(resp)
813813
}
814814

815+
#[tracing::instrument(err)]
816+
#[allow(clippy::blocks_in_conditions)]
817+
async fn list_objects(
818+
&self,
819+
req: S3Request<ListObjectsInput>,
820+
) -> S3Result<S3Response<ListObjectsOutput>> {
821+
let CheckAccessResult { headers, .. } = req
822+
.extensions
823+
.get::<CheckAccessResult>()
824+
.cloned()
825+
.ok_or_else(|| {
826+
error!(error = "No context found");
827+
s3_error!(InternalError, "No context found")
828+
})?;
829+
// Fetch the project name, delimiter and prefix from the request
830+
let project_name = &req.input.bucket;
831+
let delimiter = req.input.delimiter;
832+
let prefix = req.input.prefix.filter(|prefix| !prefix.is_empty());
833+
834+
// Check if bucket exists as root in cache of paths
835+
match self.cache.get_path(project_name.as_str()) {
836+
Some(_) => {}
837+
None => {
838+
error!("No bucket found");
839+
return Err(s3_error!(NoSuchBucket, "No bucket found"));
840+
}
841+
};
842+
843+
let max_keys = match req.input.max_keys {
844+
Some(k) if k < 1000 => k as usize,
845+
_ => 1000usize,
846+
};
847+
848+
let start_at = match &req.input.marker {
849+
Some(marker) => marker,
850+
None => "",
851+
};
852+
853+
let (keys, common_prefixes, next_marker) = list_response(
854+
&self.cache,
855+
&delimiter,
856+
&prefix,
857+
project_name,
858+
start_at,
859+
max_keys,
860+
true,
861+
)
862+
.await
863+
.map_err(|_| {
864+
error!(error = "Keys not found in ListObjects");
865+
s3_error!(NoSuchKey, "Keys not found in ListObjects")
866+
})?;
867+
868+
let common_prefixes = Some(
869+
common_prefixes
870+
.into_iter()
871+
.map(|e| CommonPrefix { prefix: Some(e) })
872+
.collect(),
873+
);
874+
let contents = Some(
875+
keys.into_iter()
876+
.map(|e| Object {
877+
checksum_algorithm: None,
878+
e_tag: Some(e.etag.to_string()),
879+
key: Some(e.key),
880+
last_modified: e.created_at.map(|t| {
881+
s3s::dto::Timestamp::from(
882+
time::OffsetDateTime::from_unix_timestamp(t.and_utc().timestamp())
883+
.unwrap_or_else(|_| {
884+
error!(error = "Unable to parse timestamp");
885+
time::OffsetDateTime::now_utc()
886+
}),
887+
)
888+
}),
889+
owner: None,
890+
size: Some(e.size),
891+
..Default::default()
892+
})
893+
.collect(),
894+
);
895+
896+
let result = ListObjectsOutput {
897+
common_prefixes,
898+
contents,
899+
marker: req.input.marker,
900+
delimiter,
901+
encoding_type: None,
902+
is_truncated: Some(next_marker.is_some()),
903+
max_keys: Some(max_keys.try_into().map_err(|err| {
904+
error!(error = ?err, "Conversion failure");
905+
s3_error!(InternalError, "[BACKEND] Conversion failure: {}", err)
906+
})?),
907+
name: Some(project_name.clone()),
908+
next_marker,
909+
prefix,
910+
..Default::default()
911+
};
912+
debug!(?result);
913+
914+
let mut resp = S3Response::new(result);
915+
916+
if let Some(headers) = headers {
917+
for (k, v) in headers {
918+
resp.headers.insert(
919+
HeaderName::from_bytes(k.as_bytes()).map_err(|_| {
920+
error!(error = "Unable to parse header name");
921+
s3_error!(InternalError, "Unable to parse header name")
922+
})?,
923+
HeaderValue::from_str(&v).map_err(|_| {
924+
error!(error = "Unable to parse header value");
925+
s3_error!(InternalError, "Unable to parse header value")
926+
})?,
927+
);
928+
}
929+
}
930+
931+
Ok(resp)
932+
}
933+
815934
#[tracing::instrument(err)]
816935
#[allow(clippy::blocks_in_conditions)]
817936
async fn list_objects_v2(
@@ -878,6 +997,7 @@ impl S3 for ArunaS3Service {
878997
project_name,
879998
&start_after,
880999
max_keys,
1000+
false,
8811001
)
8821002
.await
8831003
.map_err(|_| {

‎components/data_proxy/src/s3_frontend/utils/list_objects.rs‎

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ pub async fn list_response(
4242
bucket_name: &str,
4343
start_at: &str,
4444
max_keys: usize,
45+
v1: bool, // If true, returns raw path marker instead of continuation token
4546
) -> Result<(BTreeSet<Contents>, BTreeSet<String>, Option<String>)> {
4647
let mut keys: BTreeSet<Contents> = BTreeSet::default();
4748
let mut common_prefixes: BTreeSet<String> = BTreeSet::default();
@@ -53,7 +54,11 @@ pub async fn list_response(
5354
// Breaks with next path to start at after max_keys is reached
5455
let num_keys = keys.len() + common_prefixes.len();
5556
if num_keys == max_keys {
56-
new_continuation_token = Some(general_purpose::STANDARD_NO_PAD.encode(path));
57+
new_continuation_token = if v1 {
58+
Some(path)
59+
} else {
60+
Some(general_purpose::STANDARD_NO_PAD.encode(path))
61+
};
5762
break;
5863
}
5964

@@ -85,7 +90,11 @@ pub async fn list_response(
8590
// Breaks with next path to start at after max_keys is reached
8691
let num_keys = keys.len() + common_prefixes.len();
8792
if num_keys == max_keys {
88-
new_continuation_token = Some(general_purpose::STANDARD_NO_PAD.encode(path));
93+
new_continuation_token = if v1 {
94+
Some(path)
95+
} else {
96+
Some(general_purpose::STANDARD_NO_PAD.encode(path))
97+
};
8998
break;
9099
}
91100

@@ -111,7 +120,11 @@ pub async fn list_response(
111120
// Breaks with next path to start at after max_keys is reached
112121
let num_keys = keys.len() + common_prefixes.len();
113122
if num_keys == max_keys {
114-
new_continuation_token = Some(general_purpose::STANDARD_NO_PAD.encode(path));
123+
new_continuation_token = if v1 {
124+
Some(path)
125+
} else {
126+
Some(general_purpose::STANDARD_NO_PAD.encode(path))
127+
};
115128
break;
116129
}
117130

@@ -135,7 +148,11 @@ pub async fn list_response(
135148
// Breaks with next path to start at after max_keys is reached
136149
let num_keys = keys.len() + common_prefixes.len();
137150
if num_keys == max_keys {
138-
new_continuation_token = Some(general_purpose::STANDARD_NO_PAD.encode(path));
151+
new_continuation_token = if v1 {
152+
Some(path)
153+
} else {
154+
Some(general_purpose::STANDARD_NO_PAD.encode(path))
155+
};
139156
break;
140157
}
141158

0 commit comments

Comments
 (0)