1use anyhow::Result;
5use derive_builder::Builder;
6use figment::{
7 Figment,
8 providers::{Env, Format, Serialized, Toml},
9};
10use serde::{Deserialize, Serialize};
11use std::fmt;
12use std::sync::OnceLock;
13use validator::Validate;
14
15pub mod environment_names;
16
17const DEFAULT_SYSTEM_HOST: &str = "0.0.0.0";
19
20const DEFAULT_SYSTEM_PORT: i16 = -1;
22
23const DEFAULT_SYSTEM_HEALTH_PATH: &str = "/health";
25const DEFAULT_SYSTEM_LIVE_PATH: &str = "/live";
26
27pub const DEFAULT_CANARY_WAIT_TIME_SECS: u64 = 10;
30pub const DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS: u64 = 3;
32
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct WorkerConfig {
35 pub graceful_shutdown_timeout: u64,
37}
38
39impl WorkerConfig {
40 pub fn from_settings() -> Self {
43 Figment::new()
45 .merge(Serialized::defaults(Self::default()))
46 .merge(Env::prefixed("DYN_WORKER_"))
47 .extract()
48 .unwrap() }
50}
51
52impl Default for WorkerConfig {
53 fn default() -> Self {
54 WorkerConfig {
55 graceful_shutdown_timeout: if cfg!(debug_assertions) {
56 1 } else {
58 30 },
60 }
61 }
62}
63
64#[derive(Debug, Deserialize, Serialize, PartialEq, Clone)]
65#[serde(rename_all = "lowercase")]
66pub enum HealthStatus {
67 Ready,
68 NotReady,
69}
70
71#[derive(Serialize, Deserialize, Validate, Debug, Builder, Clone)]
74#[builder(build_fn(private, name = "build_internal"), derive(Debug, Serialize))]
75pub struct RuntimeConfig {
76 #[validate(range(min = 1))]
81 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
82 pub num_worker_threads: Option<usize>,
83
84 #[validate(range(min = 1))]
94 #[builder(default = "512")]
95 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
96 pub max_blocking_threads: usize,
97
98 #[builder(default = "DEFAULT_SYSTEM_HOST.to_string()")]
101 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
102 pub system_host: String,
103
104 #[builder(default = "DEFAULT_SYSTEM_PORT")]
110 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
111 pub system_port: i16,
112
113 #[deprecated(
117 note = "Use system_port instead. Set DYN_SYSTEM_PORT to enable the system metrics server."
118 )]
119 #[builder(default = "false")]
120 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
121 pub system_enabled: bool,
122
123 #[builder(default = "HealthStatus::NotReady")]
126 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
127 pub starting_health_status: HealthStatus,
128
129 #[builder(default = "vec![]")]
135 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
136 pub use_endpoint_health_status: Vec<String>,
137
138 #[builder(default = "DEFAULT_SYSTEM_HEALTH_PATH.to_string()")]
141 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
142 pub system_health_path: String,
143 #[builder(default = "DEFAULT_SYSTEM_LIVE_PATH.to_string()")]
145 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
146 pub system_live_path: String,
147
148 #[builder(default = "None")]
152 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
153 pub compute_threads: Option<usize>,
154
155 #[builder(default = "Some(2 * 1024 * 1024)")]
159 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
160 pub compute_stack_size: Option<usize>,
161
162 #[builder(default = "\"compute\".to_string()")]
165 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
166 pub compute_thread_prefix: String,
167
168 #[builder(default = "false")]
171 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
172 pub health_check_enabled: bool,
173
174 #[builder(default = "DEFAULT_CANARY_WAIT_TIME_SECS")]
177 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
178 pub canary_wait_time_secs: u64,
179
180 #[builder(default = "DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS")]
183 #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
184 pub health_check_request_timeout_secs: u64,
185}
186
187impl fmt::Display for RuntimeConfig {
188 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
189 match self.num_worker_threads {
191 Some(val) => write!(f, "num_worker_threads={val}, ")?,
192 None => write!(f, "num_worker_threads=default (num_cores), ")?,
193 }
194
195 write!(f, "max_blocking_threads={}, ", self.max_blocking_threads)?;
196 write!(f, "system_host={}, ", self.system_host)?;
197 write!(f, "system_port={}, ", self.system_port)?;
198 write!(
199 f,
200 "use_endpoint_health_status={:?}",
201 self.use_endpoint_health_status
202 )?;
203 write!(
204 f,
205 "starting_health_status={:?}",
206 self.starting_health_status
207 )?;
208 write!(f, ", system_health_path={}", self.system_health_path)?;
209 write!(f, ", system_live_path={}", self.system_live_path)?;
210 write!(f, ", health_check_enabled={}", self.health_check_enabled)?;
211 write!(f, ", canary_wait_time_secs={}", self.canary_wait_time_secs)?;
212 write!(
213 f,
214 ", health_check_request_timeout_secs={}",
215 self.health_check_request_timeout_secs
216 )?;
217
218 Ok(())
219 }
220}
221
222impl RuntimeConfig {
223 pub fn builder() -> RuntimeConfigBuilder {
224 RuntimeConfigBuilder::default()
225 }
226
227 pub(crate) fn figment() -> Figment {
228 Figment::new()
229 .merge(Serialized::defaults(RuntimeConfig::default()))
230 .merge(Toml::file("/opt/dynamo/defaults/runtime.toml"))
231 .merge(Toml::file("/opt/dynamo/etc/runtime.toml"))
232 .merge(Env::prefixed("DYN_RUNTIME_").filter_map(|k| {
233 let full_key = format!("DYN_RUNTIME_{}", k.as_str());
234 match std::env::var(&full_key) {
236 Ok(v) if !v.is_empty() => Some(k.into()),
237 _ => None,
238 }
239 }))
240 .merge(Env::prefixed("DYN_SYSTEM_").filter_map(|k| {
241 let full_key = format!("DYN_SYSTEM_{}", k.as_str());
242 match std::env::var(&full_key) {
244 Ok(v) if !v.is_empty() => {
245 let mapped_key = match k.as_str() {
247 "HOST" => "system_host",
248 "PORT" => "system_port",
249 "ENABLED" => "system_enabled",
250 "USE_ENDPOINT_HEALTH_STATUS" => "use_endpoint_health_status",
251 "STARTING_HEALTH_STATUS" => "starting_health_status",
252 "HEALTH_PATH" => "system_health_path",
253 "LIVE_PATH" => "system_live_path",
254 _ => k.as_str(),
255 };
256 Some(mapped_key.into())
257 }
258 _ => None,
259 }
260 }))
261 .merge(Env::prefixed("DYN_COMPUTE_").filter_map(|k| {
262 let full_key = format!("DYN_COMPUTE_{}", k.as_str());
263 match std::env::var(&full_key) {
265 Ok(v) if !v.is_empty() => {
266 let mapped_key = match k.as_str() {
268 "THREADS" => "compute_threads",
269 "STACK_SIZE" => "compute_stack_size",
270 "THREAD_PREFIX" => "compute_thread_prefix",
271 _ => k.as_str(),
272 };
273 Some(mapped_key.into())
274 }
275 _ => None,
276 }
277 }))
278 .merge(Env::prefixed("DYN_HEALTH_CHECK_").filter_map(|k| {
279 let full_key = format!("DYN_HEALTH_CHECK_{}", k.as_str());
280 match std::env::var(&full_key) {
282 Ok(v) if !v.is_empty() => {
283 let mapped_key = match k.as_str() {
285 "ENABLED" => "health_check_enabled",
286 "REQUEST_TIMEOUT" => "health_check_request_timeout_secs",
287 _ => k.as_str(),
288 };
289 Some(mapped_key.into())
290 }
291 _ => None,
292 }
293 }))
294 .merge(Env::prefixed("DYN_CANARY_").filter_map(|k| {
295 let full_key = format!("DYN_CANARY_{}", k.as_str());
296 match std::env::var(&full_key) {
298 Ok(v) if !v.is_empty() => {
299 let mapped_key = match k.as_str() {
301 "WAIT_TIME" => "canary_wait_time_secs",
302 _ => k.as_str(),
303 };
304 Some(mapped_key.into())
305 }
306 _ => None,
307 }
308 }))
309 }
310
311 pub fn from_settings() -> Result<RuntimeConfig> {
320 use environment_names::runtime::system as env_system;
321 if std::env::var(env_system::DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS).is_ok() {
323 tracing::warn!(
324 "DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS is deprecated and no longer used. \
325 System health is now determined by endpoints that register with health check payloads. \
326 Please update your configuration to register health check payloads directly on endpoints."
327 );
328 }
329
330 if std::env::var(env_system::DYN_SYSTEM_ENABLED).is_ok() {
331 tracing::warn!(
332 "DYN_SYSTEM_ENABLED is deprecated. \
333 System metrics server is now controlled solely by DYN_SYSTEM_PORT. \
334 Set DYN_SYSTEM_PORT to a positive value to enable the server, or set to -1 to disable (default)."
335 );
336 }
337
338 let config: RuntimeConfig = Self::figment().extract()?;
339 config.validate()?;
340 Ok(config)
341 }
342
343 pub fn system_server_enabled(&self) -> bool {
348 self.system_port >= 0
349 }
350
351 pub fn single_threaded() -> Self {
352 RuntimeConfig {
353 num_worker_threads: Some(1),
354 max_blocking_threads: 1,
355 system_host: DEFAULT_SYSTEM_HOST.to_string(),
356 system_port: DEFAULT_SYSTEM_PORT,
357 #[allow(deprecated)]
358 system_enabled: false,
359 starting_health_status: HealthStatus::NotReady,
360 use_endpoint_health_status: vec![],
361 system_health_path: DEFAULT_SYSTEM_HEALTH_PATH.to_string(),
362 system_live_path: DEFAULT_SYSTEM_LIVE_PATH.to_string(),
363 compute_threads: Some(1),
364 compute_stack_size: Some(2 * 1024 * 1024),
365 compute_thread_prefix: "compute".to_string(),
366 health_check_enabled: false,
367 canary_wait_time_secs: DEFAULT_CANARY_WAIT_TIME_SECS,
368 health_check_request_timeout_secs: DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS,
369 }
370 }
371
372 pub fn tokio_builder(&self) -> tokio::runtime::Builder {
379 let mut builder = tokio::runtime::Builder::new_multi_thread();
380 builder
381 .worker_threads(
382 self.num_worker_threads
383 .unwrap_or_else(|| std::thread::available_parallelism().unwrap().get()),
384 )
385 .max_blocking_threads(self.max_blocking_threads)
386 .enable_all();
387 if env_is_truthy(environment_names::runtime::DYN_ENABLE_POLL_HISTOGRAM) {
388 tracing::info!(
389 "Tokio poll-time histogram enabled (DYN_ENABLE_POLL_HISTOGRAM); \
390 expect ~2× Instant::now() overhead per task poll"
391 );
392 builder.enable_metrics_poll_time_histogram();
393 }
394 builder
395 }
396
397 pub(crate) fn create_runtime(&self) -> std::io::Result<tokio::runtime::Runtime> {
399 self.tokio_builder().build()
400 }
401}
402
403impl Default for RuntimeConfig {
404 fn default() -> Self {
405 let num_cores = std::thread::available_parallelism().unwrap().get();
406 Self {
407 num_worker_threads: Some(num_cores),
408 max_blocking_threads: num_cores,
409 system_host: DEFAULT_SYSTEM_HOST.to_string(),
410 system_port: DEFAULT_SYSTEM_PORT,
411 #[allow(deprecated)]
412 system_enabled: false,
413 starting_health_status: HealthStatus::NotReady,
414 use_endpoint_health_status: vec![],
415 system_health_path: DEFAULT_SYSTEM_HEALTH_PATH.to_string(),
416 system_live_path: DEFAULT_SYSTEM_LIVE_PATH.to_string(),
417 compute_threads: None,
418 compute_stack_size: Some(2 * 1024 * 1024),
419 compute_thread_prefix: "compute".to_string(),
420 health_check_enabled: false,
421 canary_wait_time_secs: DEFAULT_CANARY_WAIT_TIME_SECS,
422 health_check_request_timeout_secs: DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS,
423 }
424 }
425}
426
427impl RuntimeConfigBuilder {
428 pub fn build(&self) -> Result<RuntimeConfig> {
430 let config = self.build_internal()?;
431 config.validate()?;
432 Ok(config)
433 }
434}
435
436pub use dynamo_truthy::{
442 env_is_falsey, env_is_truthy, is_falsey, is_truthy, parse_bool, parse_bool_opt,
443};
444
445#[derive(Debug, Clone, Copy, PartialEq, Eq)]
446pub enum ConsoleLogFormat {
447 Readable,
448 Jsonl,
449}
450
451impl ConsoleLogFormat {
452 fn from_env_value(value: &str) -> Option<Self> {
453 match value.to_ascii_lowercase().as_str() {
454 "readable" => Some(Self::Readable),
455 "jsonl" => Some(Self::Jsonl),
456 _ => None,
457 }
458 }
459
460 pub fn as_str(self) -> &'static str {
461 match self {
462 Self::Readable => "readable",
463 Self::Jsonl => "jsonl",
464 }
465 }
466}
467
468pub(crate) fn legacy_jsonl_logging_enabled() -> bool {
473 env_is_truthy(environment_names::logging::DYN_LOGGING_JSONL)
474}
475
476pub fn console_log_format() -> ConsoleLogFormat {
481 let legacy_format = || {
482 if legacy_jsonl_logging_enabled() {
483 ConsoleLogFormat::Jsonl
484 } else {
485 ConsoleLogFormat::Readable
486 }
487 };
488
489 match std::env::var(environment_names::logging::DYN_LOGGING_CONSOLE_FORMAT) {
490 Ok(value) if value.trim().is_empty() => legacy_format(),
491 Ok(value) => match ConsoleLogFormat::from_env_value(value.trim()) {
492 Some(format) => format,
493 None => {
494 eprintln!(
495 "Invalid {} value '{}'; using readable console logs",
496 environment_names::logging::DYN_LOGGING_CONSOLE_FORMAT,
497 value
498 );
499 ConsoleLogFormat::Readable
500 }
501 },
502 Err(_) => legacy_format(),
503 }
504}
505
506pub fn jsonl_logging_enabled() -> bool {
508 console_log_format() == ConsoleLogFormat::Jsonl
509}
510
511pub fn disable_ansi_logging() -> bool {
514 env_is_truthy(environment_names::logging::DYN_SDK_DISABLE_ANSI_LOGGING)
515}
516
517pub fn use_local_timezone() -> bool {
520 env_is_truthy(environment_names::logging::DYN_LOG_USE_LOCAL_TZ)
521}
522
523pub fn span_events_enabled() -> bool {
525 env_is_truthy(environment_names::logging::DYN_LOGGING_SPAN_EVENTS)
526}
527
528#[cfg(test)]
529mod tests {
530 use super::*;
531
532 #[test]
533 fn test_runtime_config_builder_overrides_related_fields() -> Result<()> {
534 let config = RuntimeConfig::builder()
535 .num_worker_threads(Some(24))
536 .max_blocking_threads(32)
537 .system_host("127.0.0.1".to_string())
538 .system_port(9090)
539 .build()?;
540
541 assert_eq!(config.num_worker_threads, Some(24));
542 assert_eq!(config.max_blocking_threads, 32);
543 assert_eq!(config.system_host, "127.0.0.1");
544 assert_eq!(config.system_port, 9090);
545 Ok(())
546 }
547
548 #[test]
555 fn test_from_settings_reads_both_thread_env_vars() {
556 const WORKERS: &str = "DYN_RUNTIME_NUM_WORKER_THREADS";
557 const BLOCKING: &str = "DYN_RUNTIME_MAX_BLOCKING_THREADS";
558
559 temp_env::with_vars([(WORKERS, Some("7")), (BLOCKING, Some("11"))], || {
560 let config = RuntimeConfig::from_settings().expect("from_settings failed");
561 assert_eq!(config.num_worker_threads, Some(7), "{WORKERS} was not read");
562 assert_eq!(config.max_blocking_threads, 11, "{BLOCKING} was not read");
563 });
564 }
565
566 #[test]
572 fn test_tokio_builder_applies_configured_worker_threads() -> Result<()> {
573 let config = RuntimeConfig::builder()
574 .num_worker_threads(Some(3))
575 .max_blocking_threads(5)
576 .build()?;
577
578 let runtime = config.tokio_builder().build()?;
579 assert_eq!(runtime.metrics().num_workers(), 3);
580 Ok(())
581 }
582
583 #[test]
585 fn test_tokio_builder_defaults_worker_threads_to_core_count() -> Result<()> {
586 let config = RuntimeConfig {
587 num_worker_threads: None,
588 ..RuntimeConfig::default()
589 };
590
591 let runtime = config.tokio_builder().build()?;
592 assert_eq!(
593 runtime.metrics().num_workers(),
594 std::thread::available_parallelism()?.get()
595 );
596 Ok(())
597 }
598
599 #[test]
611 fn test_tokio_builder_applies_max_blocking_threads() -> Result<()> {
612 use std::sync::Arc;
613 use std::sync::atomic::{AtomicUsize, Ordering};
614
615 const CAP: usize = 2;
616
617 let config = RuntimeConfig::builder()
618 .num_worker_threads(Some(2))
619 .max_blocking_threads(CAP)
620 .build()?;
621 let runtime = config.tokio_builder().build()?;
622
623 let in_flight = Arc::new(AtomicUsize::new(0));
624 let peak = Arc::new(AtomicUsize::new(0));
625
626 runtime.block_on(async {
627 let tasks: Vec<_> = (0..CAP * 4)
628 .map(|_| {
629 let in_flight = Arc::clone(&in_flight);
630 let peak = Arc::clone(&peak);
631 tokio::task::spawn_blocking(move || {
632 let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
633 peak.fetch_max(now, Ordering::SeqCst);
634 std::thread::sleep(std::time::Duration::from_millis(20));
636 in_flight.fetch_sub(1, Ordering::SeqCst);
637 })
638 })
639 .collect();
640
641 for task in tasks {
642 task.await.expect("blocking task panicked");
643 }
644 });
645
646 let observed = peak.load(Ordering::SeqCst);
647 assert!(
648 observed <= CAP,
649 "{observed} blocking tasks ran at once, but the cap was {CAP}"
650 );
651 Ok(())
652 }
653
654 #[test]
657 fn test_default_max_blocking_threads_is_core_count() {
658 let cores = std::thread::available_parallelism().unwrap().get();
659 let config = RuntimeConfig::default();
660 assert_eq!(config.max_blocking_threads, cores);
661 assert_eq!(config.num_worker_threads, Some(cores));
662 }
663
664 #[test]
665 fn test_runtime_config_rejects_invalid_thread_count() -> Result<()> {
666 let result = RuntimeConfig::builder()
667 .num_worker_threads(Some(0))
668 .max_blocking_threads(0)
669 .build();
670
671 let error = result.unwrap_err().to_string();
672 assert!(error.contains("num_worker_threads: Validation error"));
673 assert!(error.contains("max_blocking_threads: Validation error"));
674 Ok(())
675 }
676
677 #[test]
678 fn test_system_server_enabled_by_nonnegative_port() {
679 let mut config = RuntimeConfig::default();
680 for (port, enabled) in [(-1, false), (0, true), (9527, true)] {
681 config.system_port = port;
682 assert_eq!(config.system_server_enabled(), enabled);
683 }
684 }
685
686 #[test]
687 fn test_is_truthy_and_falsey() {
688 assert!(is_truthy("1"));
690 assert!(is_truthy("true"));
691 assert!(is_truthy("TRUE"));
692 assert!(is_truthy("on"));
693 assert!(is_truthy("yes"));
694
695 assert!(is_falsey("0"));
697 assert!(is_falsey("false"));
698 assert!(is_falsey("FALSE"));
699 assert!(is_falsey("off"));
700 assert!(is_falsey("no"));
701
702 assert!(!is_truthy("0"));
704 assert!(!is_falsey("1"));
705 }
706
707 #[test]
708 fn test_console_log_format() {
709 use environment_names::logging;
710
711 for (console_format, legacy_jsonl, expected) in [
712 (None, None, ConsoleLogFormat::Readable),
713 (None, Some("true"), ConsoleLogFormat::Jsonl),
714 (Some(""), Some("true"), ConsoleLogFormat::Jsonl),
715 (Some(" "), Some("true"), ConsoleLogFormat::Jsonl),
716 (Some(" jsonl "), Some("false"), ConsoleLogFormat::Jsonl),
717 (Some("readable"), Some("true"), ConsoleLogFormat::Readable),
718 (Some("jsonl"), Some("false"), ConsoleLogFormat::Jsonl),
719 (
720 Some("unsupported"),
721 Some("true"),
722 ConsoleLogFormat::Readable,
723 ),
724 ] {
725 temp_env::with_vars(
726 [
727 (logging::DYN_LOGGING_CONSOLE_FORMAT, console_format),
728 (logging::DYN_LOGGING_JSONL, legacy_jsonl),
729 ],
730 || {
731 assert_eq!(console_log_format(), expected);
732 assert_eq!(jsonl_logging_enabled(), expected == ConsoleLogFormat::Jsonl);
733 },
734 );
735 }
736 }
737}