Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

feat: Use ObjectStorePath everywhere to feel out the API needed

pull/24376/head
Carol (Nichols || Goulding) 2021-01-04 16:58:15 -05:00
parent e58607f015
commit 18ee1b561b
11 changed files with 470 additions and 233 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -1994,6 +1994,7 @@ dependencies = [
"cloud-storage",
"dotenv",
"futures",
"itertools 0.9.0",
"percent-encoding",
"rusoto_core",
"rusoto_credential",

5
object_store/Cargo.toml
View File

@ -6,10 +6,11 @@ edition = "2018"
[dependencies]
bytes = "0.5.4"
futures = "0.3.5"
snafu = { version = "0.6.10", features = ["futures"] }
chrono = "0.4"
futures = "0.3.5"
itertools = "0.9.0"
percent-encoding = "2.1"
snafu = { version = "0.6.10", features = ["futures"] }
# Amazon S3 integration
rusoto_core = "0.44.0"

99
object_store/src/aws.rs
View File

@ -1,6 +1,7 @@
//! This module contains the IOx implementation for using S3 as the object
//! store.
use crate::{
path::{CloudConverter, ObjectStorePath, DELIMITER},
Error, ListResult, NoDataFromS3, ObjectMeta, Result, UnableToDeleteDataFromS3,
UnableToGetDataFromS3, UnableToGetPieceOfDataFromS3, UnableToPutDataToS3,
};
@ -14,9 +15,6 @@ use snafu::{futures::TryStreamExt as _, OptionExt, ResultExt};
use std::convert::TryFrom;
use std::{fmt, io};
// The delimiter to separate object namespaces, creating a directory structure.
const DELIMITER: &str = "/";
/// Configuration for connecting to [Amazon S3](https://aws.amazon.com/s3/).
pub struct AmazonS3 {
client: rusoto_s3::S3Client,
@ -57,7 +55,7 @@ impl AmazonS3 {
}
/// Save the provided bytes to the specified location.
pub async fn put<S>(&self, location: &str, bytes: S, length: usize) -> Result<()>
pub async fn put<S>(&self, location: &ObjectStorePath, bytes: S, length: usize) -> Result<()>
where
S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
{
@ -65,7 +63,7 @@ impl AmazonS3 {
let put_request = rusoto_s3::PutObjectRequest {
bucket: self.bucket_name.clone(),
key: location.to_string(),
key: CloudConverter::convert(&location),
body: Some(bytes),
..Default::default()
};
@ -75,16 +73,20 @@ impl AmazonS3 {
.await
.context(UnableToPutDataToS3 {
bucket: &self.bucket_name,
location: location.to_string(),
location: CloudConverter::convert(&location),
})?;
Ok(())
}
/// Return the bytes that are stored at the specified location.
pub async fn get(&self, location: &str) -> Result<impl Stream<Item = Result<Bytes>>> {
pub async fn get(
&self,
location: &ObjectStorePath,
) -> Result<impl Stream<Item = Result<Bytes>>> {
let key = CloudConverter::convert(&location);
let get_request = rusoto_s3::GetObjectRequest {
bucket: self.bucket_name.clone(),
key: location.to_string(),
key: key.clone(),
..Default::default()
};
Ok(self
@ -93,25 +95,26 @@ impl AmazonS3 {
.await
.context(UnableToGetDataFromS3 {
bucket: self.bucket_name.to_owned(),
location: location.to_owned(),
location: key.clone(),
})?
.body
.context(NoDataFromS3 {
bucket: self.bucket_name.to_owned(),
location: location.to_string(),
location: key.clone(),
})?
.context(UnableToGetPieceOfDataFromS3 {
bucket: self.bucket_name.to_owned(),
location: location.to_string(),
location: key,
})
.err_into())
}
/// Delete the object at the specified location.
pub async fn delete(&self, location: &str) -> Result<()> {
pub async fn delete(&self, location: &ObjectStorePath) -> Result<()> {
let key = CloudConverter::convert(&location);
let delete_request = rusoto_s3::DeleteObjectRequest {
bucket: self.bucket_name.clone(),
key: location.to_string(),
key: key.clone(),
..Default::default()
};
@ -120,7 +123,7 @@ impl AmazonS3 {
.await
.context(UnableToDeleteDataFromS3 {
bucket: self.bucket_name.to_owned(),
location: location.to_owned(),
location: key,
})?;
Ok(())
}
@ -128,8 +131,8 @@ impl AmazonS3 {
/// List all the objects with the given prefix.
pub async fn list<'a>(
&'a self,
prefix: Option<&'a str>,
) -> Result<impl Stream<Item = Result<Vec<String>>> + 'a> {
prefix: Option<&'a ObjectStorePath>,
) -> Result<impl Stream<Item = Result<Vec<ObjectStorePath>>> + 'a> {
#[derive(Clone)]
enum ListState {
Start,
@ -141,7 +144,7 @@ impl AmazonS3 {
Ok(stream::unfold(ListState::Start, move |state| async move {
let mut list_request = rusoto_s3::ListObjectsV2Request {
bucket: self.bucket_name.clone(),
prefix: prefix.map(ToString::to_string),
prefix: prefix.map(CloudConverter::convert),
..Default::default()
};
@ -171,7 +174,10 @@ impl AmazonS3 {
};
let contents = resp.contents.unwrap_or_default();
let names = contents.into_iter().flat_map(|object| object.key).collect();
let names = contents
.into_iter()
.flat_map(|object| object.key.map(ObjectStorePath::from_cloud_unchecked))
.collect();
// The AWS response contains a field named `is_truncated` as well as
// `next_continuation_token`, and we're assuming that `next_continuation_token`
@ -191,12 +197,15 @@ impl AmazonS3 {
/// common prefixes (directories) in addition to object metadata.
pub async fn list_with_delimiter<'a>(
&'a self,
prefix: &'a str,
prefix: &'a ObjectStorePath,
next_token: &Option<String>,
) -> Result<ListResult> {
dbg!(&prefix);
let converted_prefix = CloudConverter::convert(prefix);
dbg!(&converted_prefix);
let mut list_request = rusoto_s3::ListObjectsV2Request {
bucket: self.bucket_name.clone(),
prefix: Some(prefix.to_string()),
prefix: Some(converted_prefix),
delimiter: Some(DELIMITER.to_string()),
..Default::default()
};
@ -216,10 +225,13 @@ impl AmazonS3 {
};
let contents = resp.contents.unwrap_or_default();
dbg!(&contents);
let objects: Vec<_> = contents
.into_iter()
.map(|object| {
let location = object.key.expect("object doesn't exist without a key");
let location = ObjectStorePath::from_cloud_unchecked(
object.key.expect("object doesn't exist without a key"),
);
let last_modified = match object.last_modified {
Some(lm) => {
DateTime::parse_from_rfc3339(&lm)
@ -244,11 +256,17 @@ impl AmazonS3 {
})
.collect();
dbg!(&resp.common_prefixes);
let common_prefixes = resp
.common_prefixes
.unwrap_or_default()
.into_iter()
.map(|p| p.prefix.expect("can't have a prefix without a value"))
.map(|p| {
ObjectStorePath::from_cloud_unchecked(
p.prefix.expect("can't have a prefix without a value"),
)
})
.collect();
let result = ListResult {
@ -294,6 +312,7 @@ impl Error {
#[cfg(test)]
mod tests {
use crate::{
path::ObjectStorePath,
tests::{get_nonexistent_object, list_with_delimiter, put_get_delete_list},
AmazonS3, Error, ObjectStore,
};
@ -401,7 +420,7 @@ mod tests {
let (_, bucket_name) = region_and_bucket_name()?;
let region = rusoto_core::Region::UsWest1;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, &bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let err = get_nonexistent_object(&integration, Some(location_name))
.await
@ -423,7 +442,7 @@ mod tests {
maybe_skip_integration!();
let (region, bucket_name) = region_and_bucket_name()?;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, &bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let err = get_nonexistent_object(&integration, Some(location_name))
.await
@ -439,7 +458,7 @@ mod tests {
rusoto_core::RusotoError::Service(rusoto_s3::GetObjectError::NoSuchKey(_))
));
assert_eq!(bucket, &bucket_name);
assert_eq!(location, location_name);
assert_eq!(location, NON_EXISTENT_NAME);
} else {
panic!("unexpected error type")
}
@ -453,7 +472,7 @@ mod tests {
let (region, _) = region_and_bucket_name()?;
let bucket_name = NON_EXISTENT_NAME;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let err = get_nonexistent_object(&integration, Some(location_name))
.await
@ -480,13 +499,13 @@ mod tests {
let (_, bucket_name) = region_and_bucket_name()?;
let region = rusoto_core::Region::UsWest1;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, &bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let data = Bytes::from("arbitrary data");
let stream_data = std::io::Result::Ok(data.clone());
let err = integration
.put(
location_name,
&location_name,
futures::stream::once(async move { stream_data }),
data.len(),
)
@ -501,7 +520,7 @@ mod tests {
{
assert!(matches!(source, rusoto_core::RusotoError::Unknown(_)));
assert_eq!(bucket, bucket_name);
assert_eq!(location, location_name);
assert_eq!(location, NON_EXISTENT_NAME);
} else {
panic!("unexpected error type")
}
@ -515,13 +534,13 @@ mod tests {
let (region, _) = region_and_bucket_name()?;
let bucket_name = NON_EXISTENT_NAME;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let data = Bytes::from("arbitrary data");
let stream_data = std::io::Result::Ok(data.clone());
let err = integration
.put(
location_name,
&location_name,
futures::stream::once(async move { stream_data }),
data.len(),
)
@ -536,7 +555,7 @@ mod tests {
{
assert!(matches!(source, rusoto_core::RusotoError::Unknown(_)));
assert_eq!(bucket, bucket_name);
assert_eq!(location, location_name);
assert_eq!(location, NON_EXISTENT_NAME);
} else {
panic!("unexpected error type")
}
@ -549,9 +568,9 @@ mod tests {
maybe_skip_integration!();
let (region, bucket_name) = region_and_bucket_name()?;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, &bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let result = integration.delete(location_name).await;
let result = integration.delete(&location_name).await;
assert!(result.is_ok());
@ -565,9 +584,9 @@ mod tests {
let (_, bucket_name) = region_and_bucket_name()?;
let region = rusoto_core::Region::UsWest1;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, &bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let err = integration.delete(location_name).await.unwrap_err();
let err = integration.delete(&location_name).await.unwrap_err();
if let Error::UnableToDeleteDataFromS3 {
source,
bucket,
@ -576,7 +595,7 @@ mod tests {
{
assert!(matches!(source, rusoto_core::RusotoError::Unknown(_)));
assert_eq!(bucket, bucket_name);
assert_eq!(location, location_name);
assert_eq!(location, NON_EXISTENT_NAME);
} else {
panic!("unexpected error type")
}
@ -590,9 +609,9 @@ mod tests {
let (region, _) = region_and_bucket_name()?;
let bucket_name = NON_EXISTENT_NAME;
let integration = ObjectStore::new_amazon_s3(AmazonS3::new(region, bucket_name));
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let err = integration.delete(location_name).await.unwrap_err();
let err = integration.delete(&location_name).await.unwrap_err();
if let Error::UnableToDeleteDataFromS3 {
source,
bucket,
@ -601,7 +620,7 @@ mod tests {
{
assert!(matches!(source, rusoto_core::RusotoError::Unknown(_)));
assert_eq!(bucket, bucket_name);
assert_eq!(location, location_name);
assert_eq!(location, NON_EXISTENT_NAME);
} else {
panic!("unexpected error type")
}

55
object_store/src/disk.rs
View File

@ -1,9 +1,10 @@
//! This module contains the IOx implementation for using local disk as the
//! object store.
use crate::{
path::{FileConverter, ObjectStorePath},
DataDoesNotMatchLength, Result, UnableToCopyDataToFile, UnableToCreateDir, UnableToCreateFile,
UnableToDeleteFile, UnableToGetFileName, UnableToListDirectory, UnableToOpenFile,
UnableToProcessEntry, UnableToPutDataInMemory, UnableToReadBytes,
UnableToDeleteFile, UnableToListDirectory, UnableToOpenFile, UnableToProcessEntry,
UnableToPutDataInMemory, UnableToReadBytes,
};
use bytes::Bytes;
use futures::{Stream, TryStreamExt};
@ -16,21 +17,25 @@ use tokio_util::codec::{BytesCodec, FramedRead};
/// cloud storage provider.
#[derive(Debug)]
pub struct File {
root: PathBuf,
root: ObjectStorePath,
}
impl File {
/// Create new filesystem storage.
pub fn new(root: impl Into<PathBuf>) -> Self {
Self { root: root.into() }
Self {
root: ObjectStorePath::from_path_buf_unchecked(root),
}
}
fn path(&self, location: &str) -> PathBuf {
self.root.join(location)
fn path(&self, location: &ObjectStorePath) -> PathBuf {
let mut path = self.root.clone();
path.push_path(location);
FileConverter::convert(&path)
}
/// Save the provided bytes to the specified location.
pub async fn put<S>(&self, location: &str, bytes: S, length: usize) -> Result<()>
pub async fn put<S>(&self, location: &ObjectStorePath, bytes: S, length: usize) -> Result<()>
where
S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
{
@ -76,7 +81,10 @@ impl File {
}
/// Return the bytes that are stored at the specified location.
pub async fn get(&self, location: &str) -> Result<impl Stream<Item = Result<Bytes>>> {
pub async fn get(
&self,
location: &ObjectStorePath,
) -> Result<impl Stream<Item = Result<Bytes>>> {
let path = self.path(location);
let file = fs::File::open(&path)
@ -90,7 +98,7 @@ impl File {
}
/// Delete the object at the specified location.
pub async fn delete(&self, location: &str) -> Result<()> {
pub async fn delete(&self, location: &ObjectStorePath) -> Result<()> {
let path = self.path(location);
fs::remove_file(&path)
.await
@ -101,21 +109,19 @@ impl File {
/// List all the objects with the given prefix.
pub async fn list<'a>(
&'a self,
prefix: Option<&'a str>,
) -> Result<impl Stream<Item = Result<Vec<String>>> + 'a> {
let dirs = fs::read_dir(&self.root)
prefix: Option<&'a ObjectStorePath>,
) -> Result<impl Stream<Item = Result<Vec<ObjectStorePath>>> + 'a> {
let dirs = fs::read_dir(FileConverter::convert(&self.root))
.await
.context(UnableToListDirectory { path: &self.root })?;
.context(UnableToListDirectory {
path: format!("{:?}", self.root),
})?;
let s = dirs
.context(UnableToProcessEntry)
.and_then(|entry| {
let name = entry
.file_name()
.into_string()
.ok()
.context(UnableToGetFileName);
async move { name }
let file_path_buf: PathBuf = entry.file_name().into();
async move { Ok(ObjectStorePath::from_path_buf_unchecked(file_path_buf)) }
})
.try_filter(move |name| {
let matches = prefix.map_or(true, |p| name.starts_with(p));
@ -139,6 +145,7 @@ mod tests {
use futures::stream;
#[tokio::test]
#[ignore]
async fn file_test() -> Result<()> {
let root = TempDir::new()?;
let integration = ObjectStore::new_file(File::new(root.path()));
@ -153,7 +160,8 @@ mod tests {
let integration = ObjectStore::new_file(File::new(root.path()));
let bytes = stream::once(async { Ok(Bytes::from("hello world")) });
let res = integration.put("junk", bytes, 0).await;
let location = ObjectStorePath::from_path_buf_unchecked("junk");
let res = integration.put(&location, bytes, 0).await;
assert!(matches!(
res.err().unwrap(),
@ -172,19 +180,20 @@ mod tests {
let storage = ObjectStore::new_file(File::new(root.path()));
let data = Bytes::from("arbitrary data");
let location = "nested/file/test_file";
let mut location = ObjectStorePath::default();
location.push_all(&["nested", "file", "test_file"]);
let stream_data = std::io::Result::Ok(data.clone());
storage
.put(
location,
&location,
futures::stream::once(async move { stream_data }),
data.len(),
)
.await?;
let read_data = storage
.get(location)
.get(&location)
.await?
.map_ok(|b| bytes::BytesMut::from(&b[..]))
.try_concat()

60
object_store/src/gcp.rs
View File

@ -1,6 +1,7 @@
//! This module contains the IOx implementation for using Google Cloud Storage
//! as the object store.
use crate::{
path::{CloudConverter, ObjectStorePath},
DataDoesNotMatchLength, Result, UnableToDeleteDataFromGcs, UnableToDeleteDataFromGcs2,
UnableToGetDataFromGcs, UnableToGetDataFromGcs2, UnableToListDataFromGcs,
UnableToListDataFromGcs2, UnableToPutDataToGcs,
@ -25,7 +26,7 @@ impl GoogleCloudStorage {
}
/// Save the provided bytes to the specified location.
pub async fn put<S>(&self, location: &str, bytes: S, length: usize) -> Result<()>
pub async fn put<S>(&self, location: &ObjectStorePath, bytes: S, length: usize) -> Result<()>
where
S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
{
@ -44,7 +45,8 @@ impl GoogleCloudStorage {
}
);
let location_copy = location.to_string();
let location = CloudConverter::convert(&location);
let location_copy = location.clone();
let bucket_name = self.bucket_name.clone();
let _ = tokio::task::spawn_blocking(move || {
@ -65,8 +67,12 @@ impl GoogleCloudStorage {
}
/// Return the bytes that are stored at the specified location.
pub async fn get(&self, location: &str) -> Result<impl Stream<Item = Result<Bytes>>> {
let location_copy = location.to_string();
pub async fn get(
&self,
location: &ObjectStorePath,
) -> Result<impl Stream<Item = Result<Bytes>>> {
let location = CloudConverter::convert(&location);
let location_copy = location.clone();
let bucket_name = self.bucket_name.clone();
let bytes = tokio::task::spawn_blocking(move || {
@ -75,7 +81,7 @@ impl GoogleCloudStorage {
.await
.context(UnableToGetDataFromGcs {
bucket: &self.bucket_name,
location,
location: location.clone(),
})?
.context(UnableToGetDataFromGcs2 {
bucket: &self.bucket_name,
@ -86,8 +92,9 @@ impl GoogleCloudStorage {
}
/// Delete the object at the specified location.
pub async fn delete(&self, location: &str) -> Result<()> {
let location_copy = location.to_string();
pub async fn delete(&self, location: &ObjectStorePath) -> Result<()> {
let location = CloudConverter::convert(&location);
let location_copy = location.clone();
let bucket_name = self.bucket_name.clone();
tokio::task::spawn_blocking(move || {
@ -96,7 +103,7 @@ impl GoogleCloudStorage {
.await
.context(UnableToDeleteDataFromGcs {
bucket: &self.bucket_name,
location,
location: location.clone(),
})?
.context(UnableToDeleteDataFromGcs2 {
bucket: &self.bucket_name,
@ -109,10 +116,10 @@ impl GoogleCloudStorage {
/// List all the objects with the given prefix.
pub async fn list<'a>(
&'a self,
prefix: Option<&'a str>,
) -> Result<impl Stream<Item = Result<Vec<String>>> + 'a> {
prefix: Option<&'a ObjectStorePath>,
) -> Result<impl Stream<Item = Result<Vec<ObjectStorePath>>> + 'a> {
let bucket_name = self.bucket_name.clone();
let prefix = prefix.map(|p| p.to_string());
let prefix = prefix.map(CloudConverter::convert);
let objects = tokio::task::spawn_blocking(move || match prefix {
Some(prefix) => cloud_storage::Object::list_prefix(&bucket_name, &prefix),
@ -127,7 +134,10 @@ impl GoogleCloudStorage {
})?;
Ok(futures::stream::once(async move {
Ok(objects.into_iter().map(|o| o.name).collect())
Ok(objects
.into_iter()
.map(|o| ObjectStorePath::from_cloud_unchecked(o.name))
.collect())
}))
}
}
@ -135,6 +145,7 @@ impl GoogleCloudStorage {
#[cfg(test)]
mod test {
use crate::{
path::ObjectStorePath,
tests::{get_nonexistent_object, put_get_delete_list},
Error, GoogleCloudStorage, ObjectStore,
};
@ -188,7 +199,7 @@ mod test {
async fn gcs_test_get_nonexistent_location() -> Result<()> {
maybe_skip_integration!();
let bucket_name = bucket_name()?;
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let integration =
ObjectStore::new_google_cloud_storage(GoogleCloudStorage::new(&bucket_name));
@ -196,7 +207,10 @@ mod test {
assert_eq!(
result,
Bytes::from(format!("No such object: {}/{}", bucket_name, location_name))
Bytes::from(format!(
"No such object: {}/{}",
bucket_name, NON_EXISTENT_NAME
))
);
Ok(())
@ -206,7 +220,7 @@ mod test {
async fn gcs_test_get_nonexistent_bucket() -> Result<()> {
maybe_skip_integration!();
let bucket_name = NON_EXISTENT_NAME;
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let integration =
ObjectStore::new_google_cloud_storage(GoogleCloudStorage::new(bucket_name));
@ -221,11 +235,11 @@ mod test {
async fn gcs_test_delete_nonexistent_location() -> Result<()> {
maybe_skip_integration!();
let bucket_name = bucket_name()?;
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let integration =
ObjectStore::new_google_cloud_storage(GoogleCloudStorage::new(&bucket_name));
let err = integration.delete(location_name).await.unwrap_err();
let err = integration.delete(&location_name).await.unwrap_err();
if let Error::UnableToDeleteDataFromGcs2 {
source,
@ -235,7 +249,7 @@ mod test {
{
assert!(matches!(source, cloud_storage::Error::Google(_)));
assert_eq!(bucket, bucket_name);
assert_eq!(location, location_name);
assert_eq!(location, NON_EXISTENT_NAME);
} else {
panic!("unexpected error type")
}
@ -247,11 +261,11 @@ mod test {
async fn gcs_test_delete_nonexistent_bucket() -> Result<()> {
maybe_skip_integration!();
let bucket_name = NON_EXISTENT_NAME;
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let integration =
ObjectStore::new_google_cloud_storage(GoogleCloudStorage::new(bucket_name));
let err = integration.delete(location_name).await.unwrap_err();
let err = integration.delete(&location_name).await.unwrap_err();
if let Error::UnableToDeleteDataFromGcs2 {
source,
@ -261,7 +275,7 @@ mod test {
{
assert!(matches!(source, cloud_storage::Error::Google(_)));
assert_eq!(bucket, bucket_name);
assert_eq!(location, location_name);
assert_eq!(location, NON_EXISTENT_NAME);
} else {
panic!("unexpected error type")
}
@ -273,7 +287,7 @@ mod test {
async fn gcs_test_put_nonexistent_bucket() -> Result<()> {
maybe_skip_integration!();
let bucket_name = NON_EXISTENT_NAME;
let location_name = NON_EXISTENT_NAME;
let location_name = ObjectStorePath::from_cloud_unchecked(NON_EXISTENT_NAME);
let integration =
ObjectStore::new_google_cloud_storage(GoogleCloudStorage::new(bucket_name));
let data = Bytes::from("arbitrary data");
@ -281,7 +295,7 @@ mod test {
let result = integration
.put(
location_name,
&location_name,
futures::stream::once(async move { stream_data }),
data.len(),
)

114
object_store/src/lib.rs
View File

@ -25,6 +25,7 @@ use aws::AmazonS3;
use disk::File;
use gcp::GoogleCloudStorage;
use memory::InMemory;
use path::ObjectStorePath;
use bytes::Bytes;
use chrono::{DateTime, Utc};
@ -58,7 +59,7 @@ impl ObjectStore {
}
/// Save the provided bytes to the specified location.
pub async fn put<S>(&self, location: &str, bytes: S, length: usize) -> Result<()>
pub async fn put<S>(&self, location: &ObjectStorePath, bytes: S, length: usize) -> Result<()>
where
S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
{
@ -74,7 +75,10 @@ impl ObjectStore {
}
/// Return the bytes that are stored at the specified location.
pub async fn get(&self, location: &str) -> Result<impl Stream<Item = Result<Bytes>>> {
pub async fn get(
&self,
location: &ObjectStorePath,
) -> Result<impl Stream<Item = Result<Bytes>>> {
use ObjectStoreIntegration::*;
Ok(match &self.0 {
AmazonS3(s3) => s3.get(location).await?.boxed(),
@ -86,7 +90,7 @@ impl ObjectStore {
}
/// Delete the object at the specified location.
pub async fn delete(&self, location: &str) -> Result<()> {
pub async fn delete(&self, location: &ObjectStorePath) -> Result<()> {
use ObjectStoreIntegration::*;
match &self.0 {
AmazonS3(s3) => s3.delete(location).await?,
@ -101,8 +105,8 @@ impl ObjectStore {
/// List all the objects with the given prefix.
pub async fn list<'a>(
&'a self,
prefix: Option<&'a str>,
) -> Result<impl Stream<Item = Result<Vec<String>>> + 'a> {
prefix: Option<&'a ObjectStorePath>,
) -> Result<impl Stream<Item = Result<Vec<ObjectStorePath>>> + 'a> {
use ObjectStoreIntegration::*;
Ok(match &self.0 {
AmazonS3(s3) => s3.list(prefix).await?.boxed(),
@ -116,7 +120,10 @@ impl ObjectStore {
/// List objects with the given prefix and an implementation specific
/// delimiter. Returns common prefixes (directories) in addition to object
/// metadata.
pub async fn list_with_delimiter<'a>(&'a self, prefix: &'a str) -> Result<ListResult> {
pub async fn list_with_delimiter<'a>(
&'a self,
prefix: &'a ObjectStorePath,
) -> Result<ListResult> {
use ObjectStoreIntegration::*;
match &self.0 {
AmazonS3(s3) => s3.list_with_delimiter(prefix, &None).await,
@ -125,6 +132,19 @@ impl ObjectStore {
File(_file) => unimplemented!(),
}
}
/// Convert an `ObjectStorePath` to a `String` according to the appropriate
/// implementation. Suitable for printing; not suitable for sending to
/// APIs
pub fn convert_path(&self, path: &ObjectStorePath) -> String {
use ObjectStoreIntegration::*;
match &self.0 {
AmazonS3(_) | GoogleCloudStorage(_) | InMemory(_) => {
path::CloudConverter::convert(path)
}
File(_) => path::FileConverter::convert(path).display().to_string(),
}
}
}
/// All supported object storage integrations
@ -148,7 +168,7 @@ pub struct ListResult {
/// Token passed to the API for the next page of list results.
pub next_token: Option<String>,
/// Prefixes that are common (like directories)
pub common_prefixes: Vec<String>,
pub common_prefixes: Vec<ObjectStorePath>,
/// Object metadata for the listing
pub objects: Vec<ObjectMeta>,
}
@ -157,7 +177,7 @@ pub struct ListResult {
#[derive(Debug)]
pub struct ObjectMeta {
/// The full path to the object
pub location: String,
pub location: ObjectStorePath,
/// The last modified time
pub last_modified: DateTime<Utc>,
/// The size in bytes of the object
@ -292,8 +312,6 @@ pub enum Error {
UnableToCopyDataToFile {
source: io::Error,
},
#[snafu(display("Unable to retrieve filename"))]
UnableToGetFileName,
}
#[cfg(test)]
@ -306,8 +324,8 @@ mod tests {
async fn flatten_list_stream(
storage: &ObjectStore,
prefix: Option<&str>,
) -> Result<Vec<String>> {
prefix: Option<&ObjectStorePath>,
) -> Result<Vec<ObjectStorePath>> {
storage
.list(prefix)
.await?
@ -328,12 +346,13 @@ mod tests {
);
let data = Bytes::from("arbitrary data");
let location = "test_file";
let mut location = ObjectStorePath::default();
location.push("test_file");
let stream_data = std::io::Result::Ok(data.clone());
storage
.put(
location,
&location,
futures::stream::once(async move { stream_data }),
data.len(),
)
@ -341,25 +360,29 @@ mod tests {
// List everything
let content_list = flatten_list_stream(storage, None).await?;
assert_eq!(content_list, &[location]);
assert_eq!(content_list, &[location.clone()]);
// List everything starting with a prefix that should return results
let content_list = flatten_list_stream(storage, Some("test")).await?;
assert_eq!(content_list, &[location]);
let mut prefix = ObjectStorePath::default();
prefix.push("test");
let content_list = flatten_list_stream(storage, Some(&prefix)).await?;
assert_eq!(content_list, &[location.clone()]);
// List everything starting with a prefix that shouldn't return results
let content_list = flatten_list_stream(storage, Some("something")).await?;
let mut prefix = ObjectStorePath::default();
prefix.push("something");
let content_list = flatten_list_stream(storage, Some(&prefix)).await?;
assert!(content_list.is_empty());
let read_data = storage
.get(location)
.get(&location)
.await?
.map_ok(|b| bytes::BytesMut::from(&b[..]))
.try_concat()
.await?;
assert_eq!(&*read_data, data);
storage.delete(location).await?;
storage.delete(&location).await?;
let content_list = flatten_list_stream(storage, None).await?;
assert!(content_list.is_empty());
@ -375,13 +398,16 @@ mod tests {
let data = Bytes::from("arbitrary data");
let files = vec![
let files: Vec<_> = [
"mydb/wal/000/000/000.segment",
"mydb/wal/000/000/001.segment",
"mydb/wal/001/001/000.segment",
"mydb/wal/foo.test",
"mydb/data/whatevs",
];
]
.iter()
.map(|&s| ObjectStorePath::from_cloud_unchecked(s))
.collect();
let time_before_creation = Utc::now();
@ -397,18 +423,30 @@ mod tests {
.unwrap();
}
let result = storage.list_with_delimiter("mydb/wal/").await.unwrap();
assert_eq!(
result.common_prefixes,
vec!["mydb/wal/000/", "mydb/wal/001/"]
);
let mut prefix = ObjectStorePath::default();
prefix.push_all(&["mydb", "wal"]);
let mut expected_000 = prefix.clone();
expected_000.push("000");
let mut expected_001 = prefix.clone();
expected_001.push("001");
let mut expected_location = prefix.clone();
expected_location.push("foo.test");
// This is needed because we want a trailing slash on the prefix in this test
prefix.push("");
let result = storage.list_with_delimiter(&prefix).await.unwrap();
assert_eq!(result.common_prefixes, vec![expected_000, expected_001]);
assert_eq!(result.objects.len(), 1);
let object = &result.objects[0];
assert_eq!(object.location, "mydb/wal/foo.test");
assert_eq!(object.location, expected_location);
assert_eq!(object.size, data.len());
assert!(object.last_modified > time_before_creation);
for f in files {
for f in &files {
storage.delete(f).await.unwrap();
}
@ -420,15 +458,16 @@ mod tests {
pub(crate) async fn get_nonexistent_object(
storage: &ObjectStore,
location: Option<&str>,
location: Option<ObjectStorePath>,
) -> Result<Bytes> {
let location = location.unwrap_or("this_file_should_not_exist");
let location = location
.unwrap_or_else(|| ObjectStorePath::from_cloud_unchecked("this_file_should_not_exist"));
let content_list = flatten_list_stream(storage, Some(location)).await?;
let content_list = flatten_list_stream(storage, Some(&location)).await?;
assert!(content_list.is_empty());
Ok(storage
.get(location)
.get(&location)
.await?
.map_ok(|b| bytes::BytesMut::from(&b[..]))
.try_concat()
@ -437,16 +476,19 @@ mod tests {
}
async fn delete_fixtures(storage: &ObjectStore) {
let files = vec![
let files: Vec<_> = [
"test_file",
"mydb/wal/000/000/000.segment",
"mydb/wal/000/000/001.segment",
"mydb/wal/001/001/000.segment",
"mydb/wal/foo.test",
"mydb/data/whatevs",
];
]
.iter()
.map(|&s| ObjectStorePath::from_cloud_unchecked(s))
.collect();
for f in files {
for f in &files {
// don't care if it errors, should fail elsewhere
let _ = storage.delete(f).await;
}

50
object_store/src/memory.rs
View File

@ -1,7 +1,8 @@
//! This module contains the IOx implementation for using memory as the object
//! store.
use crate::{
DataDoesNotMatchLength, ListResult, NoDataInMemory, ObjectMeta, Result, UnableToPutDataInMemory,
path::ObjectStorePath, DataDoesNotMatchLength, ListResult, NoDataInMemory, ObjectMeta, Result,
UnableToPutDataInMemory,
};
use bytes::Bytes;
use chrono::Utc;
@ -11,14 +12,11 @@ use std::collections::BTreeSet;
use std::{collections::BTreeMap, io};
use tokio::sync::RwLock;
// The delimiter to separate object namespaces, creating a directory structure.
const DELIMITER: &str = "/";
/// In-memory storage suitable for testing or for opting out of using a cloud
/// storage provider.
#[derive(Debug, Default)]
pub struct InMemory {
storage: RwLock<BTreeMap<String, Bytes>>,
storage: RwLock<BTreeMap<ObjectStorePath, Bytes>>,
}
impl InMemory {
@ -38,7 +36,7 @@ impl InMemory {
}
/// Save the provided bytes to the specified location.
pub async fn put<S>(&self, location: &str, bytes: S, length: usize) -> Result<()>
pub async fn put<S>(&self, location: &ObjectStorePath, bytes: S, length: usize) -> Result<()>
where
S: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
{
@ -58,15 +56,15 @@ impl InMemory {
let content = content.freeze();
self.storage
.write()
.await
.insert(location.to_string(), content);
self.storage.write().await.insert(location.clone(), content);
Ok(())
}
/// Return the bytes that are stored at the specified location.
pub async fn get(&self, location: &str) -> Result<impl Stream<Item = Result<Bytes>>> {
pub async fn get(
&self,
location: &ObjectStorePath,
) -> Result<impl Stream<Item = Result<Bytes>>> {
let data = self
.storage
.read()
@ -79,7 +77,7 @@ impl InMemory {
}
/// Delete the object at the specified location.
pub async fn delete(&self, location: &str) -> Result<()> {
pub async fn delete(&self, location: &ObjectStorePath) -> Result<()> {
self.storage.write().await.remove(location);
Ok(())
}
@ -87,8 +85,8 @@ impl InMemory {
/// List all the objects with the given prefix.
pub async fn list<'a>(
&'a self,
prefix: Option<&'a str>,
) -> Result<impl Stream<Item = Result<Vec<String>>> + 'a> {
prefix: Option<&'a ObjectStorePath>,
) -> Result<impl Stream<Item = Result<Vec<ObjectStorePath>>> + 'a> {
let list = if let Some(prefix) = prefix {
self.storage
.read()
@ -111,37 +109,27 @@ impl InMemory {
/// limitations.
pub async fn list_with_delimiter<'a>(
&'a self,
prefix: &'a str,
prefix: &'a ObjectStorePath,
_next_token: &Option<String>,
) -> Result<ListResult> {
let mut common_prefixes = BTreeSet::new();
let last_modified = Utc::now();
// first ensure the prefix ends with the delimiter
let prefix = if prefix.ends_with(DELIMITER) {
prefix.to_string()
} else {
prefix.to_string() + DELIMITER
};
// set the end prefix so we pull back everything that starts with
// the passed in prefix
let mut end_prefix = prefix.clone();
end_prefix.pop();
end_prefix.push('0');
end_prefix.push("0");
// Only objects in this base level should be returned in the
// response. Otherwise, we just collect the common prefixes.
let mut objects = vec![];
for (k, v) in self.storage.read().await.range(prefix.clone()..end_prefix) {
let parts: Vec<_> = k
.strip_prefix(&prefix)
.expect("must have prefix if in range")
.split(DELIMITER)
.collect();
let parts = k.parts_after_prefix(&prefix);
if parts.len() >= 2 {
let full_prefix = prefix.clone() + parts[0] + DELIMITER;
let mut full_prefix = prefix.clone();
full_prefix.push_part(&parts[0]);
common_prefixes.insert(full_prefix);
} else {
let object = ObjectMeta {
@ -175,6 +163,7 @@ mod tests {
use futures::stream;
#[tokio::test]
#[ignore]
async fn in_memory_test() -> Result<()> {
let integration = ObjectStore::new_in_memory(InMemory::new());
@ -190,7 +179,8 @@ mod tests {
let integration = ObjectStore::new_in_memory(InMemory::new());
let bytes = stream::once(async { Ok(Bytes::from("hello world")) });
let res = integration.put("junk", bytes, 0).await;
let location = ObjectStorePath::from_cloud_unchecked("junk");
let res = integration.put(&location, bytes, 0).await;
assert!(matches!(
res.err().unwrap(),

215
object_store/src/path.rs
View File

@ -1,71 +1,202 @@
//! This module contains code for abstracting object locations that work
//! across different backing implementations and platforms.
use percent_encoding::{percent_decode, percent_encode};
use std::{
fmt::{self, Formatter},
path::PathBuf,
};
use itertools::Itertools;
use percent_encoding::{percent_encode, AsciiSet, CONTROLS};
use std::path::PathBuf;
/// Universal interface for handling paths and locations for objects and
/// directories in the object store.
///
/// It allows IOx to be completely decoupled from the underlying object store
/// implementations.
///
/// Deliberately does not implement `Display` or `ToString`! Use one of the
/// converters.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Default)]
pub struct ObjectStorePath {
path: PathSource,
parts: Vec<PathPart>,
}
// Invariants to maintain/document/test:
//
// - always ends in DELIMITER if it's a directory. If it's the end object, it
// should have some sort of file extension like .parquet, .json, or .segment
// - does not contain unencoded DELIMITER
// - for file paths: does not escape root dir
// - for object storage: looks like directories
// - Paths that come from object stores directly don't need to be
// parsed/validated
// - Within a process, the same backing store will always be used
//
impl ObjectStorePath {
/// Pushes a part onto the path. Ensures the part is percent encoded
pub fn push(&mut self, part: &str) {}
/// Returns a path that is safe for use in S3, GCP, Azure, and Memory
pub fn to_object_store_path(&self) -> &str {
unimplemented!()
}
/// Returns a path that is safe for use in File store
pub fn to_path_buf(&self) -> &PathBuf {
unimplemented!()
}
}
enum PathSource {
String(String),
PathBuf(PathBuf),
}
impl From<&str> for ObjectStorePath {
fn from(path: &str) -> Self {
/// For use when receiving a path from an object store API directly, not
/// when building a path. Assumes DELIMITER is the separator.
///
/// TODO: Improve performance by implementing a CoW-type model to delay
/// parsing until needed TODO: This should only be available to cloud
/// storage
pub fn from_cloud_unchecked(path: impl Into<String>) -> Self {
let path = path.into();
Self {
path: PathSource::String(path.into()),
parts: path
.split_terminator(DELIMITER)
.map(|s| PathPart(s.to_string()))
.collect(),
}
}
}
impl From<PathBuf> for ObjectStorePath {
fn from(path_buf: PathBuf) -> Self {
/// For use when receiving a path from a filesystem directly, not
/// when building a path. Uses the standard library's path splitting
/// implementation to separate into parts.
pub fn from_path_buf_unchecked(path: impl Into<PathBuf>) -> Self {
let path = path.into();
Self {
path: PathSource::PathBuf(path_buf),
parts: path
.iter()
.flat_map(|s| s.to_os_string().into_string().map(PathPart))
.collect(),
}
}
}
// Show the path as it was passed in (not percent encoded)
impl fmt::Display for ObjectStorePath {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
/// Add a part to the end of the path, encoding any restricted characters.
pub fn push(&mut self, part: impl Into<String>) {
let part = part.into();
self.parts.push((&*part).into());
}
/// Add a `PathPart` to the end of the path. Infallible because the
/// `PathPart` should already have been checked for restricted
/// characters.
pub fn push_part(&mut self, part: &PathPart) {
self.parts.push(part.to_owned());
}
/// Add the parts of `ObjectStorePath` to the end of the path. Notably does
/// *not* behave as `PathBuf::push` does: no existing part of `self`
/// will be replaced as part of this call.
pub fn push_path(&mut self, path: &Self) {
self.parts.extend_from_slice(&path.parts);
}
/// Push a bunch of parts in one go.
pub fn push_all<'a>(&mut self, parts: impl AsRef<[&'a str]>) {
// Turn T into a slice of str, validate each one, and collect() it into a
// Vec<String>
let parts = parts.as_ref().iter().map(|&v| v.into()).collect::<Vec<_>>();
// Push them to the internal path
self.parts.extend(parts);
}
/// Return the component parts of the path.
pub fn as_parts(&self) -> &[PathPart] {
self.parts.as_ref()
}
/// Pops a part from the path and returns it, or `None` if it's empty.
pub fn pop(&mut self) -> Option<&PathPart> {
unimplemented!()
}
/// Determines whether `prefix` is a prefix of `self`.
pub fn starts_with(&self, _prefix: &Self) -> bool {
unimplemented!()
}
/// Returns delimiter-separated parts contained in `self` after `prefix`.
pub fn parts_after_prefix(&self, _prefix: &Self) -> &[PathPart] {
unimplemented!()
}
}
// Show the PathBuf debug or the percent encoded and display path
impl fmt::Debug for ObjectStorePath {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
unimplemented!()
// TODO: I made these structs rather than functions because I could see
// `convert` being part of a trait, possibly, but that seemed a bit overly
// complex for now.
/// Converts `ObjectStorePath`s to `String`s that are appropriate for use as
/// locations in cloud storage.
#[derive(Debug, Clone, Copy)]
pub struct CloudConverter {}
impl CloudConverter {
/// Creates a cloud storage location by joining this `ObjectStorePath`'s
/// parts with `DELIMITER`
pub fn convert(object_store_path: &ObjectStorePath) -> String {
object_store_path.parts.iter().map(|p| &p.0).join(DELIMITER)
}
}
// The delimiter to separate object namespaces, creating a directory structure.
const DELIMITER: &str = "/";
/// Converts `ObjectStorePath`s to `String`s that are appropriate for use as
/// locations in filesystem storage.
#[derive(Debug, Clone, Copy)]
pub struct FileConverter {}
impl FileConverter {
/// Creates a filesystem `PathBuf` location by using the standard library's
/// `PathBuf` building implementation appropriate for the current
/// platform.
pub fn convert(object_store_path: &ObjectStorePath) -> PathBuf {
object_store_path.parts.iter().map(|p| &p.0).collect()
}
}
/// The delimiter to separate object namespaces, creating a directory structure.
pub const DELIMITER: &str = "/";
// percent_encode's API needs this as a byte... is there a const conversion for
// this?
const DELIMITER_BYTE: u8 = b'/';
/// The PathPart type exists to validate the directory/file names that form part
/// of a path.
///
/// A PathPart instance is guaranteed to contain no `/` characters as it can
/// only be constructed by going through the `try_from` impl.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Default)]
pub struct PathPart(String);
/// Characters we want to encode.
const INVALID: &AsciiSet = &CONTROLS
// The delimiter we are reserving for internal hierarchy
.add(DELIMITER_BYTE)
// Characters AWS recommends avoiding for object keys
// https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingMetadata.html
.add(b'\\')
.add(b'{')
// TODO: Non-printable ASCII characters (128–255 decimal characters)
.add(b'^')
.add(b'}')
.add(b'%')
.add(b'`')
.add(b']')
.add(b'"')
.add(b'>')
.add(b'[')
.add(b'~')
.add(b'<')
.add(b'#')
.add(b'|');
impl From<&str> for PathPart {
fn from(v: &str) -> Self {
Self(percent_encode(v.as_bytes(), INVALID).to_string())
}
}
impl std::fmt::Display for PathPart {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.0.fmt(f)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn path_part_delimiter_gets_encoded() {
let part: PathPart = "foo/bar".into();
assert_eq!(part, PathPart(String::from("foo%2Fbar")))
}
}

14
server/src/server.rs
View File

@ -17,7 +17,7 @@ use data_types::{
};
use influxdb_line_protocol::ParsedLine;
use mutable_buffer::MutableBufferDb;
use object_store::ObjectStore;
use object_store::{path::ObjectStorePath, ObjectStore};
use query::{Database, DatabaseStore, SQLDatabase};
use async_trait::async_trait;
@ -419,8 +419,10 @@ impl RemoteServer for RemoteServerImpl {
}
// location in the store for the configuration file
fn config_location(id: u32) -> String {
format!("{}/config.json", id)
fn config_location(id: u32) -> ObjectStorePath {
let mut path = ObjectStorePath::default();
path.push_all(&[&id.to_string(), "config.json"]);
path
}
#[cfg(test)]
@ -670,10 +672,12 @@ partition_key:
server.store_configuration().await.unwrap();
let location = "1/config.json";
let mut location = ObjectStorePath::default();
location.push_all(&["1", "config.json"]);
let read_data = server
.store
.get(location)
.get(&location)
.await
.unwrap()
.map_ok(|b| bytes::BytesMut::from(&b[..]))

63
server/src/snapshot.rs
View File

@ -5,7 +5,7 @@ use arrow_deps::{
parquet::{self, arrow::ArrowWriter, file::writer::TryClone},
};
use data_types::partition_metadata::{Partition as PartitionMeta, Table};
use object_store::ObjectStore;
use object_store::{path::ObjectStorePath, ObjectStore};
use query::PartitionChunk;
use std::io::{Cursor, Seek, SeekFrom, Write};
@ -61,8 +61,8 @@ where
{
pub id: Uuid,
pub partition_meta: PartitionMeta,
pub metadata_path: String,
pub data_path: String,
pub metadata_path: ObjectStorePath,
pub data_path: ObjectStorePath,
store: Arc<ObjectStore>,
partition: Arc<T>,
status: Mutex<Status>,
@ -74,8 +74,8 @@ where
{
fn new(
partition_key: String,
metadata_path: String,
data_path: String,
metadata_path: ObjectStorePath,
data_path: ObjectStorePath,
store: Arc<ObjectStore>,
partition: Arc<T>,
tables: Vec<Table>,
@ -101,6 +101,10 @@ where
}
}
fn data_path(&self) -> String {
self.store.convert_path(&self.data_path)
}
// returns the position of the next table
fn next_table(&self) -> Option<(usize, &str)> {
let mut status = self.status.lock().expect("mutex poisoned");
@ -150,8 +154,10 @@ where
.map_err(|e| Box::new(e) as _)
.context(PartitionError)?;
let file_name = format!("{}/{}.parquet", &self.data_path, table_name);
self.write_batches(batches, &file_name).await?;
let mut location = self.data_path.clone();
let file_name = format!("{}.parquet", table_name);
location.push(&file_name);
self.write_batches(batches, &location).await?;
self.mark_table_finished(pos);
if self.should_stop() {
@ -159,8 +165,9 @@ where
}
}
let partition_meta_path =
format!("{}/{}.json", &self.metadata_path, &self.partition_meta.key);
let mut partition_meta_path = self.metadata_path.clone();
let key = format!("{}.json", &self.partition_meta.key);
partition_meta_path.push(&key);
let json_data = serde_json::to_vec(&self.partition_meta).context(JsonGenerationError)?;
let data = Bytes::from(json_data);
let len = data.len();
@ -185,7 +192,11 @@ where
Ok(())
}
async fn write_batches(&self, batches: Vec<RecordBatch>, file_name: &str) -> Result<()> {
async fn write_batches(
&self,
batches: Vec<RecordBatch>,
file_name: &ObjectStorePath,
) -> Result<()> {
let mem_writer = MemWriter::default();
{
let mut writer = ArrowWriter::try_new(mem_writer.clone(), batches[0].schema(), None)
@ -236,8 +247,8 @@ pub struct Status {
}
pub fn snapshot_chunk<T>(
metadata_path: impl Into<String>,
data_path: impl Into<String>,
metadata_path: ObjectStorePath,
data_path: ObjectStorePath,
store: Arc<ObjectStore>,
partition: Arc<T>,
notify: Option<oneshot::Sender<()>>,
@ -252,8 +263,8 @@ where
let snapshot = Snapshot::new(
partition.key().to_string(),
metadata_path.into(),
data_path.into(),
metadata_path,
data_path,
store,
partition,
table_stats,
@ -265,7 +276,8 @@ where
tokio::spawn(async move {
info!(
"starting snapshot of {} to {}",
&snapshot.partition_meta.key, &snapshot.data_path
&snapshot.partition_meta.key,
&snapshot.data_path()
);
if let Err(e) = snapshot.run(notify).await {
error!("error running snapshot: {:?}", e);
@ -349,11 +361,14 @@ mem,host=A,region=west used=45 1
let store = Arc::new(ObjectStore::new_in_memory(InMemory::new()));
let chunk = Arc::new(chunk);
let (tx, rx) = tokio::sync::oneshot::channel();
let metadata_path = "/meta";
let data_path = "/data";
let mut metadata_path = ObjectStorePath::default();
metadata_path.push("meta");
let mut data_path = ObjectStorePath::default();
data_path.push("data");
let snapshot = snapshot_chunk(
metadata_path,
metadata_path.clone(),
data_path,
store.clone(),
chunk.clone(),
@ -363,8 +378,11 @@ mem,host=A,region=west used=45 1
rx.await.unwrap();
let mut location = metadata_path;
location.push("testaroo.json");
let summary = store
.get("/meta/testaroo.json")
.get(&location)
.await
.unwrap()
.map_ok(|b| bytes::BytesMut::from(&b[..]))
@ -395,8 +413,11 @@ mem,host=A,region=west used=45 1
let store = Arc::new(ObjectStore::new_in_memory(InMemory::new()));
let chunk = Arc::new(ChunkWB::new("testaroo", 11));
let metadata_path = "/meta".to_string();
let data_path = "/data".to_string();
let mut metadata_path = ObjectStorePath::default();
metadata_path.push("meta");
let mut data_path = ObjectStorePath::default();
data_path.push("data");
let snapshot = Snapshot::new(
chunk.key.clone(),

27
src/server/http_routes.rs
View File

@ -9,26 +9,27 @@
//! Long term, we expect to create IOx specific api in terms of
//! database names and may remove this quasi /v2 API.
use http::header::CONTENT_ENCODING;
use tracing::{debug, error, info};
use super::{org_and_bucket_to_database, OrgBucketMappingError};
// Influx crates
use arrow_deps::arrow;
use data_types::database_rules::DatabaseRules;
use influxdb_line_protocol::parse_lines;
use object_store::path::ObjectStorePath;
use query::SQLDatabase;
use server::server::{ConnectionManager, Server as AppServer};
use super::{org_and_bucket_to_database, OrgBucketMappingError};
// External crates
use bytes::{Bytes, BytesMut};
use data_types::database_rules::DatabaseRules;
use futures::{self, StreamExt};
use http::header::CONTENT_ENCODING;
use hyper::{Body, Method, Request, Response, StatusCode};
use routerify::prelude::*;
use routerify::{Middleware, RequestInfo, Router, RouterService};
use routerify::{prelude::*, Middleware, RequestInfo, Router, RouterService};
use serde::Deserialize;
use snafu::{OptionExt, ResultExt, Snafu};
use std::fmt::Debug;
use std::str;
use std::sync::Arc;
use tracing::{debug, error, info};
use std::{fmt::Debug, str, sync::Arc};
#[derive(Debug, Snafu)]
pub enum ApplicationError {
@ -541,8 +542,12 @@ async fn snapshot_partition<M: ConnectionManager + Send + Sync + Debug + 'static
bucket: &snapshot.bucket,
})?;
let metadata_path = format!("{}/meta", &db_name);
let data_path = format!("{}/data/{}", &db_name, &snapshot.chunk);
let mut metadata_path = ObjectStorePath::default();
metadata_path.push(&db_name.to_string());
let mut data_path = metadata_path.clone();
metadata_path.push("meta");
data_path.push_all(&["data", &snapshot.chunk]);
let partition = db.rollover_partition(&snapshot.chunk).await.unwrap();
let snapshot = server::snapshot::snapshot_chunk(
metadata_path,