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/// Number of open connections at which balancerd leaves load balancer rotation.
59pub const CONNECTION_HIGH_WATERMARK: Config<usize> = Config::new(
60    "balancerd_connection_high_watermark",
61    0,
62    "Number of open connections at which balancerd reports not ready on /api/readyz, so \
63    load balancers stop sending it new connections while existing connections continue. \
64    Takes effect only when balancerd_connection_low_watermark is also non-zero, the low \
65    watermark is below the high watermark, and the high watermark is below a non-zero \
66    balancerd_max_connections. Zero disables the watermarks.",
67    ParameterScope::Environment,
68);
69
70/// Number of open connections below which balancerd returns to load balancer rotation.
71pub const CONNECTION_LOW_WATERMARK: Config<usize> = Config::new(
72    "balancerd_connection_low_watermark",
73    0,
74    "Number of open connections below which balancerd reports ready again after reaching \
75    balancerd_connection_high_watermark. Must be below the high watermark. The gap \
76    between the two keeps a balancer near the threshold from repeatedly entering and \
77    leaving load balancer rotation. Zero disables the watermarks.",
78    ParameterScope::Environment,
79);
80
81/// How long a client has to reach a resolved backend before the connection is closed.
82///
83/// Zero disables the deadline. The default allows for authentication's roughly 35-second
84/// retry budget, plus TLS negotiation and client startup.
85pub const PRE_RESOLVED_TIMEOUT: Config<Duration> = Config::new(
86    "balancerd_pre_resolved_timeout",
87    Duration::from_secs(60),
88    "How long a client has to complete the TLS handshake, the startup sequence and any credential \
89    exchange before the connection is closed. Zero disables the deadline.",
90    ParameterScope::Environment,
91);
92
93/// Sets the filter to apply to stderr logging.
94pub const LOGGING_FILTER: Config<&str> = Config::new(
95    "balancerd_log_filter",
96    "info",
97    "Sets the filter to apply to stderr logging.",
98    ParameterScope::Environment,
99);
100
101/// Sets the filter to apply to OpenTelemetry-backed distributed tracing.
102pub const OPENTELEMETRY_FILTER: Config<&str> = Config::new(
103    "balancerd_opentelemetry_filter",
104    "info",
105    "Sets the filter to apply to OpenTelemetry-backed distributed tracing.",
106    ParameterScope::Environment,
107);
108
109/// Sets additional default directives to apply to stderr logging.
110/// These apply to all variations of `log_filter`. Directives other than
111/// `module=off` are likely incorrect. Comma separated list.
112pub const LOGGING_FILTER_DEFAULTS: Config<fn() -> String> = Config::new(
113    "balancerd_log_filter_defaults",
114    || mz_ore::tracing::LOGGING_DEFAULTS_STR.join(","),
115    "Sets additional default directives to apply to stderr logging. \
116    These apply to all variations of `log_filter`. Directives other than \
117    `module=off` are likely incorrect. Comma separated list.",
118    ParameterScope::Environment,
119);
120
121/// Sets additional default directives to apply to OpenTelemetry-backed
122/// distributed tracing.
123/// These apply to all variations of `opentelemetry_filter`. Directives other than
124/// `module=off` are likely incorrect. Comma separated list.
125pub const OPENTELEMETRY_FILTER_DEFAULTS: Config<fn() -> String> = Config::new(
126    "balancerd_opentelemetry_filter_defaults",
127    || mz_ore::tracing::OPENTELEMETRY_DEFAULTS_STR.join(","),
128    "Sets additional default directives to apply to OpenTelemetry-backed \
129    distributed tracing. \
130    These apply to all variations of `opentelemetry_filter`. Directives other than \
131    `module=off` are likely incorrect. Comma separated list.",
132    ParameterScope::Environment,
133);
134
135/// Sets additional default directives to apply to sentry logging. \
136/// These apply on top of a default `info` directive. Directives other than \
137/// `module=off` are likely incorrect. Comma separated list.
138pub const SENTRY_FILTERS: Config<fn() -> String> = Config::new(
139    "balancerd_sentry_filters",
140    || mz_ore::tracing::SENTRY_DEFAULTS_STR.join(","),
141    "Sets additional default directives to apply to sentry logging. \
142    These apply on top of a default `info` directive. Directives other than \
143    `module=off` are likely incorrect. Comma separated list.",
144    ParameterScope::Environment,
145);
146
147/// Adds the full set of all balancer `Config`s.
148pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
149    configs
150        .add(&SIGTERM_CONNECTION_WAIT)
151        .add(&SIGTERM_LISTEN_WAIT)
152        .add(&INJECT_PROXY_PROTOCOL_HEADER_HTTP)
153        .add(&MAX_CONNECTIONS)
154        .add(&CONNECTION_HIGH_WATERMARK)
155        .add(&CONNECTION_LOW_WATERMARK)
156        .add(&PRE_RESOLVED_TIMEOUT)
157        .add(&LOGGING_FILTER)
158        .add(&OPENTELEMETRY_FILTER)
159        .add(&LOGGING_FILTER_DEFAULTS)
160        .add(&OPENTELEMETRY_FILTER_DEFAULTS)
161        .add(&SENTRY_FILTERS)
162}
163
164/// Overrides default values for the Balancerd ConfigSet.
165///
166/// This is meant to be used in combination with clap cli flag
167/// `--default-config key=value`
168/// Not all ConfigSet values can be defaulted with this
169/// function. An error will be returned if a key does
170/// not accept default overrides, or if there is a value
171/// parsing error..
172pub(crate) fn set_defaults(
173    config_set: &ConfigSet,
174    default_config: Vec<(String, String)>,
175) -> Result<(), anyhow::Error> {
176    let mut config_updates = ConfigUpdates::default();
177    for (k, v) in default_config.iter() {
178        if k.as_str() == INJECT_PROXY_PROTOCOL_HEADER_HTTP.name() {
179            config_updates.add_dynamic(
180                INJECT_PROXY_PROTOCOL_HEADER_HTTP.name(),
181                mz_dyncfg::ConfigVal::Bool(bool::from_str(v)?),
182            )
183        } else if k.as_str() == MAX_CONNECTIONS.name() {
184            config_updates.add_dynamic(
185                MAX_CONNECTIONS.name(),
186                mz_dyncfg::ConfigVal::U32(u32::from_str(v)?),
187            )
188        } else if k.as_str() == CONNECTION_HIGH_WATERMARK.name() {
189            config_updates.add_dynamic(
190                CONNECTION_HIGH_WATERMARK.name(),
191                mz_dyncfg::ConfigVal::Usize(usize::from_str(v)?),
192            )
193        } else if k.as_str() == CONNECTION_LOW_WATERMARK.name() {
194            config_updates.add_dynamic(
195                CONNECTION_LOW_WATERMARK.name(),
196                mz_dyncfg::ConfigVal::Usize(usize::from_str(v)?),
197            )
198        } else if k.as_str() == PRE_RESOLVED_TIMEOUT.name() {
199            config_updates.add_dynamic(
200                PRE_RESOLVED_TIMEOUT.name(),
201                mz_dyncfg::ConfigVal::Duration(humantime::parse_duration(v)?),
202            )
203        } else {
204            return Err(anyhow!("Invalid default config value {k}"));
205        }
206    }
207    config_updates.apply(config_set);
208    Ok(())
209}
210
211/// Get all dynamic tracing config parameters from this [`ConfigSet`].
212pub fn tracing_config(configs: &ConfigSet) -> Result<TracingParameters, String> {
213    fn to_serializable_directives(
214        config: &Config<fn() -> String>,
215        configs: &ConfigSet,
216    ) -> Result<Vec<SerializableDirective>, String> {
217        let directives = config.get(configs);
218        let directives: Vec<_> = directives
219            .split(',')
220            .map(Directive::from_str)
221            .collect::<Result<_, _>>()
222            .map_err(|e| e.to_string())?;
223        Ok(directives.into_iter().map(|d| d.into()).collect())
224    }
225
226    let log_filter = LOGGING_FILTER.get(configs);
227    let log_filter = CloneableEnvFilter::from_str(&log_filter).map_err(|e| e.to_string())?;
228
229    let opentelemetry_filter = OPENTELEMETRY_FILTER.get(configs);
230    let opentelemetry_filter =
231        CloneableEnvFilter::from_str(&opentelemetry_filter).map_err(|e| e.to_string())?;
232
233    let log_filter_defaults = to_serializable_directives(&LOGGING_FILTER_DEFAULTS, configs)?;
234
235    let opentelemetry_filter_defaults =
236        to_serializable_directives(&OPENTELEMETRY_FILTER_DEFAULTS, configs)?;
237
238    let sentry_filters = to_serializable_directives(&SENTRY_FILTERS, configs)?;
239
240    Ok(TracingParameters {
241        log_filter: Some(log_filter),
242        opentelemetry_filter: Some(opentelemetry_filter),
243        log_filter_defaults,
244        opentelemetry_filter_defaults,
245        sentry_filters,
246    })
247}
248
249/// Returns true if `updates` contains an update to a tracing config, false otherwise.
250pub fn has_tracing_config_update(updates: &ConfigUpdates) -> bool {
251    [
252        LOGGING_FILTER.name(),
253        OPENTELEMETRY_FILTER.name(),
254        LOGGING_FILTER_DEFAULTS.name(),
255        OPENTELEMETRY_FILTER_DEFAULTS.name(),
256        SENTRY_FILTERS.name(),
257    ]
258    .into_iter()
259    .any(|name| updates.updates.contains_key(name))
260}
mz_balancerd/dyncfgs.rs

mz_balancerd/
dyncfgs.rs
