feat: Optional Storage health check

.cargo/config.toml

@@ -2,3 +2,27 @@
 rustflags = [
     "-Dclippy::suspicious",
 ]
+
+[env]
+OS_URL="0.0.0.0"
+OS_PORT="5000"
+DB_CONNECTION_POOL_SIZE="3"
+DB_HOST="localhost"
+DB_PORT="5432"
+DB_USER="postgres"
+DB_PASS="password1"
+DB_NAME="object-store"
+DB_SCHEMA="public"
+URI_HOST="example.com"
+TRACE_HEADER="x-trace-header"
+LOGGING_THRESHOLD_SECONDS="1"
+RUST_LOG="warn,object_store=debug"
+DD_AGENT_ENABLED="false"
+DD_VERSION="undefined"
+DD_SERVICE="object-store"
+DD_ENV="local"
+HEALTH_SERVICE_ENABLED="true"
+
+# STORAGE
+# Check connectivity to storage on startup
+STORAGE_HEALTH_CHECK="false"

.cargo/fs.config.toml

@@ -0,0 +1,3 @@
+[env]
+STORAGE_TYPE="file_system"
+STORAGE_BASE_PATH="/tmp"

.cargo/google_cloud.config.toml

@@ -0,0 +1,4 @@
+[env]
+STORAGE_TYPE="google_cloud"
+# Required - replace/run with actual bucket name
+# STORAGE_BASE_PATH="bucket-test"

README.md

@@ -1,40 +1,32 @@
 # Object Store
