vector/sinks/gcs_common/
config.rs

1use std::marker::PhantomData;
2
3use futures::FutureExt;
4use http::StatusCode;
5use hyper::Body;
6use snafu::Snafu;
7use vector_lib::configurable::configurable_component;
8
9use crate::{
10    gcp::{GcpAuthenticator, GcpError},
11    http::HttpClient,
12    sinks::{
13        Healthcheck, HealthcheckError,
14        gcs_common::service::GcsResponse,
15        util::{
16            HttpEndpoint,
17            retries::{RetryAction, RetryLogic},
18        },
19    },
20};
21
22pub fn default_endpoint() -> HttpEndpoint {
23    HttpEndpoint::parse("https://storage.googleapis.com")
24        .expect("static default endpoint should be a valid http(s) URL")
25}
26
27/// GCS Predefined ACLs.
28///
29/// For more information, see [Predefined ACLs][predefined_acls].
30///
31/// [predefined_acls]: https://cloud.google.com/storage/docs/access-control/lists#predefined-acl
32#[configurable_component]
33#[derive(Clone, Copy, Debug, Default)]
34#[serde(rename_all = "kebab-case")]
35pub enum GcsPredefinedAcl {
36    /// Bucket/object can be read by authenticated users.
37    ///
38    /// The bucket/object owner is granted the `OWNER` permission, and anyone authenticated Google
39    /// account holder is granted the `READER` permission.
40    AuthenticatedRead,
41
42    /// Object is semi-private.
43    ///
44    /// Both the object owner and bucket owner are granted the `OWNER` permission.
45    ///
46    /// Only relevant when specified for an object: this predefined ACL is otherwise ignored when
47    /// specified for a bucket.
48    BucketOwnerFullControl,
49
50    /// Object is private, except to the bucket owner.
51    ///
52    /// The object owner is granted the `OWNER` permission, and the bucket owner is granted the
53    /// `READER` permission.
54    ///
55    /// Only relevant when specified for an object: this predefined ACL is otherwise ignored when
56    /// specified for a bucket.
57    BucketOwnerRead,
58
59    /// Bucket/object are private.
60    ///
61    /// The bucket/object owner is granted the `OWNER` permission, and no one else has
62    /// access.
63    Private,
64
65    /// Bucket/object are private within the project.
66    ///
67    /// Project owners and project editors are granted the `OWNER` permission, and anyone who is
68    /// part of the project team is granted the `READER` permission.
69    ///
70    /// This is the default.
71    #[default]
72    ProjectPrivate,
73
74    /// Bucket/object can be read publicly.
75    ///
76    /// The bucket/object owner is granted the `OWNER` permission, and all other users, whether
77    /// authenticated or anonymous, are granted the `READER` permission.
78    PublicRead,
79}
80
81/// GCS storage classes.
82///
83/// For more information, see [Storage classes][storage_classes].
84///
85/// [storage_classes]: https://cloud.google.com/storage/docs/storage-classes
86#[configurable_component]
87#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
88#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
89pub enum GcsStorageClass {
90    /// Standard storage.
91    ///
92    /// This is the default.
93    #[default]
94    Standard,
95
96    /// Nearline storage.
97    Nearline,
98
99    /// Coldline storage.
100    Coldline,
101
102    /// Archive storage.
103    Archive,
104}
105
106#[derive(Debug, Snafu)]
107pub enum GcsError {
108    #[snafu(display("Bucket {:?} not found", bucket))]
109    BucketNotFound { bucket: String },
110}
111
112pub fn build_healthcheck(
113    bucket: String,
114    client: HttpClient,
115    base_url: HttpEndpoint,
116    auth: GcpAuthenticator,
117) -> crate::Result<Healthcheck> {
118    let healthcheck = async move {
119        let uri = base_url.into_uri();
120        let mut request = http::Request::head(uri).body(Body::empty())?;
121
122        auth.apply(&mut request);
123
124        let not_found_error = GcsError::BucketNotFound { bucket }.into();
125
126        let response = client.send(request).await?;
127        healthcheck_response(response, not_found_error)
128    };
129
130    Ok(healthcheck.boxed())
131}
132
133pub fn healthcheck_response(
134    response: http::Response<hyper::Body>,
135    not_found_error: crate::Error,
136) -> crate::Result<()> {
137    match response.status() {
138        StatusCode::OK => Ok(()),
139        StatusCode::FORBIDDEN => Err(GcpError::HealthcheckForbidden.into()),
140        StatusCode::NOT_FOUND => Err(not_found_error),
141        status => Err(HealthcheckError::UnexpectedStatus { status }.into()),
142    }
143}
144
145pub struct GcsRetryLogic<Request> {
146    request: PhantomData<Request>,
147}
148
149impl<Request> Default for GcsRetryLogic<Request> {
150    fn default() -> Self {
151        Self {
152            request: PhantomData,
153        }
154    }
155}
156
157impl<Request> Clone for GcsRetryLogic<Request> {
158    fn clone(&self) -> Self {
159        Self {
160            request: PhantomData,
161        }
162    }
163}
164
165// This is a clone of HttpRetryLogic for the Body type, should get merged
166impl<Request: Clone + Send + Sync + 'static> RetryLogic for GcsRetryLogic<Request> {
167    type Error = hyper::Error;
168    type Request = Request;
169    type Response = GcsResponse;
170
171    fn is_retriable_error(&self, _error: &Self::Error) -> bool {
172        true
173    }
174
175    fn should_retry_response(&self, response: &Self::Response) -> RetryAction<Self::Request> {
176        let status = response.inner.status();
177
178        match status {
179            StatusCode::UNAUTHORIZED => RetryAction::Retry("unauthorized".into()),
180            StatusCode::REQUEST_TIMEOUT => RetryAction::Retry("request timeout".into()),
181            StatusCode::TOO_MANY_REQUESTS => RetryAction::Retry("too many requests".into()),
182            StatusCode::NOT_IMPLEMENTED => {
183                RetryAction::DontRetry("endpoint not implemented".into())
184            }
185            _ if status.is_server_error() => RetryAction::Retry(status.to_string().into()),
186            _ if status.is_success() => RetryAction::Successful,
187            _ => RetryAction::DontRetry(format!("response status: {status}").into()),
188        }
189    }
190}
vector/sinks/gcs_common/config.rs

vector/sinks/gcs_common/
config.rs
