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::error::ErrorExt;
19use mz_ore::url::SensitiveUrl;
20use tracing::warn;
21
22use mz_postgres_client::PostgresClientKnobs;
23use mz_postgres_client::metrics::PostgresClientMetrics;
24
25use crate::azure::{AzureBlob, AzureBlobConfig};
26use crate::file::{FileBlob, FileBlobConfig};
27use crate::gcs::{GcsBlob, GcsBlobConfig};
28use crate::hedge::HedgeSibling;
29use crate::location::{Blob, Consensus, Determinate, ExternalError};
30use crate::mem::{MemBlob, MemBlobConfig, MemConsensus};
31use crate::metrics::S3BlobMetrics;
32use crate::postgres::{PostgresConsensus, PostgresConsensusConfig};
33use crate::s3::{BlobHttpPool, S3Blob, S3BlobConfig};
34
35/// Adds the full set of all mz_persist `Config`s.
36pub fn all_dyn_configs(configs: ConfigSet) -> ConfigSet {
37    configs
38        .add(&crate::postgres::PG_CONSENSUS_READ_COMMITTED)
39        .add(&crate::hedge::BLOB_HEDGED_GET_ENABLED)
40        .add(&crate::hedge::BLOB_HEDGED_GET_DELAY)
41        .add(&crate::hedge::BLOB_HEDGED_GET_MAX_CONCURRENT)
42        .add(&crate::hedge::BLOB_HEDGED_GET_BUDGET_RATIO)
43        .add(&crate::hedge::BLOB_HEDGED_GET_WARM_INTERVAL)
44        .add(&crate::s3::BLOB_HTTP_POOL)
45        .add(&crate::s3::BLOB_HTTP_POOL_MAX_IN_FLIGHT_PER_CLIENT)
46}
47
48/// Opens the sibling handle that [crate::hedge::HedgedBlob] runs hedge
49/// requests on for `url`.
50///
51/// Contract:
52/// - An [HedgeSibling::Isolated] handle observes exactly the same durable
53///   store as a handle opened from the same `url`, but is built from a
54///   scratch client: it shares no HTTP connection pool, DNS state, or
55///   credential chain, so a hedge request on it can never be assigned a
56///   connection the primary's pool has already half-killed.
57/// - Backends where a second open would observe an independent store (mem,
58///   turmoil's simulated store), or that have no connection state to isolate
59///   (file), return [HedgeSibling::SharedWithPrimary] instead.
60/// - Callers must use the handle only for idempotent reads.
61///
62/// Errors opening the sibling degrade to [HedgeSibling::Unavailable] with a
63/// warning rather than failing: persist must come up even if hedging cannot.
64/// What a caller does with that is its own contract, see
65/// [crate::hedge::HedgedBlob::new_arming].
66pub async fn open_hedge_sibling(
67    url: &SensitiveUrl,
68    knobs: Box<dyn BlobKnobs>,
69    metrics: S3BlobMetrics,
70) -> HedgeSibling {
71    // The HTTP pool metrics describe only the primary's requests.
72    let metrics = metrics.without_http_pool_metrics();
73    let config = match BlobConfig::try_from(url, knobs, metrics).await {
74        Ok(config) => config,
75        Err(err) => {
76            warn!(
77                "hedged blob gets unavailable, sibling config failed: {}",
78                err.display_with_causes()
79            );
80            return HedgeSibling::Unavailable;
81        }
82    };
83    match config {
84        // A second S3/Azure/GCS config builds its own SDK client and therefore
85        // its own connection pool, with DNS resolved per connect.
86        config @ (BlobConfig::S3(_) | BlobConfig::Azure(_) | BlobConfig::Gcs(_)) => {
87            match config.open().await {
88                Ok(blob) => HedgeSibling::Isolated(blob),
89                Err(err) => {
90                    warn!(
91                        "hedged blob gets unavailable, sibling open failed: {}",
92                        err.display_with_causes()
93                    );
94                    HedgeSibling::Unavailable
95                }
96            }
97        }
98        // File has no connection pool to isolate, so a second instance would
99        // buy nothing. A second open of Mem (or of turmoil's simulated
100        // store) would be actively wrong: it creates an INDEPENDENT store,
101        // and a hedged get against a different store can return `Ok(None)`
102        // for data that exists.
103        BlobConfig::File(_) | BlobConfig::Mem(_) => HedgeSibling::SharedWithPrimary,
104        #[cfg(feature = "turmoil")]
105        BlobConfig::Turmoil(_) => HedgeSibling::SharedWithPrimary,
106    }
107}
108
109/// Config for an implementation of [Blob].
110#[derive(Debug, Clone)]
111pub enum BlobConfig {
112    /// Config for [FileBlob].
113    File(FileBlobConfig),
114    /// Config for [S3Blob].
115    S3(S3BlobConfig),
116    /// Config for [MemBlob], only available in testing to prevent
117    /// footguns.
118    Mem(bool),
119    /// Config for [AzureBlob].
120    Azure(AzureBlobConfig),
121    /// Config for [GcsBlob].
122    Gcs(GcsBlobConfig),
123    #[cfg(feature = "turmoil")]
124    /// Config for [crate::turmoil::TurmoilBlob].
125    Turmoil(crate::turmoil::BlobConfig),
126}
127
128/// Configuration knobs for [Blob].
129pub trait BlobKnobs: std::fmt::Debug + Send + Sync {
130    /// Maximum time allowed for a network call, including retry attempts.
131    fn operation_timeout(&self) -> Duration;
132    /// Maximum time allowed for a single network call.
133    fn operation_attempt_timeout(&self) -> Duration;
134    /// Maximum time to wait for a socket connection to be made.
135    fn connect_timeout(&self) -> Duration;
136    /// Maximum time to wait to read the first byte of a response, including connection time.
137    fn read_timeout(&self) -> Duration;
138    /// Whether this is running in a "cc" sized cluster.
139    fn is_cc_active(&self) -> bool;
140    /// Which S3 endpoints send requests through a pool of HTTP clients.
141    fn http_pool(&self) -> BlobHttpPool {
142        BlobHttpPool::default()
143    }
144    /// See [`crate::s3::BLOB_HTTP_POOL_MAX_IN_FLIGHT_PER_CLIENT`].
145    fn http_pool_max_in_flight_per_client(&self) -> usize {
146        *crate::s3::BLOB_HTTP_POOL_MAX_IN_FLIGHT_PER_CLIENT.default()
147    }
148}
149
150impl BlobConfig {
151    /// Opens the associated implementation of [Blob].
152    pub async fn open(self) -> Result<Arc<dyn Blob>, ExternalError> {
153        match self {
154            BlobConfig::File(config) => Ok(Arc::new(FileBlob::open(config).await?)),
155            BlobConfig::S3(config) => Ok(Arc::new(S3Blob::open(config).await?)),
156            BlobConfig::Azure(config) => Ok(Arc::new(AzureBlob::open(config).await?)),
157            BlobConfig::Gcs(config) => Ok(Arc::new(GcsBlob::open(config).await?)),
158            BlobConfig::Mem(tombstone) => {
159                Ok(Arc::new(MemBlob::open(MemBlobConfig::new(tombstone))))
160            }
161            #[cfg(feature = "turmoil")]
162            BlobConfig::Turmoil(config) => Ok(Arc::new(crate::turmoil::TurmoilBlob::open(config))),
163        }
164    }
165
166    /// Parses a [Blob] config from a uri string.
167    pub async fn try_from(
168        url: &SensitiveUrl,
169        knobs: Box<dyn BlobKnobs>,
170        metrics: S3BlobMetrics,
171    ) -> Result<Self, ExternalError> {
172        let mut query_params = url.query_pairs().collect::<BTreeMap<_, _>>();
173
174        let config = match url.scheme() {
175            "file" => {
176                let mut config = FileBlobConfig::from(url.path());
177                if query_params.remove("tombstone").is_some() {
178                    config.tombstone = true;
179                }
180                Ok(BlobConfig::File(config))
181            }
182            "s3" => {
183                let bucket = url
184                    .host()
185                    .ok_or_else(|| anyhow!("missing bucket: {}", url))?
186                    .to_string();
187                let prefix = url
188                    .path()
189                    .strip_prefix('/')
190                    .unwrap_or_else(|| url.path())
191                    .to_string();
192                let role_arn = query_params.remove("role_arn").map(|x| x.into_owned());
193                let endpoint = query_params.remove("endpoint").map(|x| x.into_owned());
194                let region = query_params.remove("region").map(|x| x.into_owned());
195
196                let credentials = match url.password() {
197                    None => None,
198                    Some(password) => Some((
199                        String::from_utf8_lossy(&urlencoding::decode_binary(
200                            url.username().as_bytes(),
201                        ))
202                        .into_owned(),
203                        String::from_utf8_lossy(&urlencoding::decode_binary(password.as_bytes()))
204                            .into_owned(),
205                    )),
206                };
207
208                let config = S3BlobConfig::new(
209                    bucket,
210                    prefix,
211                    role_arn,
212                    endpoint,
213                    region,
214                    credentials,
215                    knobs,
216                    metrics,
217                )
218                .await?;
219
220                Ok(BlobConfig::S3(config))
221            }
222            "gcs" => {
223                let bucket = url
224                    .host()
225                    .ok_or_else(|| anyhow!("missing bucket: {}", url.as_str()))?
226                    .to_string();
227                let prefix = url
228                    .path()
229                    .strip_prefix('/')
230                    .unwrap_or_else(|| url.path())
231                    .to_string();
232
233                query_params.clear();
234                let config = GcsBlobConfig::new(bucket, prefix, knobs, metrics).await?;
235
236                Ok(BlobConfig::Gcs(config))
237            }
238            "mem" => {
239                if !cfg!(debug_assertions) {
240                    warn!("persist unexpectedly using in-mem blob in a release binary");
241                }
242                let tombstone = match query_params.remove("tombstone").as_deref() {
243                    None | Some("true") => true,
244                    Some("false") => false,
245                    Some(other) => Err(Determinate::new(anyhow!(
246                        "invalid tombstone param value: {other}"
247                    )))?,
248                };
249                query_params.clear();
250                Ok(BlobConfig::Mem(tombstone))
251            }
252            "http" | "https" => match url
253                .host()
254                .ok_or_else(|| anyhow!("missing protocol: {}", url))?
255                .to_string()
256                .split_once('.')
257            {
258                // The Azurite emulator always uses the well-known account name devstoreaccount1
259                Some((account, root))
260                    if account == "devstoreaccount1" || root == "blob.core.windows.net" =>
261                {
262                    if let Some(container) = url
263                        .path_segments()
264                        .expect("azure blob storage container")
265                        .next()
266                    {
267                        query_params.clear();
268                        Ok(BlobConfig::Azure(AzureBlobConfig::new(
269                            account.to_string(),
270                            container.to_string(),
271                            // Azure doesn't support prefixes in the way S3 does.
272                            // This is always empty, but we leave the field for
273                            // compatibility with our existing test suite.
274                            "".to_string(),
275                            metrics,
276                            url.clone().into_redacted(),
277                            knobs,
278                        )?))
279                    } else {
280                        Err(anyhow!("unknown persist blob scheme: {}", url))
281                    }
282                }
283                _ => Err(anyhow!("unknown persist blob scheme: {}", url)),
284            },
285            #[cfg(feature = "turmoil")]
286            "turmoil" => {
287                let cfg = crate::turmoil::BlobConfig::new(url);
288                Ok(BlobConfig::Turmoil(cfg))
289            }
290            p => Err(anyhow!("unknown persist blob scheme {}: {}", p, url)),
291        }?;
292
293        if !query_params.is_empty() {
294            return Err(ExternalError::from(anyhow!(
295                "unknown blob location params {}: {}",
296                query_params
297                    .keys()
298                    .map(|x| x.as_ref())
299                    .collect::<Vec<_>>()
300                    .join(" "),
301                url,
302            )));
303        }
304
305        Ok(config)
306    }
307}
308
309/// Config for an implementation of [Consensus].
310#[derive(Debug, Clone)]
311pub enum ConsensusConfig {
312    /// Config for [PostgresConsensus].
313    Postgres(PostgresConsensusConfig),
314    /// Config for [MemConsensus], only available in testing.
315    Mem,
316    #[cfg(feature = "turmoil")]
317    /// Config for [crate::turmoil::TurmoilConsensus].
318    Turmoil(crate::turmoil::ConsensusConfig),
319}
320
321impl ConsensusConfig {
322    /// Opens the associated implementation of [Consensus].
323    pub async fn open(self) -> Result<Arc<dyn Consensus>, ExternalError> {
324        match self {
325            ConsensusConfig::Postgres(config) => {
326                Ok(Arc::new(PostgresConsensus::open(config).await?))
327            }
328            ConsensusConfig::Mem => Ok(Arc::new(MemConsensus::default())),
329            #[cfg(feature = "turmoil")]
330            ConsensusConfig::Turmoil(config) => {
331                Ok(Arc::new(crate::turmoil::TurmoilConsensus::open(config)))
332            }
333        }
334    }
335
336    /// Parses a [Consensus] config from a uri string.
337    pub fn try_from(
338        url: &SensitiveUrl,
339        knobs: Box<dyn PostgresClientKnobs>,
340        metrics: PostgresClientMetrics,
341        dyncfg: Arc<ConfigSet>,
342    ) -> Result<Self, ExternalError> {
343        let config = match url.scheme() {
344            "postgres" | "postgresql" => Ok(ConsensusConfig::Postgres(
345                PostgresConsensusConfig::new(url, knobs, metrics, dyncfg)?,
346            )),
347            "mem" => {
348                if !cfg!(debug_assertions) {
349                    warn!("persist unexpectedly using in-mem consensus in a release binary");
350                }
351                Ok(ConsensusConfig::Mem)
352            }
353            #[cfg(feature = "turmoil")]
354            "turmoil" => {
355                let cfg = crate::turmoil::ConsensusConfig::new(url);
356                Ok(ConsensusConfig::Turmoil(cfg))
357            }
358            p => Err(anyhow!("unknown persist consensus scheme {}: {}", p, url)),
359        }?;
360        Ok(config)
361    }
362}
mz_persist/cfg.rs

mz_persist/
cfg.rs
