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

mz_persist/
cfg.rs
