krabka_remote_storage/
gcs.rs

1//! Native Google Cloud Storage [`RemoteStorageManager`] backend (KIP-405).
2//!
3//! This module is the GCS sibling of [`S3RemoteStorage`]. It builds an
4//! `object_store::gcp::GoogleCloudStorage` client from a [`GcsConfig`] and
5//! wraps it in the same generic [`S3RemoteStorage`] engine with
6//! [`S3RemoteStorage::with_store`].
7//!
8//! That engine is backend-agnostic: the object-key layout, the byte-range
9//! fetch, the multipart upload stream, and the `object_store` error mapping
10//! are all generic over `dyn ObjectStore`, so GCS reuses the whole copy,
11//! fetch, delete, and multipart implementation. There is no separate trait
12//! impl or storage struct.
13//!
14//! ## Authentication
15//!
16//! GCS credentials follow `object_store`'s resolution order, which matches
17//! Google's Application Default Credentials (ADC):
18//!
19//! 1. An explicit service-account JSON key file ([`GcsConfig::service_account_path`]).
20//! 2. An inline service-account JSON key ([`GcsConfig::service_account_key`]).
21//! 3. An application-default-credentials JSON file
22//!    ([`GcsConfig::application_credentials_path`]; when unset, the gcloud
23//!    well-known ADC file under `$HOME/.config/gcloud` if present).
24//! 4. The GKE / GCE metadata server, that is **Workload Identity**. This is
25//!    the keyless production path. Leave all credential fields unset, and the
26//!    metadata server exchanges the pod's bound Kubernetes service account
27//!    for GCS access tokens. No secret material is on disk or in the broker
28//!    config.
29//!
30//! This backend does not need the S3-compatibility shim that reaches GCS
31//! through [`S3RemoteStorage::from_s3_config`]. That shim cannot use Workload
32//! Identity and needs HMAC interoperability keys.
33
34use {
35    crate::{
36        error::RemoteStorageError,
37        s3::{S3RemoteStorage, WormBucket},
38    },
39    krabka_object_store::{GcsConfig, ObjectStoreConfig},
40};
41
42impl S3RemoteStorage {
43    /// Builds a `GoogleCloudStorage` client from `cfg` and wraps it in the
44    /// generic [`S3RemoteStorage`] engine.
45    ///
46    /// With no credential fields set, authentication uses Workload Identity
47    /// or ADC through the metadata server. This is the keyless GKE path.
48    ///
49    /// # Errors
50    ///
51    /// Returns [`RemoteStorageError::InvalidArgument`] if `object_store`'s
52    /// builder rejects the bucket, credential, and endpoint combination. For
53    /// example, the caller supplied both a service-account path and a key, a
54    /// credential file is unreadable, or the bucket name is empty.
55    pub fn from_gcs_config(cfg: &GcsConfig) -> Result<Self, RemoteStorageError> {
56        Self::from_backend_config(
57            &ObjectStoreConfig::Gcs(cfg.clone()),
58            cfg.prefix.clone(),
59            cfg.multipart_threshold,
60            cfg.multipart_chunk_size,
61            WormBucket::Gcs(cfg.clone()),
62        )
63    }
64}
65
66#[cfg(test)]
67mod tests {
68    use {
69        super::*,
70        crate::{
71            storage_manager::{IndexType, RemoteStorageManager},
72            test_support::{TransactionIndex, sample_data, sample_metadata},
73            worm::WormConfig,
74        },
75        assert2::assert,
76        object_store::memory::InMemory,
77        std::{
78            io::{Read, Write},
79            net::TcpListener,
80            sync::Arc,
81            thread,
82        },
83        tempfile::TempDir,
84        tokio::task,
85    };
86
87    // The GCS backend reuses the generic `S3RemoteStorage` engine, so the
88    // copy / fetch / delete round-trip behaviour is already covered by the
89    // `InMemory`-backed suite in `s3.rs`. This test pins that the engine shape
90    // used for GCS still applies prefixes correctly.
91
92    /// End-to-end round-trip against the generic engine through the GCS
93    /// construction path. The test asserts that the engine applies the
94    /// operator-visible prefix. It uses `with_store(InMemory)` because the
95    /// real GCS client needs a live bucket.
96    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
97    async fn engine_round_trips_with_prefix() {
98        let store =
99            S3RemoteStorage::with_store(Arc::new(InMemory::new()), Some("cluster-a".to_owned()));
100        let src = TempDir::new().unwrap();
101        let md = sample_metadata(uuid::Uuid::from_u128(10));
102        task::spawn_blocking(move || {
103            store
104                .copy_log_segment_data(&md, &sample_data(src.path(), TransactionIndex::Omitted))
105                .unwrap();
106            assert!(store.fetch_log_segment(&md, 0, None).unwrap() == b"0123456789");
107            assert!(store.fetch_index(&md, IndexType::Offset).unwrap() == b"OFFSET-IDX");
108        })
109        .await
110        .unwrap();
111    }
112
113    #[test]
114    fn worm_accepts_gcs_with_versioning_and_locked_retention() {
115        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
116        let endpoint = format!("http://{}", listener.local_addr().unwrap());
117        let server = thread::spawn(move || {
118            let (mut socket, _) = listener.accept().unwrap();
119            let mut request = [0; 2048];
120            let read = socket.read(&mut request).unwrap();
121            assert!(
122                String::from_utf8_lossy(&request[..read])
123                    .starts_with("GET /storage/v1/b/archive?fields=")
124            );
125            let body = r#"{"versioning":{"enabled":true},"retentionPolicy":{"retentionPeriod":"86400","isLocked":true}}"#;
126            write!(
127                socket,
128                "HTTP/1.1 200 OK\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
129                body.len()
130            )
131            .unwrap();
132        });
133        let config = GcsConfig {
134            bucket: "archive".into(),
135            endpoint: Some(endpoint),
136            service_account_key: Some(
137                r#"{"private_key":"unused","private_key_id":"unused","client_email":"unused","disable_oauth":true}"#
138                    .into(),
139            ),
140            allow_http: true,
141            ..Default::default()
142        };
143
144        S3RemoteStorage::from_gcs_config(&config)
145            .unwrap()
146            .with_worm(&WormConfig::default())
147            .unwrap();
148        server.join().unwrap();
149    }
150}
krabka_remote_storage/gcs.rs

krabka_remote_storage/
gcs.rs
