Merge pull request #65 from threefoldtech/development_http_store

Add HTTP store for fetching data using HTTP requests

rfs/Cargo.toml

@@ -55,6 +55,7 @@ rust-s3 = "0.34.0-rc3"
 openssl = { version = "0.10", features = ["vendored"] }
 regex = "1.9.6"
 which = "6.0"
+reqwest = "0.11"
 
 [dependencies.polyfuse]
 branch = "master"

rfs/README.md

@@ -22,7 +22,7 @@ to be able to use from anywhere on your system.
 
 ## Stores
 
-A store in where the actual data lives. A store can be as simple as a `directory` on your local machine in that case the files on the `fl` are only 'accessible' on your local machine. A store can also be a `zdb` running remotely or a cluster of `zdb`. Right now only `dir`, `zdb` and `s3` stores are supported but this will change in the future to support even more stores.
+A store in where the actual data lives. A store can be as simple as a `directory` on your local machine in that case the files on the `fl` are only 'accessible' on your local machine. A store can also be a `zdb` running remotely or a cluster of `zdb`. Right now only `dir`, `http`, `zdb` and `s3` stores are supported but this will change in the future to support even more stores.
 
 ## Usage
 
@@ -41,6 +41,8 @@ The simplest form of `<store-specs>` is a `url`. the store `url` defines the sto
 - `s3`: aws-s3 is used for storing and retrieving large amounts of data (blobs) in buckets (directories). An example `s3://<username>:<password>@<host>:<port>/<bucket-name>`
 
   `region` is an optional param for s3 stores, if you want to provide one you can add it as a query to the url `?region=<region-name>`
+- `http`: http is a store mostly used for wrapping a dir store to fetch data through http requests. It does not support uploading, just fetching the data.
+  It can be set in the FL file as the store to fetch the data with `rfs config`. Example: `http://localhost:9000/store` (https works too).
 
 `<store-specs>` can also be of the form `<start>-<end>=<url>` where `start` and `end` are a hex bytes for partitioning of blob keys. rfs will then store a set of blobs on the defined store if they blob key falls in the `[start:end]` range (inclusive).
 

rfs/src/store/http.rs

@@ -0,0 +1,73 @@
+use super::{Error, Result, Route, Store};
+use reqwest::{self, StatusCode};
+use url::Url;
+
+#[derive(Clone)]
+pub struct HTTPStore {
+    url: Url,
+}
+
+impl HTTPStore {
+    pub async fn make<U: AsRef<str>>(url: &U) -> Result<HTTPStore> {
+        let u = Url::parse(url.as_ref())?;
+        if u.scheme() != "http" && u.scheme() != "https" {
+            return Err(Error::Other(anyhow::Error::msg("invalid scheme")));
+        }
+
+        Ok(HTTPStore::new(u).await?)
+    }
+    pub async fn new<U: Into<Url>>(url: U) -> Result<Self> {
+        let url = url.into();
+        Ok(Self { url })
+    }
+}
+
+#[async_trait::async_trait]
+impl Store for HTTPStore {
+    async fn get(&self, key: &[u8]) -> Result<Vec<u8>> {
+        let file = hex::encode(key);
+        let mut file_path = self.url.clone();
+        file_path
+            .path_segments_mut()
+            .map_err(|_| Error::Other(anyhow::Error::msg("cannot be base")))?
+            .push(&file[0..2])
+            .push(&file);
+        let mut legacy_path = self.url.clone();
+
+        legacy_path
+            .path_segments_mut()
+            .map_err(|_| Error::Other(anyhow::Error::msg("cannot be base")))?
+            .push(&file);
+
+        let data = match reqwest::get(file_path).await {
+            Ok(mut response) => {
+                if response.status() == StatusCode::NOT_FOUND {
+                    response = reqwest::get(legacy_path)
+                        .await
+                        .map_err(|_| Error::KeyNotFound)?;
+                    if response.status() != StatusCode::OK {
+                        return Err(Error::KeyNotFound);
+                    }
+                }
+                if response.status() != StatusCode::OK {
+                    return Err(Error::Unavailable);
+                }
+                response.bytes().await.map_err(|e| Error::Other(e.into()))?
+            }
+            Err(err) => return Err(Error::Other(err.into())),
+        };
+        Ok(data.into())
+    }
+
+    async fn set(&self, _key: &[u8], _blob: &[u8]) -> Result<()> {
+        Err(Error::Other(anyhow::Error::msg(
+            "http store doesn't support uploading",
+        )))
+    }
+
+    fn routes(&self) -> Vec<Route> {
+        let r = Route::url(self.url.clone());
+
+        vec![r]
+    }
+}

rfs/src/store/mod.rs

@@ -1,5 +1,6 @@
 mod bs;
 pub mod dir;
+pub mod http;
 mod router;
 pub mod s3store;
 pub mod zdb;
@@ -16,25 +17,10 @@ pub async fn make<U: AsRef<str>>(u: U) -> Result<Stores> {
     let parsed = url::Url::parse(u.as_ref())?;
 
     match parsed.scheme() {
-        dir::SCHEME => {
-            return Ok(Stores::Dir(
-                dir::DirStore::make(&u)
-                    .await
-                    .expect("failed to make dir store"),
-            ))
-        }
-        "s3" | "s3s" | "s3s+tls" => {
-            return Ok(Stores::S3(s3store::S3Store::make(&u).await.expect(
-                format!("failed to make {} store", parsed.scheme()).as_str(),
-            )))
-        }
-        "zdb" => {
-            return Ok(Stores::ZDB(
-                zdb::ZdbStore::make(&u)
-                    .await
-                    .expect("failed to make zdb store"),
-            ))
-        }
+        dir::SCHEME => return Ok(Stores::Dir(dir::DirStore::make(&u).await?)),
+        "s3" | "s3s" | "s3s+tls" => return Ok(Stores::S3(s3store::S3Store::make(&u).await?)),
+        "zdb" => return Ok(Stores::ZDB(zdb::ZdbStore::make(&u).await?)),
+        "http" | "https" => return Ok(Stores::HTTP(http::HTTPStore::make(&u).await?)),
         _ => return Err(Error::UnknownStore(parsed.scheme().into())),
     }
 }
@@ -207,6 +193,7 @@ pub enum Stores {
     S3(s3store::S3Store),
     Dir(dir::DirStore),
     ZDB(zdb::ZdbStore),
+    HTTP(http::HTTPStore),
 }
 
 #[async_trait::async_trait]
@@ -216,20 +203,23 @@ impl Store for Stores {
             self::Stores::S3(s3_store) => s3_store.get(key).await,
             self::Stores::Dir(dir_store) => dir_store.get(key).await,
             self::Stores::ZDB(zdb_store) => zdb_store.get(key).await,
+            self::Stores::HTTP(http_store) => http_store.get(key).await,
         }
     }
     async fn set(&self, key: &[u8], blob: &[u8]) -> Result<()> {
         match self {
             self::Stores::S3(s3_store) => s3_store.set(key, blob).await,
             self::Stores::Dir(dir_store) => dir_store.set(key, blob).await,
             self::Stores::ZDB(zdb_store) => zdb_store.set(key, blob).await,
+            self::Stores::HTTP(http_store) => http_store.set(key, blob).await,
         }
     }
     fn routes(&self) -> Vec<Route> {
         match self {
             self::Stores::S3(s3_store) => s3_store.routes(),
             self::Stores::Dir(dir_store) => dir_store.routes(),
             self::Stores::ZDB(zdb_store) => zdb_store.routes(),
+            self::Stores::HTTP(http_store) => http_store.routes(),
         }
     }
 }