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 pub miss_expiry: u32,
1100 pub batch_interval: u32,
1102 pub downstream_relays_batch_interval: u32,
1104 pub batch_size: usize,
1108 pub file_interval: u32,
1110 pub global_config_fetch_interval: u32,
1112}
1113
1114impl Default for Cache {
1115 fn default() -> Self {
1116 Cache {
1117 project_request_full_config: false,
1118 project_expiry: 300, project_grace_period: 120, project_refresh_interval: None,
1121 relay_expiry: 3600, miss_expiry: 60, batch_interval: 100, downstream_relays_batch_interval: 100, batch_size: 500,
1126 file_interval: 10, global_config_fetch_interval: 10, }
1129 }
1130}
1131
1132#[derive(Serialize, Deserialize, Debug, Clone)]
1134#[serde(default)]
1135pub struct Processing {
1136 pub enabled: bool,
1138 pub geoip_path: Option<PathBuf>,
1140 pub max_secs_in_future: u32,
1142 pub max_session_secs_in_past: u32,
1144 pub kafka_config: Vec<KafkaConfigParam>,
1146 pub secondary_kafka_configs: BTreeMap<String, Vec<KafkaConfigParam>>,
1166 pub topics: TopicAssignments,
1168 pub kafka_validate_topics: bool,
1170 pub redis: Option<RedisConfigs>,
1172 pub attachment_chunk_size: ByteSize,
1174 pub projectconfig_cache_prefix: String,
1176 pub max_rate_limit: Option<u32>,
1178 pub quota_cache_ratio: Option<f32>,
1189 pub quota_cache_max: Option<f32>,
1196 #[serde(alias = "upload")]
1198 pub objectstore: ObjectstoreServiceConfig,
1199}
1200
1201impl Default for Processing {
1202 fn default() -> Self {
1204 Self {
1205 enabled: false,
1206 geoip_path: None,
1207 max_secs_in_future: 60, max_session_secs_in_past: 5 * 24 * 3600, kafka_config: Vec::new(),
1210 secondary_kafka_configs: BTreeMap::new(),
1211 topics: TopicAssignments::default(),
1212 kafka_validate_topics: false,
1213 redis: None,
1214 attachment_chunk_size: ByteSize::mebibytes(1),
1215 projectconfig_cache_prefix: "relayconfig".to_owned(),
1216 max_rate_limit: Some(300), quota_cache_ratio: None,
1218 quota_cache_max: None,
1219 objectstore: ObjectstoreServiceConfig::default(),
1220 }
1221 }
1222}
1223
1224#[derive(Debug, Default, Serialize, Deserialize, Clone)]
1226#[serde(default)]
1227pub struct Normalization {
1228 pub level: NormalizationLevel,
1230}
1231
1232#[derive(Copy, Clone, Debug, Default, Serialize, Deserialize, Eq, PartialEq)]
1234#[serde(rename_all = "lowercase")]
1235pub enum NormalizationLevel {
1236 #[default]
1240 Default,
1241 Full,
1246}
1247
1248#[derive(Serialize, Deserialize, Clone)]
1250pub struct ObjectstoreAuthConfig {
1251 pub key_id: String,
1254
1255 pub signing_key: String,
1257}
1258
1259impl fmt::Debug for ObjectstoreAuthConfig {
1260 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1261 f.debug_struct("ObjectstoreAuthConfig")
1262 .field("key_id", &self.key_id)
1263 .field("signing_key", &"[redacted]")
1264 .finish()
1265 }
1266}
1267
1268#[derive(Serialize, Deserialize, Debug, Clone)]
1270#[serde(default)]
1271pub struct ObjectstoreServiceConfig {
1272 pub objectstore_url: Option<String>,
1277
1278 pub max_concurrent_requests: usize,
1280
1281 pub max_backlog: usize,
1285
1286 pub timeout: u64,
1291
1292 pub stream_timeout: u64,
1297
1298 pub retry_delay: f64,
1300
1301 pub max_attempts: NonZeroU16,
1303
1304 pub fallback_to_kafka: bool,
1309
1310 pub auth: Option<ObjectstoreAuthConfig>,
1312}
1313
1314impl Default for ObjectstoreServiceConfig {
1315 fn default() -> Self {
1316 Self {
1317 objectstore_url: None,
1318 max_concurrent_requests: 10,
1319 max_backlog: 20,
1320 timeout: 60,
1321 stream_timeout: 5 * 60, retry_delay: 1.0,
1323 max_attempts: NonZeroU16::new(5).unwrap(),
1324 fallback_to_kafka: true,
1325 auth: None,
1326 }
1327 }
1328}
1329
1330#[derive(Copy, Clone, Debug, PartialEq, Eq)]
1333
1334pub enum EmitOutcomes {
1335 None,
1337 AsClientReports,
1339 AsOutcomes,
1341}
1342
1343impl EmitOutcomes {
1344 pub fn any(&self) -> bool {
1346 !matches!(self, EmitOutcomes::None)
1347 }
1348}
1349
1350impl Serialize for EmitOutcomes {
1351 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1352 where
1353 S: Serializer,
1354 {
1355 match self {
1357 Self::None => serializer.serialize_bool(false),
1358 Self::AsClientReports => serializer.serialize_str("as_client_reports"),
1359 Self::AsOutcomes => serializer.serialize_bool(true),
1360 }
1361 }
1362}
1363
1364struct EmitOutcomesVisitor;
1365
1366impl Visitor<'_> for EmitOutcomesVisitor {
1367 type Value = EmitOutcomes;
1368
1369 fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
1370 formatter.write_str("true, false, 'as_client_reports'")
1371 }
1372
1373 fn visit_bool<E>(self, v: bool) -> Result<Self::Value, E>
1374 where
1375 E: serde::de::Error,
1376 {
1377 Ok(if v {
1378 EmitOutcomes::AsOutcomes
1379 } else {
1380 EmitOutcomes::None
1381 })
1382 }
1383
1384 fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
1385 where
1386 E: serde::de::Error,
1387 {
1388 match v {
1389 "as_client_reports" => Ok(EmitOutcomes::AsClientReports),
1390 _ => Err(E::invalid_value(Unexpected::Str(v), &self)),
1391 }
1392 }
1393}
1394
1395impl<'de> Deserialize<'de> for EmitOutcomes {
1396 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1397 where
1398 D: Deserializer<'de>,
1399 {
1400 deserializer.deserialize_any(EmitOutcomesVisitor)
1401 }
1402}
1403
1404#[derive(Serialize, Deserialize, Debug, Clone)]
1406#[serde(default)]
1407pub struct Outcomes {
1408 pub emit_outcomes: EmitOutcomes,
1412 pub batch_size: usize,
1415 pub batch_interval: u64,
1418 pub source: Option<String>,
1421}
1422
1423impl Default for Outcomes {
1424 fn default() -> Self {
1425 Outcomes {
1426 emit_outcomes: EmitOutcomes::AsClientReports,
1427 batch_size: 1000,
1428 batch_interval: 500,
1429 source: None,
1430 }
1431 }
1432}
1433
1434#[derive(Serialize, Deserialize, Debug, Default)]
1436pub struct MinimalConfig {
1437 pub relay: Relay,
1439}
1440
1441impl MinimalConfig {
1442 pub fn save_in_folder<P: AsRef<Path>>(&self, p: P) -> anyhow::Result<()> {
1444 let path = p.as_ref();
1445 if fs::metadata(path).is_err() {
1446 fs::create_dir_all(path)
1447 .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotOpenFile, path))?;
1448 }
1449 self.save(path)
1450 }
1451}
1452
1453impl ConfigObject for MinimalConfig {
1454 fn format() -> ConfigFormat {
1455 ConfigFormat::Yaml
1456 }
1457
1458 fn name() -> &'static str {
1459 "config"
1460 }
1461}
1462
1463mod config_relay_info {
1465 use serde::ser::SerializeMap;
1466
1467 use super::*;
1468
1469 #[derive(Debug, Serialize, Deserialize, Clone)]
1471 struct RelayInfoConfig {
1472 public_key: PublicKey,
1473 #[serde(default)]
1474 internal: bool,
1475 }
1476
1477 impl From<RelayInfoConfig> for RelayInfo {
1478 fn from(v: RelayInfoConfig) -> Self {
1479 RelayInfo {
1480 public_key: v.public_key,
1481 internal: v.internal,
1482 }
1483 }
1484 }
1485
1486 impl From<RelayInfo> for RelayInfoConfig {
1487 fn from(v: RelayInfo) -> Self {
1488 RelayInfoConfig {
1489 public_key: v.public_key,
1490 internal: v.internal,
1491 }
1492 }
1493 }
1494
1495 pub(super) fn deserialize<'de, D>(des: D) -> Result<HashMap<RelayId, RelayInfo>, D::Error>
1496 where
1497 D: Deserializer<'de>,
1498 {
1499 let map = HashMap::<RelayId, RelayInfoConfig>::deserialize(des)?;
1500 Ok(map.into_iter().map(|(k, v)| (k, v.into())).collect())
1501 }
1502
1503 pub(super) fn serialize<S>(elm: &HashMap<RelayId, RelayInfo>, ser: S) -> Result<S::Ok, S::Error>
1504 where
1505 S: Serializer,
1506 {
1507 let mut map = ser.serialize_map(Some(elm.len()))?;
1508
1509 for (k, v) in elm {
1510 map.serialize_entry(k, &RelayInfoConfig::from(v.clone()))?;
1511 }
1512
1513 map.end()
1514 }
1515}
1516
1517#[derive(Serialize, Deserialize, Debug, Clone)]
1519#[serde(default)]
1520pub struct AuthConfig {
1521 #[serde(skip_serializing_if = "is_default")]
1523 pub ready: ReadinessCondition,
1524
1525 #[serde(with = "config_relay_info")]
1527 pub static_relays: HashMap<RelayId, RelayInfo>,
1528
1529 pub signature_max_age: u64,
1533}
1534
1535impl Default for AuthConfig {
1536 fn default() -> Self {
1537 Self {
1538 ready: ReadinessCondition::default(),
1539 static_relays: HashMap::new(),
1540 signature_max_age: 300, }
1542 }
1543}
1544
1545#[derive(Serialize, Deserialize, Debug, Default, Clone)]
1547pub struct GeoIpConfig {
1548 pub path: Option<PathBuf>,
1550}
1551
1552#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
1557#[serde(default)]
1558pub struct Health {
1559 pub refresh_interval_ms: u64,
1566 pub max_memory_bytes: Option<ByteSize>,
1571 pub max_memory_percent: f32,
1575 pub probe_timeout_ms: u64,
1582 pub memory_stat_refresh_frequency_ms: u64,
1588}
1589
1590impl Default for Health {
1591 fn default() -> Self {
1592 Self {
1593 refresh_interval_ms: 3000,
1594 max_memory_bytes: None,
1595 max_memory_percent: 0.95,
1596 probe_timeout_ms: 900,
1597 memory_stat_refresh_frequency_ms: 100,
1598 }
1599 }
1600}
1601
1602#[derive(Serialize, Deserialize, Debug, Clone)]
1604#[serde(default)]
1605pub struct Cogs {
1606 pub max_queue_size: u64,
1612 pub relay_resource_id: String,
1618}
1619
1620impl Default for Cogs {
1621 fn default() -> Self {
1622 Self {
1623 max_queue_size: 10_000,
1624 relay_resource_id: "relay_service".to_owned(),
1625 }
1626 }
1627}
1628
1629#[derive(Debug, Clone, Serialize, Deserialize)]
1631#[serde(default)]
1632pub struct Upload {
1633 pub max_concurrent_requests: usize,
1637 pub timeout: u64,
1639 pub max_age: i64,
1643
1644 pub credentials: Option<UploadCredentials>,
1648}
1649
1650impl Default for Upload {
1651 fn default() -> Self {
1652 Self {
1653 max_concurrent_requests: 100,
1654 timeout: 5 * 60, max_age: 60 * 60, credentials: None,
1657 }
1658 }
1659}
1660
1661#[derive(Clone, Serialize, Deserialize)]
1663pub struct UploadCredentials {
1664 #[cfg(feature = "processing")]
1666 pub signing_key: SecretKey,
1667
1668 pub verification_key: PublicKey,
1670}
1671
1672impl fmt::Debug for UploadCredentials {
1673 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1674 let Self {
1675 #[cfg(feature = "processing")]
1676 signing_key: _,
1677 verification_key,
1678 } = self;
1679 let mut b = f.debug_struct("UploadCredentials");
1680 #[cfg(feature = "processing")]
1681 b.field("signing_key", &"[redacted]");
1682 b.field("verification_key", verification_key).finish()
1683 }
1684}
1685
1686#[derive(Serialize, Deserialize, Debug, Default, Clone)]
1688#[serde(default)]
1689#[allow(missing_docs)]
1690pub struct ConfigValues {
1691 pub relay: Relay,
1692 pub http: Http,
1693 pub cache: Cache,
1694 pub spool: Spool,
1695 pub limits: Limits,
1696 pub logging: relay_log::LogConfig,
1697 pub routing: Routing,
1698 pub metrics: Metrics,
1699 pub sentry: relay_log::SentryConfig,
1700 pub processing: Processing,
1701 pub outcomes: Outcomes,
1702 pub aggregator: AggregatorServiceConfig,
1703 pub secondary_aggregators: Vec<ScopedAggregatorConfig>,
1704 pub auth: AuthConfig,
1705 pub geoip: GeoIpConfig,
1706 pub normalization: Normalization,
1707 pub health: Health,
1708 pub cogs: Cogs,
1709 pub upload: Upload,
1710}
1711
1712impl ConfigObject for ConfigValues {
1713 fn format() -> ConfigFormat {
1714 ConfigFormat::Yaml
1715 }
1716
1717 fn name() -> &'static str {
1718 "config"
1719 }
1720}
1721
1722#[derive(Default, Clone)]
1723struct ConfigInner {
1724 values: ConfigValues,
1726 credentials: Option<Credentials>,
1730}
1731
1732impl ConfigInner {
1733 fn apply_overrides(&mut self, overrides: &OverridableConfig) -> anyhow::Result<()> {
1734 if let Some(log_level) = &overrides.log_level {
1735 self.values.logging.level = log_level.parse()?;
1736 }
1737
1738 if let Some(log_format) = &overrides.log_format {
1739 self.values.logging.format = log_format.parse()?;
1740 }
1741
1742 let relay = &mut self.values.relay;
1743 if let Some(mode) = &overrides.mode {
1744 relay.mode = mode
1745 .parse::<RelayMode>()
1746 .with_context(|| ConfigError::field("mode"))?;
1747 }
1748 if let Some(deployment) = &overrides.instance {
1749 relay.instance = deployment
1750 .parse::<RelayInstance>()
1751 .with_context(|| ConfigError::field("deployment"))?;
1752 }
1753 if let Some(upstream) = &overrides.upstream {
1754 relay.upstream = upstream
1755 .parse::<UpstreamDescriptor>()
1756 .with_context(|| ConfigError::field("upstream"))?;
1757 } else if let Some(upstream_dsn) = &overrides.upstream_dsn {
1758 relay.upstream = upstream_dsn
1759 .parse::<Dsn>()
1760 .map(|dsn| UpstreamDescriptor::from_dsn(&dsn))
1761 .with_context(|| ConfigError::field("upstream_dsn"))?;
1762 }
1763 if let Some(host) = &overrides.host {
1764 relay.host = host
1765 .parse::<IpAddr>()
1766 .with_context(|| ConfigError::field("host"))?;
1767 }
1768 if let Some(port) = &overrides.port {
1769 relay.port = port
1770 .as_str()
1771 .parse()
1772 .with_context(|| ConfigError::field("port"))?;
1773 }
1774
1775 let processing = &mut self.values.processing;
1776 if let Some(enabled) = &overrides.processing {
1777 match enabled.to_lowercase().as_str() {
1778 "true" | "1" => processing.enabled = true,
1779 "false" | "0" | "" => processing.enabled = false,
1780 _ => return Err(ConfigError::field("processing").into()),
1781 }
1782 }
1783 if let Some(redis) = overrides.redis_url.clone() {
1784 processing.redis = Some(RedisConfigs::Unified(RedisConfig::single(redis)))
1785 }
1786 if let Some(kafka_url) = overrides.kafka_url.clone() {
1787 let existing = processing
1788 .kafka_config
1789 .iter_mut()
1790 .find(|e| e.name == "bootstrap.servers");
1791
1792 if let Some(config_param) = existing {
1793 config_param.value = kafka_url;
1794 } else {
1795 self.values.processing.kafka_config.push(KafkaConfigParam {
1796 name: "bootstrap.servers".to_owned(),
1797 value: kafka_url,
1798 })
1799 }
1800 }
1801
1802 if overrides.outcome_source.is_some() {
1803 self.values.outcomes.source = overrides.outcome_source.clone();
1804 }
1805
1806 if let Some(shutdown_timeout) = &overrides.shutdown_timeout
1807 && let Ok(shutdown_timeout) = shutdown_timeout.parse::<u64>()
1808 {
1809 self.values.limits.shutdown_timeout = shutdown_timeout;
1810 }
1811
1812 if let Some(server_name) = overrides.server_name.clone() {
1813 self.values.sentry.server_name = Some(server_name.into());
1814 }
1815
1816 let id = if let Some(id) = &overrides.id {
1817 let id = Uuid::parse_str(id).with_context(|| ConfigError::field("id"))?;
1818 Some(id)
1819 } else {
1820 None
1821 };
1822 let public_key = if let Some(public_key) = &overrides.public_key {
1823 let public_key = public_key
1824 .parse::<PublicKey>()
1825 .with_context(|| ConfigError::field("public_key"))?;
1826 Some(public_key)
1827 } else {
1828 None
1829 };
1830
1831 let secret_key = if let Some(secret_key) = &overrides.secret_key {
1832 let secret_key = secret_key
1833 .parse::<SecretKey>()
1834 .with_context(|| ConfigError::field("secret_key"))?;
1835 Some(secret_key)
1836 } else {
1837 None
1838 };
1839
1840 if let Some(credentials) = &mut self.credentials {
1841 if let Some(id) = id {
1843 credentials.id = id;
1844 }
1845 if let Some(public_key) = public_key {
1846 credentials.public_key = public_key;
1847 }
1848 if let Some(secret_key) = secret_key {
1849 credentials.secret_key = secret_key
1850 }
1851 } else {
1852 match (id, public_key, secret_key) {
1854 (Some(id), Some(public_key), Some(secret_key)) => {
1855 self.credentials = Some(Credentials {
1856 secret_key,
1857 public_key,
1858 id,
1859 })
1860 }
1861 (None, None, None) => {
1862 }
1865 _ => {
1866 return Err(ConfigError::field("incomplete credentials").into());
1867 }
1868 }
1869 }
1870
1871 Ok(())
1872 }
1873
1874 fn reload_with(&mut self, other: &Self) -> bool {
1878 let mut changed = false;
1879
1880 if self.values.health != other.values.health {
1881 relay_log::debug!("updating health");
1882 self.values.health = other.values.health.clone();
1883 changed = true;
1884 }
1885
1886 changed
1887 }
1888}
1889
1890pub struct Config {
1892 inner_access: Mutex<()>,
1901 inner: ArcSwap<ConfigInner>,
1903 overrides: Vec<OverridableConfig>,
1908 path: PathBuf,
1912}
1913
1914impl fmt::Debug for Config {
1915 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1916 let inner = self.inner.load();
1917
1918 f.debug_struct("Config")
1919 .field("path", &self.path)
1920 .field("values", &inner.values)
1922 .finish()
1923 }
1924}
1925
1926impl Config {
1927 pub fn from_path<P: AsRef<Path>>(path: P) -> anyhow::Result<Config> {
1929 let path = env::current_dir()
1930 .map(|x| x.join(path.as_ref()))
1931 .unwrap_or_else(|_| path.as_ref().to_path_buf());
1932
1933 let inner = ConfigInner {
1934 values: ConfigValues::load(&path)?,
1935 credentials: match Credentials::path(&path).exists() {
1936 true => Some(Credentials::load(&path)?),
1937 false => None,
1938 },
1939 };
1940
1941 let config = Config {
1942 inner_access: Mutex::new(()),
1943 inner: ArcSwap::from_pointee(inner),
1944 overrides: Vec::new(),
1945 path: path.clone(),
1946 };
1947
1948 if cfg!(not(feature = "processing")) && config.current().processing_enabled() {
1949 return Err(ConfigError::file(ConfigErrorKind::ProcessingNotAvailable, &path).into());
1950 }
1951
1952 Ok(config)
1953 }
1954
1955 pub fn from_json_value(value: serde_json::Value) -> anyhow::Result<Config> {
1959 Ok(Config {
1960 inner_access: Mutex::new(()),
1961 inner: ArcSwap::from_pointee(ConfigInner {
1962 values: serde_json::from_value(value)
1963 .with_context(|| ConfigError::new(ConfigErrorKind::BadJson))?,
1964 credentials: None,
1965 }),
1966 overrides: Vec::new(),
1967 path: PathBuf::new(),
1968 })
1969 }
1970
1971 pub fn apply_override(&mut self, overrides: OverridableConfig) -> anyhow::Result<&mut Self> {
1976 crate::utils::try_rcu(&self.inner, |inner| {
1979 let mut new = ConfigInner::clone(inner);
1980 new.apply_overrides(&overrides)?;
1981 Ok::<_, anyhow::Error>(Arc::new(new))
1982 })?;
1983
1984 self.overrides.push(overrides);
1986
1987 Ok(self)
1988 }
1989
1990 pub fn config_exists<P: AsRef<Path>>(path: P) -> bool {
1992 fs::metadata(ConfigValues::path(path.as_ref())).is_ok()
1993 }
1994
1995 pub fn path(&self) -> &Path {
1997 &self.path
1998 }
1999
2000 pub fn to_yaml_string(&self) -> anyhow::Result<String> {
2002 serde_yaml::to_string(&self.inner.load().values)
2003 .with_context(|| ConfigError::new(ConfigErrorKind::CouldNotWriteFile))
2004 }
2005
2006 pub fn replace_credentials(
2010 &mut self,
2011 credentials: Option<Credentials>,
2012 ) -> anyhow::Result<bool> {
2013 if self.inner.load().credentials == credentials {
2014 return Ok(false);
2015 }
2016
2017 if !self.path.is_empty() {
2018 match &credentials {
2019 Some(creds) => {
2020 creds.save(&self.path)?;
2021 }
2022 None => {
2023 let path = Credentials::path(&self.path);
2024 if fs::metadata(&path).is_ok() {
2025 fs::remove_file(&path).with_context(|| {
2026 ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path)
2027 })?;
2028 }
2029 }
2030 }
2031 }
2032
2033 self.inner.rcu(|inner| {
2038 let mut inner = ConfigInner::clone(inner);
2039 inner.credentials = credentials.clone();
2040 Arc::new(inner)
2041 });
2042
2043 Ok(true)
2044 }
2045
2046 pub fn reload(&self) -> anyhow::Result<bool> {
2056 if self.path.is_empty() {
2057 return Ok(false);
2058 }
2059
2060 let _access = self
2061 .inner_access
2062 .lock()
2063 .unwrap_or_else(PoisonError::into_inner);
2064
2065 let mut new_config = Self::from_path(&self.path)?;
2066 for overrides in &self.overrides {
2067 new_config.apply_override(overrides.clone())?;
2068 }
2069 let new_config = new_config.current();
2070
2071 let mut changed = false;
2072
2073 self.inner.rcu(|inner| {
2075 let mut new_inner = ConfigInner::clone(inner);
2076 changed = new_inner.reload_with(&new_config.inner);
2077 match changed {
2078 true => Arc::new(new_inner),
2079 false => Arc::clone(inner),
2080 }
2081 });
2082
2083 Ok(changed)
2084 }
2085
2086 pub fn current(&self) -> ConfigSnapshot {
2091 let inner = self.inner.load();
2092 ConfigSnapshot { inner }
2093 }
2094}
2095
2096impl Default for Config {
2097 fn default() -> Self {
2098 Self {
2099 inner_access: Mutex::new(()),
2100 inner: ArcSwap::from_pointee(Default::default()),
2101 overrides: Vec::new(),
2102 path: PathBuf::new(),
2103 }
2104 }
2105}
2106
2107pub struct ConfigSnapshot {
2115 inner: arc_swap::Guard<Arc<ConfigInner>>,
2116}
2117
2118impl ConfigSnapshot {
2119 pub fn has_credentials(&self) -> bool {
2121 self.inner.credentials.is_some()
2122 }
2123
2124 pub fn credentials(&self) -> Option<&Credentials> {
2126 self.inner.credentials.as_ref()
2127 }
2128
2129 pub fn secret_key(&self) -> Option<&SecretKey> {
2131 self.inner.credentials.as_ref().map(|x| &x.secret_key)
2132 }
2133
2134 pub fn public_key(&self) -> Option<&PublicKey> {
2136 self.inner.credentials.as_ref().map(|x| &x.public_key)
2137 }
2138
2139 pub fn relay_id(&self) -> Option<&RelayId> {
2141 self.inner.credentials.as_ref().map(|x| &x.id)
2142 }
2143
2144 pub fn relay_mode(&self) -> RelayMode {
2146 self.inner.values.relay.mode
2147 }
2148
2149 pub fn relay_instance(&self) -> RelayInstance {
2151 self.inner.values.relay.instance
2152 }
2153
2154 pub fn upstream(&self) -> &UpstreamDescriptor {
2156 &self.inner.values.relay.upstream
2157 }
2158
2159 pub fn advertised_upstream(&self) -> Option<&UpstreamDescriptor> {
2161 self.inner.values.relay.advertised_upstream.as_ref()
2162 }
2163
2164 pub fn http_host_header(&self) -> Option<&str> {
2166 self.inner.values.http.host_header.as_deref()
2167 }
2168
2169 pub fn listen_addr(&self) -> SocketAddr {
2171 (self.inner.values.relay.host, self.inner.values.relay.port).into()
2172 }
2173
2174 pub fn listen_addr_internal(&self) -> Option<SocketAddr> {
2182 match (
2183 self.inner.values.relay.internal_host,
2184 self.inner.values.relay.internal_port,
2185 ) {
2186 (Some(host), None) => Some((host, self.inner.values.relay.port).into()),
2187 (None, Some(port)) => Some((self.inner.values.relay.host, port).into()),
2188 (Some(host), Some(port)) => Some((host, port).into()),
2189 (None, None) => None,
2190 }
2191 }
2192
2193 pub fn tls_listen_addr(&self) -> Option<SocketAddr> {
2195 if self.inner.values.relay.tls_identity_path.is_some() {
2196 let port = self.inner.values.relay.tls_port.unwrap_or(3443);
2197 Some((self.inner.values.relay.host, port).into())
2198 } else {
2199 None
2200 }
2201 }
2202
2203 pub fn tls_identity_path(&self) -> Option<&Path> {
2205 self.inner.values.relay.tls_identity_path.as_deref()
2206 }
2207
2208 pub fn tls_identity_password(&self) -> Option<&str> {
2210 self.inner.values.relay.tls_identity_password.as_deref()
2211 }
2212
2213 pub fn override_project_ids(&self) -> bool {
2217 self.inner.values.relay.override_project_ids
2218 }
2219
2220 pub fn config_reload_interval(&self) -> Option<Duration> {
2224 let interval = self.inner.values.relay.config_reload_interval?;
2225 Some(match interval {
2226 0 => Duration::from_millis(50),
2229 secs => Duration::from_secs(secs),
2230 })
2231 }
2232
2233 pub fn requires_auth(&self) -> bool {
2237 match self.inner.values.auth.ready {
2238 ReadinessCondition::Authenticated => self.relay_mode() == RelayMode::Managed,
2239 ReadinessCondition::Always => false,
2240 }
2241 }
2242
2243 pub fn http_auth_interval(&self) -> Option<Duration> {
2247 if self.processing_enabled() {
2248 return None;
2249 }
2250
2251 match self.inner.values.http.auth_interval {
2252 None | Some(0) => None,
2253 Some(secs) => Some(Duration::from_secs(secs)),
2254 }
2255 }
2256
2257 pub fn http_outage_grace_period(&self) -> Duration {
2260 Duration::from_secs(self.inner.values.http.outage_grace_period)
2261 }
2262
2263 pub fn http_retry_delay(&self) -> Duration {
2268 Duration::from_secs(self.inner.values.http.retry_delay)
2269 }
2270
2271 pub fn http_project_failure_interval(&self) -> Duration {
2273 Duration::from_secs(self.inner.values.http.project_failure_interval)
2274 }
2275
2276 pub fn http_encoding(&self) -> HttpEncoding {
2278 self.inner.values.http.encoding
2279 }
2280
2281 pub fn http_global_metrics(&self) -> bool {
2283 self.inner.values.http.global_metrics
2284 }
2285
2286 pub fn http_forward(&self) -> bool {
2291 self.inner.values.http.forward && !self.processing_enabled()
2292 }
2293
2294 pub fn emit_outcomes(&self) -> EmitOutcomes {
2299 if self.processing_enabled() {
2300 return EmitOutcomes::AsOutcomes;
2301 }
2302 self.inner.values.outcomes.emit_outcomes
2303 }
2304
2305 pub fn outcome_batch_size(&self) -> usize {
2307 self.inner.values.outcomes.batch_size
2308 }
2309
2310 pub fn outcome_batch_interval(&self) -> Duration {
2312 Duration::from_millis(self.inner.values.outcomes.batch_interval)
2313 }
2314
2315 pub fn outcome_source(&self) -> Option<&str> {
2317 self.inner.values.outcomes.source.as_deref()
2318 }
2319
2320 pub fn logging(&self) -> &relay_log::LogConfig {
2322 &self.inner.values.logging
2323 }
2324
2325 pub fn sentry(&self) -> &relay_log::SentryConfig {
2327 &self.inner.values.sentry
2328 }
2329
2330 pub fn statsd_addr(&self) -> Option<&str> {
2332 self.inner.values.metrics.statsd.as_deref()
2333 }
2334
2335 pub fn statsd_buffer_size(&self) -> Option<usize> {
2337 self.inner.values.metrics.statsd_buffer_size
2338 }
2339
2340 pub fn metrics_prefix(&self) -> &str {
2342 &self.inner.values.metrics.prefix
2343 }
2344
2345 pub fn metrics_default_tags(&self) -> &BTreeMap<String, String> {
2347 &self.inner.values.metrics.default_tags
2348 }
2349
2350 pub fn metrics_hostname_tag(&self) -> Option<&str> {
2352 self.inner.values.metrics.hostname_tag.as_deref()
2353 }
2354
2355 pub fn metrics_periodic_interval(&self) -> Option<Duration> {
2359 match self.inner.values.metrics.periodic_secs {
2360 0 => None,
2361 secs => Some(Duration::from_secs(secs)),
2362 }
2363 }
2364
2365 pub fn http_timeout(&self) -> Duration {
2367 Duration::from_secs(self.inner.values.http.timeout.into())
2368 }
2369
2370 pub fn http_connection_timeout(&self) -> Duration {
2372 Duration::from_secs(self.inner.values.http.connection_timeout.into())
2373 }
2374
2375 pub fn http_max_retry_interval(&self) -> Duration {
2377 Duration::from_secs(self.inner.values.http.max_retry_interval.into())
2378 }
2379
2380 pub fn http_dns_cache(&self) -> bool {
2382 self.inner.values.http.dns_cache
2383 }
2384
2385 pub fn project_cache_expiry(&self) -> Duration {
2387 Duration::from_secs(self.inner.values.cache.project_expiry.into())
2388 }
2389
2390 pub fn request_full_project_config(&self) -> bool {
2392 self.inner.values.cache.project_request_full_config
2393 }
2394
2395 pub fn relay_cache_expiry(&self) -> Duration {
2397 Duration::from_secs(self.inner.values.cache.relay_expiry.into())
2398 }
2399
2400 pub fn cache_miss_expiry(&self) -> Duration {
2402 Duration::from_secs(self.inner.values.cache.miss_expiry.into())
2403 }
2404
2405 pub fn project_grace_period(&self) -> Duration {
2407 Duration::from_secs(self.inner.values.cache.project_grace_period.into())
2408 }
2409
2410 pub fn project_refresh_interval(&self) -> Option<Duration> {
2414 self.inner
2415 .values
2416 .cache
2417 .project_refresh_interval
2418 .map(Into::into)
2419 .map(Duration::from_secs)
2420 }
2421
2422 pub fn query_batch_interval(&self) -> Duration {
2425 Duration::from_millis(self.inner.values.cache.batch_interval.into())
2426 }
2427
2428 pub fn downstream_relays_batch_interval(&self) -> Duration {
2430 Duration::from_millis(
2431 self.inner
2432 .values
2433 .cache
2434 .downstream_relays_batch_interval
2435 .into(),
2436 )
2437 }
2438
2439 pub fn local_cache_interval(&self) -> Duration {
2441 Duration::from_secs(self.inner.values.cache.file_interval.into())
2442 }
2443
2444 pub fn global_config_fetch_interval(&self) -> Duration {
2447 Duration::from_secs(self.inner.values.cache.global_config_fetch_interval.into())
2448 }
2449
2450 pub fn spool_envelopes_path(&self, partition_id: u8) -> Option<PathBuf> {
2455 let mut path = self
2456 .inner
2457 .values
2458 .spool
2459 .envelopes
2460 .path
2461 .as_ref()
2462 .map(|path| path.to_owned())?;
2463
2464 if partition_id == 0 {
2465 return Some(path);
2466 }
2467
2468 let file_name = path.file_name().and_then(|f| f.to_str())?;
2469 let new_file_name = format!("{file_name}.{partition_id}");
2470 path.set_file_name(new_file_name);
2471
2472 Some(path)
2473 }
2474
2475 pub fn spool_envelopes_max_disk_size(&self) -> usize {
2477 self.inner.values.spool.envelopes.max_disk_size.as_bytes()
2478 }
2479
2480 pub fn spool_envelopes_batch_size_bytes(&self) -> usize {
2483 self.inner
2484 .values
2485 .spool
2486 .envelopes
2487 .batch_size_bytes
2488 .as_bytes()
2489 }
2490
2491 pub fn spool_envelopes_max_age(&self) -> Duration {
2493 Duration::from_secs(self.inner.values.spool.envelopes.max_envelope_delay_secs)
2494 }
2495
2496 pub fn spool_disk_usage_refresh_frequency_ms(&self) -> Duration {
2498 Duration::from_millis(
2499 self.inner
2500 .values
2501 .spool
2502 .envelopes
2503 .disk_usage_refresh_frequency_ms,
2504 )
2505 }
2506
2507 pub fn spool_max_backpressure_memory_percent(&self) -> f32 {
2509 self.inner
2510 .values
2511 .spool
2512 .envelopes
2513 .max_backpressure_memory_percent
2514 }
2515
2516 pub fn spool_partitions(&self) -> NonZeroU8 {
2518 self.inner.values.spool.envelopes.partitions
2519 }
2520
2521 pub fn spool_partitioning(&self) -> EnvelopeSpoolPartitioning {
2523 self.inner.values.spool.envelopes.partitioning
2524 }
2525
2526 pub fn spool_ephemeral(&self) -> bool {
2528 self.inner.values.spool.envelopes.ephemeral
2529 }
2530
2531 pub fn max_event_size(&self) -> usize {
2533 self.inner.values.limits.max_event_size.as_bytes()
2534 }
2535
2536 pub fn max_attachment_size(&self) -> usize {
2538 self.inner.values.limits.max_attachment_size.as_bytes()
2539 }
2540
2541 pub fn max_attachment_count(&self) -> usize {
2543 self.inner.values.limits.max_attachment_count
2544 }
2545
2546 pub fn max_attachments_size(&self) -> usize {
2549 self.inner.values.limits.max_attachments_size.as_bytes()
2550 }
2551
2552 pub fn max_upload_size(&self) -> usize {
2554 self.inner.values.limits.max_upload_size.as_bytes()
2555 }
2556
2557 pub fn max_client_reports_count(&self) -> usize {
2559 self.inner.values.limits.max_client_reports_count
2560 }
2561
2562 pub fn max_client_reports_size(&self) -> usize {
2564 self.inner.values.limits.max_client_reports_size.as_bytes()
2565 }
2566
2567 pub fn max_check_in_size(&self) -> usize {
2569 self.inner.values.limits.max_check_in_size.as_bytes()
2570 }
2571
2572 pub fn max_log_size(&self) -> usize {
2574 self.inner.values.limits.max_log_size.as_bytes()
2575 }
2576
2577 pub fn max_span_size(&self) -> usize {
2579 self.inner.values.limits.max_span_size.as_bytes()
2580 }
2581
2582 pub fn max_standalone_span_count(&self) -> usize {
2584 self.inner.values.limits.max_standalone_span_count
2585 }
2586
2587 pub fn max_container_size(&self) -> usize {
2589 self.inner.values.limits.max_container_size.as_bytes()
2590 }
2591
2592 pub fn max_envelope_size(&self) -> usize {
2596 self.inner.values.limits.max_envelope_size.as_bytes()
2597 }
2598
2599 pub fn max_session_count(&self) -> usize {
2601 self.inner.values.limits.max_session_count
2602 }
2603
2604 pub fn max_sessions_size(&self) -> usize {
2606 self.inner.values.limits.max_sessions_size.as_bytes()
2607 }
2608
2609 pub fn max_statsd_size(&self) -> usize {
2611 self.inner.values.limits.max_statsd_size.as_bytes()
2612 }
2613
2614 pub fn max_metric_buckets_size(&self) -> usize {
2616 self.inner.values.limits.max_metric_buckets_size.as_bytes()
2617 }
2618
2619 pub fn max_api_payload_size(&self) -> usize {
2621 self.inner.values.limits.max_api_payload_size.as_bytes()
2622 }
2623
2624 pub fn max_api_file_upload_size(&self) -> usize {
2626 self.inner.values.limits.max_api_file_upload_size.as_bytes()
2627 }
2628
2629 pub fn max_api_chunk_upload_size(&self) -> usize {
2631 self.inner
2632 .values
2633 .limits
2634 .max_api_chunk_upload_size
2635 .as_bytes()
2636 }
2637
2638 pub fn max_profile_size(&self) -> usize {
2640 self.inner.values.limits.max_profile_size.as_bytes()
2641 }
2642
2643 pub fn max_trace_metric_size(&self) -> usize {
2645 self.inner.values.limits.max_trace_metric_size.as_bytes()
2646 }
2647
2648 pub fn max_replay_compressed_size(&self) -> usize {
2650 self.inner
2651 .values
2652 .limits
2653 .max_replay_compressed_size
2654 .as_bytes()
2655 }
2656
2657 pub fn max_replay_uncompressed_size(&self) -> usize {
2659 self.inner
2660 .values
2661 .limits
2662 .max_replay_uncompressed_size
2663 .as_bytes()
2664 }
2665
2666 pub fn max_replay_message_size(&self) -> usize {
2672 self.inner.values.limits.max_replay_message_size.as_bytes()
2673 }
2674
2675 pub fn max_concurrent_requests(&self) -> usize {
2677 self.inner.values.limits.max_concurrent_requests
2678 }
2679
2680 pub fn max_concurrent_queries(&self) -> usize {
2682 self.inner.values.limits.max_concurrent_queries
2683 }
2684
2685 pub fn max_removed_attribute_key_size(&self) -> usize {
2687 self.inner
2688 .values
2689 .limits
2690 .max_removed_attribute_key_size
2691 .as_bytes()
2692 }
2693
2694 pub fn query_timeout(&self) -> Duration {
2696 Duration::from_secs(self.inner.values.limits.query_timeout)
2697 }
2698
2699 pub fn shutdown_timeout(&self) -> Duration {
2702 Duration::from_secs(self.inner.values.limits.shutdown_timeout)
2703 }
2704
2705 pub fn keepalive_timeout(&self) -> Duration {
2709 Duration::from_secs(self.inner.values.limits.keepalive_timeout)
2710 }
2711
2712 pub fn idle_timeout(&self) -> Option<Duration> {
2714 self.inner
2715 .values
2716 .limits
2717 .idle_timeout
2718 .map(Duration::from_secs)
2719 }
2720
2721 pub fn max_connections(&self) -> Option<usize> {
2723 self.inner.values.limits.max_connections
2724 }
2725
2726 pub fn tcp_listen_backlog(&self) -> u32 {
2728 self.inner.values.limits.tcp_listen_backlog
2729 }
2730
2731 pub fn cpu_concurrency(&self) -> usize {
2733 self.inner.values.limits.max_thread_count
2734 }
2735
2736 pub fn pool_concurrency(&self) -> usize {
2738 self.inner.values.limits.max_pool_concurrency
2739 }
2740
2741 pub fn query_batch_size(&self) -> usize {
2743 self.inner.values.cache.batch_size
2744 }
2745
2746 pub fn processing_enabled(&self) -> bool {
2748 self.inner.values.processing.enabled
2749 }
2750
2751 pub fn normalization_level(&self) -> NormalizationLevel {
2753 self.inner.values.normalization.level
2754 }
2755
2756 pub fn geoip_path(&self) -> Option<&Path> {
2758 self.inner.values.geoip.path.as_deref().or(self
2759 .inner
2760 .values
2761 .processing
2762 .geoip_path
2763 .as_deref())
2764 }
2765
2766 pub fn max_secs_in_future(&self) -> i64 {
2770 self.inner.values.processing.max_secs_in_future.into()
2771 }
2772
2773 pub fn max_session_secs_in_past(&self) -> i64 {
2775 self.inner.values.processing.max_session_secs_in_past.into()
2776 }
2777
2778 pub fn kafka_configs(
2780 &self,
2781 topic: KafkaTopic,
2782 ) -> Result<KafkaTopicConfig<'_>, KafkaConfigError> {
2783 self.inner
2784 .values
2785 .processing
2786 .topics
2787 .get(topic)
2788 .kafka_configs(
2789 &self.inner.values.processing.kafka_config,
2790 &self.inner.values.processing.secondary_kafka_configs,
2791 )
2792 }
2793
2794 pub fn kafka_validate_topics(&self) -> bool {
2796 self.inner.values.processing.kafka_validate_topics
2797 }
2798
2799 pub fn unused_topic_assignments(&self) -> &relay_kafka::Unused {
2801 &self.inner.values.processing.topics.unused
2802 }
2803
2804 pub fn objectstore(&self) -> &ObjectstoreServiceConfig {
2806 &self.inner.values.processing.objectstore
2807 }
2808
2809 pub fn upload(&self) -> &Upload {
2811 &self.inner.values.upload
2812 }
2813
2814 #[cfg(feature = "processing")]
2816 pub fn upload_signing_key(&self) -> Option<&SecretKey> {
2817 self.upload()
2818 .credentials
2819 .as_ref()
2820 .map(|c| &c.signing_key)
2821 .or(self.credentials().map(|c| &c.secret_key))
2822 }
2823
2824 #[cfg(feature = "processing")]
2826 pub fn upload_verification_key(&self) -> Option<&PublicKey> {
2827 self.upload()
2828 .credentials
2829 .as_ref()
2830 .map(|c| &c.verification_key)
2831 .or(self.credentials().map(|c| &c.public_key))
2832 }
2833
2834 pub fn redis(&self) -> Option<RedisConfigsRef<'_>> {
2836 let redis_configs = self.inner.values.processing.redis.as_ref()?;
2837
2838 Some(build_redis_configs(
2839 redis_configs,
2840 self.cpu_concurrency() as u32,
2841 self.pool_concurrency() as u32,
2842 ))
2843 }
2844
2845 pub fn attachment_chunk_size(&self) -> usize {
2847 self.inner
2848 .values
2849 .processing
2850 .attachment_chunk_size
2851 .as_bytes()
2852 }
2853
2854 pub fn metrics_max_batch_size_bytes(&self) -> usize {
2856 self.inner.values.aggregator.max_flush_bytes
2857 }
2858
2859 pub fn projectconfig_cache_prefix(&self) -> &str {
2862 &self.inner.values.processing.projectconfig_cache_prefix
2863 }
2864
2865 pub fn max_rate_limit(&self) -> Option<u64> {
2867 self.inner.values.processing.max_rate_limit.map(u32::into)
2868 }
2869
2870 pub fn quota_cache_ratio(&self) -> Option<f32> {
2872 self.inner.values.processing.quota_cache_ratio
2873 }
2874
2875 pub fn quota_cache_max(&self) -> Option<f32> {
2877 self.inner.values.processing.quota_cache_max
2878 }
2879
2880 pub fn health_refresh_interval(&self) -> Duration {
2882 Duration::from_millis(self.inner.values.health.refresh_interval_ms)
2883 }
2884
2885 pub fn health_max_memory_watermark_bytes(&self) -> u64 {
2887 self.inner
2888 .values
2889 .health
2890 .max_memory_bytes
2891 .as_ref()
2892 .map_or(u64::MAX, |b| b.as_bytes() as u64)
2893 }
2894
2895 pub fn health_max_memory_watermark_percent(&self) -> f32 {
2897 self.inner.values.health.max_memory_percent
2898 }
2899
2900 pub fn health_probe_timeout(&self) -> Duration {
2902 Duration::from_millis(self.inner.values.health.probe_timeout_ms)
2903 }
2904
2905 pub fn memory_stat_refresh_frequency_ms(&self) -> u64 {
2907 self.inner.values.health.memory_stat_refresh_frequency_ms
2908 }
2909
2910 pub fn cogs_max_queue_size(&self) -> u64 {
2912 self.inner.values.cogs.max_queue_size
2913 }
2914
2915 pub fn cogs_relay_resource_id(&self) -> &str {
2917 &self.inner.values.cogs.relay_resource_id
2918 }
2919
2920 pub fn default_aggregator_config(&self) -> &AggregatorServiceConfig {
2922 &self.inner.values.aggregator
2923 }
2924
2925 pub fn secondary_aggregator_configs(&self) -> &Vec<ScopedAggregatorConfig> {
2927 &self.inner.values.secondary_aggregators
2928 }
2929
2930 pub fn aggregator_config_for(&self, namespace: MetricNamespace) -> &AggregatorServiceConfig {
2932 for entry in &self.inner.values.secondary_aggregators {
2933 if entry.condition.matches(Some(namespace)) {
2934 return &entry.config;
2935 }
2936 }
2937 &self.inner.values.aggregator
2938 }
2939
2940 pub fn static_relays(&self) -> &HashMap<RelayId, RelayInfo> {
2942 &self.inner.values.auth.static_relays
2943 }
2944
2945 pub fn signature_max_age(&self) -> Duration {
2947 Duration::from_secs(self.inner.values.auth.signature_max_age)
2948 }
2949
2950 pub fn accept_unknown_items(&self) -> bool {
2952 let forward = self.inner.values.routing.accept_unknown_items;
2953 forward.unwrap_or_else(|| !self.processing_enabled())
2954 }
2955}
2956
2957impl fmt::Debug for ConfigSnapshot {
2958 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2959 f.debug_struct("ConfigSnapshot")
2960 .field("values", &self.inner.values)
2961 .finish()
2962 }
2963}
2964
2965#[cfg(test)]
2966mod tests {
2967 use super::*;
2968
2969 #[cfg(feature = "processing")]
2970 #[test]
2971 fn test_upload_secret_key_from_file() {
2972 let path = env::temp_dir().join(Uuid::new_v4().to_string());
2973 fs::create_dir(&path).unwrap();
2974 fs::write(
2975 path.join("my_secret.txt"),
2976 "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks",
2977 )
2978 .unwrap();
2979 fs::write(
2980 ConfigValues::path(&path),
2981 r#"
2982 upload:
2983 credentials:
2984 signing_key: ${file:my_secret.txt}
2985 verification_key: "VNS8haF0VTnuMMDR2t-f7AgnmUcXmcdzV3SVksSk34s""#,
2986 )
2987 .unwrap();
2988
2989 let config = Config::from_path(&path).unwrap().current();
2990
2991 fs::remove_dir_all(path).unwrap();
2992
2993 let signing_key = &config.upload().credentials.as_ref().unwrap().signing_key;
2994 assert_eq!(
2995 signing_key.to_string(),
2996 "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks"
2997 );
2998 }
2999
3000 #[test]
3001 fn test_emit_outcomes() {
3002 for (serialized, deserialized) in &[
3003 ("true", EmitOutcomes::AsOutcomes),
3004 ("false", EmitOutcomes::None),
3005 ("\"as_client_reports\"", EmitOutcomes::AsClientReports),
3006 ] {
3007 let value: EmitOutcomes = serde_json::from_str(serialized).unwrap();
3008 assert_eq!(value, *deserialized);
3009 assert_eq!(serde_json::to_string(&value).unwrap(), *serialized);
3010 }
3011 }
3012
3013 #[test]
3014 fn test_emit_outcomes_invalid() {
3015 assert!(serde_json::from_str::<EmitOutcomes>("asdf").is_err());
3016 }
3017}