1use std::collections::{BTreeMap, HashMap};
2use std::error::Error;
3use std::io::Write;
4use std::net::{IpAddr, SocketAddr};
5use std::num::{NonZeroU8, NonZeroU16};
6use std::path::{Path, PathBuf};
7use std::str::FromStr;
8use std::sync::{Arc, Mutex, PoisonError};
9use std::time::Duration;
10use std::{env, fmt, fs, io};
11
12use anyhow::Context;
13use arc_swap::ArcSwap;
14use relay_auth::{PublicKey, RelayId, SecretKey, generate_key_pair, generate_relay_id};
15use relay_common::Dsn;
16use relay_kafka::{
17 ConfigError as KafkaConfigError, KafkaConfigParam, KafkaTopic, KafkaTopicConfig,
18 TopicAssignments,
19};
20use relay_metrics::MetricNamespace;
21use serde::de::{DeserializeOwned, Unexpected, Visitor};
22use serde::{Deserialize, Deserializer, Serialize, Serializer};
23use uuid::Uuid;
24
25use crate::aggregator::{AggregatorServiceConfig, ScopedAggregatorConfig};
26use crate::byte_size::ByteSize;
27use crate::upstream::UpstreamDescriptor;
28use crate::{RedisConfig, RedisConfigs, RedisConfigsRef, build_redis_configs};
29
30const DEFAULT_NETWORK_OUTAGE_GRACE_PERIOD: u64 = 10;
31
32static CONFIG_YAML_HEADER: &str = r###"# Please see the relevant documentation.
33# Performance tuning: https://docs.sentry.io/product/relay/operating-guidelines/
34# All config options: https://docs.sentry.io/product/relay/options/
35"###;
36
37#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
39#[non_exhaustive]
40pub enum ConfigErrorKind {
41 CouldNotOpenFile,
43 CouldNotWriteFile,
45 BadYaml,
47 BadJson,
49 InvalidValue,
51 ProcessingNotAvailable,
54}
55
56impl fmt::Display for ConfigErrorKind {
57 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
58 match self {
59 Self::CouldNotOpenFile => write!(f, "could not open config file"),
60 Self::CouldNotWriteFile => write!(f, "could not write config file"),
61 Self::BadYaml => write!(f, "could not parse yaml config file"),
62 Self::BadJson => write!(f, "could not parse json config file"),
63 Self::InvalidValue => write!(f, "invalid config value"),
64 Self::ProcessingNotAvailable => write!(
65 f,
66 "was not compiled with processing, cannot enable processing"
67 ),
68 }
69 }
70}
71
72#[derive(Debug, Default)]
74enum ConfigErrorSource {
75 #[default]
77 None,
78 File(PathBuf),
80 FieldOverride(String),
82}
83
84impl fmt::Display for ConfigErrorSource {
85 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86 match self {
87 ConfigErrorSource::None => Ok(()),
88 ConfigErrorSource::File(file_name) => {
89 write!(f, " (file {})", file_name.display())
90 }
91 ConfigErrorSource::FieldOverride(name) => write!(f, " (field {name})"),
92 }
93 }
94}
95
96#[derive(Debug)]
98pub struct ConfigError {
99 source: ConfigErrorSource,
100 kind: ConfigErrorKind,
101}
102
103impl ConfigError {
104 #[inline]
105 fn new(kind: ConfigErrorKind) -> Self {
106 Self {
107 source: ConfigErrorSource::None,
108 kind,
109 }
110 }
111
112 #[inline]
113 fn field(field: &'static str) -> Self {
114 Self {
115 source: ConfigErrorSource::FieldOverride(field.to_owned()),
116 kind: ConfigErrorKind::InvalidValue,
117 }
118 }
119
120 #[inline]
121 fn file(kind: ConfigErrorKind, p: impl AsRef<Path>) -> Self {
122 Self {
123 source: ConfigErrorSource::File(p.as_ref().to_path_buf()),
124 kind,
125 }
126 }
127
128 pub fn kind(&self) -> ConfigErrorKind {
130 self.kind
131 }
132}
133
134impl fmt::Display for ConfigError {
135 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136 write!(f, "{}{}", self.kind(), self.source)
137 }
138}
139
140impl Error for ConfigError {}
141
142enum ConfigFormat {
143 Yaml,
144 Json,
145}
146
147impl ConfigFormat {
148 pub fn extension(&self) -> &'static str {
149 match self {
150 ConfigFormat::Yaml => "yml",
151 ConfigFormat::Json => "json",
152 }
153 }
154}
155
156trait ConfigObject: DeserializeOwned + Serialize {
157 fn format() -> ConfigFormat;
159
160 fn name() -> &'static str;
162
163 fn path(base: &Path) -> PathBuf {
165 base.join(format!("{}.{}", Self::name(), Self::format().extension()))
166 }
167
168 fn load(base: &Path) -> anyhow::Result<Self> {
170 let path = Self::path(base);
171
172 let f = fs::File::open(&path)
173 .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotOpenFile, &path))?;
174 let f = io::BufReader::new(f);
175
176 let mut source = {
177 let file = serde_vars::FileSource::default()
178 .with_variable_prefix("${file:")
179 .with_variable_suffix("}")
180 .with_base_path(base);
181 let env = serde_vars::EnvSource::default()
182 .with_variable_prefix("${")
183 .with_variable_suffix("}");
184 (file, env)
185 };
186 match Self::format() {
187 ConfigFormat::Yaml => {
188 serde_vars::deserialize(serde_yaml::Deserializer::from_reader(f), &mut source)
189 .with_context(|| ConfigError::file(ConfigErrorKind::BadYaml, &path))
190 }
191 ConfigFormat::Json => {
192 serde_vars::deserialize(&mut serde_json::Deserializer::from_reader(f), &mut source)
193 .with_context(|| ConfigError::file(ConfigErrorKind::BadJson, &path))
194 }
195 }
196 }
197
198 fn save(&self, base: &Path) -> anyhow::Result<()> {
200 let path = Self::path(base);
201 let mut options = fs::OpenOptions::new();
202 options.write(true).truncate(true).create(true);
203
204 #[cfg(unix)]
206 {
207 use std::os::unix::fs::OpenOptionsExt;
208 options.mode(0o600);
209 }
210
211 let mut f = options
212 .open(&path)
213 .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?;
214
215 match Self::format() {
216 ConfigFormat::Yaml => {
217 f.write_all(CONFIG_YAML_HEADER.as_bytes())?;
218 serde_yaml::to_writer(&mut f, self)
219 .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?
220 }
221 ConfigFormat::Json => serde_json::to_writer_pretty(&mut f, self)
222 .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?,
223 }
224
225 f.write_all(b"\n").ok();
226
227 Ok(())
228 }
229}
230
231#[derive(Debug, Default, Clone)]
234pub struct OverridableConfig {
235 pub mode: Option<String>,
237 pub instance: Option<String>,
239 pub log_level: Option<String>,
241 pub log_format: Option<String>,
243 pub upstream: Option<String>,
245 pub upstream_dsn: Option<String>,
247 pub host: Option<String>,
249 pub port: Option<String>,
251 pub processing: Option<String>,
253 pub kafka_url: Option<String>,
255 pub redis_url: Option<String>,
257 pub id: Option<String>,
259 pub secret_key: Option<String>,
261 pub public_key: Option<String>,
263 pub outcome_source: Option<String>,
265 pub shutdown_timeout: Option<String>,
267 pub server_name: Option<String>,
269}
270
271#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
273pub struct Credentials {
274 pub secret_key: SecretKey,
276 pub public_key: PublicKey,
278 pub id: RelayId,
280}
281
282impl Credentials {
283 pub fn generate() -> Self {
285 relay_log::info!("generating new relay credentials");
286 let (secret_key, public_key) = generate_key_pair();
287 Self {
288 secret_key,
289 public_key,
290 id: generate_relay_id(),
291 }
292 }
293
294 pub fn to_json_string(&self) -> anyhow::Result<String> {
296 serde_json::to_string(self)
297 .with_context(|| ConfigError::new(ConfigErrorKind::CouldNotWriteFile))
298 }
299}
300
301impl ConfigObject for Credentials {
302 fn format() -> ConfigFormat {
303 ConfigFormat::Json
304 }
305 fn name() -> &'static str {
306 "credentials"
307 }
308}
309
310#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
312#[serde(rename_all = "camelCase")]
313pub struct RelayInfo {
314 pub public_key: PublicKey,
316
317 #[serde(default)]
319 pub internal: bool,
320}
321
322impl RelayInfo {
323 pub fn new(public_key: PublicKey) -> Self {
325 Self {
326 public_key,
327 internal: false,
328 }
329 }
330}
331
332#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
334#[serde(rename_all = "camelCase")]
335pub enum RelayMode {
336 Proxy,
342
343 Managed,
349}
350
351impl<'de> Deserialize<'de> for RelayMode {
352 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
353 where
354 D: Deserializer<'de>,
355 {
356 let s = String::deserialize(deserializer)?;
357 match s.as_str() {
358 "proxy" => Ok(RelayMode::Proxy),
359 "managed" => Ok(RelayMode::Managed),
360 "static" => Err(serde::de::Error::custom(
361 "Relay mode 'static' has been removed. Please use 'managed' or 'proxy' instead.",
362 )),
363 other => Err(serde::de::Error::unknown_variant(
364 other,
365 &["proxy", "managed"],
366 )),
367 }
368 }
369}
370
371impl fmt::Display for RelayMode {
372 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
373 match self {
374 RelayMode::Proxy => write!(f, "proxy"),
375 RelayMode::Managed => write!(f, "managed"),
376 }
377 }
378}
379
380#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)]
382#[serde(rename_all = "camelCase")]
383pub enum RelayInstance {
384 Default,
386
387 Canary,
389}
390
391impl RelayInstance {
392 pub fn is_canary(&self) -> bool {
394 matches!(self, RelayInstance::Canary)
395 }
396}
397
398impl fmt::Display for RelayInstance {
399 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
400 match self {
401 RelayInstance::Default => write!(f, "default"),
402 RelayInstance::Canary => write!(f, "canary"),
403 }
404 }
405}
406
407impl FromStr for RelayInstance {
408 type Err = fmt::Error;
409
410 fn from_str(s: &str) -> Result<Self, Self::Err> {
411 match s {
412 "canary" => Ok(RelayInstance::Canary),
413 _ => Ok(RelayInstance::Default),
414 }
415 }
416}
417
418#[derive(Clone, Copy, Debug, Eq, PartialEq)]
420pub struct ParseRelayModeError;
421
422impl fmt::Display for ParseRelayModeError {
423 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
424 write!(f, "Relay mode must be one of: managed or proxy")
425 }
426}
427
428impl Error for ParseRelayModeError {}
429
430impl FromStr for RelayMode {
431 type Err = ParseRelayModeError;
432
433 fn from_str(s: &str) -> Result<Self, Self::Err> {
434 match s {
435 "proxy" => Ok(RelayMode::Proxy),
436 "managed" => Ok(RelayMode::Managed),
437 _ => Err(ParseRelayModeError),
438 }
439 }
440}
441
442fn is_default<T: Default + PartialEq>(t: &T) -> bool {
444 *t == T::default()
445}
446
447fn is_docker() -> bool {
449 if fs::metadata("/.dockerenv").is_ok() {
450 return true;
451 }
452
453 fs::read_to_string("/proc/self/cgroup").is_ok_and(|s| s.contains("/docker"))
454}
455
456fn default_host() -> IpAddr {
458 if is_docker() {
459 "0.0.0.0".parse().unwrap()
461 } else {
462 "127.0.0.1".parse().unwrap()
463 }
464}
465
466#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)]
470#[serde(rename_all = "lowercase")]
471#[derive(Default)]
472pub enum ReadinessCondition {
473 #[default]
482 Authenticated,
483 Always,
485}
486
487#[derive(Serialize, Deserialize, Debug, Clone)]
489#[serde(default)]
490pub struct Relay {
491 pub mode: RelayMode,
493 pub instance: RelayInstance,
495 pub upstream: UpstreamDescriptor,
497 pub advertised_upstream: Option<UpstreamDescriptor>,
506 pub host: IpAddr,
508 pub port: u16,
510 pub internal_host: Option<IpAddr>,
524 pub internal_port: Option<u16>,
528 #[serde(skip_serializing)]
530 pub tls_port: Option<u16>,
531 #[serde(skip_serializing)]
533 pub tls_identity_path: Option<PathBuf>,
534 #[serde(skip_serializing)]
536 pub tls_identity_password: Option<String>,
537 #[serde(skip_serializing_if = "is_default")]
542 pub override_project_ids: bool,
543 pub config_reload_interval: Option<u64>,
552}
553
554impl Default for Relay {
555 fn default() -> Self {
556 Relay {
557 mode: RelayMode::Managed,
558 instance: RelayInstance::Default,
559 upstream: "https://sentry.io/".parse().unwrap(),
560 advertised_upstream: None,
561 host: default_host(),
562 port: 3000,
563 internal_host: None,
564 internal_port: None,
565 tls_port: None,
566 tls_identity_path: None,
567 tls_identity_password: None,
568 override_project_ids: false,
569 config_reload_interval: None,
570 }
571 }
572}
573
574#[derive(Serialize, Deserialize, Debug, Clone)]
576#[serde(default)]
577pub struct Metrics {
578 pub statsd: Option<String>,
582 pub statsd_buffer_size: Option<usize>,
586 pub prefix: String,
590 pub default_tags: BTreeMap<String, String>,
592 pub hostname_tag: Option<String>,
594 pub periodic_secs: u64,
599}
600
601impl Default for Metrics {
602 fn default() -> Self {
603 Metrics {
604 statsd: None,
605 statsd_buffer_size: None,
606 prefix: "sentry.relay".into(),
607 default_tags: BTreeMap::new(),
608 hostname_tag: None,
609 periodic_secs: 5,
610 }
611 }
612}
613
614#[derive(Serialize, Deserialize, Debug, Clone)]
616#[serde(default)]
617pub struct Limits {
618 pub max_concurrent_requests: usize,
621 pub max_concurrent_queries: usize,
626 pub max_event_size: ByteSize,
628 pub max_attachment_size: ByteSize,
630 pub max_attachment_count: usize,
632 pub max_attachments_size: ByteSize,
634 pub max_upload_size: ByteSize,
636 pub max_client_reports_size: ByteSize,
638 pub max_client_reports_count: usize,
640 pub max_check_in_size: ByteSize,
642 pub max_envelope_size: ByteSize,
644 pub max_sessions_size: ByteSize,
646 pub max_session_count: usize,
648 pub max_api_payload_size: ByteSize,
650 pub max_api_file_upload_size: ByteSize,
652 pub max_api_chunk_upload_size: ByteSize,
654 pub max_profile_size: ByteSize,
656 pub max_trace_metric_size: ByteSize,
658 pub max_log_size: ByteSize,
660 pub max_span_size: ByteSize,
662 pub max_standalone_span_count: usize,
664 pub max_container_size: ByteSize,
666 pub max_statsd_size: ByteSize,
668 pub max_metric_buckets_size: ByteSize,
670 pub max_replay_compressed_size: ByteSize,
672 #[serde(alias = "max_replay_size")]
674 max_replay_uncompressed_size: ByteSize,
675 pub max_replay_message_size: ByteSize,
677 pub max_removed_attribute_key_size: ByteSize,
687 pub max_thread_count: usize,
692 pub max_pool_concurrency: usize,
699 pub query_timeout: u64,
702 pub shutdown_timeout: u64,
705 pub keepalive_timeout: u64,
709 pub idle_timeout: Option<u64>,
716 pub max_connections: Option<usize>,
722 pub tcp_listen_backlog: u32,
730}
731
732impl Default for Limits {
733 fn default() -> Self {
734 Limits {
735 max_concurrent_requests: 100,
736 max_concurrent_queries: 5,
737 max_event_size: ByteSize::mebibytes(1),
738 max_attachment_size: ByteSize::mebibytes(200),
739 max_attachment_count: 30,
740 max_attachments_size: ByteSize::mebibytes(200),
741 max_upload_size: ByteSize::mebibytes(1024),
742 max_client_reports_size: ByteSize::kibibytes(100),
743 max_client_reports_count: 100,
744 max_check_in_size: ByteSize::kibibytes(100),
745 max_envelope_size: ByteSize::mebibytes(200),
746 max_sessions_size: ByteSize::mebibytes(10),
747 max_session_count: 100,
748 max_api_payload_size: ByteSize::mebibytes(20),
749 max_api_file_upload_size: ByteSize::mebibytes(40),
750 max_api_chunk_upload_size: ByteSize::mebibytes(100),
751 max_profile_size: ByteSize::mebibytes(50),
752 max_trace_metric_size: ByteSize::mebibytes(1),
753 max_log_size: ByteSize::mebibytes(2),
754 max_span_size: ByteSize::mebibytes(10),
755 max_standalone_span_count: 25,
756 max_container_size: ByteSize::mebibytes(12),
757 max_statsd_size: ByteSize::mebibytes(1),
758 max_metric_buckets_size: ByteSize::mebibytes(1),
759 max_replay_compressed_size: ByteSize::mebibytes(10),
760 max_replay_uncompressed_size: ByteSize::mebibytes(100),
761 max_replay_message_size: ByteSize::mebibytes(15),
762 max_thread_count: num_cpus::get(),
763 max_pool_concurrency: 1,
764 query_timeout: 30,
765 shutdown_timeout: 10,
766 keepalive_timeout: 5,
767 idle_timeout: None,
768 max_connections: None,
769 tcp_listen_backlog: 1024,
770 max_removed_attribute_key_size: ByteSize::kibibytes(10),
771 }
772 }
773}
774
775#[derive(Debug, Default, Deserialize, Serialize, Clone)]
777#[serde(default)]
778pub struct Routing {
779 pub accept_unknown_items: Option<bool>,
789}
790
791#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)]
793#[serde(rename_all = "lowercase")]
794pub enum HttpEncoding {
795 #[default]
800 Identity,
801 Deflate,
807 Gzip,
814 Br,
816 Zstd,
818}
819
820impl HttpEncoding {
821 pub fn parse(str: &str) -> Self {
823 let str = str.trim();
824 if str.eq_ignore_ascii_case("zstd") {
825 Self::Zstd
826 } else if str.eq_ignore_ascii_case("br") {
827 Self::Br
828 } else if str.eq_ignore_ascii_case("gzip") || str.eq_ignore_ascii_case("x-gzip") {
829 Self::Gzip
830 } else if str.eq_ignore_ascii_case("deflate") {
831 Self::Deflate
832 } else {
833 Self::Identity
834 }
835 }
836
837 pub fn name(&self) -> Option<&'static str> {
841 match self {
842 Self::Identity => None,
843 Self::Deflate => Some("deflate"),
844 Self::Gzip => Some("gzip"),
845 Self::Br => Some("br"),
846 Self::Zstd => Some("zstd"),
847 }
848 }
849}
850
851#[derive(Serialize, Deserialize, Debug, Clone)]
853#[serde(default)]
854pub struct Http {
855 pub timeout: u32,
861 pub connection_timeout: u32,
866 pub max_retry_interval: u32,
868 pub host_header: Option<String>,
870 pub auth_interval: Option<u64>,
878 pub outage_grace_period: u64,
884 pub retry_delay: u64,
888 pub project_failure_interval: u64,
893 pub encoding: HttpEncoding,
909 pub global_metrics: bool,
916 pub forward: bool,
923 pub dns_cache: bool,
927}
928
929impl Default for Http {
930 fn default() -> Self {
931 Http {
932 timeout: 5,
933 connection_timeout: 3,
934 max_retry_interval: 60, host_header: None,
936 auth_interval: Some(600), outage_grace_period: DEFAULT_NETWORK_OUTAGE_GRACE_PERIOD,
938 retry_delay: 1,
939 project_failure_interval: 90,
940 encoding: HttpEncoding::Zstd,
941 global_metrics: false,
942 forward: true,
943 dns_cache: true,
944 }
945 }
946}
947
948#[derive(Clone, Copy, Debug, Eq, PartialEq, Default, Deserialize, Serialize)]
950#[serde(rename_all = "snake_case")]
951pub enum EnvelopeSpoolPartitioning {
952 ProjectKeyPair,
956 #[default]
963 RoundRobin,
964}
965
966#[derive(Debug, Serialize, Deserialize, Clone)]
968#[serde(default)]
969pub struct EnvelopeSpool {
970 pub path: Option<PathBuf>,
976 pub max_disk_size: ByteSize,
982 pub batch_size_bytes: ByteSize,
989 pub max_envelope_delay_secs: u64,
996 pub disk_usage_refresh_frequency_ms: u64,
1001 pub max_backpressure_memory_percent: f32,
1031 pub partitions: NonZeroU8,
1038 pub partitioning: EnvelopeSpoolPartitioning,
1044 pub ephemeral: bool,
1051}
1052
1053impl Default for EnvelopeSpool {
1054 fn default() -> Self {
1055 Self {
1056 path: None,
1057 max_disk_size: ByteSize::mebibytes(500),
1058 batch_size_bytes: ByteSize::kibibytes(10),
1059 max_envelope_delay_secs: 24 * 60 * 60,
1060 disk_usage_refresh_frequency_ms: 100,
1061 max_backpressure_memory_percent: 0.8,
1062 partitions: NonZeroU8::new(1).unwrap(),
1063 partitioning: EnvelopeSpoolPartitioning::default(),
1064 ephemeral: false,
1065 }
1066 }
1067}
1068
1069#[derive(Debug, Serialize, Deserialize, Default, Clone)]
1071#[serde(default)]
1072pub struct Spool {
1073 pub envelopes: EnvelopeSpool,
1075}
1076
1077#[derive(Serialize, Deserialize, Debug, Clone)]
1079#[serde(default)]
1080pub struct Cache {
1081 pub project_request_full_config: bool,
1083 pub project_expiry: u32,
1085 pub project_grace_period: u32,
1090 pub project_refresh_interval: Option<u32>,
1096 pub relay_expiry: u32,
1098 #[serde(alias = "event_expiry")]
1104 envelope_expiry: u32,
1105 #[serde(alias = "event_buffer_size")]
1107 envelope_buffer_size: u32,
1108 pub miss_expiry: u32,
1110 pub batch_interval: u32,
1112 pub downstream_relays_batch_interval: u32,
1114 pub batch_size: usize,
1118 pub file_interval: u32,
1120 pub global_config_fetch_interval: u32,
1122}
1123
1124impl Default for Cache {
1125 fn default() -> Self {
1126 Cache {
1127 project_request_full_config: false,
1128 project_expiry: 300, project_grace_period: 120, project_refresh_interval: None,
1131 relay_expiry: 3600, envelope_expiry: 600, envelope_buffer_size: 1000,
1134 miss_expiry: 60, batch_interval: 100, downstream_relays_batch_interval: 100, batch_size: 500,
1138 file_interval: 10, global_config_fetch_interval: 10, }
1141 }
1142}
1143
1144#[derive(Serialize, Deserialize, Debug, Clone)]
1146#[serde(default)]
1147pub struct Processing {
1148 pub enabled: bool,
1150 pub geoip_path: Option<PathBuf>,
1152 pub max_secs_in_future: u32,
1154 pub max_session_secs_in_past: u32,
1156 pub kafka_config: Vec<KafkaConfigParam>,
1158 pub secondary_kafka_configs: BTreeMap<String, Vec<KafkaConfigParam>>,
1178 pub topics: TopicAssignments,
1180 pub kafka_validate_topics: bool,
1182 pub redis: Option<RedisConfigs>,
1184 pub attachment_chunk_size: ByteSize,
1186 pub projectconfig_cache_prefix: String,
1188 pub max_rate_limit: Option<u32>,
1190 pub quota_cache_ratio: Option<f32>,
1201 pub quota_cache_max: Option<f32>,
1208 #[serde(alias = "upload")]
1210 pub objectstore: ObjectstoreServiceConfig,
1211}
1212
1213impl Default for Processing {
1214 fn default() -> Self {
1216 Self {
1217 enabled: false,
1218 geoip_path: None,
1219 max_secs_in_future: 60, max_session_secs_in_past: 5 * 24 * 3600, kafka_config: Vec::new(),
1222 secondary_kafka_configs: BTreeMap::new(),
1223 topics: TopicAssignments::default(),
1224 kafka_validate_topics: false,
1225 redis: None,
1226 attachment_chunk_size: ByteSize::mebibytes(1),
1227 projectconfig_cache_prefix: "relayconfig".to_owned(),
1228 max_rate_limit: Some(300), quota_cache_ratio: None,
1230 quota_cache_max: None,
1231 objectstore: ObjectstoreServiceConfig::default(),
1232 }
1233 }
1234}
1235
1236#[derive(Debug, Default, Serialize, Deserialize, Clone)]
1238#[serde(default)]
1239pub struct Normalization {
1240 pub level: NormalizationLevel,
1242}
1243
1244#[derive(Copy, Clone, Debug, Default, Serialize, Deserialize, Eq, PartialEq)]
1246#[serde(rename_all = "lowercase")]
1247pub enum NormalizationLevel {
1248 #[default]
1252 Default,
1253 Full,
1258}
1259
1260#[derive(Serialize, Deserialize, Clone)]
1262pub struct ObjectstoreAuthConfig {
1263 pub key_id: String,
1266
1267 pub signing_key: String,
1269}
1270
1271impl fmt::Debug for ObjectstoreAuthConfig {
1272 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1273 f.debug_struct("ObjectstoreAuthConfig")
1274 .field("key_id", &self.key_id)
1275 .field("signing_key", &"[redacted]")
1276 .finish()
1277 }
1278}
1279
1280#[derive(Serialize, Deserialize, Debug, Clone)]
1282#[serde(default)]
1283pub struct ObjectstoreServiceConfig {
1284 pub objectstore_url: Option<String>,
1289
1290 pub max_concurrent_requests: usize,
1292
1293 pub max_backlog: usize,
1297
1298 pub timeout: u64,
1303
1304 pub stream_timeout: u64,
1309
1310 pub retry_delay: f64,
1312
1313 pub max_attempts: NonZeroU16,
1315
1316 pub fallback_to_kafka: bool,
1321
1322 pub auth: Option<ObjectstoreAuthConfig>,
1324}
1325
1326impl Default for ObjectstoreServiceConfig {
1327 fn default() -> Self {
1328 Self {
1329 objectstore_url: None,
1330 max_concurrent_requests: 10,
1331 max_backlog: 20,
1332 timeout: 60,
1333 stream_timeout: 5 * 60, retry_delay: 1.0,
1335 max_attempts: NonZeroU16::new(5).unwrap(),
1336 fallback_to_kafka: true,
1337 auth: None,
1338 }
1339 }
1340}
1341
1342#[derive(Copy, Clone, Debug, PartialEq, Eq)]
1345
1346pub enum EmitOutcomes {
1347 None,
1349 AsClientReports,
1351 AsOutcomes,
1353}
1354
1355impl EmitOutcomes {
1356 pub fn any(&self) -> bool {
1358 !matches!(self, EmitOutcomes::None)
1359 }
1360}
1361
1362impl Serialize for EmitOutcomes {
1363 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1364 where
1365 S: Serializer,
1366 {
1367 match self {
1369 Self::None => serializer.serialize_bool(false),
1370 Self::AsClientReports => serializer.serialize_str("as_client_reports"),
1371 Self::AsOutcomes => serializer.serialize_bool(true),
1372 }
1373 }
1374}
1375
1376struct EmitOutcomesVisitor;
1377
1378impl Visitor<'_> for EmitOutcomesVisitor {
1379 type Value = EmitOutcomes;
1380
1381 fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
1382 formatter.write_str("true, false, 'as_client_reports'")
1383 }
1384
1385 fn visit_bool<E>(self, v: bool) -> Result<Self::Value, E>
1386 where
1387 E: serde::de::Error,
1388 {
1389 Ok(if v {
1390 EmitOutcomes::AsOutcomes
1391 } else {
1392 EmitOutcomes::None
1393 })
1394 }
1395
1396 fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
1397 where
1398 E: serde::de::Error,
1399 {
1400 match v {
1401 "as_client_reports" => Ok(EmitOutcomes::AsClientReports),
1402 _ => Err(E::invalid_value(Unexpected::Str(v), &self)),
1403 }
1404 }
1405}
1406
1407impl<'de> Deserialize<'de> for EmitOutcomes {
1408 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1409 where
1410 D: Deserializer<'de>,
1411 {
1412 deserializer.deserialize_any(EmitOutcomesVisitor)
1413 }
1414}
1415
1416#[derive(Serialize, Deserialize, Debug, Clone)]
1418#[serde(default)]
1419pub struct Outcomes {
1420 pub emit_outcomes: EmitOutcomes,
1424 pub batch_size: usize,
1427 pub batch_interval: u64,
1430 pub source: Option<String>,
1433}
1434
1435impl Default for Outcomes {
1436 fn default() -> Self {
1437 Outcomes {
1438 emit_outcomes: EmitOutcomes::AsClientReports,
1439 batch_size: 1000,
1440 batch_interval: 500,
1441 source: None,
1442 }
1443 }
1444}
1445
1446#[derive(Serialize, Deserialize, Debug, Default)]
1448pub struct MinimalConfig {
1449 pub relay: Relay,
1451}
1452
1453impl MinimalConfig {
1454 pub fn save_in_folder<P: AsRef<Path>>(&self, p: P) -> anyhow::Result<()> {
1456 let path = p.as_ref();
1457 if fs::metadata(path).is_err() {
1458 fs::create_dir_all(path)
1459 .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotOpenFile, path))?;
1460 }
1461 self.save(path)
1462 }
1463}
1464
1465impl ConfigObject for MinimalConfig {
1466 fn format() -> ConfigFormat {
1467 ConfigFormat::Yaml
1468 }
1469
1470 fn name() -> &'static str {
1471 "config"
1472 }
1473}
1474
1475mod config_relay_info {
1477 use serde::ser::SerializeMap;
1478
1479 use super::*;
1480
1481 #[derive(Debug, Serialize, Deserialize, Clone)]
1483 struct RelayInfoConfig {
1484 public_key: PublicKey,
1485 #[serde(default)]
1486 internal: bool,
1487 }
1488
1489 impl From<RelayInfoConfig> for RelayInfo {
1490 fn from(v: RelayInfoConfig) -> Self {
1491 RelayInfo {
1492 public_key: v.public_key,
1493 internal: v.internal,
1494 }
1495 }
1496 }
1497
1498 impl From<RelayInfo> for RelayInfoConfig {
1499 fn from(v: RelayInfo) -> Self {
1500 RelayInfoConfig {
1501 public_key: v.public_key,
1502 internal: v.internal,
1503 }
1504 }
1505 }
1506
1507 pub(super) fn deserialize<'de, D>(des: D) -> Result<HashMap<RelayId, RelayInfo>, D::Error>
1508 where
1509 D: Deserializer<'de>,
1510 {
1511 let map = HashMap::<RelayId, RelayInfoConfig>::deserialize(des)?;
1512 Ok(map.into_iter().map(|(k, v)| (k, v.into())).collect())
1513 }
1514
1515 pub(super) fn serialize<S>(elm: &HashMap<RelayId, RelayInfo>, ser: S) -> Result<S::Ok, S::Error>
1516 where
1517 S: Serializer,
1518 {
1519 let mut map = ser.serialize_map(Some(elm.len()))?;
1520
1521 for (k, v) in elm {
1522 map.serialize_entry(k, &RelayInfoConfig::from(v.clone()))?;
1523 }
1524
1525 map.end()
1526 }
1527}
1528
1529#[derive(Serialize, Deserialize, Debug, Clone)]
1531#[serde(default)]
1532pub struct AuthConfig {
1533 #[serde(skip_serializing_if = "is_default")]
1535 pub ready: ReadinessCondition,
1536
1537 #[serde(with = "config_relay_info")]
1539 pub static_relays: HashMap<RelayId, RelayInfo>,
1540
1541 pub signature_max_age: u64,
1545}
1546
1547impl Default for AuthConfig {
1548 fn default() -> Self {
1549 Self {
1550 ready: ReadinessCondition::default(),
1551 static_relays: HashMap::new(),
1552 signature_max_age: 300, }
1554 }
1555}
1556
1557#[derive(Serialize, Deserialize, Debug, Default, Clone)]
1559pub struct GeoIpConfig {
1560 pub path: Option<PathBuf>,
1562}
1563
1564#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
1569#[serde(default)]
1570pub struct Health {
1571 pub refresh_interval_ms: u64,
1578 pub max_memory_bytes: Option<ByteSize>,
1583 pub max_memory_percent: f32,
1587 pub probe_timeout_ms: u64,
1594 pub memory_stat_refresh_frequency_ms: u64,
1600}
1601
1602impl Default for Health {
1603 fn default() -> Self {
1604 Self {
1605 refresh_interval_ms: 3000,
1606 max_memory_bytes: None,
1607 max_memory_percent: 0.95,
1608 probe_timeout_ms: 900,
1609 memory_stat_refresh_frequency_ms: 100,
1610 }
1611 }
1612}
1613
1614#[derive(Serialize, Deserialize, Debug, Clone)]
1616#[serde(default)]
1617pub struct Cogs {
1618 pub max_queue_size: u64,
1624 pub relay_resource_id: String,
1630}
1631
1632impl Default for Cogs {
1633 fn default() -> Self {
1634 Self {
1635 max_queue_size: 10_000,
1636 relay_resource_id: "relay_service".to_owned(),
1637 }
1638 }
1639}
1640
1641#[derive(Debug, Clone, Serialize, Deserialize)]
1643#[serde(default)]
1644pub struct Upload {
1645 pub max_concurrent_requests: usize,
1649 pub timeout: u64,
1651 pub max_age: i64,
1655
1656 pub credentials: Option<UploadCredentials>,
1660}
1661
1662impl Default for Upload {
1663 fn default() -> Self {
1664 Self {
1665 max_concurrent_requests: 100,
1666 timeout: 5 * 60, max_age: 60 * 60, credentials: None,
1669 }
1670 }
1671}
1672
1673#[derive(Clone, Serialize, Deserialize)]
1675pub struct UploadCredentials {
1676 #[cfg(feature = "processing")]
1678 pub signing_key: SecretKey,
1679
1680 pub verification_key: PublicKey,
1682}
1683
1684impl fmt::Debug for UploadCredentials {
1685 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1686 let Self {
1687 #[cfg(feature = "processing")]
1688 signing_key: _,
1689 verification_key,
1690 } = self;
1691 let mut b = f.debug_struct("UploadCredentials");
1692 #[cfg(feature = "processing")]
1693 b.field("signing_key", &"[redacted]");
1694 b.field("verification_key", verification_key).finish()
1695 }
1696}
1697
1698#[derive(Serialize, Deserialize, Debug, Default, Clone)]
1700#[serde(default)]
1701#[allow(missing_docs)]
1702pub struct ConfigValues {
1703 pub relay: Relay,
1704 pub http: Http,
1705 pub cache: Cache,
1706 pub spool: Spool,
1707 pub limits: Limits,
1708 pub logging: relay_log::LogConfig,
1709 pub routing: Routing,
1710 pub metrics: Metrics,
1711 pub sentry: relay_log::SentryConfig,
1712 pub processing: Processing,
1713 pub outcomes: Outcomes,
1714 pub aggregator: AggregatorServiceConfig,
1715 pub secondary_aggregators: Vec<ScopedAggregatorConfig>,
1716 pub auth: AuthConfig,
1717 pub geoip: GeoIpConfig,
1718 pub normalization: Normalization,
1719 pub health: Health,
1720 pub cogs: Cogs,
1721 pub upload: Upload,
1722}
1723
1724impl ConfigObject for ConfigValues {
1725 fn format() -> ConfigFormat {
1726 ConfigFormat::Yaml
1727 }
1728
1729 fn name() -> &'static str {
1730 "config"
1731 }
1732}
1733
1734#[derive(Default, Clone)]
1735struct ConfigInner {
1736 values: ConfigValues,
1738 credentials: Option<Credentials>,
1742}
1743
1744impl ConfigInner {
1745 fn apply_overrides(&mut self, overrides: &OverridableConfig) -> anyhow::Result<()> {
1746 if let Some(log_level) = &overrides.log_level {
1747 self.values.logging.level = log_level.parse()?;
1748 }
1749
1750 if let Some(log_format) = &overrides.log_format {
1751 self.values.logging.format = log_format.parse()?;
1752 }
1753
1754 let relay = &mut self.values.relay;
1755 if let Some(mode) = &overrides.mode {
1756 relay.mode = mode
1757 .parse::<RelayMode>()
1758 .with_context(|| ConfigError::field("mode"))?;
1759 }
1760 if let Some(deployment) = &overrides.instance {
1761 relay.instance = deployment
1762 .parse::<RelayInstance>()
1763 .with_context(|| ConfigError::field("deployment"))?;
1764 }
1765 if let Some(upstream) = &overrides.upstream {
1766 relay.upstream = upstream
1767 .parse::<UpstreamDescriptor>()
1768 .with_context(|| ConfigError::field("upstream"))?;
1769 } else if let Some(upstream_dsn) = &overrides.upstream_dsn {
1770 relay.upstream = upstream_dsn
1771 .parse::<Dsn>()
1772 .map(|dsn| UpstreamDescriptor::from_dsn(&dsn))
1773 .with_context(|| ConfigError::field("upstream_dsn"))?;
1774 }
1775 if let Some(host) = &overrides.host {
1776 relay.host = host
1777 .parse::<IpAddr>()
1778 .with_context(|| ConfigError::field("host"))?;
1779 }
1780 if let Some(port) = &overrides.port {
1781 relay.port = port
1782 .as_str()
1783 .parse()
1784 .with_context(|| ConfigError::field("port"))?;
1785 }
1786
1787 let processing = &mut self.values.processing;
1788 if let Some(enabled) = &overrides.processing {
1789 match enabled.to_lowercase().as_str() {
1790 "true" | "1" => processing.enabled = true,
1791 "false" | "0" | "" => processing.enabled = false,
1792 _ => return Err(ConfigError::field("processing").into()),
1793 }
1794 }
1795 if let Some(redis) = overrides.redis_url.clone() {
1796 processing.redis = Some(RedisConfigs::Unified(RedisConfig::single(redis)))
1797 }
1798 if let Some(kafka_url) = overrides.kafka_url.clone() {
1799 let existing = processing
1800 .kafka_config
1801 .iter_mut()
1802 .find(|e| e.name == "bootstrap.servers");
1803
1804 if let Some(config_param) = existing {
1805 config_param.value = kafka_url;
1806 } else {
1807 self.values.processing.kafka_config.push(KafkaConfigParam {
1808 name: "bootstrap.servers".to_owned(),
1809 value: kafka_url,
1810 })
1811 }
1812 }
1813
1814 if overrides.outcome_source.is_some() {
1815 self.values.outcomes.source = overrides.outcome_source.clone();
1816 }
1817
1818 if let Some(shutdown_timeout) = &overrides.shutdown_timeout
1819 && let Ok(shutdown_timeout) = shutdown_timeout.parse::<u64>()
1820 {
1821 self.values.limits.shutdown_timeout = shutdown_timeout;
1822 }
1823
1824 if let Some(server_name) = overrides.server_name.clone() {
1825 self.values.sentry.server_name = Some(server_name.into());
1826 }
1827
1828 let id = if let Some(id) = &overrides.id {
1829 let id = Uuid::parse_str(id).with_context(|| ConfigError::field("id"))?;
1830 Some(id)
1831 } else {
1832 None
1833 };
1834 let public_key = if let Some(public_key) = &overrides.public_key {
1835 let public_key = public_key
1836 .parse::<PublicKey>()
1837 .with_context(|| ConfigError::field("public_key"))?;
1838 Some(public_key)
1839 } else {
1840 None
1841 };
1842
1843 let secret_key = if let Some(secret_key) = &overrides.secret_key {
1844 let secret_key = secret_key
1845 .parse::<SecretKey>()
1846 .with_context(|| ConfigError::field("secret_key"))?;
1847 Some(secret_key)
1848 } else {
1849 None
1850 };
1851
1852 if let Some(credentials) = &mut self.credentials {
1853 if let Some(id) = id {
1855 credentials.id = id;
1856 }
1857 if let Some(public_key) = public_key {
1858 credentials.public_key = public_key;
1859 }
1860 if let Some(secret_key) = secret_key {
1861 credentials.secret_key = secret_key
1862 }
1863 } else {
1864 match (id, public_key, secret_key) {
1866 (Some(id), Some(public_key), Some(secret_key)) => {
1867 self.credentials = Some(Credentials {
1868 secret_key,
1869 public_key,
1870 id,
1871 })
1872 }
1873 (None, None, None) => {
1874 }
1877 _ => {
1878 return Err(ConfigError::field("incomplete credentials").into());
1879 }
1880 }
1881 }
1882
1883 Ok(())
1884 }
1885
1886 fn reload_with(&mut self, other: &Self) -> bool {
1890 let mut changed = false;
1891
1892 if self.values.health != other.values.health {
1893 relay_log::debug!("updating health");
1894 self.values.health = other.values.health.clone();
1895 changed = true;
1896 }
1897
1898 changed
1899 }
1900}
1901
1902pub struct Config {
1904 inner_access: Mutex<()>,
1913 inner: ArcSwap<ConfigInner>,
1915 overrides: Vec<OverridableConfig>,
1920 path: PathBuf,
1924}
1925
1926impl fmt::Debug for Config {
1927 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1928 let inner = self.inner.load();
1929
1930 f.debug_struct("Config")
1931 .field("path", &self.path)
1932 .field("values", &inner.values)
1934 .finish()
1935 }
1936}
1937
1938impl Config {
1939 pub fn from_path<P: AsRef<Path>>(path: P) -> anyhow::Result<Config> {
1941 let path = env::current_dir()
1942 .map(|x| x.join(path.as_ref()))
1943 .unwrap_or_else(|_| path.as_ref().to_path_buf());
1944
1945 let inner = ConfigInner {
1946 values: ConfigValues::load(&path)?,
1947 credentials: match Credentials::path(&path).exists() {
1948 true => Some(Credentials::load(&path)?),
1949 false => None,
1950 },
1951 };
1952
1953 let config = Config {
1954 inner_access: Mutex::new(()),
1955 inner: ArcSwap::from_pointee(inner),
1956 overrides: Vec::new(),
1957 path: path.clone(),
1958 };
1959
1960 if cfg!(not(feature = "processing")) && config.current().processing_enabled() {
1961 return Err(ConfigError::file(ConfigErrorKind::ProcessingNotAvailable, &path).into());
1962 }
1963
1964 Ok(config)
1965 }
1966
1967 pub fn from_json_value(value: serde_json::Value) -> anyhow::Result<Config> {
1971 Ok(Config {
1972 inner_access: Mutex::new(()),
1973 inner: ArcSwap::from_pointee(ConfigInner {
1974 values: serde_json::from_value(value)
1975 .with_context(|| ConfigError::new(ConfigErrorKind::BadJson))?,
1976 credentials: None,
1977 }),
1978 overrides: Vec::new(),
1979 path: PathBuf::new(),
1980 })
1981 }
1982
1983 pub fn apply_override(&mut self, overrides: OverridableConfig) -> anyhow::Result<&mut Self> {
1988 crate::utils::try_rcu(&self.inner, |inner| {
1991 let mut new = ConfigInner::clone(inner);
1992 new.apply_overrides(&overrides)?;
1993 Ok::<_, anyhow::Error>(Arc::new(new))
1994 })?;
1995
1996 self.overrides.push(overrides);
1998
1999 Ok(self)
2000 }
2001
2002 pub fn config_exists<P: AsRef<Path>>(path: P) -> bool {
2004 fs::metadata(ConfigValues::path(path.as_ref())).is_ok()
2005 }
2006
2007 pub fn path(&self) -> &Path {
2009 &self.path
2010 }
2011
2012 pub fn to_yaml_string(&self) -> anyhow::Result<String> {
2014 serde_yaml::to_string(&self.inner.load().values)
2015 .with_context(|| ConfigError::new(ConfigErrorKind::CouldNotWriteFile))
2016 }
2017
2018 pub fn replace_credentials(
2022 &mut self,
2023 credentials: Option<Credentials>,
2024 ) -> anyhow::Result<bool> {
2025 if self.inner.load().credentials == credentials {
2026 return Ok(false);
2027 }
2028
2029 if !self.path.is_empty() {
2030 match &credentials {
2031 Some(creds) => {
2032 creds.save(&self.path)?;
2033 }
2034 None => {
2035 let path = Credentials::path(&self.path);
2036 if fs::metadata(&path).is_ok() {
2037 fs::remove_file(&path).with_context(|| {
2038 ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path)
2039 })?;
2040 }
2041 }
2042 }
2043 }
2044
2045 self.inner.rcu(|inner| {
2050 let mut inner = ConfigInner::clone(inner);
2051 inner.credentials = credentials.clone();
2052 Arc::new(inner)
2053 });
2054
2055 Ok(true)
2056 }
2057
2058 pub fn reload(&self) -> anyhow::Result<bool> {
2068 if self.path.is_empty() {
2069 return Ok(false);
2070 }
2071
2072 let _access = self
2073 .inner_access
2074 .lock()
2075 .unwrap_or_else(PoisonError::into_inner);
2076
2077 let mut new_config = Self::from_path(&self.path)?;
2078 for overrides in &self.overrides {
2079 new_config.apply_override(overrides.clone())?;
2080 }
2081 let new_config = new_config.current();
2082
2083 let mut changed = false;
2084
2085 self.inner.rcu(|inner| {
2087 let mut new_inner = ConfigInner::clone(inner);
2088 changed = new_inner.reload_with(&new_config.inner);
2089 match changed {
2090 true => Arc::new(new_inner),
2091 false => Arc::clone(inner),
2092 }
2093 });
2094
2095 Ok(changed)
2096 }
2097
2098 pub fn current(&self) -> ConfigSnapshot {
2103 let inner = self.inner.load();
2104 ConfigSnapshot { inner }
2105 }
2106}
2107
2108impl Default for Config {
2109 fn default() -> Self {
2110 Self {
2111 inner_access: Mutex::new(()),
2112 inner: ArcSwap::from_pointee(Default::default()),
2113 overrides: Vec::new(),
2114 path: PathBuf::new(),
2115 }
2116 }
2117}
2118
2119pub struct ConfigSnapshot {
2127 inner: arc_swap::Guard<Arc<ConfigInner>>,
2128}
2129
2130impl ConfigSnapshot {
2131 pub fn has_credentials(&self) -> bool {
2133 self.inner.credentials.is_some()
2134 }
2135
2136 pub fn credentials(&self) -> Option<&Credentials> {
2138 self.inner.credentials.as_ref()
2139 }
2140
2141 pub fn secret_key(&self) -> Option<&SecretKey> {
2143 self.inner.credentials.as_ref().map(|x| &x.secret_key)
2144 }
2145
2146 pub fn public_key(&self) -> Option<&PublicKey> {
2148 self.inner.credentials.as_ref().map(|x| &x.public_key)
2149 }
2150
2151 pub fn relay_id(&self) -> Option<&RelayId> {
2153 self.inner.credentials.as_ref().map(|x| &x.id)
2154 }
2155
2156 pub fn relay_mode(&self) -> RelayMode {
2158 self.inner.values.relay.mode
2159 }
2160
2161 pub fn relay_instance(&self) -> RelayInstance {
2163 self.inner.values.relay.instance
2164 }
2165
2166 pub fn upstream(&self) -> &UpstreamDescriptor {
2168 &self.inner.values.relay.upstream
2169 }
2170
2171 pub fn advertised_upstream(&self) -> Option<&UpstreamDescriptor> {
2173 self.inner.values.relay.advertised_upstream.as_ref()
2174 }
2175
2176 pub fn http_host_header(&self) -> Option<&str> {
2178 self.inner.values.http.host_header.as_deref()
2179 }
2180
2181 pub fn listen_addr(&self) -> SocketAddr {
2183 (self.inner.values.relay.host, self.inner.values.relay.port).into()
2184 }
2185
2186 pub fn listen_addr_internal(&self) -> Option<SocketAddr> {
2194 match (
2195 self.inner.values.relay.internal_host,
2196 self.inner.values.relay.internal_port,
2197 ) {
2198 (Some(host), None) => Some((host, self.inner.values.relay.port).into()),
2199 (None, Some(port)) => Some((self.inner.values.relay.host, port).into()),
2200 (Some(host), Some(port)) => Some((host, port).into()),
2201 (None, None) => None,
2202 }
2203 }
2204
2205 pub fn tls_listen_addr(&self) -> Option<SocketAddr> {
2207 if self.inner.values.relay.tls_identity_path.is_some() {
2208 let port = self.inner.values.relay.tls_port.unwrap_or(3443);
2209 Some((self.inner.values.relay.host, port).into())
2210 } else {
2211 None
2212 }
2213 }
2214
2215 pub fn tls_identity_path(&self) -> Option<&Path> {
2217 self.inner.values.relay.tls_identity_path.as_deref()
2218 }
2219
2220 pub fn tls_identity_password(&self) -> Option<&str> {
2222 self.inner.values.relay.tls_identity_password.as_deref()
2223 }
2224
2225 pub fn override_project_ids(&self) -> bool {
2229 self.inner.values.relay.override_project_ids
2230 }
2231
2232 pub fn config_reload_interval(&self) -> Option<Duration> {
2236 let interval = self.inner.values.relay.config_reload_interval?;
2237 Some(match interval {
2238 0 => Duration::from_millis(50),
2241 secs => Duration::from_secs(secs),
2242 })
2243 }
2244
2245 pub fn requires_auth(&self) -> bool {
2249 match self.inner.values.auth.ready {
2250 ReadinessCondition::Authenticated => self.relay_mode() == RelayMode::Managed,
2251 ReadinessCondition::Always => false,
2252 }
2253 }
2254
2255 pub fn http_auth_interval(&self) -> Option<Duration> {
2259 if self.processing_enabled() {
2260 return None;
2261 }
2262
2263 match self.inner.values.http.auth_interval {
2264 None | Some(0) => None,
2265 Some(secs) => Some(Duration::from_secs(secs)),
2266 }
2267 }
2268
2269 pub fn http_outage_grace_period(&self) -> Duration {
2272 Duration::from_secs(self.inner.values.http.outage_grace_period)
2273 }
2274
2275 pub fn http_retry_delay(&self) -> Duration {
2280 Duration::from_secs(self.inner.values.http.retry_delay)
2281 }
2282
2283 pub fn http_project_failure_interval(&self) -> Duration {
2285 Duration::from_secs(self.inner.values.http.project_failure_interval)
2286 }
2287
2288 pub fn http_encoding(&self) -> HttpEncoding {
2290 self.inner.values.http.encoding
2291 }
2292
2293 pub fn http_global_metrics(&self) -> bool {
2295 self.inner.values.http.global_metrics
2296 }
2297
2298 pub fn http_forward(&self) -> bool {
2303 self.inner.values.http.forward && !self.processing_enabled()
2304 }
2305
2306 pub fn emit_outcomes(&self) -> EmitOutcomes {
2311 if self.processing_enabled() {
2312 return EmitOutcomes::AsOutcomes;
2313 }
2314 self.inner.values.outcomes.emit_outcomes
2315 }
2316
2317 pub fn outcome_batch_size(&self) -> usize {
2319 self.inner.values.outcomes.batch_size
2320 }
2321
2322 pub fn outcome_batch_interval(&self) -> Duration {
2324 Duration::from_millis(self.inner.values.outcomes.batch_interval)
2325 }
2326
2327 pub fn outcome_source(&self) -> Option<&str> {
2329 self.inner.values.outcomes.source.as_deref()
2330 }
2331
2332 pub fn logging(&self) -> &relay_log::LogConfig {
2334 &self.inner.values.logging
2335 }
2336
2337 pub fn sentry(&self) -> &relay_log::SentryConfig {
2339 &self.inner.values.sentry
2340 }
2341
2342 pub fn statsd_addr(&self) -> Option<&str> {
2344 self.inner.values.metrics.statsd.as_deref()
2345 }
2346
2347 pub fn statsd_buffer_size(&self) -> Option<usize> {
2349 self.inner.values.metrics.statsd_buffer_size
2350 }
2351
2352 pub fn metrics_prefix(&self) -> &str {
2354 &self.inner.values.metrics.prefix
2355 }
2356
2357 pub fn metrics_default_tags(&self) -> &BTreeMap<String, String> {
2359 &self.inner.values.metrics.default_tags
2360 }
2361
2362 pub fn metrics_hostname_tag(&self) -> Option<&str> {
2364 self.inner.values.metrics.hostname_tag.as_deref()
2365 }
2366
2367 pub fn metrics_periodic_interval(&self) -> Option<Duration> {
2371 match self.inner.values.metrics.periodic_secs {
2372 0 => None,
2373 secs => Some(Duration::from_secs(secs)),
2374 }
2375 }
2376
2377 pub fn http_timeout(&self) -> Duration {
2379 Duration::from_secs(self.inner.values.http.timeout.into())
2380 }
2381
2382 pub fn http_connection_timeout(&self) -> Duration {
2384 Duration::from_secs(self.inner.values.http.connection_timeout.into())
2385 }
2386
2387 pub fn http_max_retry_interval(&self) -> Duration {
2389 Duration::from_secs(self.inner.values.http.max_retry_interval.into())
2390 }
2391
2392 pub fn http_dns_cache(&self) -> bool {
2394 self.inner.values.http.dns_cache
2395 }
2396
2397 pub fn project_cache_expiry(&self) -> Duration {
2399 Duration::from_secs(self.inner.values.cache.project_expiry.into())
2400 }
2401
2402 pub fn request_full_project_config(&self) -> bool {
2404 self.inner.values.cache.project_request_full_config
2405 }
2406
2407 pub fn relay_cache_expiry(&self) -> Duration {
2409 Duration::from_secs(self.inner.values.cache.relay_expiry.into())
2410 }
2411
2412 pub fn envelope_buffer_size(&self) -> usize {
2414 self.inner
2415 .values
2416 .cache
2417 .envelope_buffer_size
2418 .try_into()
2419 .unwrap_or(usize::MAX)
2420 }
2421
2422 pub fn cache_miss_expiry(&self) -> Duration {
2424 Duration::from_secs(self.inner.values.cache.miss_expiry.into())
2425 }
2426
2427 pub fn project_grace_period(&self) -> Duration {
2429 Duration::from_secs(self.inner.values.cache.project_grace_period.into())
2430 }
2431
2432 pub fn project_refresh_interval(&self) -> Option<Duration> {
2436 self.inner
2437 .values
2438 .cache
2439 .project_refresh_interval
2440 .map(Into::into)
2441 .map(Duration::from_secs)
2442 }
2443
2444 pub fn query_batch_interval(&self) -> Duration {
2447 Duration::from_millis(self.inner.values.cache.batch_interval.into())
2448 }
2449
2450 pub fn downstream_relays_batch_interval(&self) -> Duration {
2452 Duration::from_millis(
2453 self.inner
2454 .values
2455 .cache
2456 .downstream_relays_batch_interval
2457 .into(),
2458 )
2459 }
2460
2461 pub fn local_cache_interval(&self) -> Duration {
2463 Duration::from_secs(self.inner.values.cache.file_interval.into())
2464 }
2465
2466 pub fn global_config_fetch_interval(&self) -> Duration {
2469 Duration::from_secs(self.inner.values.cache.global_config_fetch_interval.into())
2470 }
2471
2472 pub fn spool_envelopes_path(&self, partition_id: u8) -> Option<PathBuf> {
2477 let mut path = self
2478 .inner
2479 .values
2480 .spool
2481 .envelopes
2482 .path
2483 .as_ref()
2484 .map(|path| path.to_owned())?;
2485
2486 if partition_id == 0 {
2487 return Some(path);
2488 }
2489
2490 let file_name = path.file_name().and_then(|f| f.to_str())?;
2491 let new_file_name = format!("{file_name}.{partition_id}");
2492 path.set_file_name(new_file_name);
2493
2494 Some(path)
2495 }
2496
2497 pub fn spool_envelopes_max_disk_size(&self) -> usize {
2499 self.inner.values.spool.envelopes.max_disk_size.as_bytes()
2500 }
2501
2502 pub fn spool_envelopes_batch_size_bytes(&self) -> usize {
2505 self.inner
2506 .values
2507 .spool
2508 .envelopes
2509 .batch_size_bytes
2510 .as_bytes()
2511 }
2512
2513 pub fn spool_envelopes_max_age(&self) -> Duration {
2515 Duration::from_secs(self.inner.values.spool.envelopes.max_envelope_delay_secs)
2516 }
2517
2518 pub fn spool_disk_usage_refresh_frequency_ms(&self) -> Duration {
2520 Duration::from_millis(
2521 self.inner
2522 .values
2523 .spool
2524 .envelopes
2525 .disk_usage_refresh_frequency_ms,
2526 )
2527 }
2528
2529 pub fn spool_max_backpressure_memory_percent(&self) -> f32 {
2531 self.inner
2532 .values
2533 .spool
2534 .envelopes
2535 .max_backpressure_memory_percent
2536 }
2537
2538 pub fn spool_partitions(&self) -> NonZeroU8 {
2540 self.inner.values.spool.envelopes.partitions
2541 }
2542
2543 pub fn spool_partitioning(&self) -> EnvelopeSpoolPartitioning {
2545 self.inner.values.spool.envelopes.partitioning
2546 }
2547
2548 pub fn spool_ephemeral(&self) -> bool {
2550 self.inner.values.spool.envelopes.ephemeral
2551 }
2552
2553 pub fn max_event_size(&self) -> usize {
2555 self.inner.values.limits.max_event_size.as_bytes()
2556 }
2557
2558 pub fn max_attachment_size(&self) -> usize {
2560 self.inner.values.limits.max_attachment_size.as_bytes()
2561 }
2562
2563 pub fn max_attachment_count(&self) -> usize {
2565 self.inner.values.limits.max_attachment_count
2566 }
2567
2568 pub fn max_attachments_size(&self) -> usize {
2571 self.inner.values.limits.max_attachments_size.as_bytes()
2572 }
2573
2574 pub fn max_upload_size(&self) -> usize {
2576 self.inner.values.limits.max_upload_size.as_bytes()
2577 }
2578
2579 pub fn max_client_reports_count(&self) -> usize {
2581 self.inner.values.limits.max_client_reports_count
2582 }
2583
2584 pub fn max_client_reports_size(&self) -> usize {
2586 self.inner.values.limits.max_client_reports_size.as_bytes()
2587 }
2588
2589 pub fn max_check_in_size(&self) -> usize {
2591 self.inner.values.limits.max_check_in_size.as_bytes()
2592 }
2593
2594 pub fn max_log_size(&self) -> usize {
2596 self.inner.values.limits.max_log_size.as_bytes()
2597 }
2598
2599 pub fn max_span_size(&self) -> usize {
2601 self.inner.values.limits.max_span_size.as_bytes()
2602 }
2603
2604 pub fn max_standalone_span_count(&self) -> usize {
2606 self.inner.values.limits.max_standalone_span_count
2607 }
2608
2609 pub fn max_container_size(&self) -> usize {
2611 self.inner.values.limits.max_container_size.as_bytes()
2612 }
2613
2614 pub fn max_envelope_size(&self) -> usize {
2618 self.inner.values.limits.max_envelope_size.as_bytes()
2619 }
2620
2621 pub fn max_session_count(&self) -> usize {
2623 self.inner.values.limits.max_session_count
2624 }
2625
2626 pub fn max_sessions_size(&self) -> usize {
2628 self.inner.values.limits.max_sessions_size.as_bytes()
2629 }
2630
2631 pub fn max_statsd_size(&self) -> usize {
2633 self.inner.values.limits.max_statsd_size.as_bytes()
2634 }
2635
2636 pub fn max_metric_buckets_size(&self) -> usize {
2638 self.inner.values.limits.max_metric_buckets_size.as_bytes()
2639 }
2640
2641 pub fn max_api_payload_size(&self) -> usize {
2643 self.inner.values.limits.max_api_payload_size.as_bytes()
2644 }
2645
2646 pub fn max_api_file_upload_size(&self) -> usize {
2648 self.inner.values.limits.max_api_file_upload_size.as_bytes()
2649 }
2650
2651 pub fn max_api_chunk_upload_size(&self) -> usize {
2653 self.inner
2654 .values
2655 .limits
2656 .max_api_chunk_upload_size
2657 .as_bytes()
2658 }
2659
2660 pub fn max_profile_size(&self) -> usize {
2662 self.inner.values.limits.max_profile_size.as_bytes()
2663 }
2664
2665 pub fn max_trace_metric_size(&self) -> usize {
2667 self.inner.values.limits.max_trace_metric_size.as_bytes()
2668 }
2669
2670 pub fn max_replay_compressed_size(&self) -> usize {
2672 self.inner
2673 .values
2674 .limits
2675 .max_replay_compressed_size
2676 .as_bytes()
2677 }
2678
2679 pub fn max_replay_uncompressed_size(&self) -> usize {
2681 self.inner
2682 .values
2683 .limits
2684 .max_replay_uncompressed_size
2685 .as_bytes()
2686 }
2687
2688 pub fn max_replay_message_size(&self) -> usize {
2694 self.inner.values.limits.max_replay_message_size.as_bytes()
2695 }
2696
2697 pub fn max_concurrent_requests(&self) -> usize {
2699 self.inner.values.limits.max_concurrent_requests
2700 }
2701
2702 pub fn max_concurrent_queries(&self) -> usize {
2704 self.inner.values.limits.max_concurrent_queries
2705 }
2706
2707 pub fn max_removed_attribute_key_size(&self) -> usize {
2709 self.inner
2710 .values
2711 .limits
2712 .max_removed_attribute_key_size
2713 .as_bytes()
2714 }
2715
2716 pub fn query_timeout(&self) -> Duration {
2718 Duration::from_secs(self.inner.values.limits.query_timeout)
2719 }
2720
2721 pub fn shutdown_timeout(&self) -> Duration {
2724 Duration::from_secs(self.inner.values.limits.shutdown_timeout)
2725 }
2726
2727 pub fn keepalive_timeout(&self) -> Duration {
2731 Duration::from_secs(self.inner.values.limits.keepalive_timeout)
2732 }
2733
2734 pub fn idle_timeout(&self) -> Option<Duration> {
2736 self.inner
2737 .values
2738 .limits
2739 .idle_timeout
2740 .map(Duration::from_secs)
2741 }
2742
2743 pub fn max_connections(&self) -> Option<usize> {
2745 self.inner.values.limits.max_connections
2746 }
2747
2748 pub fn tcp_listen_backlog(&self) -> u32 {
2750 self.inner.values.limits.tcp_listen_backlog
2751 }
2752
2753 pub fn cpu_concurrency(&self) -> usize {
2755 self.inner.values.limits.max_thread_count
2756 }
2757
2758 pub fn pool_concurrency(&self) -> usize {
2760 self.inner.values.limits.max_pool_concurrency
2761 }
2762
2763 pub fn query_batch_size(&self) -> usize {
2765 self.inner.values.cache.batch_size
2766 }
2767
2768 pub fn processing_enabled(&self) -> bool {
2770 self.inner.values.processing.enabled
2771 }
2772
2773 pub fn normalization_level(&self) -> NormalizationLevel {
2775 self.inner.values.normalization.level
2776 }
2777
2778 pub fn geoip_path(&self) -> Option<&Path> {
2780 self.inner.values.geoip.path.as_deref().or(self
2781 .inner
2782 .values
2783 .processing
2784 .geoip_path
2785 .as_deref())
2786 }
2787
2788 pub fn max_secs_in_future(&self) -> i64 {
2792 self.inner.values.processing.max_secs_in_future.into()
2793 }
2794
2795 pub fn max_session_secs_in_past(&self) -> i64 {
2797 self.inner.values.processing.max_session_secs_in_past.into()
2798 }
2799
2800 pub fn kafka_configs(
2802 &self,
2803 topic: KafkaTopic,
2804 ) -> Result<KafkaTopicConfig<'_>, KafkaConfigError> {
2805 self.inner
2806 .values
2807 .processing
2808 .topics
2809 .get(topic)
2810 .kafka_configs(
2811 &self.inner.values.processing.kafka_config,
2812 &self.inner.values.processing.secondary_kafka_configs,
2813 )
2814 }
2815
2816 pub fn kafka_validate_topics(&self) -> bool {
2818 self.inner.values.processing.kafka_validate_topics
2819 }
2820
2821 pub fn unused_topic_assignments(&self) -> &relay_kafka::Unused {
2823 &self.inner.values.processing.topics.unused
2824 }
2825
2826 pub fn objectstore(&self) -> &ObjectstoreServiceConfig {
2828 &self.inner.values.processing.objectstore
2829 }
2830
2831 pub fn upload(&self) -> &Upload {
2833 &self.inner.values.upload
2834 }
2835
2836 #[cfg(feature = "processing")]
2838 pub fn upload_signing_key(&self) -> Option<&SecretKey> {
2839 self.upload()
2840 .credentials
2841 .as_ref()
2842 .map(|c| &c.signing_key)
2843 .or(self.credentials().map(|c| &c.secret_key))
2844 }
2845
2846 #[cfg(feature = "processing")]
2848 pub fn upload_verification_key(&self) -> Option<&PublicKey> {
2849 self.upload()
2850 .credentials
2851 .as_ref()
2852 .map(|c| &c.verification_key)
2853 .or(self.credentials().map(|c| &c.public_key))
2854 }
2855
2856 pub fn redis(&self) -> Option<RedisConfigsRef<'_>> {
2858 let redis_configs = self.inner.values.processing.redis.as_ref()?;
2859
2860 Some(build_redis_configs(
2861 redis_configs,
2862 self.cpu_concurrency() as u32,
2863 self.pool_concurrency() as u32,
2864 ))
2865 }
2866
2867 pub fn attachment_chunk_size(&self) -> usize {
2869 self.inner
2870 .values
2871 .processing
2872 .attachment_chunk_size
2873 .as_bytes()
2874 }
2875
2876 pub fn metrics_max_batch_size_bytes(&self) -> usize {
2878 self.inner.values.aggregator.max_flush_bytes
2879 }
2880
2881 pub fn projectconfig_cache_prefix(&self) -> &str {
2884 &self.inner.values.processing.projectconfig_cache_prefix
2885 }
2886
2887 pub fn max_rate_limit(&self) -> Option<u64> {
2889 self.inner.values.processing.max_rate_limit.map(u32::into)
2890 }
2891
2892 pub fn quota_cache_ratio(&self) -> Option<f32> {
2894 self.inner.values.processing.quota_cache_ratio
2895 }
2896
2897 pub fn quota_cache_max(&self) -> Option<f32> {
2899 self.inner.values.processing.quota_cache_max
2900 }
2901
2902 pub fn health_refresh_interval(&self) -> Duration {
2904 Duration::from_millis(self.inner.values.health.refresh_interval_ms)
2905 }
2906
2907 pub fn health_max_memory_watermark_bytes(&self) -> u64 {
2909 self.inner
2910 .values
2911 .health
2912 .max_memory_bytes
2913 .as_ref()
2914 .map_or(u64::MAX, |b| b.as_bytes() as u64)
2915 }
2916
2917 pub fn health_max_memory_watermark_percent(&self) -> f32 {
2919 self.inner.values.health.max_memory_percent
2920 }
2921
2922 pub fn health_probe_timeout(&self) -> Duration {
2924 Duration::from_millis(self.inner.values.health.probe_timeout_ms)
2925 }
2926
2927 pub fn memory_stat_refresh_frequency_ms(&self) -> u64 {
2929 self.inner.values.health.memory_stat_refresh_frequency_ms
2930 }
2931
2932 pub fn cogs_max_queue_size(&self) -> u64 {
2934 self.inner.values.cogs.max_queue_size
2935 }
2936
2937 pub fn cogs_relay_resource_id(&self) -> &str {
2939 &self.inner.values.cogs.relay_resource_id
2940 }
2941
2942 pub fn default_aggregator_config(&self) -> &AggregatorServiceConfig {
2944 &self.inner.values.aggregator
2945 }
2946
2947 pub fn secondary_aggregator_configs(&self) -> &Vec<ScopedAggregatorConfig> {
2949 &self.inner.values.secondary_aggregators
2950 }
2951
2952 pub fn aggregator_config_for(&self, namespace: MetricNamespace) -> &AggregatorServiceConfig {
2954 for entry in &self.inner.values.secondary_aggregators {
2955 if entry.condition.matches(Some(namespace)) {
2956 return &entry.config;
2957 }
2958 }
2959 &self.inner.values.aggregator
2960 }
2961
2962 pub fn static_relays(&self) -> &HashMap<RelayId, RelayInfo> {
2964 &self.inner.values.auth.static_relays
2965 }
2966
2967 pub fn signature_max_age(&self) -> Duration {
2969 Duration::from_secs(self.inner.values.auth.signature_max_age)
2970 }
2971
2972 pub fn accept_unknown_items(&self) -> bool {
2974 let forward = self.inner.values.routing.accept_unknown_items;
2975 forward.unwrap_or_else(|| !self.processing_enabled())
2976 }
2977}
2978
2979impl fmt::Debug for ConfigSnapshot {
2980 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2981 f.debug_struct("ConfigSnapshot")
2982 .field("values", &self.inner.values)
2983 .finish()
2984 }
2985}
2986
2987#[cfg(test)]
2988mod tests {
2989 use super::*;
2990
2991 #[test]
2993 fn test_event_buffer_size() {
2994 let yaml = r###"
2995cache:
2996 event_buffer_size: 1000000
2997 event_expiry: 1800
2998"###;
2999
3000 let values: ConfigValues = serde_yaml::from_str(yaml).unwrap();
3001 assert_eq!(values.cache.envelope_buffer_size, 1_000_000);
3002 assert_eq!(values.cache.envelope_expiry, 1800);
3003 }
3004
3005 #[cfg(feature = "processing")]
3006 #[test]
3007 fn test_upload_secret_key_from_file() {
3008 let path = env::temp_dir().join(Uuid::new_v4().to_string());
3009 fs::create_dir(&path).unwrap();
3010 fs::write(
3011 path.join("my_secret.txt"),
3012 "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks",
3013 )
3014 .unwrap();
3015 fs::write(
3016 ConfigValues::path(&path),
3017 r#"
3018 upload:
3019 credentials:
3020 signing_key: ${file:my_secret.txt}
3021 verification_key: "VNS8haF0VTnuMMDR2t-f7AgnmUcXmcdzV3SVksSk34s""#,
3022 )
3023 .unwrap();
3024
3025 let config = Config::from_path(&path).unwrap().current();
3026
3027 fs::remove_dir_all(path).unwrap();
3028
3029 let signing_key = &config.upload().credentials.as_ref().unwrap().signing_key;
3030 assert_eq!(
3031 signing_key.to_string(),
3032 "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks"
3033 );
3034 }
3035
3036 #[test]
3037 fn test_emit_outcomes() {
3038 for (serialized, deserialized) in &[
3039 ("true", EmitOutcomes::AsOutcomes),
3040 ("false", EmitOutcomes::None),
3041 ("\"as_client_reports\"", EmitOutcomes::AsClientReports),
3042 ] {
3043 let value: EmitOutcomes = serde_json::from_str(serialized).unwrap();
3044 assert_eq!(value, *deserialized);
3045 assert_eq!(serde_json::to_string(&value).unwrap(), *serialized);
3046 }
3047 }
3048
3049 #[test]
3050 fn test_emit_outcomes_invalid() {
3051 assert!(serde_json::from_str::<EmitOutcomes>("asdf").is_err());
3052 }
3053}