Struct S3ObjectStore

pub struct S3ObjectStore {
    client: Client,
    bucket: String,
    metrics: Arc<ObjectStoreMetrics>,
    config: Arc<ObjectStoreConfig>,
}

Object store with S3 backend The full path to a file on S3 would be s3://bucket/<data_directory>/prefix/file

Fields

client: Client

bucket: String

metrics: Arc<ObjectStoreMetrics>

For S3 specific metrics.

config: Arc<ObjectStoreConfig>

Implementations

impl S3ObjectStore

pub fn new_http_client(config: &ObjectStoreConfig) -> impl HttpClient

pub async fn new_with_config( bucket: String, metrics: Arc<ObjectStoreMetrics>, config: Arc<ObjectStoreConfig>, ) -> Self

Creates an S3 object store from environment variable.

See AWS Docs on how to provide credentials and region from env variable. If you are running compute-node on EC2, no configuration is required.

pub async fn new_minio_engine( server: &str, metrics: Arc<ObjectStoreMetrics>, object_store_config: Arc<ObjectStoreConfig>, ) -> Self

Creates a minio client. The server should be like minio://key:secret@address:port/bucket.

fn obj_store_request( &self, path: &str, range: impl ObjectRangeBounds, ) -> GetObjectFluentBuilder

Generates an HTTP GET request to download the object specified in path. If given, start_pos and end_pos specify the first and last byte to download, respectively. Both are inclusive and 0-based. For example, set start_pos = 0 and end_pos = 7 to download the first 8 bytes. If neither is given, the request will download the whole object.

pub async fn configure_bucket_lifecycle(&self, data_directory: &str) -> bool

Note: This configuration only works for S3. MinIO automatically enables this feature, and it is not configurable with S3 sdk. To verify that this feature is enabled, use mc admin config get <alias> api.

Reference:

• S3
  • https://docs.aws.amazon.com/AmazonS3/latest/API/API_AbortMultipartUpload.html
  • https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html
• MinIO
  • https://github.com/minio/minio/issues/15681#issuecomment-1245126561

Trait Implementations

impl Clone for S3ObjectStore

fn clone(&self) -> S3ObjectStore

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl ObjectStore for S3ObjectStore

fn read<'life0, 'life1, 'async_trait>( &'life0 self, path: &'life1 str, range: impl 'async_trait + ObjectRangeBounds, ) -> Pin<Box<dyn Future<Output = ObjectResult<Bytes>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Amazon S3 doesn’t support retrieving multiple ranges of data per GET request.

fn streaming_read<'life0, 'life1, 'async_trait>( &'life0 self, path: &'life1 str, range: Range<usize>, ) -> Pin<Box<dyn Future<Output = ObjectResult<ObjectDataStream>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Returns a stream reading the object specified in path. If given, the stream starts at the byte with index start_pos (0-based). As far as possible, the stream only loads the amount of data into memory that is read from the stream.

fn delete<'life0, 'life1, 'async_trait>( &'life0 self, path: &'life1 str, ) -> Pin<Box<dyn Future<Output = ObjectResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Permanently deletes the whole object. According to Amazon S3, this will simply return Ok if the object does not exist.

fn delete_objects<'life0, 'life1, 'async_trait>( &'life0 self, paths: &'life1 [String], ) -> Pin<Box<dyn Future<Output = ObjectResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Deletes the objects with the given paths permanently from the storage. If an object specified in the request is not found, it will be considered as successfully deleted.

Uses AWS’ DeleteObjects API. See AWS Docs for more details.

type StreamingUploader = S3StreamingUploader

fn get_object_prefix( &self, obj_id: u64, _use_new_object_prefix_strategy: bool, ) -> String

Get the key prefix for object, the prefix is determined by the type of object store and devise_object_prefix.

fn upload<'life0, 'life1, 'async_trait>( &'life0 self, path: &'life1 str, obj: Bytes, ) -> Pin<Box<dyn Future<Output = ObjectResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Uploads the object to ObjectStore.

fn streaming_upload<'life0, 'life1, 'async_trait>( &'life0 self, path: &'life1 str, ) -> Pin<Box<dyn Future<Output = ObjectResult<Self::StreamingUploader>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn metadata<'life0, 'life1, 'async_trait>( &'life0 self, path: &'life1 str, ) -> Pin<Box<dyn Future<Output = ObjectResult<ObjectMetadata>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Obtains the object metadata.

fn list<'life0, 'life1, 'async_trait>( &'life0 self, prefix: &'life1 str, start_after: Option<String>, limit: Option<usize>, ) -> Pin<Box<dyn Future<Output = ObjectResult<ObjectMetadataIter>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn store_media_type(&self) -> &'static str

fn monitored( self, metrics: Arc<ObjectStoreMetrics>, config: Arc<ObjectStoreConfig>, ) -> MonitoredObjectStore<Self>

where Self: Sized,

fn support_streaming_upload(&self) -> bool