feat: add gcs implementation for storage

Author: solomonope

Cargo.lock

@@ -10,14 +10,19 @@ build = "build.rs"
 [dependencies]
 # Arrow and DataFusion ecosystem
 arrow = "54.0.0"
-arrow-array = "54.0.0" 
+arrow-array = "54.0.0"
 arrow-flight = { version = "54.0.0", features = ["tls"] }
 arrow-ipc = { version = "54.0.0", features = ["zstd"] }
 arrow-json = "54.0.0"
 arrow-schema = { version = "54.0.0", features = ["serde"] }
 arrow-select = "54.0.0"
 datafusion = "45.0.0"
-object_store = { version = "0.11.2", features = ["cloud", "aws", "azure"] }
+object_store = { version = "0.11.2", features = [
+    "cloud",
+    "aws",
+    "azure",
+    "gcp",
+] }
 parquet = "54.0.0"
 
 # Web server and HTTP-related
@@ -34,7 +39,11 @@ tower-http = { version = "0.6.1", features = ["cors"] }
 url = "2.4.0"
 
 # Connectors dependencies
-rdkafka = { version = "0.37", optional = true, features = ["cmake-build", "tracing", "libz-static"] }
+rdkafka = { version = "0.37", optional = true, features = [
+    "cmake-build",
+    "tracing",
+    "libz-static",
+] }
 sasl2-sys = { version = "0.1.22", optional = true, features = ["vendored"] }
 
 # Authentication and Security
@@ -144,7 +153,14 @@ assets-sha1 = "3e703ef8bedf8ae55fd31713f6267ad14ad3d29d"
 
 [features]
 debug = []
-kafka = ["rdkafka", "rdkafka/ssl-vendored", "rdkafka/ssl", "rdkafka/sasl", "sasl2-sys", "sasl2-sys/vendored"]
+kafka = [
+    "rdkafka",
+    "rdkafka/ssl-vendored",
+    "rdkafka/ssl",
+    "rdkafka/sasl",
+    "sasl2-sys",
+    "sasl2-sys/vendored",
+]
 
 [profile.release-lto]
 inherits = "release"

Cargo.toml

@@ -14,7 +14,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 # build stage
-FROM rust:1.84.0-bookworm AS builder
+FROM docker.io/rust:1.84.0-bookworm AS builder
 
 LABEL org.opencontainers.image.title="Parseable"
 LABEL maintainer="Parseable Team <[email protected]>"

Dockerfile.debug

@@ -0,0 +1,114 @@
+networks:
+  parseable-internal:
+
+services:
+  # query server
+  parseable-query:
+    container_name: parseable-query
+    build:
+      context: .
+      dockerfile: Dockerfile.debug
+    platform: linux/amd64
+    command: ["parseable", "gcs-store"]
+    ports:
+      - "8000:8000"
+    environment:
+      - P_S3_URL=http://minio:9000
+      - P_S3_ACCESS_KEY=parseable
+      - P_S3_SECRET_KEY=supersecret
+      - P_S3_REGION=us-east-1
+      - P_S3_BUCKET=parseable
+      - P_STAGING_DIR=/tmp/data
+      - P_USERNAME=parseableadmin
+      - P_PASSWORD=parseableadmin
+      - P_CHECK_UPDATE=false
+      - P_PARQUET_COMPRESSION_ALGO=snappy
+      - P_MODE=query
+      - RUST_LOG=warn
+    networks:
+      - parseable-internal
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/liveness"]
+      interval: 15s
+      timeout: 20s
+      retries: 5
+    deploy:
+      restart_policy:
+        condition: on-failure
+        delay: 20s
+        max_attempts: 3
+  # ingest server one
+  parseable-ingest-one:
+    container_name: parseable-ingest-one
+    build:
+      context: .
+      dockerfile: Dockerfile.debug
+    platform: linux/amd64
+    command: ["parseable", "gcs-store"]
+    ports:
+      - "8000"
+    environment:
+      - P_S3_URL=http://minio:9000
+      - P_S3_ACCESS_KEY=parseable
+      - P_S3_SECRET_KEY=supersecret
+      - P_S3_REGION=us-east-1
+      - P_S3_BUCKET=parseable
+      - P_STAGING_DIR=/tmp/data
+      - P_USERNAME=parseableadmin
+      - P_PASSWORD=parseableadmin
+      - P_CHECK_UPDATE=false
+      - P_PARQUET_COMPRESSION_ALGO=snappy
+      - P_MODE=ingest
+      - P_INGESTOR_ENDPOINT=parseable-ingest-one:8000
+      - RUST_LOG=warn
+    networks:
+      - parseable-internal
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/liveness"]
+      interval: 15s
+      timeout: 20s
+      retries: 5
+    depends_on:
+      parseable-query:
+        condition: service_healthy
+    deploy:
+      restart_policy:
+        condition: on-failure
+        delay: 20s
+        max_attempts: 3
+
+  quest:
+    platform: linux/amd64
+    image: ghcr.io/parseablehq/quest:main
+    pull_policy: always
+    command:
+      [
+        "load",
+        "http://parseable-query:8000",
+        "parseableadmin",
+        "parseableadmin",
+        "20",
+        "10",
+        "5m",
+        "minio:9000",
+        "parseable",
+        "supersecret",
+        "parseable",
+        "http://parseable-ingest-one:8000",
+        "parseableadmin",
+        "parseableadmin",
+      ]
+    networks:
+      - parseable-internal
+    depends_on:
+      parseable-query:
+        condition: service_healthy
+      parseable-ingest-one:
+        condition: service_healthy
+      minio:
+        condition: service_healthy
+    deploy:
+      restart_policy:
+        condition: on-failure
+        delay: 20s
+        max_attempts: 3

