feat: fix coderabbit feedback

Author: solomonope

docker-compose-gcs-distributed-test.yaml

@@ -56,7 +56,7 @@ services:
       - P_MODE=ingest
       - P_INGESTOR_ENDPOINT=parseable-ingest-one:8000
       - RUST_LOG=warn
-      - GOOGLE_APPLICATION_CREDENTIALS=/parseable/svc/mpt-randd-8217aef869fd.json
+      - GOOGLE_APPLICATION_CREDENTIALS=/parseable/svc/${GCS_CREDENTIALS_FILE:-key.json}
     networks:
       - parseable-internal
     healthcheck:
@@ -73,7 +73,7 @@ services:
         delay: 20s
         max_attempts: 3
     volumes:
-      - "/home/opeyemi/Downloads/:/parseable/svc/:ro,z"
+      - "${GCS_CREDENTIALS_PATH:-./credentials}:/parseable/svc/:ro,z"
 
   quest:
     platform: linux/amd64

src/cli.rs

@@ -27,7 +27,7 @@ use crate::connectors::kafka::config::KafkaConfig;
 use crate::{
     oidc::{self, OpenidConfig},
     option::{validation, Compression, Mode},
-    storage::{AzureBlobConfig, FSConfig, GCSConfig, S3Config},
+    storage::{AzureBlobConfig, FSConfig, GcsConfig, S3Config},
 };
 
 /// Default username and password for Parseable server, used by default for local mode.
@@ -82,7 +82,7 @@ pub enum StorageOptions {
     Blob(BlobStoreArgs),
 
     #[command(name = "gcs-store")]
-    GCS(GCSStoreArgs),
+    Gcs(GcsStoreArgs),
 }
 
 #[derive(Parser)]
@@ -119,11 +119,11 @@ pub struct BlobStoreArgs {
 }
 
 #[derive(Parser)]
-pub struct GCSStoreArgs {
+pub struct GcsStoreArgs {
     #[command(flatten)]
     pub options: Options,
     #[command(flatten)]
-    pub storage: GCSConfig,
+    pub storage: GcsConfig,
     #[cfg(feature = "kafka")]
     #[command(flatten)]
     pub kafka: KafkaConfig,

src/metrics/storage.rs

@@ -127,15 +127,15 @@ pub mod azureblob {
 }
 
 pub mod gcs {
-    use crate::{metrics::METRICS_NAMESPACE, storage::GCSConfig};
+    use crate::{metrics::METRICS_NAMESPACE, storage::GcsConfig};
     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("gcs_response_time", "gcs Request Latency")
+            HistogramOpts::new("gcs_response_time", "GCS Request Latency")
                 .namespace(METRICS_NAMESPACE),
             &["method", "status"],
         )
@@ -151,7 +151,7 @@ pub mod gcs {
         .expect("metric can be created")
     });
 
-    impl StorageMetrics for GCSConfig {
+    impl StorageMetrics for GcsConfig {
         fn register_metrics(&self, handler: &actix_web_prometheus::PrometheusMetrics) {
             handler
                 .registry

src/parseable/mod.rs

@@ -117,7 +117,7 @@ pub static PARSEABLE: Lazy<Parseable> = Lazy::new(|| match Cli::parse().storage
         args.kafka,
         Arc::new(args.storage),
     ),
-    StorageOptions::GCS(args) => Parseable::new(
+    StorageOptions::Gcs(args) => Parseable::new(
         args.options,
         #[cfg(feature = "kafka")]
         args.kafka,

src/storage/gcs.rs

@@ -66,8 +66,7 @@ use super::{
 {all-args}
 "
 )]
-
-pub struct GCSConfig {
+pub struct GcsConfig {
     /// The endpoint to GCS or compatible object storage platform
     #[arg(
         long,
@@ -90,14 +89,14 @@ pub struct GCSConfig {
     /// Set client to skip tls verification
     #[arg(
         long,
-        env = "P_S3_TLS_SKIP_VERIFY",
+        env = "P_GCS_TLS_SKIP_VERIFY",
         value_name = "bool",
         default_value = "false"
     )]
     pub skip_tls: bool,
 }
 
-impl GCSConfig {
+impl GcsConfig {
     fn get_default_builder(&self) -> GoogleCloudStorageBuilder {
         let mut client_options = ClientOptions::default()
             .with_allow_http(true)
@@ -121,7 +120,7 @@ impl GCSConfig {
     }
 }
 
-impl ObjectStorageProvider for GCSConfig {
+impl ObjectStorageProvider for GcsConfig {
     fn name(&self) -> &'static str {
         "gcs"
     }
@@ -134,6 +133,8 @@ impl ObjectStorageProvider for GCSConfig {
         let gcs = MetricLayer::new(gcs);
 
         let object_store_registry = DefaultObjectStoreRegistry::new();
+        // Register GCS client under the "s3://" scheme so DataFusion can route
+        // object store calls to our GoogleCloudStorage implementatio
         let url = ObjectStoreUrl::parse(format!("s3://{}", &self.bucket_name)).unwrap();
         object_store_registry.register_store(url.as_ref(), Arc::new(gcs));
 
@@ -143,7 +144,7 @@ impl ObjectStorageProvider for GCSConfig {
     fn construct_client(&self) -> Arc<dyn ObjectStorage> {
         let gcs = self.get_default_builder().build().unwrap();
 
-        Arc::new(GCS {
+        Arc::new(Gcs {
             client: Arc::new(gcs),
             bucket: self.bucket_name.clone(),
             root: StorePath::from(""),
@@ -167,13 +168,13 @@ impl ObjectStorageProvider for GCSConfig {
 }
 
 #[derive(Debug)]
-pub struct GCS {
+pub struct Gcs {
     client: Arc<GoogleCloudStorage>,
     bucket: String,
     root: StorePath,
 }
 
-impl GCS {
+impl Gcs {
     async fn _get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError> {
         let instant = Instant::now();
 
@@ -215,7 +216,7 @@ impl GCS {
             let source_str = source.to_string();
             if source_str.contains("<Code>NoSuchBucket</Code>") {
                 return Err(ObjectStorageError::Custom(
-                    format!("Bucket '{}' does not exist in S3.", self.bucket).to_string(),
+                    format!("Bucket '{}' does not exist in GCS.", self.bucket).to_string(),
                 ));
             }
         }
@@ -335,7 +336,7 @@ impl GCS {
         } else {
             let bytes = tokio::fs::read(path).await?;
             let result = self.client.put(&key.into(), bytes.into()).await?;
-            info!("Uploaded file to S3: {:?}", result);
+            info!("Uploaded file to GCS: {:?}", result);
             Ok(())
         };
 
@@ -406,7 +407,7 @@ impl GCS {
 }
 
 #[async_trait]
-impl ObjectStorage for GCS {
+impl ObjectStorage for Gcs {
     async fn get_buffered_reader(
         &self,
         path: &RelativePath,

src/storage/mod.rs

@@ -47,7 +47,7 @@ pub mod store_metadata;
 
 use self::retention::Retention;
 pub use azure_blob::AzureBlobConfig;
-pub use gcs::GCSConfig;
+pub use gcs::GcsConfig;
 pub use localfs::FSConfig;
 pub use object_storage::{ObjectStorage, ObjectStorageProvider};
 pub use s3::S3Config;