mz_persist/
cfg.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Configuration for [crate::location] implementations.
11
12use std::collections::BTreeMap;
13use std::sync::Arc;
14use std::time::Duration;
15
16use anyhow::anyhow;
17use mz_dyncfg::ConfigSet;
18use mz_ore::url::SensitiveUrl;
19use tracing::warn;
20
21use mz_postgres_client::PostgresClientKnobs;
22use mz_postgres_client::metrics::PostgresClientMetrics;
23
24use crate::azure::{AzureBlob, AzureBlobConfig};
25use crate::file::{FileBlob, FileBlobConfig};
26#[cfg(feature = "foundationdb")]
27use crate::foundationdb::{FdbConsensus, FdbConsensusConfig};
28use crate::location::{Blob, Consensus, Determinate, ExternalError};
29use crate::mem::{MemBlob, MemBlobConfig, MemConsensus};
30use crate::metrics::S3BlobMetrics;
31use crate::postgres::{PostgresConsensus, PostgresConsensusConfig};
32use crate::s3::{S3Blob, S3BlobConfig};
33
34/// Adds the full set of all mz_persist `Config`s.
35pub fn all_dyn_configs(configs: ConfigSet) -> ConfigSet {
36    configs.add(&crate::postgres::USE_POSTGRES_TUNED_QUERIES)
37}
38
39/// Config for an implementation of [Blob].
40#[derive(Debug, Clone)]
41pub enum BlobConfig {
42    /// Config for [FileBlob].
43    File(FileBlobConfig),
44    /// Config for [S3Blob].
45    S3(S3BlobConfig),
46    /// Config for [MemBlob], only available in testing to prevent
47    /// footguns.
48    Mem(bool),
49    /// Config for [AzureBlob].
50    Azure(AzureBlobConfig),
51    #[cfg(feature = "turmoil")]
52    /// Config for [crate::turmoil::TurmoilBlob].
53    Turmoil(crate::turmoil::BlobConfig),
54}
55
56/// Configuration knobs for [Blob].
57pub trait BlobKnobs: std::fmt::Debug + Send + Sync {
58    /// Maximum time allowed for a network call, including retry attempts.
59    fn operation_timeout(&self) -> Duration;
60    /// Maximum time allowed for a single network call.
61    fn operation_attempt_timeout(&self) -> Duration;
62    /// Maximum time to wait for a socket connection to be made.
63    fn connect_timeout(&self) -> Duration;
64    /// Maximum time to wait to read the first byte of a response, including connection time.
65    fn read_timeout(&self) -> Duration;
66    /// Whether this is running in a "cc" sized cluster.
67    fn is_cc_active(&self) -> bool;
68}
69
70impl BlobConfig {
71    /// Opens the associated implementation of [Blob].
72    pub async fn open(self) -> Result<Arc<dyn Blob>, ExternalError> {
73        match self {
74            BlobConfig::File(config) => Ok(Arc::new(FileBlob::open(config).await?)),
75            BlobConfig::S3(config) => Ok(Arc::new(S3Blob::open(config).await?)),
76            BlobConfig::Azure(config) => Ok(Arc::new(AzureBlob::open(config).await?)),
77            BlobConfig::Mem(tombstone) => {
78                Ok(Arc::new(MemBlob::open(MemBlobConfig::new(tombstone))))
79            }
80            #[cfg(feature = "turmoil")]
81            BlobConfig::Turmoil(config) => Ok(Arc::new(crate::turmoil::TurmoilBlob::open(config))),
82        }
83    }
84
85    /// Parses a [Blob] config from a uri string.
86    pub async fn try_from(
87        url: &SensitiveUrl,
88        knobs: Box<dyn BlobKnobs>,
89        metrics: S3BlobMetrics,
90    ) -> Result<Self, ExternalError> {
91        let mut query_params = url.query_pairs().collect::<BTreeMap<_, _>>();
92
93        let config = match url.scheme() {
94            "file" => {
95                let mut config = FileBlobConfig::from(url.path());
96                if query_params.remove("tombstone").is_some() {
97                    config.tombstone = true;
98                }
99                Ok(BlobConfig::File(config))
100            }
101            "s3" => {
102                let bucket = url
103                    .host()
104                    .ok_or_else(|| anyhow!("missing bucket: {}", url))?
105                    .to_string();
106                let prefix = url
107                    .path()
108                    .strip_prefix('/')
109                    .unwrap_or_else(|| url.path())
110                    .to_string();
111                let role_arn = query_params.remove("role_arn").map(|x| x.into_owned());
112                let endpoint = query_params.remove("endpoint").map(|x| x.into_owned());
113                let region = query_params.remove("region").map(|x| x.into_owned());
114
115                let credentials = match url.password() {
116                    None => None,
117                    Some(password) => Some((
118                        String::from_utf8_lossy(&urlencoding::decode_binary(
119                            url.username().as_bytes(),
120                        ))
121                        .into_owned(),
122                        String::from_utf8_lossy(&urlencoding::decode_binary(password.as_bytes()))
123                            .into_owned(),
124                    )),
125                };
126
127                let config = S3BlobConfig::new(
128                    bucket,
129                    prefix,
130                    role_arn,
131                    endpoint,
132                    region,
133                    credentials,
134                    knobs,
135                    metrics,
136                )
137                .await?;
138
139                Ok(BlobConfig::S3(config))
140            }
141            "mem" => {
142                if !cfg!(debug_assertions) {
143                    warn!("persist unexpectedly using in-mem blob in a release binary");
144                }
145                let tombstone = match query_params.remove("tombstone").as_deref() {
146                    None | Some("true") => true,
147                    Some("false") => false,
148                    Some(other) => Err(Determinate::new(anyhow!(
149                        "invalid tombstone param value: {other}"
150                    )))?,
151                };
152                query_params.clear();
153                Ok(BlobConfig::Mem(tombstone))
154            }
155            "http" | "https" => match url
156                .host()
157                .ok_or_else(|| anyhow!("missing protocol: {}", url))?
158                .to_string()
159                .split_once('.')
160            {
161                // The Azurite emulator always uses the well-known account name devstoreaccount1
162                Some((account, root))
163                    if account == "devstoreaccount1" || root == "blob.core.windows.net" =>
164                {
165                    if let Some(container) = url
166                        .path_segments()
167                        .expect("azure blob storage container")
168                        .next()
169                    {
170                        query_params.clear();
171                        Ok(BlobConfig::Azure(AzureBlobConfig::new(
172                            account.to_string(),
173                            container.to_string(),
174                            // Azure doesn't support prefixes in the way S3 does.
175                            // This is always empty, but we leave the field for
176                            // compatibility with our existing test suite.
177                            "".to_string(),
178                            metrics,
179                            url.clone().into_redacted(),
180                            knobs,
181                        )?))
182                    } else {
183                        Err(anyhow!("unknown persist blob scheme: {}", url))
184                    }
185                }
186                _ => Err(anyhow!("unknown persist blob scheme: {}", url)),
187            },
188            #[cfg(feature = "turmoil")]
189            "turmoil" => {
190                let cfg = crate::turmoil::BlobConfig::new(url);
191                Ok(BlobConfig::Turmoil(cfg))
192            }
193            p => Err(anyhow!("unknown persist blob scheme {}: {}", p, url)),
194        }?;
195
196        if !query_params.is_empty() {
197            return Err(ExternalError::from(anyhow!(
198                "unknown blob location params {}: {}",
199                query_params
200                    .keys()
201                    .map(|x| x.as_ref())
202                    .collect::<Vec<_>>()
203                    .join(" "),
204                url,
205            )));
206        }
207
208        Ok(config)
209    }
210}
211
212/// Config for an implementation of [Consensus].
213#[derive(Debug, Clone)]
214pub enum ConsensusConfig {
215    #[cfg(feature = "foundationdb")]
216    /// Config for FoundationDB.
217    FoundationDB(FdbConsensusConfig),
218    /// Config for [PostgresConsensus].
219    Postgres(PostgresConsensusConfig),
220    /// Config for [MemConsensus], only available in testing.
221    Mem,
222    #[cfg(feature = "turmoil")]
223    /// Config for [crate::turmoil::TurmoilConsensus].
224    Turmoil(crate::turmoil::ConsensusConfig),
225}
226
227impl ConsensusConfig {
228    /// Opens the associated implementation of [Consensus].
229    pub async fn open(self) -> Result<Arc<dyn Consensus>, ExternalError> {
230        match self {
231            #[cfg(feature = "foundationdb")]
232            ConsensusConfig::FoundationDB(config) => {
233                Ok(Arc::new(FdbConsensus::open(config).await?))
234            }
235            ConsensusConfig::Postgres(config) => {
236                Ok(Arc::new(PostgresConsensus::open(config).await?))
237            }
238            ConsensusConfig::Mem => Ok(Arc::new(MemConsensus::default())),
239            #[cfg(feature = "turmoil")]
240            ConsensusConfig::Turmoil(config) => {
241                Ok(Arc::new(crate::turmoil::TurmoilConsensus::open(config)))
242            }
243        }
244    }
245
246    /// Parses a [Consensus] config from a uri string.
247    pub fn try_from(
248        url: &SensitiveUrl,
249        knobs: Box<dyn PostgresClientKnobs>,
250        metrics: PostgresClientMetrics,
251        dyncfg: Arc<ConfigSet>,
252    ) -> Result<Self, ExternalError> {
253        let config = match url.scheme() {
254            #[cfg(feature = "foundationdb")]
255            "foundationdb" => Ok(ConsensusConfig::FoundationDB(FdbConsensusConfig::new(
256                url.clone(),
257            )?)),
258            "postgres" | "postgresql" => Ok(ConsensusConfig::Postgres(
259                PostgresConsensusConfig::new(url, knobs, metrics, dyncfg)?,
260            )),
261            "mem" => {
262                if !cfg!(debug_assertions) {
263                    warn!("persist unexpectedly using in-mem consensus in a release binary");
264                }
265                Ok(ConsensusConfig::Mem)
266            }
267            #[cfg(feature = "turmoil")]
268            "turmoil" => {
269                let cfg = crate::turmoil::ConsensusConfig::new(url);
270                Ok(ConsensusConfig::Turmoil(cfg))
271            }
272            p => Err(anyhow!("unknown persist consensus scheme {}: {}", p, url)),
273        }?;
274        Ok(config)
275    }
276}
mz_persist/cfg.rs

mz_persist/
cfg.rs