-
-## Status
+An object storage system with strong encryption properties and peer-to-peer replication
 
 [![stability-release-candidate](https://img.shields.io/badge/stability-pre--release-48c9b0.svg)](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#release-candidate)
 [![Latest Release][release-badge]][release-latest]
 [![License][license-badge]][license-url]
 
 [release-badge]: https://img.shields.io/github/v/tag/provenance-io/object-store.svg?sort=semver
 [release-latest]: https://github.com/provenance-io/object-store/releases/latest
-
 [license-badge]: https://img.shields.io/github/license/provenance-io/object-store.svg
 [license-url]: https://github.com/provenance-io/object-store/blob/main/LICENSE
 
-A object storage system with a gRPC interface and strong encryption properties. Currently, only the
-[Provenance DIME](https://developer.provenance.io/docs/pb/p8e/overview/encrypted-object-store/dime-encryption-envelope-specification/)
-format is accepted. Supporting material can be found [here](https://developer.provenance.io/docs/pb/p8e/overview/encrypted-object-store/).
-
-This service can be used directly, but the most common case is to use this alongside the [P8e Execution Environment](https://github.com/provenance-io/p8e-scope-sdk)
-in order to process Provenance scopes and memorialize them on chain.
-
 ## Features
-
-- Strong end-to-end encryption.
+- Strong end-to-end encryption - Currently, only the [Provenance DIME](https://developer.provenance.io/docs/pb/p8e/overview/encrypted-object-store/dime-encryption-envelope-specification/) format is accepted. Supporting material can be found [here](https://developer.provenance.io/docs/pb/p8e/overview/encrypted-object-store/).
 - Peer-to-peer replication to parties (third party `object-store`s) you want to share data with.
-- Capable of publishing traces to Datadog.
+- Configurable storage backends
+- Tracing with [Datadog](https://docs.datadoghq.com).
 - [Additional features](./docs/FEATURES.md)
 
-## Authentication
+## Overview
+An object store can be used directly, but the most common case is to use this alongside the [P8e Execution Environment](https://github.com/provenance-io/p8e-scope-sdk) in order to process Provenance scopes and memorialize them on chain.
 
-gRPC metadata based authentication is provided on a per key basis. When adding a public key to the database, an `auth_type` and `auth_data` can be provided. These
-can either be leveraged directly as an api key or indirectly be combining it with a proxy capable of authentication and header forwarding. Setting both of these fields
-to `null` and a service level config property of `USER_AUTH_ENABLED=false` disables all authentication - this can be used if the object store is meant for internal use
-and not exposed publicly.
-
-- Example authentication configuration
+### Authentication
+gRPC metadata based authentication is provided on a per key basis.
+When adding a public key to the database, an `auth_type` and `auth_data` can be provided.
+These can either be leveraged directly as an api key or indirectly be combining it with a proxy capable of authentication and header forwarding.
+Setting both of these fields to `null` and a service level config property of `USER_AUTH_ENABLED=false` disables all authentication - this can be used if the object store is meant for internal use and not exposed publicly.
 
+#### Example authentication configuration
 NOTE: Requires settings the service level configuration to `USER_AUTH_ENABLED=true`.
 
 ```
@@ -44,22 +36,30 @@ auth_type=header
 auth_data=x-custom-header:6eace982-f682-4b1d-9f8e-82ed9ab15813
 ```
 
-With such a configuration all requests for this public key will have to contain this metadata.
-
-## Backends
+With such a configuration, all requests for this public key _must_ contain this metadata.
 
-This service was designed to support many underlying storage backends. The currently supported backends are `postgres`, `google cloud storage`, and the local `file system`.
-In practice, a sizeable number of objects this system stores are very small. For this reason the `postgres` backend, along with a byte threshold, is provided
-so that items smaller than the threshold specified can have thier bytes stored directly in the database.
+### Storage
+This system supports different storage backends. Currently native file system and Google Cloud are supported.
+In practice, many objects are very small. Therefore the `postgres` backend, along with a byte threshold, is provided where items smaller than the threshold are stored directly in the database.
 
 ## Development
+The most common use case for `object-store` is to run it alongside `p8e` in order to write to the [Provenance blockchain](https://provenance.io).
+The simplest way to get this up and running is [here](https://github.com/provenance-io/p8e-scope-sdk/tree/main/dev-tools/compose).
 
 ### Running Locally
+A postgres connection and data directory is required to run locally.
+A base set of environment variables are defined in [.cargo/config.toml](./.cargo/config.toml).
+Then, specify additional config with one of the sibling config files and/or env variables.
 
-A postgres connection and data directory is required to run locally. A base set of environment variables can be edited and sourced from `./bin/env`.
-
-The most common use case for `object-store` is to run it alongside `p8e` in order to write to the [Provenance blockchain](https://provenance.io). The simplest way to get this
-up and running is [here](https://github.com/provenance-io/p8e-scope-sdk/tree/main/dev-tools/compose).
+**Run with file system storage**
+```shell
+cargo --config .cargo/fs.config.toml run       
+```
+**Run with google cloud storage and health check**
+```shell
+STORAGE_HEALTH_CHECK=true STORAGE_BASE_PATH=my-custom-bucket \
+  cargo --config .cargo/google_cloud.config.toml run     
+```
 
 ### Contributing
 It is recommended to have `rust-analyzer` installed and configured for your IDE.

src/config.rs

@@ -26,6 +26,43 @@ pub struct ReplicationConfig {
     pub snapshot_cache_refresh_frequency: TimeDelta,
 }
 
+#[derive(Clone, Debug)]
+pub struct StorageConfig {
+    /// One of: `file_system`, `google_cloud`
+    pub storage_type: String,
+    pub storage_base_url: Option<String>,
+    pub storage_base_path: String,
+    /// Objects with size, in bytes, below this threshold will be stored in database.
+    /// Larger objects will be in configured storage
+    pub storage_threshold: usize,
+    pub health_check: bool,
+}
+
+impl StorageConfig {
+    pub fn from_env() -> Self {
+        let storage_type = env::var("STORAGE_TYPE").expect("STORAGE_TYPE not set");
+        let storage_base_url = env::var("STORAGE_BASE_URL").ok();
+        let storage_base_path = env::var("STORAGE_BASE_PATH").expect("STORAGE_BASE_PATH not set");
+        let storage_threshold = env::var("STORAGE_THRESHOLD")
+            // default to ~ 5KB
+            .unwrap_or("5000".to_owned())
+            .parse()
+            .expect("STORAGE_THRESHOLD could not be parsed into a u32");
+        let health_check = env::var("STORAGE_HEALTH_CHECK")
+            .unwrap_or("false".to_owned())
+            .parse()
+            .expect("STORAGE_HEALTH_CHECK could not be parsed into a bool");
+
+        Self {
+            storage_type,
+            storage_base_url,
+            storage_base_path,
+            storage_threshold,
+            health_check,
+        }
+    }
+}
+
 impl ReplicationConfig {
     pub fn new(
         replication_enabled: bool,
@@ -57,13 +94,7 @@ pub struct Config {
     pub db_password: String,
     pub db_database: String,
     pub db_schema: String,
-    /// One of: `file_system`, `google_cloud`
-    pub storage_type: String,
-    pub storage_base_url: Option<String>,
-    pub storage_base_path: String,
-    /// Objects with size, in bytes, below this threshold will be stored in database.
-    /// Larger objects will be in configured storage
-    pub storage_threshold: usize,
+    pub storage_config: StorageConfig,
     pub replication_config: ReplicationConfig,
     /// If None, trace middleware [MinitraceGrpcMiddlewareLayer][crate::middleware::MinitraceGrpcMiddlewareLayer] disabled
     pub dd_config: Option<DatadogConfig>,
@@ -105,14 +136,7 @@ impl Config {
         let db_password = env::var("DB_PASS").expect("DB_PASS not set");
         let db_database = env::var("DB_NAME").expect("DB_NAME not set");
         let db_schema = env::var("DB_SCHEMA").expect("DB_SCHEMA not set");
-        let storage_type = env::var("STORAGE_TYPE").expect("STORAGE_TYPE not set");
-        let storage_base_url = env::var("STORAGE_BASE_URL").ok();
-        let storage_base_path = env::var("STORAGE_BASE_PATH").expect("STORAGE_BASE_PATH not set");
-        let storage_threshold = env::var("STORAGE_THRESHOLD")
-            // default to ~ 5KB
-            .unwrap_or("5000".to_owned())
-            .parse()
-            .expect("STORAGE_THRESHOLD could not be parsed into a u32");
+
         let replication_batch_size = env::var("REPLICATION_BATCH_SIZE")
             .unwrap_or("10".to_owned())
             .parse()
@@ -221,10 +245,7 @@ impl Config {
             db_password,
             db_database,
             db_schema,
-            storage_type,
-            storage_base_url,
-            storage_base_path,
-            storage_threshold,
+            storage_config: StorageConfig::from_env(),
             replication_config,
             dd_config,
             logging_threshold_seconds,

src/domain.rs

@@ -33,7 +33,7 @@ impl ObjectApiResponse for Object {
             }),
             hash: self.hash.decoded()?,
             uri: format!("object://{}/{}", &config.uri_host, &self.hash),
-            bucket: config.storage_base_path.clone(),
+            bucket: config.storage_config.storage_base_path.clone(),
             name: self.name.clone(),
             metadata: Some(ObjectMetadata {
                 sha512: Vec::new(), // TODO get hash of whole dime?

src/lib.rs

@@ -60,7 +60,7 @@ impl AppContext {
     pub async fn new(config: Arc<Config>) -> Result<Self, OsError> {
         let db_pool = connect_and_migrate(&config).await?;
         let cache = Cache::new(db_pool.clone()).await?;
-        let storage = new_storage(&config).await?;
+        let storage = new_storage(&config.storage_config).await?;
 
         let admin_service = AdminGrpc::new(config.clone());
         let public_key_service = PublicKeyGrpc::new(cache.clone(), config.clone(), db_pool.clone());

src/object.rs

@@ -277,7 +277,8 @@ impl ObjectService for ObjectGrpc {
         // mail items should be under any reasonable threshold set, but explicitly check for
         // mail and always used database storage
         let is_mail = dime.metadata.contains_key(consts::MAILBOX_KEY);
-        let above_storage_threshold = dime_properties.dime_length > self.config.storage_threshold;
+        let above_storage_threshold =
+            dime_properties.dime_length > self.config.storage_config.storage_threshold;
 
         let response = if !is_mail && above_storage_threshold {
             let response = datastore::put_object(

src/storage/error.rs

@@ -5,6 +5,9 @@ quick_error! {
     pub enum StorageError {
         IoError(message: String) { }
         ContentLengthError(message: String) { }
+        Utf8Error(err: std::string::FromUtf8Error) {
+            from()
+        }
     }
 }
 

src/storage/file_system.rs

@@ -28,38 +28,61 @@ impl FileSystem {
             Ok(_) => Ok(()),
             Err(e) => match e.kind() {
                 std::io::ErrorKind::AlreadyExists => Ok(()),
-                _ => Err(StorageError::IoError(format!("{:?}", e))),
+                _ => Err(StorageError::IoError(format!("{:?} - {:?}", e, path_buf))),
             },
         }
     }
 }
 
 #[async_trait::async_trait]
 impl Storage for FileSystem {
+    #[trace(name = "file_system::health_check")]
+    async fn health_check(&self) -> Result<()> {
+        let test_file = StoragePath {
+            dir: "test".to_owned(),
+            file: "test.txt".to_owned(),
+        };
+
+        self.store(&test_file, 12, b"fs connected").await?;
+
+        let response = self.fetch(&test_file, 12).await?;
+
+        let contents = String::from_utf8(response)?;
+
+        log::debug!("health check - {} contents: {}", test_file, contents);
+
+        // Ignore delete errors
+        let _ = self.delete(&test_file).await;
+
+        Ok(())
+    }
+
     #[trace(name = "file_system::store")]
     async fn store(&self, path: &StoragePath, content_length: usize, data: &[u8]) -> Result<()> {
         if let Err(e) = self.validate_content_length(path, content_length, data) {
             log::warn!("{:?}", e);
         }
 
+        let file_path = self.get_path(path);
+
         let file = OpenOptions::new()
             .write(true)
             .create_new(true)
-            .open(self.get_path(path))
+            .open(&file_path)
             .await;
 
         match file {
             Ok(mut file) => file
                 .write_all(data)
                 .await
-                .map_err(|e| StorageError::IoError(format!("{:?}", e))),
+                .map_err(|e| StorageError::IoError(format!("{:?}: {:?}", e, file_path))),
             Err(e) => match e.kind() {
                 std::io::ErrorKind::AlreadyExists => Ok(()),
                 std::io::ErrorKind::NotFound => {
                     self.create_dir(path).await?;
                     self.store(path, content_length, data).await
                 }
-                _ => Err(StorageError::IoError(format!("{:?}", e))),
+                _ => Err(StorageError::IoError(format!("{:?}: {:?}", e, file_path))),
             },
         }
     }
@@ -79,6 +102,21 @@ impl Storage for FileSystem {
 
         Ok(data)
     }
+
+    #[trace(name = "file_system::delete")]
+    async fn delete(&self, path: &StoragePath) -> Result<()> {
+        let full_path = self.get_path(path);
+        let absolute_path = full_path.canonicalize().unwrap_or(full_path.clone());
+
+        tokio::fs::remove_file(full_path).await.map_err(|e| {
+            StorageError::IoError(format!(
+                "Unable to delete file: {:?} {:?}",
+                absolute_path, e
+            ))
+        })?;
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]