Consistently use GCP XML API

.github/workflows/object_store.yml

@@ -96,7 +96,7 @@ jobs:
 
       - name: Configure Fake GCS Server (GCP emulation)
         run: |
-          docker run -d -p 4443:4443 fsouza/fake-gcs-server -scheme http
+          docker run -d -p 4443:4443 tustvold/fake-gcs-server -scheme http -backend memory -public-host localhost:4443
           # Give the container a moment to start up prior to configuring it
           sleep 1
           curl -v -X POST --data-binary '{"name":"test-bucket"}' -H "Content-Type: application/json" "http://localhost:4443/storage/v1/b"

object_store/Cargo.toml

@@ -67,7 +67,7 @@ tokio = { version = "1.25.0", features = ["sync", "macros", "rt", "time", "io-ut
 nix = "0.26.1"
 
 [features]
-cloud = ["serde", "serde_json", "quick-xml",  "hyper", "reqwest", "reqwest/json","reqwest/stream", "chrono/serde", "base64", "rand", "ring"]
+cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/json", "reqwest/stream", "chrono/serde", "base64", "rand", "ring"]
 azure = ["cloud"]
 gcp = ["cloud", "rustls-pemfile"]
 aws = ["cloud"]

object_store/src/aws/client.rs

@@ -18,19 +18,19 @@
 use crate::aws::checksum::Checksum;
 use crate::aws::credential::{AwsCredential, CredentialExt, CredentialProvider};
 use crate::aws::STRICT_PATH_ENCODE_SET;
+use crate::client::list::ListResponse;
 use crate::client::pagination::stream_paginated;
 use crate::client::retry::RetryExt;
 use crate::multipart::UploadPart;
 use crate::path::DELIMITER;
 use crate::util::{format_http_range, format_prefix};
 use crate::{
-    BoxStream, ClientOptions, ListResult, MultipartId, ObjectMeta, Path, Result,
-    RetryConfig, StreamExt,
+    BoxStream, ClientOptions, ListResult, MultipartId, Path, Result, RetryConfig,
+    StreamExt,
 };
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::{Buf, Bytes};
-use chrono::{DateTime, Utc};
 use percent_encoding::{utf8_percent_encode, PercentEncode};
 use reqwest::{
     header::CONTENT_TYPE, Client as ReqwestClient, Method, Response, StatusCode,
@@ -118,69 +118,6 @@ impl From<Error> for crate::Error {
     }
 }
 
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "PascalCase")]
-pub struct ListResponse {
-    #[serde(default)]
-    pub contents: Vec<ListContents>,
-    #[serde(default)]
-    pub common_prefixes: Vec<ListPrefix>,
-    #[serde(default)]
-    pub next_continuation_token: Option<String>,
-}
-
-impl TryFrom<ListResponse> for ListResult {
-    type Error = crate::Error;
-
-    fn try_from(value: ListResponse) -> Result<Self> {
-        let common_prefixes = value
-            .common_prefixes
-            .into_iter()
-            .map(|x| Ok(Path::parse(x.prefix)?))
-            .collect::<Result<_>>()?;
-
-        let objects = value
-            .contents
-            .into_iter()
-            .map(TryFrom::try_from)
-            .collect::<Result<_>>()?;
-
-        Ok(Self {
-            common_prefixes,
-            objects,
-        })
-    }
-}
-
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "PascalCase")]
-pub struct ListPrefix {
-    pub prefix: String,
-}
-
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "PascalCase")]
-pub struct ListContents {
-    pub key: String,
-    pub size: usize,
-    pub last_modified: DateTime<Utc>,
-    #[serde(rename = "ETag")]
-    pub e_tag: Option<String>,
-}
-
-impl TryFrom<ListContents> for ObjectMeta {
-    type Error = crate::Error;
-
-    fn try_from(value: ListContents) -> Result<Self> {
-        Ok(Self {
-            location: Path::parse(value.key)?,
-            last_modified: value.last_modified,
-            size: value.size,
-            e_tag: value.e_tag,
-        })
-    }
-}
-
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "PascalCase")]
 struct InitiateMultipart {

object_store/src/aws/mod.rs

@@ -33,7 +33,6 @@
 
 use async_trait::async_trait;
 use bytes::Bytes;
-use chrono::{DateTime, Utc};
 use futures::stream::BoxStream;
 use futures::TryStreamExt;
 use itertools::Itertools;
@@ -53,6 +52,7 @@ use crate::aws::credential::{
     AwsCredential, CredentialProvider, InstanceCredentialProvider,
     StaticCredentialProvider, WebIdentityProvider,
 };
+use crate::client::header::header_meta;
 use crate::client::ClientConfigKey;
 use crate::config::ConfigValue;
 use crate::multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart};
