From 78f2d86fc8be8e923d91f069406acd48edcae180 Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Fri, 22 Dec 2023 21:52:20 +0100 Subject: WIP k2v --- src/storage/garage.rs | 118 +++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 98 insertions(+), 20 deletions(-) (limited to 'src/storage') diff --git a/src/storage/garage.rs b/src/storage/garage.rs index ec26e80..bc123b5 100644 --- a/src/storage/garage.rs +++ b/src/storage/garage.rs @@ -34,7 +34,7 @@ impl GarageBuilder { #[async_trait] impl IBuilder for GarageBuilder { async fn build(&self) -> Result { - let creds = s3::config::Credentials::new( + let s3_creds = s3::config::Credentials::new( self.conf.aws_access_key_id.clone(), self.conf.aws_secret_access_key.clone(), None, @@ -42,17 +42,35 @@ impl IBuilder for GarageBuilder { "aerogramme" ); - let config = aws_config::from_env() + let s3_config = aws_config::from_env() .region(aws_config::Region::new(self.conf.region.clone())) - .credentials_provider(creds) + .credentials_provider(s3_creds) .endpoint_url(self.conf.s3_endpoint.clone()) .load() .await; + let s3_client = aws_sdk_s3::Client::new(&s3_config); + + let k2v_config = k2v_client::K2vClientConfig { + endpoint: self.conf.k2v_endpoint.clone(), + region: self.conf.region.clone(), + aws_access_key_id: self.conf.aws_access_key_id.clone(), + aws_secret_access_key: self.conf.aws_secret_access_key.clone(), + bucket: self.conf.bucket.clone(), + user_agent: None, + }; + + let k2v_client = match k2v_client::K2vClient::new(k2v_config) { + Err(e) => { + tracing::error!("unable to build k2v client: {}", e); + return Err(StorageError::Internal); + } + Ok(v) => v, + }; - let s3_client = aws_sdk_s3::Client::new(&config); Ok(Box::new(GarageStore { - s3_bucket: self.conf.bucket.clone(), - s3: s3_client + bucket: self.conf.bucket.clone(), + s3: s3_client, + k2v: k2v_client, })) } fn unique(&self) -> UnicityBuffer { @@ -61,13 +79,73 @@ impl IBuilder for GarageBuilder { } pub struct GarageStore { - s3_bucket: String, + bucket: String, s3: s3::Client, + k2v: k2v_client::K2vClient, +} + +fn causal_to_row_val(row_ref: RowRef, causal_value: k2v_client::CausalValue) -> RowVal { + let new_row_ref = row_ref.with_causality(causal_value.causality.into()); + let row_values = causal_value.value.into_iter().map(|k2v_value| match k2v_value { + k2v_client::K2vValue::Tombstone => Alternative::Tombstone, + k2v_client::K2vValue::Value(v) => Alternative::Value(v), + }).collect::>(); + + RowVal { row_ref: new_row_ref, value: row_values } } #[async_trait] impl IStore for GarageStore { async fn row_fetch<'a>(&self, select: &Selector<'a>) -> Result, StorageError> { + let batch_op = match select { + Selector::Range { shard, sort_begin, sort_end } => vec![k2v_client::BatchReadOp { + partition_key: shard, + filter: k2v_client::Filter { + start: Some(sort_begin), + end: Some(sort_end), + ..k2v_client::Filter::default() + }, + ..k2v_client::BatchReadOp::default() + }], + Selector::List(row_ref_list) => row_ref_list.iter().map(|row_ref| k2v_client::BatchReadOp { + partition_key: &row_ref.uid.shard, + filter: k2v_client::Filter { + start: Some(&row_ref.uid.sort), + ..k2v_client::Filter::default() + }, + single_item: true, + ..k2v_client::BatchReadOp::default() + }).collect::>(), + Selector::Prefix { shard, sort_prefix } => vec![k2v_client::BatchReadOp { + partition_key: shard, + filter: k2v_client::Filter { + prefix: Some(sort_prefix), + ..k2v_client::Filter::default() + }, + ..k2v_client::BatchReadOp::default() + }], + Selector::Single(row_ref) => { + let causal_value = match self.k2v.read_item(&row_ref.uid.shard, &row_ref.uid.sort).await { + Err(e) => { + tracing::error!("K2V read item shard={}, sort={}, bucket={} failed: {}", row_ref.uid.shard, row_ref.uid.sort, self.bucket, e); + return Err(StorageError::Internal); + }, + Ok(v) => v, + }; + + let row_val = causal_to_row_val((*row_ref).clone(), causal_value); + return Ok(vec![row_val]) + }, + }; + + let all_res = match self.k2v.read_batch(&batch_op).await { + Err(e) => { + tracing::error!("k2v read batch failed for {:?}, bucket {} with err: {}", select, self.bucket, e); + return Err(StorageError::Internal); + }, + Ok(v) => v, + }; + unimplemented!(); } async fn row_rm<'a>(&self, select: &Selector<'a>) -> Result<(), StorageError> { @@ -89,7 +167,7 @@ impl IStore for GarageStore { async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result { let maybe_out = self.s3 .get_object() - .bucket(self.s3_bucket.to_string()) + .bucket(self.bucket.to_string()) .key(blob_ref.0.to_string()) .send() .await; @@ -117,7 +195,7 @@ impl IStore for GarageStore { } }; - tracing::debug!("Fetched {}/{}", self.s3_bucket, blob_ref.0); + tracing::debug!("Fetched {}/{}", self.bucket, blob_ref.0); Ok(BlobVal::new(blob_ref.clone(), buffer)) } async fn blob_insert(&self, blob_val: BlobVal) -> Result<(), StorageError> { @@ -125,7 +203,7 @@ impl IStore for GarageStore { let maybe_send = self.s3 .put_object() - .bucket(self.s3_bucket.to_string()) + .bucket(self.bucket.to_string()) .key(blob_val.blob_ref.0.to_string()) .body(streamable_value) .send() @@ -137,7 +215,7 @@ impl IStore for GarageStore { Err(StorageError::Internal) } Ok(_) => { - tracing::debug!("Inserted {}/{}", self.s3_bucket, blob_val.blob_ref.0); + tracing::debug!("Inserted {}/{}", self.bucket, blob_val.blob_ref.0); Ok(()) } } @@ -145,19 +223,19 @@ impl IStore for GarageStore { async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<(), StorageError> { let maybe_copy = self.s3 .copy_object() - .bucket(self.s3_bucket.to_string()) + .bucket(self.bucket.to_string()) .key(dst.0.clone()) - .copy_source(format!("/{}/{}", self.s3_bucket.to_string(), src.0.clone())) + .copy_source(format!("/{}/{}", self.bucket.to_string(), src.0.clone())) .send() .await; match maybe_copy { Err(e) => { - tracing::error!("unable to copy object {} to {} (bucket: {}), error: {}", src.0, dst.0, self.s3_bucket, e); + tracing::error!("unable to copy object {} to {} (bucket: {}), error: {}", src.0, dst.0, self.bucket, e); Err(StorageError::Internal) }, Ok(_) => { - tracing::debug!("copied {} to {} (bucket: {})", src.0, dst.0, self.s3_bucket); + tracing::debug!("copied {} to {} (bucket: {})", src.0, dst.0, self.bucket); Ok(()) } } @@ -166,7 +244,7 @@ impl IStore for GarageStore { async fn blob_list(&self, prefix: &str) -> Result, StorageError> { let maybe_list = self.s3 .list_objects_v2() - .bucket(self.s3_bucket.to_string()) + .bucket(self.bucket.to_string()) .prefix(prefix) .into_paginator() .send() @@ -175,7 +253,7 @@ impl IStore for GarageStore { match maybe_list { Err(e) => { - tracing::error!("listing prefix {} on bucket {} failed: {}", prefix, self.s3_bucket, e); + tracing::error!("listing prefix {} on bucket {} failed: {}", prefix, self.bucket, e); Err(StorageError::Internal) } Ok(pagin_list_out) => Ok(pagin_list_out @@ -189,18 +267,18 @@ impl IStore for GarageStore { async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError> { let maybe_delete = self.s3 .delete_object() - .bucket(self.s3_bucket.to_string()) + .bucket(self.bucket.to_string()) .key(blob_ref.0.clone()) .send() .await; match maybe_delete { Err(e) => { - tracing::error!("unable to delete {} (bucket: {}), error {}", blob_ref.0, self.s3_bucket, e); + tracing::error!("unable to delete {} (bucket: {}), error {}", blob_ref.0, self.bucket, e); Err(StorageError::Internal) }, Ok(_) => { - tracing::debug!("deleted {} (bucket: {})", blob_ref.0, self.s3_bucket); + tracing::debug!("deleted {} (bucket: {})", blob_ref.0, self.bucket); Ok(()) } } -- cgit v1.2.3