opendal_service_alluxio/
core.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19
20use bytes::Buf;
21use http::Request;
22use http::Response;
23use http::StatusCode;
24use serde::Deserialize;
25use serde::Serialize;
26
27use opendal_core::raw::*;
28use opendal_core::*;
29
30/// Alluxio core
31#[derive(Clone)]
32pub struct AlluxioCore {
33    pub info: ServiceInfo,
34    pub capability: Capability,
35    /// root of this backend.
36    pub root: String,
37    /// endpoint of alluxio
38    pub endpoint: String,
39}
40
41impl Debug for AlluxioCore {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.debug_struct("AlluxioCore")
44            .field("root", &self.root)
45            .field("endpoint", &self.endpoint)
46            .finish_non_exhaustive()
47    }
48}
49
50impl AlluxioCore {
51    pub async fn create_dir(&self, ctx: &OperationContext, path: &str) -> Result<()> {
52        let path = build_rooted_abs_path(&self.root, path);
53
54        let r = CreateDirRequest {
55            recursive: Some(true),
56            allow_exists: Some(true),
57        };
58
59        let body = serde_json::to_vec(&r).map_err(new_json_serialize_error)?;
60        let body = bytes::Bytes::from(body);
61
62        let mut req = Request::post(format!(
63            "{}/api/v1/paths/{}/create-directory",
64            self.endpoint,
65            percent_encode_path(&path)
66        ));
67
68        req = req.header("Content-Type", "application/json");
69
70        let req = req
71            .extension(Operation::CreateDir)
72            .extension(ServiceOperation("CreateDirectory"));
73
74        let req = req
75            .body(Buffer::from(body))
76            .map_err(new_request_build_error)?;
77
78        let resp = ctx.http_transport().send(req).await?;
79
80        let status = resp.status();
81        match status {
82            StatusCode::OK => Ok(()),
83            _ => Err(parse_error(resp)),
84        }
85    }
86
87    pub async fn create_file(&self, ctx: &OperationContext, path: &str) -> Result<u64> {
88        let path = build_rooted_abs_path(&self.root, path);
89
90        let r = CreateFileRequest {
91            recursive: Some(true),
92        };
93
94        let body = serde_json::to_vec(&r).map_err(new_json_serialize_error)?;
95        let body = bytes::Bytes::from(body);
96        let mut req = Request::post(format!(
97            "{}/api/v1/paths/{}/create-file",
98            self.endpoint,
99            percent_encode_path(&path)
100        ));
101
102        req = req.header("Content-Type", "application/json");
103
104        let req = req
105            .extension(Operation::Write)
106            .extension(ServiceOperation("CreateFile"));
107
108        let req = req
109            .body(Buffer::from(body))
110            .map_err(new_request_build_error)?;
111
112        let resp = ctx.http_transport().send(req).await?;
113        let status = resp.status();
114
115        match status {
116            StatusCode::OK => {
117                let body = resp.into_body();
118                let steam_id: u64 =
119                    serde_json::from_reader(body.reader()).map_err(new_json_serialize_error)?;
120                Ok(steam_id)
121            }
122            _ => Err(parse_error(resp)),
123        }
124    }
125
126    pub(super) async fn open_file(&self, ctx: &OperationContext, path: &str) -> Result<u64> {
127        let path = build_rooted_abs_path(&self.root, path);
128
129        let req = Request::post(format!(
130            "{}/api/v1/paths/{}/open-file",
131            self.endpoint,
132            percent_encode_path(&path)
133        ));
134
135        let req = req
136            .extension(Operation::Read)
137            .extension(ServiceOperation("OpenFile"));
138
139        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
140        let resp = ctx.http_transport().send(req).await?;
141
142        let status = resp.status();
143
144        match status {
145            StatusCode::OK => {
146                let body = resp.into_body();
147                let steam_id: u64 =
148                    serde_json::from_reader(body.reader()).map_err(new_json_serialize_error)?;
149                Ok(steam_id)
150            }
151            _ => Err(parse_error(resp)),
152        }
153    }
154
155    pub(super) async fn delete(&self, ctx: &OperationContext, path: &str) -> Result<()> {
156        let path = build_rooted_abs_path(&self.root, path);
157
158        let req = Request::post(format!(
159            "{}/api/v1/paths/{}/delete",
160            self.endpoint,
161            percent_encode_path(&path)
162        ));
163
164        let req = req
165            .extension(Operation::Delete)
166            .extension(ServiceOperation("Delete"));
167
168        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
169        let resp = ctx.http_transport().send(req).await?;
170
171        let status = resp.status();
172
173        match status {
174            StatusCode::OK => Ok(()),
175            _ => {
176                let err = parse_error(resp);
177                if err.kind() == ErrorKind::NotFound {
178                    return Ok(());
179                }
180                Err(err)
181            }
182        }
183    }
184
185    pub(super) async fn rename(&self, ctx: &OperationContext, path: &str, dst: &str) -> Result<()> {
186        let path = build_rooted_abs_path(&self.root, path);
187        let dst = build_rooted_abs_path(&self.root, dst);
188
189        let req = Request::post(format!(
190            "{}/api/v1/paths/{}/rename?dst={}",
191            self.endpoint,
192            percent_encode_path(&path),
193            percent_encode_path(&dst)
194        ));
195
196        let req = req
197            .extension(Operation::Rename)
198            .extension(ServiceOperation("Rename"));
199
200        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
201
202        let resp = ctx.http_transport().send(req).await?;
203
204        let status = resp.status();
205
206        match status {
207            StatusCode::OK => Ok(()),
208            _ => Err(parse_error(resp)),
209        }
210    }
211
212    pub(super) async fn get_status(&self, ctx: &OperationContext, path: &str) -> Result<FileInfo> {
213        let path = build_rooted_abs_path(&self.root, path);
214
215        let req = Request::post(format!(
216            "{}/api/v1/paths/{}/get-status",
217            self.endpoint,
218            percent_encode_path(&path)
219        ));
220
221        let req = req
222            .extension(Operation::Stat)
223            .extension(ServiceOperation("GetStatus"));
224
225        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
226
227        let resp = ctx.http_transport().send(req).await?;
228
229        let status = resp.status();
230
231        match status {
232            StatusCode::OK => {
233                let body = resp.into_body();
234                let file_info: FileInfo =
235                    serde_json::from_reader(body.reader()).map_err(new_json_serialize_error)?;
236                Ok(file_info)
237            }
238            _ => Err(parse_error(resp)),
239        }
240    }
241
242    pub(super) async fn list_status(
243        &self,
244        ctx: &OperationContext,
245        path: &str,
246    ) -> Result<Vec<FileInfo>> {
247        let path = build_rooted_abs_path(&self.root, path);
248
249        let req = Request::post(format!(
250            "{}/api/v1/paths/{}/list-status",
251            self.endpoint,
252            percent_encode_path(&path)
253        ));
254
255        let req = req
256            .extension(Operation::List)
257            .extension(ServiceOperation("ListStatus"));
258
259        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
260
261        let resp = ctx.http_transport().send(req).await?;
262
263        let status = resp.status();
264
265        match status {
266            StatusCode::OK => {
267                let body = resp.into_body();
268                let file_infos: Vec<FileInfo> =
269                    serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
270                Ok(file_infos)
271            }
272            _ => Err(parse_error(resp)),
273        }
274    }
275
276    pub async fn read(
277        &self,
278        ctx: &OperationContext,
279        stream_id: u64,
280        range: BytesRange,
281    ) -> Result<Response<HttpBody>> {
282        if !range.is_full() {
283            return Err(Error::new(
284                ErrorKind::Unsupported,
285                "alluxio stream read doesn't support range",
286            )
287            .with_context("range", format!("{range:?}")));
288        }
289
290        let req = Request::post(format!(
291            "{}/api/v1/streams/{}/read",
292            self.endpoint, stream_id,
293        ));
294
295        let req = req
296            .extension(Operation::Read)
297            .extension(ServiceOperation("ReadStream"));
298
299        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
300
301        ctx.http_transport().fetch(req).await
302    }
303
304    pub(super) async fn write(
305        &self,
306        ctx: &OperationContext,
307        stream_id: u64,
308        body: Buffer,
309    ) -> Result<usize> {
310        let req = Request::post(format!(
311            "{}/api/v1/streams/{}/write",
312            self.endpoint, stream_id
313        ));
314
315        let req = req
316            .extension(Operation::Write)
317            .extension(ServiceOperation("WriteStream"));
318
319        let req = req.body(body).map_err(new_request_build_error)?;
320
321        let resp = ctx.http_transport().send(req).await?;
322
323        let status = resp.status();
324
325        match status {
326            StatusCode::OK => {
327                let body = resp.into_body();
328                let size: usize =
329                    serde_json::from_reader(body.reader()).map_err(new_json_serialize_error)?;
330                Ok(size)
331            }
332            _ => Err(parse_error(resp)),
333        }
334    }
335
336    pub(super) async fn close(&self, ctx: &OperationContext, stream_id: u64) -> Result<()> {
337        let req = Request::post(format!(
338            "{}/api/v1/streams/{}/close",
339            self.endpoint, stream_id
340        ));
341
342        let req = req
343            .extension(Operation::Write)
344            .extension(ServiceOperation("CloseStream"));
345
346        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
347
348        let resp = ctx.http_transport().send(req).await?;
349
350        let status = resp.status();
351
352        match status {
353            StatusCode::OK => Ok(()),
354            _ => Err(parse_error(resp)),
355        }
356    }
357}
358
359#[cfg(test)]
360mod tests {
361    use super::*;
362
363    #[tokio::test]
364    async fn test_read_rejects_range() {
365        let core = AlluxioCore {
366            info: ServiceInfo::new("alluxio", "", ""),
367            capability: Capability::default(),
368            root: "/".to_string(),
369            endpoint: "http://127.0.0.1:1".to_string(),
370        };
371
372        let ctx = OperationContext::new();
373        let err = match core.read(&ctx, 1, BytesRange::from(0_u64..1)).await {
374            Ok(_) => panic!("range read should be rejected"),
375            Err(err) => err,
376        };
377
378        assert_eq!(err.kind(), ErrorKind::Unsupported);
379    }
380}
381
382#[derive(Debug, Serialize)]
383struct CreateFileRequest {
384    #[serde(skip_serializing_if = "Option::is_none")]
385    recursive: Option<bool>,
386}
387
388#[derive(Debug, Serialize)]
389#[serde(rename_all = "camelCase")]
390struct CreateDirRequest {
391    #[serde(skip_serializing_if = "Option::is_none")]
392    recursive: Option<bool>,
393    #[serde(skip_serializing_if = "Option::is_none")]
394    allow_exists: Option<bool>,
395}
396
397/// Metadata of alluxio object
398#[derive(Debug, Deserialize)]
399#[serde(rename_all = "camelCase")]
400pub(super) struct FileInfo {
401    /// The path of the object
402    pub path: String,
403    /// The last modification time of the object
404    pub last_modification_time_ms: i64,
405    /// Whether the object is a folder
406    pub folder: bool,
407    /// The length of the object in bytes
408    pub length: u64,
409}
410
411impl TryFrom<FileInfo> for Metadata {
412    type Error = Error;
413
414    fn try_from(file_info: FileInfo) -> Result<Metadata> {
415        let mut metadata = if file_info.folder {
416            Metadata::new(EntryMode::DIR)
417        } else {
418            Metadata::new(EntryMode::FILE)
419        };
420        metadata
421            .set_content_length(file_info.length)
422            .set_last_modified(Timestamp::from_millisecond(
423                file_info.last_modification_time_ms,
424            )?);
425        Ok(metadata)
426    }
427}
428
429mod error {
430    use bytes::Buf;
431    use http::Response;
432    use serde::Deserialize;
433
434    use opendal_core::raw::*;
435    use opendal_core::*;
436
437    /// the error response of alluxio
438    #[derive(Default, Debug, Deserialize)]
439    #[serde(rename_all = "camelCase")]
440    #[allow(dead_code)]
441    struct AlluxioError {
442        status_code: String,
443        message: String,
444    }
445
446    pub(crate) fn parse_error(resp: Response<Buffer>) -> Error {
447        let (parts, body) = resp.into_parts();
448        let bs = body.to_bytes();
449
450        let mut kind = match parts.status.as_u16() {
451            500 => ErrorKind::Unexpected,
452            _ => ErrorKind::Unexpected,
453        };
454
455        let (message, alluxio_err) =
456            serde_json::from_reader::<_, AlluxioError>(bs.clone().reader())
457                .map(|alluxio_err| (format!("{alluxio_err:?}"), Some(alluxio_err)))
458                .unwrap_or_else(|_| (String::from_utf8_lossy(&bs).into_owned(), None));
459
460        if let Some(alluxio_err) = alluxio_err {
461            kind = match alluxio_err.status_code.as_str() {
462                "ALREADY_EXISTS" => ErrorKind::AlreadyExists,
463                "NOT_FOUND" => ErrorKind::NotFound,
464                _ => ErrorKind::Unexpected,
465            }
466        }
467
468        let mut err = Error::new(kind, message);
469
470        err = with_error_response_context(err, parts);
471
472        err
473    }
474
475    #[cfg(test)]
476    mod tests {
477        use http::StatusCode;
478
479        use super::*;
480
481        /// Error response example is from https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
482        #[test]
483        fn test_parse_error() {
484            let err_res = vec![
485                (
486                    r#"{"statusCode":"ALREADY_EXISTS","message":"The resource you requested already exist"}"#,
487                    ErrorKind::AlreadyExists,
488                ),
489                (
490                    r#"{"statusCode":"NOT_FOUND","message":"The resource you requested does not exist"}"#,
491                    ErrorKind::NotFound,
492                ),
493                (
494                    r#"{"statusCode":"INTERNAL_SERVER_ERROR","message":"Internal server error"}"#,
495                    ErrorKind::Unexpected,
496                ),
497            ];
498
499            for res in err_res {
500                let bs = bytes::Bytes::from(res.0);
501                let body = Buffer::from(bs);
502                let resp = Response::builder()
503                    .status(StatusCode::INTERNAL_SERVER_ERROR)
504                    .body(body)
505                    .unwrap();
506
507                let err = parse_error(resp);
508
509                assert_eq!(err.kind(), res.1);
510            }
511        }
512    }
513}
514
515pub(super) use error::*;
opendal_service_alluxio/core.rs

opendal_service_alluxio/
core.rs
