feat: Azure blob storage support (#955)

add below env vars -
P_AZR_URL=https://.blob.core.windows.net
P_AZR_ACCOUNT=
P_AZR_CONTAINER=
P_AZR_ACCESS_KEY=

command to start Parseable with azure blob storage- parseable blob-store

Fixes: #687

Author: nikhilsinhaparseable

server/Cargo.toml

@@ -16,7 +16,7 @@ arrow-json = "53.0.0"
 arrow-ipc = { version = "53.0.0", features = ["zstd"] }
 arrow-select = "53.0.0"
 datafusion = "42.0.0"
-object_store = { version = "0.11.0", features = ["cloud", "aws"] }
+object_store = { version = "0.11.0", features = ["cloud", "aws", "azure"] }
 parquet = "53.0.0"
 arrow-flight = { version = "53.0.0", features = [ "tls" ] }
 tonic = {version = "0.12.3", features = ["tls", "transport", "gzip", "zstd"] }

server/src/metrics/storage.rs

@@ -86,3 +86,42 @@ pub mod s3 {
         }
     }
 }
+
+pub mod azureblob {
+    use crate::{metrics::METRICS_NAMESPACE, storage::AzureBlobConfig};
+    use once_cell::sync::Lazy;
+    use prometheus::{HistogramOpts, HistogramVec};
+
+    use super::StorageMetrics;
+
+    pub static REQUEST_RESPONSE_TIME: Lazy<HistogramVec> = Lazy::new(|| {
+        HistogramVec::new(
+            HistogramOpts::new("azr_blob_response_time", "AzureBlob Request Latency")
+                .namespace(METRICS_NAMESPACE),
+            &["method", "status"],
+        )
+        .expect("metric can be created")
+    });
+
+    pub static QUERY_LAYER_STORAGE_REQUEST_RESPONSE_TIME: Lazy<HistogramVec> = Lazy::new(|| {
+        HistogramVec::new(
+            HistogramOpts::new("query_azr_blob_response_time", "AzureBlob Request Latency")
+                .namespace(METRICS_NAMESPACE),
+            &["method", "status"],
+        )
+        .expect("metric can be created")
+    });
+
+    impl StorageMetrics for AzureBlobConfig {
+        fn register_metrics(&self, handler: &actix_web_prometheus::PrometheusMetrics) {
+            handler
+                .registry
+                .register(Box::new(REQUEST_RESPONSE_TIME.clone()))
+                .expect("metric can be registered");
+            handler
+                .registry
+                .register(Box::new(QUERY_LAYER_STORAGE_REQUEST_RESPONSE_TIME.clone()))
+                .expect("metric can be registered");
+        }
+    }
+}

server/src/option.rs

@@ -18,7 +18,9 @@
 
 use crate::cli::Cli;
 use crate::storage::object_storage::parseable_json_path;
-use crate::storage::{FSConfig, ObjectStorageError, ObjectStorageProvider, S3Config};
+use crate::storage::{
+    AzureBlobConfig, FSConfig, ObjectStorageError, ObjectStorageProvider, S3Config,
+};
 use bytes::Bytes;
 use clap::error::ErrorKind;
 use clap::{command, Args, Command, FromArgMatches};
@@ -105,6 +107,22 @@ Cloud Native, log analytics platform for modern applications."#,
                     storage_name: "s3",
                 }
             }
+            Some(("blob-store", m)) => {
+                let cli = match Cli::from_arg_matches(m) {
+                    Ok(cli) => cli,
+                    Err(err) => err.exit(),
+                };
+                let storage = match AzureBlobConfig::from_arg_matches(m) {
+                    Ok(storage) => storage,
+                    Err(err) => err.exit(),
+                };
+
+                Config {
+                    parseable: cli,
+                    storage: Arc::new(storage),
+                    storage_name: "blob_store",
+                }
+            }
             _ => unreachable!(),
         }
     }
@@ -163,11 +181,16 @@ Cloud Native, log analytics platform for modern applications."#,
     // returns the string representation of the storage mode
     // drive --> Local drive
     // s3 --> S3 bucket
+    // azure_blob --> Azure Blob Storage
     pub fn get_storage_mode_string(&self) -> &str {
         if self.storage_name == "drive" {
             return "Local drive";
+        } else if self.storage_name == "s3" {
+            return "S3 bucket";
+        } else if self.storage_name == "blob_store" {
+            return "Azure Blob Storage";
         }
-        "S3 bucket"
+        "Unknown"
     }
 
     pub fn get_server_mode_string(&self) -> &str {
@@ -193,6 +216,9 @@ fn create_parseable_cli_command() -> Command {
     let s3 = Cli::create_cli_command_with_clap("s3-store");
     let s3 = <S3Config as Args>::augment_args_for_update(s3);
 
+    let azureblob = Cli::create_cli_command_with_clap("blob-store");
+    let azureblob = <AzureBlobConfig as Args>::augment_args_for_update(azureblob);
+
     command!()
         .name("Parseable")
         .bin_name("parseable")
@@ -207,7 +233,7 @@ Join the community at https://logg.ing/community.
         "#,
         )
         .subcommand_required(true)
-        .subcommands([local, s3])
+        .subcommands([local, s3, azureblob])
 }
 
 #[derive(Debug, Default, Eq, PartialEq)]

server/src/storage.rs

@@ -24,6 +24,7 @@ use chrono::Local;
 
 use std::fmt::Debug;
 
+mod azure_blob;
 mod localfs;
 mod metrics_layer;
 pub(crate) mod object_storage;
@@ -34,6 +35,7 @@ mod store_metadata;
 
 use self::retention::Retention;
 pub use self::staging::StorageDir;
+pub use azure_blob::AzureBlobConfig;
 pub use localfs::FSConfig;
 pub use object_storage::{ObjectStorage, ObjectStorageProvider};
 pub use s3::S3Config;