1use core::fmt;
2use std::{borrow::Cow, collections::BTreeMap};
3
4use aligned_vec::{AVec, ConstAlign};
5use chrono::{DateTime, Utc};
6use eyre::Context as _;
7use serde::Deserialize;
8use uuid::Uuid;
9
10use crate::{BuildId, DataflowId, daemon_to_daemon::InterDaemonEvent, id::NodeId};
11
12pub use log::Level as LogLevel;
13
14#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
15#[must_use]
16pub struct LogMessage {
17 pub build_id: Option<BuildId>,
18 pub dataflow_id: Option<DataflowId>,
19 pub node_id: Option<NodeId>,
20 pub daemon_id: Option<DaemonId>,
21 pub level: LogLevelOrStdout,
22 pub target: Option<String>,
23 pub module_path: Option<String>,
24 pub file: Option<String>,
25 pub line: Option<u32>,
26 pub message: String,
27 pub timestamp: DateTime<Utc>,
28 pub fields: Option<BTreeMap<String, String>>,
29}
30
31#[derive(Deserialize)]
32pub struct LogMessageHelper {
33 build_id: Option<BuildId>,
34 dataflow_id: Option<DataflowId>,
35 node_id: Option<NodeId>,
36 daemon_id: Option<DaemonId>,
37 level: LogLevelOrStdout,
38 target: Option<String>,
39 module_path: Option<String>,
40 file: Option<String>,
41 line: Option<u32>,
42 message: Option<String>,
43 timestamp: DateTime<Utc>,
44 fields: Option<BTreeMap<String, String>>,
45}
46
47impl From<LogMessageHelper> for LogMessage {
48 fn from(helper: LogMessageHelper) -> Self {
49 let fields = helper.fields.as_ref();
50 LogMessage {
51 build_id: helper.build_id.or(fields
52 .and_then(|f| f.get("build_id").cloned())
53 .and_then(|id| BuildId::from_display_str(&id))),
54 dataflow_id: helper.dataflow_id.or(fields
55 .and_then(|f| f.get("dataflow_id").cloned())
56 .and_then(|id| Uuid::parse_str(&id).ok())),
57 node_id: helper.node_id.or(fields
58 .and_then(|f| f.get("node_id").cloned())
59 .and_then(|id| id.parse::<NodeId>().ok())),
60 daemon_id: helper.daemon_id.or(fields
61 .and_then(|f| f.get("daemon_id").cloned())
62 .and_then(|id| DaemonId::from_display_str(&id))),
63 level: helper.level,
64 target: helper
65 .target
66 .or(fields.and_then(|f| f.get("target").cloned())),
67 module_path: helper
68 .module_path
69 .or(fields.and_then(|f| f.get("module_path").cloned())),
70 file: helper.file.or(fields.and_then(|f| f.get("file").cloned())),
71 line: helper.line.or(fields
72 .and_then(|f| f.get("line").cloned())
73 .and_then(|s| s.parse().ok())),
74 message: helper
75 .message
76 .or(fields.and_then(|f| f.get("message").cloned()))
77 .unwrap_or_default(),
78 fields: helper.fields,
79 timestamp: helper.timestamp,
80 }
81 }
82}
83
84#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, PartialOrd, Ord)]
85#[serde(rename_all = "UPPERCASE")]
86pub enum LogLevelOrStdout {
87 #[serde(rename = "stdout")]
88 Stdout,
89 #[serde(untagged)]
90 LogLevel(LogLevel),
91}
92
93impl LogLevelOrStdout {
94 pub fn passes(&self, min: &LogLevelOrStdout) -> bool {
106 match (self, min) {
107 (LogLevelOrStdout::Stdout, LogLevelOrStdout::Stdout) => true,
108 (LogLevelOrStdout::Stdout, _) => false,
109 (LogLevelOrStdout::LogLevel(_), LogLevelOrStdout::Stdout) => true,
110 (LogLevelOrStdout::LogLevel(msg), LogLevelOrStdout::LogLevel(max)) => msg <= max,
111 }
112 }
113}
114
115impl From<LogLevel> for LogLevelOrStdout {
116 fn from(level: LogLevel) -> Self {
117 Self::LogLevel(level)
118 }
119}
120
121#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
122pub struct NodeError {
123 pub timestamp: uhlc::Timestamp,
124 pub cause: NodeErrorCause,
125 pub exit_status: NodeExitStatus,
126}
127
128impl std::fmt::Display for NodeError {
129 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
130 if let NodeErrorCause::FailedToSpawn(err) = &self.cause {
131 return write!(f, "failed to spawn node: {err}");
132 }
133 match &self.exit_status {
134 NodeExitStatus::Success => write!(f, "<success>"),
135 NodeExitStatus::IoError(err) => write!(f, "I/O error while reading exit status: {err}"),
136 NodeExitStatus::ExitCode(code) => write!(f, "exited with code {code}"),
137 NodeExitStatus::Signal(signal) => {
138 let signal_str: Cow<_> = match signal {
139 1 => "SIGHUP".into(),
140 2 => "SIGINT".into(),
141 3 => "SIGQUIT".into(),
142 4 => "SIGILL".into(),
143 6 => "SIGABRT".into(),
144 8 => "SIGFPE".into(),
145 9 => "SIGKILL".into(),
146 11 => "SIGSEGV".into(),
147 13 => "SIGPIPE".into(),
148 14 => "SIGALRM".into(),
149 15 => "SIGTERM".into(),
150 22 => "SIGTTOU".into(),
151 23 => "SIGURG".into(),
152 other => other.to_string().into(),
153 };
154 if matches!(self.cause, NodeErrorCause::GraceDuration) {
155 write!(
156 f,
157 "node was killed by dora because it didn't react to a stop message in time ({signal_str})"
158 )
159 } else {
160 write!(f, "exited because of signal {signal_str}")
161 }
162 }
163 NodeExitStatus::Unknown => write!(f, "unknown exit status"),
164 }?;
165
166 match &self.cause {
167 NodeErrorCause::GraceDuration => {} NodeErrorCause::Cascading { caused_by_node } => write!(
169 f,
170 ". This error occurred because node `{caused_by_node}` exited before connecting to dora."
171 )?,
172 NodeErrorCause::FailedToSpawn(_) => unreachable!(), NodeErrorCause::Other { stderr } if stderr.is_empty() => {}
174 NodeErrorCause::Other { stderr } => {
175 let line: &str = "---------------------------------------------------------------------------------\n";
176 let stderr = stderr.trim_end();
177 write!(f, " with stderr output:\n{line}{stderr}\n{line}")?
178 }
179 }
180
181 Ok(())
182 }
183}
184
185#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
186pub enum NodeErrorCause {
187 GraceDuration,
189 Cascading {
191 caused_by_node: NodeId,
192 },
193 FailedToSpawn(String),
194 Other {
195 stderr: String,
196 },
197}
198
199#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
200pub enum NodeExitStatus {
201 Success,
202 IoError(String),
203 ExitCode(i32),
204 Signal(i32),
205 Unknown,
206}
207
208impl NodeExitStatus {
209 pub fn is_success(&self) -> bool {
210 matches!(self, NodeExitStatus::Success)
211 }
212}
213
214impl From<Result<std::process::ExitStatus, std::io::Error>> for NodeExitStatus {
215 fn from(result: Result<std::process::ExitStatus, std::io::Error>) -> Self {
216 match result {
217 Ok(status) => {
218 if status.success() {
219 NodeExitStatus::Success
220 } else if let Some(code) = status.code() {
221 Self::ExitCode(code)
222 } else {
223 #[cfg(unix)]
224 {
225 use std::os::unix::process::ExitStatusExt;
226 if let Some(signal) = status.signal() {
227 return Self::Signal(signal);
228 }
229 }
230 Self::Unknown
231 }
232 }
233 Err(err) => Self::IoError(err.to_string()),
234 }
235 }
236}
237
238#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
239pub struct Timestamped<T> {
240 pub inner: T,
241 pub timestamp: uhlc::Timestamp,
242}
243
244impl Timestamped<InterDaemonEvent> {
245 pub fn serialize(&self) -> eyre::Result<Vec<u8>> {
247 crate::encode_presized(self, self.inner.encode_size_hint())
248 .wrap_err("failed to serialize timestamped message")
249 }
250
251 pub fn deserialize_inter_daemon_event(bytes: &[u8]) -> eyre::Result<Self> {
252 crate::decode(bytes).wrap_err("failed to deserialize InterDaemonEvent")
253 }
254}
255
256pub type SharedMemoryId = String;
257
258#[derive(serde::Serialize, serde::Deserialize, Clone)]
259pub enum DataMessage {
260 Vec(#[serde(with = "crate::bulk_bytes")] AVec<u8, ConstAlign<128>>),
261}
262
263impl DataMessage {
264 pub fn len(&self) -> usize {
266 match self {
267 Self::Vec(v) => v.len(),
268 }
269 }
270
271 pub fn is_empty(&self) -> bool {
272 self.len() == 0
273 }
274}
275
276impl fmt::Debug for DataMessage {
277 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
278 match self {
279 Self::Vec(_) => f
280 .debug_struct("Vec")
281 .field("len", &self.len())
282 .finish_non_exhaustive(),
283 }
284 }
285}
286
287#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)]
288pub struct DaemonId {
289 machine_id: Option<String>,
290 uuid: Uuid,
291}
292
293impl DaemonId {
294 pub fn new(machine_id: Option<String>) -> Self {
295 DaemonId {
296 machine_id,
297 uuid: Uuid::now_v7(),
298 }
299 }
300
301 pub fn matches_machine_id(&self, machine_id: &str) -> bool {
302 self.machine_id
303 .as_ref()
304 .map(|id| id == machine_id)
305 .unwrap_or_default()
306 }
307
308 pub fn machine_id(&self) -> Option<&str> {
309 self.machine_id.as_deref()
310 }
311
312 pub fn from_display_str(s: &str) -> Option<Self> {
325 if let Ok(uuid) = Uuid::parse_str(s) {
327 return Some(DaemonId {
328 machine_id: None,
329 uuid,
330 });
331 }
332 const UUID_LEN: usize = 36;
335 let split = s.len().checked_sub(UUID_LEN)?;
336 let uuid = Uuid::parse_str(s.get(split..)?).ok()?;
337 let machine_id = s.get(..split)?.strip_suffix('-')?;
338 Some(DaemonId {
339 machine_id: Some(machine_id.to_string()),
340 uuid,
341 })
342 }
343}
344
345impl std::fmt::Display for DaemonId {
346 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
347 if let Some(id) = &self.machine_id {
348 write!(f, "{id}-")?;
349 }
350 write!(f, "{}", self.uuid)
351 }
352}
353
354#[derive(Debug, serde::Deserialize, serde::Serialize, Clone, PartialEq, Eq)]
355pub struct GitSource {
356 pub repo: String,
357 pub commit_hash: String,
358 #[serde(default, skip_serializing_if = "Option::is_none")]
361 pub subdir: Option<String>,
362 #[serde(default, skip_serializing_if = "Option::is_none")]
367 pub hub: Option<HubProvenance>,
368}
369
370#[derive(Debug, serde::Deserialize, serde::Serialize, Clone, PartialEq, Eq)]
372pub struct HubProvenance {
373 pub name: String,
375 pub version: String,
377 #[serde(default, skip_serializing_if = "Option::is_none")]
384 pub manifest_digest: Option<String>,
385}
386
387#[derive(Debug, serde::Deserialize, serde::Serialize, Clone, PartialEq, Eq)]
392pub struct BinaryPin {
393 pub platform: String,
395 pub url: String,
397 pub sha256: String,
399 pub hub: HubProvenance,
401}
402
403#[cfg(test)]
405mod tests {
406 use super::*;
407
408 fn node_error_with_signal(signal: i32) -> NodeError {
409 NodeError {
410 timestamp: uhlc::HLC::default().new_timestamp(),
411 cause: NodeErrorCause::Other {
412 stderr: String::new(),
413 },
414 exit_status: NodeExitStatus::Signal(signal),
415 }
416 }
417
418 #[test]
419 fn node_error_signal_display_uses_linux_signal_names() {
420 for (signal, name) in [(6, "SIGABRT"), (22, "SIGTTOU"), (23, "SIGURG")] {
421 assert_eq!(
422 node_error_with_signal(signal).to_string(),
423 format!("exited because of signal {name}")
424 );
425 }
426 }
427
428 #[test]
429 fn node_error_signal_display_falls_back_to_signal_number() {
430 assert_eq!(
431 node_error_with_signal(40).to_string(),
432 "exited because of signal 40"
433 );
434 }
435
436 #[test]
437 fn test_log_message_serialization() {
438 let log_message = LogMessage {
439 build_id: Some(BuildId(Uuid::new_v4())),
440 dataflow_id: Some(DataflowId::from(Uuid::new_v4())),
441 node_id: Some(NodeId("node-1".to_string())),
442 daemon_id: Some(DaemonId::new(Some("machine-1".to_string()))),
443 level: LogLevelOrStdout::LogLevel(LogLevel::Info),
444 target: Some("target".to_string()),
445 module_path: Some("module::path".to_string()),
446 file: Some("file.rs".to_string()),
447 line: Some(42),
448 message: "This is a log message".to_string(),
449 timestamp: Utc::now(),
450 fields: Some(BTreeMap::from([("key".to_string(), "value".to_string())])),
451 };
452 let serialized = serde_yaml::to_string(&log_message).unwrap();
453 let deserialized: LogMessageHelper = serde_yaml::from_str(&serialized).unwrap();
454 assert_eq!(log_message, LogMessage::from(deserialized));
455 }
456
457 #[test]
458 fn stdout_passes_stdout_filter() {
459 let stdout = LogLevelOrStdout::Stdout;
460 assert!(stdout.passes(&LogLevelOrStdout::Stdout));
461 }
462
463 #[test]
471 fn node_id_recovered_from_fields_is_validated() {
472 let make = |node_id: &str| LogMessageHelper {
473 build_id: None,
474 dataflow_id: None,
475 node_id: None,
476 daemon_id: None,
477 level: LogLevelOrStdout::Stdout,
478 target: None,
479 module_path: None,
480 file: None,
481 line: None,
482 message: Some("m".to_string()),
483 timestamp: Utc::now(),
484 fields: Some(BTreeMap::from([(
485 "node_id".to_string(),
486 node_id.to_string(),
487 )])),
488 };
489
490 assert_eq!(
492 LogMessage::from(make("sensor")).node_id,
493 Some("sensor".parse().unwrap())
494 );
495
496 assert_eq!(LogMessage::from(make("a/b")).node_id, None);
499 assert_eq!(LogMessage::from(make("../evil")).node_id, None);
500 }
501
502 #[test]
510 fn build_id_recovered_from_fields_survives_display_round_trip() {
511 let build_id = BuildId(Uuid::new_v4());
512 let make = |value: &str| LogMessageHelper {
513 build_id: None,
514 dataflow_id: None,
515 node_id: None,
516 daemon_id: None,
517 level: LogLevelOrStdout::Stdout,
518 target: None,
519 module_path: None,
520 file: None,
521 line: None,
522 message: Some("m".to_string()),
523 timestamp: Utc::now(),
524 fields: Some(BTreeMap::from([(
525 "build_id".to_string(),
526 value.to_string(),
527 )])),
528 };
529
530 assert_eq!(
532 LogMessage::from(make(&build_id.to_string())).build_id,
533 Some(build_id)
534 );
535 assert_eq!(
537 LogMessage::from(make(&build_id.uuid().to_string())).build_id,
538 Some(build_id)
539 );
540 assert_eq!(LogMessage::from(make("not-a-build-id")).build_id, None);
542 }
543
544 #[test]
549 fn daemon_id_display_roundtrips_through_parse() {
550 let uuid = Uuid::new_v4();
551 for machine in [None, Some("host"), Some("my-host"), Some("a-b-c-d")] {
552 let id = DaemonId {
553 machine_id: machine.map(str::to_string),
554 uuid,
555 };
556 let parsed = DaemonId::from_display_str(&id.to_string())
557 .unwrap_or_else(|| panic!("failed to parse {id}"));
558 assert_eq!(parsed, id, "round-trip failed for machine_id={machine:?}");
559 }
560 }
561
562 #[test]
565 fn daemon_id_parse_edge_cases() {
566 let uuid = Uuid::new_v4();
567 let bare = DaemonId {
568 machine_id: None,
569 uuid,
570 };
571 assert_eq!(DaemonId::from_display_str(&bare.to_string()), Some(bare));
572 assert_eq!(DaemonId::from_display_str("not-a-daemon-id"), None);
573 assert_eq!(DaemonId::from_display_str(""), None);
574 }
575
576 #[test]
577 fn stdout_fails_non_stdout_filters() {
578 let stdout = LogLevelOrStdout::Stdout;
579 assert!(!stdout.passes(&LogLevelOrStdout::LogLevel(LogLevel::Info)));
580 assert!(!stdout.passes(&LogLevelOrStdout::LogLevel(LogLevel::Warn)));
581 assert!(!stdout.passes(&LogLevelOrStdout::LogLevel(LogLevel::Error)));
582 }
583
584 #[test]
585 fn any_log_level_passes_stdout_filter() {
586 let stdout_filter = LogLevelOrStdout::Stdout;
587 for level in [
588 LogLevel::Error,
589 LogLevel::Warn,
590 LogLevel::Info,
591 LogLevel::Debug,
592 LogLevel::Trace,
593 ] {
594 assert!(
595 LogLevelOrStdout::LogLevel(level).passes(&stdout_filter),
596 "{level:?} should pass stdout filter"
597 );
598 }
599 }
600
601 #[test]
602 fn error_passes_less_verbose_filters() {
603 let error = LogLevelOrStdout::LogLevel(LogLevel::Error);
604 assert!(error.passes(&LogLevelOrStdout::LogLevel(LogLevel::Error)));
605 assert!(error.passes(&LogLevelOrStdout::LogLevel(LogLevel::Warn)));
606 assert!(error.passes(&LogLevelOrStdout::LogLevel(LogLevel::Info)));
607 }
608
609 #[test]
610 fn debug_fails_info_filter() {
611 let debug = LogLevelOrStdout::LogLevel(LogLevel::Debug);
612 assert!(!debug.passes(&LogLevelOrStdout::LogLevel(LogLevel::Info)));
613 }
614
615 #[test]
616 fn same_level_passes_itself() {
617 for level in [
618 LogLevel::Error,
619 LogLevel::Warn,
620 LogLevel::Info,
621 LogLevel::Debug,
622 LogLevel::Trace,
623 ] {
624 let l = LogLevelOrStdout::LogLevel(level);
625 assert!(l.passes(&l), "{level:?} should pass itself");
626 }
627 }
628
629 #[test]
630 fn trace_passes_trace_fails_debug() {
631 let trace = LogLevelOrStdout::LogLevel(LogLevel::Trace);
632 assert!(trace.passes(&LogLevelOrStdout::LogLevel(LogLevel::Trace)));
633 assert!(!trace.passes(&LogLevelOrStdout::LogLevel(LogLevel::Debug)));
634 }
635
636 #[test]
637 fn daemon_id_uses_v7_uuid() {
638 let id = DaemonId::new(None);
639 assert_eq!(id.uuid.get_version_num(), 7);
641 }
642}