docker-compose-gcs-distributed-test.yaml

@@ -27,7 +27,7 @@ use crate::connectors::kafka::config::KafkaConfig;
 use crate::{
     oidc::{self, OpenidConfig},
     option::{validation, Compression, Mode},
-    storage::{AzureBlobConfig, FSConfig, S3Config},
+    storage::{AzureBlobConfig, FSConfig, GCSConfig, S3Config},
 };
 
 /// Default username and password for Parseable server, used by default for local mode.
@@ -80,6 +80,9 @@ pub enum StorageOptions {
 
     #[command(name = "blob-store")]
     Blob(BlobStoreArgs),
+
+    #[command(name = "gcs-store")]
+    GCS(GCSStoreArgs),
 }
 
 #[derive(Parser)]
@@ -115,6 +118,17 @@ pub struct BlobStoreArgs {
     pub kafka: KafkaConfig,
 }
 
+#[derive(Parser)]
+pub struct GCSStoreArgs {
+    #[command(flatten)]
+    pub options: Options,
+    #[command(flatten)]
+    pub storage: GCSConfig,
+    #[cfg(feature = "kafka")]
+    #[command(flatten)]
+    pub kafka: KafkaConfig,
+}
+
 #[derive(Parser, Debug, Default)]
 pub struct Options {
     // Authentication
@@ -338,7 +352,7 @@ pub struct Options {
 
     #[arg(
         long,
-        env = "P_MEMORY_THRESHOLD", 
+        env = "P_MEMORY_THRESHOLD",
         default_value = "80.0",
         value_parser = validation::validate_percentage,
         help = "Memory utilization threshold percentage (0.0-100.0) for resource monitoring"

src/cli.rs

@@ -125,3 +125,42 @@ pub mod azureblob {
         }
     }
 }
+
+pub mod gcs {
+    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")
+                .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_gcs_response_time", "GCS Request Latency")
+                .namespace(METRICS_NAMESPACE),
+            &["method", "status"],
+        )
+        .expect("metric can be created")
+    });
+
+    impl StorageMetrics for GCSConfig {
+        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");
+        }
+    }
+}

src/metrics/storage.rs

@@ -117,6 +117,12 @@ pub static PARSEABLE: Lazy<Parseable> = Lazy::new(|| match Cli::parse().storage
         args.kafka,
         Arc::new(args.storage),
     ),
+    StorageOptions::GCS(args) => Parseable::new(
+        args.options,
+        #[cfg(feature = "kafka")]
+        args.kafka,
+        Arc::new(args.storage),
+    ),
 });
 
 /// All state related to parseable, in one place.