@@ -86,24 +86,6 @@ static METADATA_ENDPOINT: &str = "http://169.254.169.254";
 #[derive(Debug, Snafu)]
 #[allow(missing_docs)]
 enum Error {
-    #[snafu(display("Last-Modified Header missing from response"))]
-    MissingLastModified,
-
-    #[snafu(display("Content-Length Header missing from response"))]
-    MissingContentLength,
-
-    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
-    InvalidLastModified {
-        last_modified: String,
-        source: chrono::ParseError,
-    },
-
-    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
-    InvalidContentLength {
-        content_length: String,
-        source: std::num::ParseIntError,
-    },
-
     #[snafu(display("Missing region"))]
     MissingRegion,
 
@@ -154,6 +136,11 @@ enum Error {
 
     #[snafu(display("Failed to parse the region for bucket '{}'", bucket))]
     RegionParse { bucket: String },
+
+    #[snafu(display("Failed to parse headers: {}", source))]
+    Header {
+        source: crate::client::header::Error,
+    },
 }
 
 impl From<Error> for super::Error {
@@ -274,40 +261,10 @@ impl ObjectStore for AmazonS3 {
     }
 
     async fn head(&self, location: &Path) -> Result<ObjectMeta> {
-        use reqwest::header::{CONTENT_LENGTH, ETAG, LAST_MODIFIED};
-
         // Extract meta from headers
         // https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html#API_HeadObject_ResponseSyntax
         let response = self.client.get_request(location, None, true).await?;
-        let headers = response.headers();
-
-        let last_modified = headers
-            .get(LAST_MODIFIED)
-            .context(MissingLastModifiedSnafu)?;
-
-        let content_length = headers
-            .get(CONTENT_LENGTH)
-            .context(MissingContentLengthSnafu)?;
-
-        let last_modified = last_modified.to_str().context(BadHeaderSnafu)?;
-        let last_modified = DateTime::parse_from_rfc2822(last_modified)
-            .context(InvalidLastModifiedSnafu { last_modified })?
-            .with_timezone(&Utc);
-
-        let content_length = content_length.to_str().context(BadHeaderSnafu)?;
-        let content_length = content_length
-            .parse()
-            .context(InvalidContentLengthSnafu { content_length })?;
-
-        let e_tag = headers.get(ETAG).context(MissingEtagSnafu)?;
-        let e_tag = e_tag.to_str().context(BadHeaderSnafu)?;
-
-        Ok(ObjectMeta {
-            location: location.clone(),
-            last_modified,
-            size: content_length,
-            e_tag: Some(e_tag.to_string()),
-        })
+        Ok(header_meta(location, response.headers()).context(HeaderSnafu)?)
     }
 
     async fn delete(&self, location: &Path) -> Result<()> {

object_store/src/azure/mod.rs

@@ -38,7 +38,6 @@ use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::Bytes;
-use chrono::{TimeZone, Utc};
 use futures::{stream::BoxStream, StreamExt, TryStreamExt};
 use percent_encoding::percent_decode_str;
 use serde::{Deserialize, Serialize};
@@ -51,9 +50,9 @@ use std::{collections::BTreeSet, str::FromStr};
 use tokio::io::AsyncWrite;
 use url::Url;
 
+use crate::client::header::header_meta;
 use crate::client::ClientConfigKey;
 use crate::config::ConfigValue;
-use crate::util::RFC1123_FMT;
 pub use credential::authority_hosts;
 
 mod client;
@@ -74,24 +73,6 @@ const MSI_ENDPOINT_ENV_KEY: &str = "IDENTITY_ENDPOINT";
 #[derive(Debug, Snafu)]
 #[allow(missing_docs)]
 enum Error {
-    #[snafu(display("Last-Modified Header missing from response"))]
-    MissingLastModified,
-
-    #[snafu(display("Content-Length Header missing from response"))]
-    MissingContentLength,
-
-    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
-    InvalidLastModified {
-        last_modified: String,
-        source: chrono::ParseError,
-    },
-
-    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
-    InvalidContentLength {
-        content_length: String,
-        source: std::num::ParseIntError,
-    },
-
     #[snafu(display("Received header containing non-ASCII data"))]
     BadHeader { source: reqwest::header::ToStrError },
 
@@ -145,6 +126,11 @@ enum Error {
 
     #[snafu(display("ETag Header missing from response"))]
     MissingEtag,
+
+    #[snafu(display("Failed to parse headers: {}", source))]
+    Header {
+        source: crate::client::header::Error,
+    },
 }
 
 impl From<Error> for super::Error {
@@ -237,43 +223,10 @@ impl ObjectStore for MicrosoftAzure {
     }
 
     async fn head(&self, location: &Path) -> Result<ObjectMeta> {
-        use reqwest::header::{CONTENT_LENGTH, ETAG, LAST_MODIFIED};
-
         // Extract meta from headers
         // https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties
         let response = self.client.get_request(location, None, true).await?;
-        let headers = response.headers();
-
-        let last_modified = headers
-            .get(LAST_MODIFIED)
-            .ok_or(Error::MissingLastModified)?
-            .to_str()
-            .context(BadHeaderSnafu)?;
-        let last_modified = Utc
-            .datetime_from_str(last_modified, RFC1123_FMT)
-            .context(InvalidLastModifiedSnafu { last_modified })?;
-
-        let content_length = headers
-            .get(CONTENT_LENGTH)
-            .ok_or(Error::MissingContentLength)?
-            .to_str()
-            .context(BadHeaderSnafu)?;
-        let content_length = content_length
-            .parse()
-            .context(InvalidContentLengthSnafu { content_length })?;
-
-        let e_tag = headers
-            .get(ETAG)
-            .ok_or(Error::MissingEtag)?
-            .to_str()
-            .context(BadHeaderSnafu)?;
-
-        Ok(ObjectMeta {
-            location: location.clone(),
-            last_modified,
-            size: content_length,
-            e_tag: Some(e_tag.to_string()),
-        })
+        Ok(header_meta(location, response.headers()).context(HeaderSnafu)?)
     }
 
     async fn delete(&self, location: &Path) -> Result<()> {

object_store/src/client/header.rs

@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Logic for extracting ObjectMeta from headers used by AWS, GCP and Azure
+
+use crate::path::Path;
+use crate::ObjectMeta;
+use chrono::{DateTime, Utc};
+use hyper::header::{CONTENT_LENGTH, ETAG, LAST_MODIFIED};
+use hyper::HeaderMap;
+use snafu::{OptionExt, ResultExt, Snafu};
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("ETag Header missing from response"))]
+    MissingEtag,
+
+    #[snafu(display("Received header containing non-ASCII data"))]
+    BadHeader { source: reqwest::header::ToStrError },
+
+    #[snafu(display("Last-Modified Header missing from response"))]
+    MissingLastModified,
+
+    #[snafu(display("Content-Length Header missing from response"))]
+    MissingContentLength,
+
+    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+    InvalidLastModified {
+        last_modified: String,
+        source: chrono::ParseError,
+    },
+
+    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+    InvalidContentLength {
+        content_length: String,
+        source: std::num::ParseIntError,
+    },
+}
+
+/// Extracts [`ObjectMeta`] from the provided [`HeaderMap`]
+pub fn header_meta(location: &Path, headers: &HeaderMap) -> Result<ObjectMeta, Error> {
+    let last_modified = headers
+        .get(LAST_MODIFIED)
+        .context(MissingLastModifiedSnafu)?;
+
+    let content_length = headers
+        .get(CONTENT_LENGTH)
+        .context(MissingContentLengthSnafu)?;
+
+    let last_modified = last_modified.to_str().context(BadHeaderSnafu)?;
+    let last_modified = DateTime::parse_from_rfc2822(last_modified)
+        .context(InvalidLastModifiedSnafu { last_modified })?
+        .with_timezone(&Utc);
+
+    let content_length = content_length.to_str().context(BadHeaderSnafu)?;
+    let content_length = content_length
+        .parse()
+        .context(InvalidContentLengthSnafu { content_length })?;
+
+    let e_tag = headers.get(ETAG).context(MissingEtagSnafu)?;
+    let e_tag = e_tag.to_str().context(BadHeaderSnafu)?;
+
+    Ok(ObjectMeta {
+        location: location.clone(),
+        last_modified,
+        size: content_length,
+        e_tag: Some(e_tag.to_string()),
+    })
+}