mz_balancerd/
dyncfgs.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//! Dyncfgs used by the balancer.
11
12use std::str::FromStr;
13use std::time::Duration;
14
15use anyhow::anyhow;
16use mz_dyncfg::{Config, ConfigSet, ConfigUpdates, ParameterScope};
17use mz_tracing::params::TracingParameters;
18use mz_tracing::{CloneableEnvFilter, SerializableDirective};
19use tracing_subscriber::filter::Directive;
20
21// The defaults here must be set to an appropriate value in case LaunchDarkly is down because we
22// continue startup even in that case.
23//
24// All configuration names should be prefixed with "balancerd_" to avoid name collisions.
25/// Duration to wait after listeners closed via SIGTERM for outstanding connections to complete.
26pub const SIGTERM_CONNECTION_WAIT: Config<Duration> = Config::new(
27    "balancerd_sigterm_connection_wait",
28    Duration::from_secs(60 * 9),
29    "Duration to wait after listeners closed via SIGTERM for outstanding connections to complete.",
30    ParameterScope::Environment,
31);
32
33/// Duration to wait after SIGTERM to begin shutdown of servers.
34pub const SIGTERM_LISTEN_WAIT: Config<Duration> = Config::new(
35    "balancerd_sigterm_listen_wait",
36    Duration::from_secs(60),
37    "Duration to wait after SIGTERM to begin shutdown of servers.",
38    ParameterScope::Environment,
39);
40
41/// Whether to inject tcp proxy protocol headers to downstream http servers.
42pub const INJECT_PROXY_PROTOCOL_HEADER_HTTP: Config<bool> = Config::new(
43    "balancerd_inject_proxy_protocol_header_http",
44    false,
45    "Whether to inject tcp proxy protocol headers to downstream http servers.",
46    ParameterScope::Environment,
47);
48
49/// Maximum number of client connections to proxy at once.
50pub const MAX_CONNECTIONS: Config<u32> = Config::new(
51    "balancerd_max_connections",
52    5000,
53    "Maximum number of client connections to proxy at once, across the pgwire and HTTPS \
54    listeners. Connections beyond this are rejected. Zero disables the limit.",
55    ParameterScope::Environment,
56);
57
58/// How long a client has to reach a resolved backend before the connection is closed.
59///
60/// Zero disables the deadline. The default allows for authentication's roughly 35-second
61/// retry budget, plus TLS negotiation and client startup.
62pub const PRE_RESOLVED_TIMEOUT: Config<Duration> = Config::new(
63    "balancerd_pre_resolved_timeout",
64    Duration::from_secs(60),
65    "How long a client has to complete the TLS handshake, the startup sequence and any credential \
66    exchange before the connection is closed. Zero disables the deadline.",
67    ParameterScope::Environment,
68);
69
70/// Sets the filter to apply to stderr logging.
71pub const LOGGING_FILTER: Config<&str> = Config::new(
72    "balancerd_log_filter",
73    "info",
74    "Sets the filter to apply to stderr logging.",
75    ParameterScope::Environment,
76);
77
78/// Sets the filter to apply to OpenTelemetry-backed distributed tracing.
79pub const OPENTELEMETRY_FILTER: Config<&str> = Config::new(
80    "balancerd_opentelemetry_filter",
81    "info",
82    "Sets the filter to apply to OpenTelemetry-backed distributed tracing.",
83    ParameterScope::Environment,
84);
85
86/// Sets additional default directives to apply to stderr logging.
87/// These apply to all variations of `log_filter`. Directives other than
88/// `module=off` are likely incorrect. Comma separated list.
89pub const LOGGING_FILTER_DEFAULTS: Config<fn() -> String> = Config::new(
90    "balancerd_log_filter_defaults",
91    || mz_ore::tracing::LOGGING_DEFAULTS_STR.join(","),
92    "Sets additional default directives to apply to stderr logging. \
93    These apply to all variations of `log_filter`. Directives other than \
94    `module=off` are likely incorrect. Comma separated list.",
95    ParameterScope::Environment,
96);
97
98/// Sets additional default directives to apply to OpenTelemetry-backed
99/// distributed tracing.
100/// These apply to all variations of `opentelemetry_filter`. Directives other than
101/// `module=off` are likely incorrect. Comma separated list.
102pub const OPENTELEMETRY_FILTER_DEFAULTS: Config<fn() -> String> = Config::new(
103    "balancerd_opentelemetry_filter_defaults",
104    || mz_ore::tracing::OPENTELEMETRY_DEFAULTS_STR.join(","),
105    "Sets additional default directives to apply to OpenTelemetry-backed \
106    distributed tracing. \
107    These apply to all variations of `opentelemetry_filter`. Directives other than \
108    `module=off` are likely incorrect. Comma separated list.",
109    ParameterScope::Environment,
110);
111
112/// Sets additional default directives to apply to sentry logging. \
113/// These apply on top of a default `info` directive. Directives other than \
114/// `module=off` are likely incorrect. Comma separated list.
115pub const SENTRY_FILTERS: Config<fn() -> String> = Config::new(
116    "balancerd_sentry_filters",
117    || mz_ore::tracing::SENTRY_DEFAULTS_STR.join(","),
118    "Sets additional default directives to apply to sentry logging. \
119    These apply on top of a default `info` directive. Directives other than \
120    `module=off` are likely incorrect. Comma separated list.",
121    ParameterScope::Environment,
122);
123
124/// Adds the full set of all balancer `Config`s.
125pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
126    configs
127        .add(&SIGTERM_CONNECTION_WAIT)
128        .add(&SIGTERM_LISTEN_WAIT)
129        .add(&INJECT_PROXY_PROTOCOL_HEADER_HTTP)
130        .add(&MAX_CONNECTIONS)
131        .add(&PRE_RESOLVED_TIMEOUT)
132        .add(&LOGGING_FILTER)
133        .add(&OPENTELEMETRY_FILTER)
134        .add(&LOGGING_FILTER_DEFAULTS)
135        .add(&OPENTELEMETRY_FILTER_DEFAULTS)
136        .add(&SENTRY_FILTERS)
137}
138
139/// Overrides default values for the Balancerd ConfigSet.
140///
141/// This is meant to be used in combination with clap cli flag
142/// `--default-config key=value`
143/// Not all ConfigSet values can be defaulted with this
144/// function. An error will be returned if a key does
145/// not accept default overrides, or if there is a value
146/// parsing error..
147pub(crate) fn set_defaults(
148    config_set: &ConfigSet,
149    default_config: Vec<(String, String)>,
150) -> Result<(), anyhow::Error> {
151    let mut config_updates = ConfigUpdates::default();
152    for (k, v) in default_config.iter() {
153        if k.as_str() == INJECT_PROXY_PROTOCOL_HEADER_HTTP.name() {
154            config_updates.add_dynamic(
155                INJECT_PROXY_PROTOCOL_HEADER_HTTP.name(),
156                mz_dyncfg::ConfigVal::Bool(bool::from_str(v)?),
157            )
158        } else if k.as_str() == MAX_CONNECTIONS.name() {
159            config_updates.add_dynamic(
160                MAX_CONNECTIONS.name(),
161                mz_dyncfg::ConfigVal::U32(u32::from_str(v)?),
162            )
163        } else if k.as_str() == PRE_RESOLVED_TIMEOUT.name() {
164            config_updates.add_dynamic(
165                PRE_RESOLVED_TIMEOUT.name(),
166                mz_dyncfg::ConfigVal::Duration(humantime::parse_duration(v)?),
167            )
168        } else {
169            return Err(anyhow!("Invalid default config value {k}"));
170        }
171    }
172    config_updates.apply(config_set);
173    Ok(())
174}
175
176/// Get all dynamic tracing config parameters from this [`ConfigSet`].
177pub fn tracing_config(configs: &ConfigSet) -> Result<TracingParameters, String> {
178    fn to_serializable_directives(
179        config: &Config<fn() -> String>,
180        configs: &ConfigSet,
181    ) -> Result<Vec<SerializableDirective>, String> {
182        let directives = config.get(configs);
183        let directives: Vec<_> = directives
184            .split(',')
185            .map(Directive::from_str)
186            .collect::<Result<_, _>>()
187            .map_err(|e| e.to_string())?;
188        Ok(directives.into_iter().map(|d| d.into()).collect())
189    }
190
191    let log_filter = LOGGING_FILTER.get(configs);
192    let log_filter = CloneableEnvFilter::from_str(&log_filter).map_err(|e| e.to_string())?;
193
194    let opentelemetry_filter = OPENTELEMETRY_FILTER.get(configs);
195    let opentelemetry_filter =
196        CloneableEnvFilter::from_str(&opentelemetry_filter).map_err(|e| e.to_string())?;
197
198    let log_filter_defaults = to_serializable_directives(&LOGGING_FILTER_DEFAULTS, configs)?;
199
200    let opentelemetry_filter_defaults =
201        to_serializable_directives(&OPENTELEMETRY_FILTER_DEFAULTS, configs)?;
202
203    let sentry_filters = to_serializable_directives(&SENTRY_FILTERS, configs)?;
204
205    Ok(TracingParameters {
206        log_filter: Some(log_filter),
207        opentelemetry_filter: Some(opentelemetry_filter),
208        log_filter_defaults,
209        opentelemetry_filter_defaults,
210        sentry_filters,
211    })
212}
213
214/// Returns true if `updates` contains an update to a tracing config, false otherwise.
215pub fn has_tracing_config_update(updates: &ConfigUpdates) -> bool {
216    [
217        LOGGING_FILTER.name(),
218        OPENTELEMETRY_FILTER.name(),
219        LOGGING_FILTER_DEFAULTS.name(),
220        OPENTELEMETRY_FILTER_DEFAULTS.name(),
221        SENTRY_FILTERS.name(),
222    ]
223    .into_iter()
224    .any(|name| updates.updates.contains_key(name))
225}
mz_balancerd/dyncfgs.rs

mz_balancerd/
dyncfgs.rs
