1use std::collections::HashMap;
2use std::hash::{DefaultHasher, Hash, Hasher};
3
4use chrono::Local;
5use colored::{Color, Colorize};
6use dora_core::build::LogLevelOrStdout;
7use dora_message::common::LogMessage;
8
9#[derive(Debug, Clone, Copy, Default, clap::ValueEnum)]
10pub enum LogFormat {
11 #[default]
12 Pretty,
13 Json,
14 Compact,
15}
16
17#[derive(Debug, Clone)]
18pub struct LogOutputConfig {
19 pub min_level: LogLevelOrStdout,
20 pub format: LogFormat,
21 pub node_filters: HashMap<String, LogLevelOrStdout>,
22 pub print_dataflow_id: bool,
23 pub print_daemon_name: bool,
24}
25
26impl Default for LogOutputConfig {
27 fn default() -> Self {
28 Self {
29 min_level: LogLevelOrStdout::Stdout,
30 format: LogFormat::Pretty,
31 node_filters: HashMap::new(),
32 print_dataflow_id: false,
33 print_daemon_name: false,
34 }
35 }
36}
37
38fn should_display(
39 msg_level: &LogLevelOrStdout,
40 msg_node: Option<&str>,
41 config: &LogOutputConfig,
42) -> bool {
43 let effective_level = msg_node
44 .and_then(|n| config.node_filters.get(n))
45 .unwrap_or(&config.min_level);
46 msg_level.passes(effective_level)
47}
48
49pub(crate) fn message_passes_level_filter(msg: &LogMessage, config: &LogOutputConfig) -> bool {
53 let node_id_str = msg.node_id.as_ref().map(|n| n.to_string());
54 should_display(&msg.level, node_id_str.as_deref(), config)
55}
56
57pub(crate) fn level_filter_is_active(config: &LogOutputConfig) -> bool {
62 !matches!(config.min_level, LogLevelOrStdout::Stdout) || !config.node_filters.is_empty()
63}
64
65pub fn print_log_message(log_message: LogMessage, config: &LogOutputConfig) {
66 let node_id_str = log_message.node_id.as_ref().map(|n| n.to_string());
67 if !should_display(&log_message.level, node_id_str.as_deref(), config) {
68 return;
69 }
70
71 match config.format {
72 LogFormat::Pretty => print_pretty(log_message, config),
73 LogFormat::Json => print_json(&log_message),
74 LogFormat::Compact => print_compact(&log_message),
75 }
76}
77
78fn print_pretty(log_message: LogMessage, config: &LogOutputConfig) {
79 let is_system = log_message.node_id.is_none();
80 let is_lifecycle = is_system && is_lifecycle_message(&log_message.message);
81 let line = format_pretty_line(&log_message, config);
82
83 if is_lifecycle {
84 println!();
85 }
86 println!("{line}");
87 if is_lifecycle {
88 println!();
89 }
90}
91
92fn format_pretty_line(log_message: &LogMessage, config: &LogOutputConfig) -> String {
96 let LogMessage {
97 build_id: _,
98 dataflow_id,
99 node_id,
100 daemon_id,
101 level,
102 target,
103 module_path: _,
104 file: _,
105 line: _,
106 message,
107 timestamp,
108 fields: _,
109 } = log_message;
110
111 let level_str = match level {
112 LogLevelOrStdout::LogLevel(level) => match level {
113 log::Level::Error => "ERROR ".red(),
114 log::Level::Warn => "WARN ".yellow(),
115 log::Level::Info => "INFO ".green(),
116 log::Level::Debug => "DEBUG ".bright_blue(),
117 log::Level::Trace => "TRACE ".dimmed(),
118 },
119 LogLevelOrStdout::Stdout => "stdout".bright_blue().italic().dimmed(),
120 };
121
122 let dataflow = match dataflow_id {
123 Some(dataflow_id) if config.print_dataflow_id => {
124 format!("dataflow `{dataflow_id}` ").cyan()
125 }
126 _ => String::new().cyan(),
127 };
128 let daemon = if config.print_daemon_name {
129 match daemon_id.as_ref().and_then(|id| id.machine_id()) {
133 Some(machine_id) => format!("on daemon `{machine_id}`"),
134 None => "on default daemon".to_string(),
135 }
136 } else {
137 String::new()
138 }
139 .bright_black();
140 let time = format!("{}", timestamp.with_timezone(&Local).format("%H:%M:%S"));
141 let colon = ":".bright_black().bold();
142 let node = match node_id {
143 Some(node_id) => {
144 let colored_id = node_id
145 .to_string()
146 .bold()
147 .color(word_to_color(node_id.as_ref()));
148 let padding = if daemon.is_empty() { "" } else { " " };
149 format!("{colored_id}{padding}{daemon}{colon} ")
150 }
151 None => {
152 let prefix = "[dora]".dimmed();
153 if daemon.is_empty() {
154 format!("{prefix}{colon} ")
155 } else {
156 format!("{prefix} {daemon}{colon} ")
157 }
158 }
159 };
160 let target = match target {
161 Some(target) => format!("{target} ").dimmed(),
162 None => "".normal(),
163 };
164
165 format!("{time} {level_str} {dataflow}{node}{target}{message}")
171}
172
173fn is_lifecycle_message(message: &str) -> bool {
174 message.contains("spawning")
175 || message.contains("node finished")
176 || message.contains("stopping")
177}
178
179fn print_json(log_message: &LogMessage) {
180 if let Ok(json) = serde_json::to_string(log_message) {
181 println!("{json}");
182 }
183}
184
185fn print_compact(log_message: &LogMessage) {
186 let time = log_message
187 .timestamp
188 .with_timezone(&Local)
189 .format("%H:%M:%S");
190 let level = match &log_message.level {
191 LogLevelOrStdout::LogLevel(l) => match l {
192 log::Level::Error => "ERROR",
193 log::Level::Warn => "WARN",
194 log::Level::Info => "INFO",
195 log::Level::Debug => "DEBUG",
196 log::Level::Trace => "TRACE",
197 },
198 LogLevelOrStdout::Stdout => "STDOUT",
199 };
200 let node = log_message
201 .node_id
202 .as_ref()
203 .map(|n| n.to_string())
204 .unwrap_or_else(|| "dora".to_string());
205 println!("{time} {level} {node}: {}", log_message.message);
206}
207
208pub fn parse_jsonl_line(line: &str) -> Option<LogMessage> {
211 if let Ok(msg) = serde_json::from_str::<LogMessage>(line) {
213 return Some(msg);
214 }
215 let v: serde_json::Value = serde_json::from_str(line).ok()?;
217 let ts = v.get("ts")?.as_str()?;
218 let timestamp = chrono::DateTime::parse_from_rfc3339(ts).ok()?.to_utc();
219 let level = v
225 .get("level")
226 .and_then(|l| l.as_str())
227 .and_then(|s| parse_log_level_str(s).ok())
228 .unwrap_or(LogLevelOrStdout::Stdout);
229 let node_id = v
235 .get("node")
236 .and_then(|n| n.as_str())
237 .and_then(|s| s.parse::<dora_message::id::NodeId>().ok());
238 let message = v
239 .get("msg")
240 .and_then(|m| m.as_str())
241 .unwrap_or("")
242 .to_string();
243 let target = v
244 .get("target")
245 .and_then(|t| t.as_str())
246 .map(|s| s.to_string());
247
248 Some(LogMessage {
249 build_id: None,
250 dataflow_id: None,
251 node_id,
252 daemon_id: None,
253 level,
254 target,
255 module_path: None,
256 file: None,
257 line: None,
258 message,
259 timestamp,
260 fields: None,
261 })
262}
263
264pub fn parse_log_filter(s: &str) -> Result<HashMap<String, LogLevelOrStdout>, String> {
288 let mut map = HashMap::new();
289 for pair in s.split(',') {
290 let pair = pair.trim();
291 if pair.is_empty() {
292 continue;
293 }
294 let (node, level) = pair
295 .split_once('=')
296 .ok_or_else(|| format!("invalid filter: '{pair}', expected 'node=level'"))?;
297 let level = parse_log_level_str(level.trim())?;
298 map.insert(node.trim().to_string(), level);
299 }
300 Ok(map)
301}
302
303pub fn parse_log_level_str(s: &str) -> Result<LogLevelOrStdout, String> {
316 match s.to_lowercase().as_str() {
317 "error" => Ok(LogLevelOrStdout::LogLevel(log::Level::Error)),
318 "warn" => Ok(LogLevelOrStdout::LogLevel(log::Level::Warn)),
319 "info" => Ok(LogLevelOrStdout::LogLevel(log::Level::Info)),
320 "debug" => Ok(LogLevelOrStdout::LogLevel(log::Level::Debug)),
321 "trace" => Ok(LogLevelOrStdout::LogLevel(log::Level::Trace)),
322 "stdout" => Ok(LogLevelOrStdout::Stdout),
323 _ => Err(format!(
324 "invalid log level: '{s}', expected one of: error, warn, info, debug, trace, stdout"
325 )),
326 }
327}
328
329pub fn word_to_color(word: &str) -> Color {
332 let word_lower = word.to_lowercase();
333
334 let mut hasher = DefaultHasher::new();
336 word_lower.hash(&mut hasher);
337 let hash = hasher.finish();
338
339 let length_factor = (word_lower.len() as f32 / 5.0).min(1.0);
341
342 let repeat_ratio = calculate_repeat_ratio(&word_lower);
344
345 let diversity = calculate_char_diversity(&word_lower);
347
348 let char_sum = calculate_char_sum(&word_lower);
350
351 let base_r = ((hash >> 16) & 0xFF) as u8;
353 let base_g = ((hash >> 8) & 0xFF) as u8;
354 let base_b = (hash & 0xFF) as u8;
355
356 let r = (base_r as f32 * 0.5 + repeat_ratio * 255.0 * 0.2 + char_sum * 255.0 * 0.3) as u8;
359 let g = (base_g as f32 * 0.5 + diversity * 255.0 * 0.25 + length_factor * 255.0 * 0.25) as u8;
360 let b = (base_b as f32 * 0.5
361 + (1.0 - repeat_ratio) * 255.0 * 0.3
362 + (1.0 - char_sum) * 255.0 * 0.2) as u8;
363
364 Color::TrueColor { r, g, b }
365}
366
367fn calculate_repeat_ratio(s: &str) -> f32 {
369 if s.is_empty() {
370 return 0.0;
371 }
372
373 let mut char_counts = std::collections::HashMap::new();
374 for c in s.chars() {
375 *char_counts.entry(c).or_insert(0) += 1;
376 }
377
378 let repeated = char_counts.values().filter(|&&count| count > 1).count();
379 repeated as f32 / char_counts.len().max(1) as f32
380}
381
382fn calculate_char_diversity(s: &str) -> f32 {
384 if s.is_empty() {
385 return 0.0;
386 }
387
388 let unique: std::collections::HashSet<_> = s.chars().collect();
389 unique.len() as f32 / s.len() as f32
390}
391
392#[cfg(test)]
395pub(crate) fn should_display_test(
396 msg_level: &LogLevelOrStdout,
397 msg_node: Option<&str>,
398 config: &LogOutputConfig,
399) -> bool {
400 should_display(msg_level, msg_node, config)
401}
402
403fn calculate_char_sum(s: &str) -> f32 {
405 if s.is_empty() {
406 return 0.0;
407 }
408
409 let sum: u32 = s
410 .chars()
411 .filter(|c| c.is_ascii_alphabetic())
412 .map(|c| c.to_ascii_lowercase() as u32 - 'a' as u32 + 1)
413 .sum();
414
415 let max_sum = s.len() as u32 * 26;
417 (sum as f32 / max_sum as f32).min(1.0)
418}
419
420#[cfg(test)]
421mod tests {
422 use super::*;
423
424 fn pretty_message(node: Option<&str>, target: Option<&str>, message: &str) -> LogMessage {
427 LogMessage {
428 build_id: None,
429 dataflow_id: None,
430 node_id: node.map(|n| dora_message::id::NodeId::from(n.to_string())),
431 daemon_id: None,
432 level: LogLevelOrStdout::LogLevel(log::Level::Info),
433 target: target.map(|t| t.to_string()),
434 module_path: None,
435 file: None,
436 line: None,
437 message: message.to_string(),
438 timestamp: chrono::Utc::now(),
439 fields: None,
440 }
441 }
442
443 fn rendered_rest(msg: LogMessage, config: &LogOutputConfig) -> String {
447 colored::control::set_override(false);
448 let line = format_pretty_line(&msg, config);
449 line.split_once(' ').unwrap().1.to_string()
450 }
451
452 #[test]
453 fn pretty_line_default_has_no_double_space() {
454 let config = LogOutputConfig::default();
458 let rest = rendered_rest(pretty_message(Some("sensor"), None, "hello"), &config);
459 assert_eq!(rest, "INFO sensor: hello");
460 assert!(
461 !rest.contains("sensor: hello"),
462 "double space before message: {rest:?}"
463 );
464 }
465
466 #[test]
467 fn pretty_line_with_target_single_space() {
468 let config = LogOutputConfig::default();
469 let rest = rendered_rest(
470 pretty_message(Some("sensor"), Some("mymod"), "hello"),
471 &config,
472 );
473 assert_eq!(rest, "INFO sensor: mymod hello");
474 assert!(
475 !rest.contains("mymod hello"),
476 "double space before message: {rest:?}"
477 );
478 }
479
480 #[test]
481 fn pretty_line_system_message_no_double_space() {
482 let config = LogOutputConfig::default();
483 let rest = rendered_rest(pretty_message(None, None, "coordinator ready"), &config);
484 assert_eq!(rest, "INFO [dora]: coordinator ready");
485 }
486
487 #[test]
488 fn pretty_line_default_daemon_no_space_before_colon() {
489 let config = LogOutputConfig {
494 print_daemon_name: true,
495 ..LogOutputConfig::default()
496 };
497 let mut msg = pretty_message(Some("sensor"), None, "hello");
498 msg.daemon_id = Some(dora_message::common::DaemonId::new(None));
499 let rest = rendered_rest(msg, &config);
500 assert_eq!(rest, "INFO sensor on default daemon: hello");
501 assert!(
502 !rest.contains("on default daemon :"),
503 "stray space before the colon: {rest:?}"
504 );
505 }
506
507 #[test]
510 fn parse_level_valid_strings() {
511 assert!(matches!(
512 parse_log_level_str("error"),
513 Ok(LogLevelOrStdout::LogLevel(log::Level::Error))
514 ));
515 assert!(matches!(
516 parse_log_level_str("warn"),
517 Ok(LogLevelOrStdout::LogLevel(log::Level::Warn))
518 ));
519 assert!(matches!(
520 parse_log_level_str("info"),
521 Ok(LogLevelOrStdout::LogLevel(log::Level::Info))
522 ));
523 assert!(matches!(
524 parse_log_level_str("debug"),
525 Ok(LogLevelOrStdout::LogLevel(log::Level::Debug))
526 ));
527 assert!(matches!(
528 parse_log_level_str("trace"),
529 Ok(LogLevelOrStdout::LogLevel(log::Level::Trace))
530 ));
531 assert!(matches!(
532 parse_log_level_str("stdout"),
533 Ok(LogLevelOrStdout::Stdout)
534 ));
535 }
536
537 #[test]
538 fn parse_level_case_insensitive() {
539 assert!(parse_log_level_str("INFO").is_ok());
540 assert!(parse_log_level_str("Info").is_ok());
541 assert!(parse_log_level_str("info").is_ok());
542 }
543
544 #[test]
545 fn parse_level_invalid() {
546 assert!(parse_log_level_str("invalid").is_err());
547 assert!(parse_log_level_str("").is_err());
548 }
549
550 #[test]
553 fn parse_filter_single_pair() {
554 let map = parse_log_filter("sensor=debug").unwrap();
555 assert_eq!(map.len(), 1);
556 assert!(matches!(
557 map.get("sensor"),
558 Some(LogLevelOrStdout::LogLevel(log::Level::Debug))
559 ));
560 }
561
562 #[test]
563 fn parse_filter_multiple_pairs() {
564 let map = parse_log_filter("sensor=debug,planner=warn").unwrap();
565 assert_eq!(map.len(), 2);
566 assert!(matches!(
567 map.get("planner"),
568 Some(LogLevelOrStdout::LogLevel(log::Level::Warn))
569 ));
570 }
571
572 #[test]
573 fn parse_filter_empty_string() {
574 let map = parse_log_filter("").unwrap();
575 assert!(map.is_empty());
576 }
577
578 #[test]
579 fn parse_filter_trailing_comma() {
580 let map = parse_log_filter("sensor=debug,").unwrap();
581 assert_eq!(map.len(), 1);
582 }
583
584 #[test]
585 fn parse_filter_invalid_no_equals() {
586 assert!(parse_log_filter("sensorDEBUG").is_err());
587 }
588
589 #[test]
590 fn parse_filter_whitespace_trimming() {
591 let map = parse_log_filter("sensor = debug , planner = warn").unwrap();
592 assert_eq!(map.len(), 2);
593 assert!(map.contains_key("sensor"));
594 assert!(map.contains_key("planner"));
595 }
596
597 #[test]
600 fn parse_jsonl_daemon_compact() {
601 let line = r#"{"ts":"2025-01-01T00:00:00Z","level":"info","node":"sensor","msg":"hello"}"#;
602 let msg = parse_jsonl_line(line).unwrap();
603 assert_eq!(msg.message, "hello");
604 assert!(matches!(
605 msg.level,
606 LogLevelOrStdout::LogLevel(log::Level::Info)
607 ));
608 assert_eq!(msg.node_id.unwrap().to_string(), "sensor");
609 }
610
611 #[test]
612 fn parse_jsonl_missing_level_defaults_to_stdout() {
613 let line = r#"{"ts":"2025-01-01T00:00:00Z","node":"sensor","msg":"hello"}"#;
616 let msg = parse_jsonl_line(line).expect("line without level should still parse");
617 assert_eq!(msg.message, "hello");
618 assert!(matches!(msg.level, LogLevelOrStdout::Stdout));
619 assert_eq!(msg.node_id.unwrap().to_string(), "sensor");
620 }
621
622 #[test]
623 fn parse_jsonl_non_string_level_defaults_to_stdout() {
624 let line = r#"{"ts":"2025-01-01T00:00:00Z","level":42,"msg":"hi"}"#;
625 let msg = parse_jsonl_line(line).expect("line with non-string level should still parse");
626 assert!(matches!(msg.level, LogLevelOrStdout::Stdout));
627 }
628
629 #[test]
630 fn parse_jsonl_capitalized_level_is_classified() {
631 let line = r#"{"ts":"2025-01-01T00:00:00Z","level":"INFO","msg":"hi"}"#;
634 let msg = parse_jsonl_line(line).unwrap();
635 assert!(matches!(
636 msg.level,
637 LogLevelOrStdout::LogLevel(log::Level::Info)
638 ));
639 }
640
641 #[test]
642 fn parse_jsonl_unknown_level_defaults_to_stdout() {
643 let line = r#"{"ts":"2025-01-01T00:00:00Z","level":"bogus","msg":"hi"}"#;
644 let msg = parse_jsonl_line(line).unwrap();
645 assert!(matches!(msg.level, LogLevelOrStdout::Stdout));
646 }
647
648 #[test]
649 fn parse_jsonl_invalid_node_id_does_not_panic() {
650 for node in ["dora", "bad id", "node/out", ".hidden", ""] {
656 let line = format!(
657 r#"{{"ts":"2025-01-01T00:00:00Z","level":"info","node":"{node}","msg":"hello"}}"#
658 );
659 let msg = parse_jsonl_line(&line)
660 .unwrap_or_else(|| panic!("line with node `{node}` should still parse"));
661 assert_eq!(msg.message, "hello");
662 assert!(
663 msg.node_id.is_none(),
664 "invalid node id `{node}` must not yield a NodeId"
665 );
666 }
667 }
668
669 #[test]
670 fn parse_jsonl_invalid_json() {
671 assert!(parse_jsonl_line("not json at all").is_none());
672 }
673
674 #[test]
675 fn parse_jsonl_empty_string() {
676 assert!(parse_jsonl_line("").is_none());
677 }
678
679 #[test]
682 fn should_display_passes_global_min_level() {
683 let config = LogOutputConfig {
684 min_level: LogLevelOrStdout::LogLevel(log::Level::Info),
685 ..Default::default()
686 };
687 assert!(should_display_test(
689 &LogLevelOrStdout::LogLevel(log::Level::Error),
690 None,
691 &config,
692 ));
693 }
694
695 #[test]
696 fn should_display_blocked_by_global_min_level() {
697 let config = LogOutputConfig {
698 min_level: LogLevelOrStdout::LogLevel(log::Level::Info),
699 ..Default::default()
700 };
701 assert!(!should_display_test(
703 &LogLevelOrStdout::LogLevel(log::Level::Debug),
704 None,
705 &config,
706 ));
707 }
708
709 #[test]
710 fn should_display_per_node_override() {
711 let mut node_filters = HashMap::new();
712 node_filters.insert(
713 "sensor".to_string(),
714 LogLevelOrStdout::LogLevel(log::Level::Debug),
715 );
716 let config = LogOutputConfig {
717 min_level: LogLevelOrStdout::LogLevel(log::Level::Error),
718 node_filters,
719 ..Default::default()
720 };
721 assert!(should_display_test(
723 &LogLevelOrStdout::LogLevel(log::Level::Debug),
724 Some("sensor"),
725 &config,
726 ));
727 assert!(!should_display_test(
729 &LogLevelOrStdout::LogLevel(log::Level::Debug),
730 Some("other"),
731 &config,
732 ));
733 }
734
735 #[test]
738 fn word_to_color_returns_true_color() {
739 assert!(matches!(word_to_color("stt"), Color::TrueColor { .. }));
741 assert!(matches!(word_to_color("vlm"), Color::TrueColor { .. }));
742 }
743
744 #[test]
745 fn word_to_color_char_sum_affects_output() {
746 let low = word_to_color("aaa");
752 let high = word_to_color("zzz");
753 assert_ne!(
754 low, high,
755 "char_sum should influence color: 'aaa' and 'zzz' must differ"
756 );
757 }
758}