start_command/
detached_output.rs

1//! Stream Docker output without buffering entire lines; timestamp command output only.
2use chrono::{DateTime, SecondsFormat, Utc};
3use std::fs::{self, OpenOptions};
4use std::io::{Read, Write};
5
6use crate::execution_attempt::activity_path;
7
8pub const INTERNAL_OUTPUT_FLAG: &str = "--internal-capture-detached-output";
9
10pub struct OutputObserver {
11    log: fs::File,
12    activity: String,
13    since: DateTime<Utc>,
14    prefix: Vec<u8>,
15    at_line_start: bool,
16    latest: Option<DateTime<Utc>>,
17}
18
19impl OutputObserver {
20    pub fn new(log_path: &str, number: u64, since: &str) -> std::io::Result<Self> {
21        let since = DateTime::parse_from_rfc3339(since)
22            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))?
23            .with_timezone(&Utc);
24        Ok(Self {
25            log: OpenOptions::new()
26                .create(true)
27                .append(true)
28                .open(log_path)?,
29            activity: activity_path(log_path, number),
30            since,
31            prefix: Vec::with_capacity(80),
32            at_line_start: true,
33            latest: None,
34        })
35    }
36
37    pub fn write(&mut self, chunk: &[u8]) -> std::io::Result<()> {
38        self.log.write_all(chunk)?;
39        let previous = self.latest;
40        for byte in chunk {
41            if *byte == b'\n' {
42                self.at_line_start = true;
43                self.prefix.clear();
44            } else if self.at_line_start {
45                if *byte == b' ' {
46                    if let Ok(text) = std::str::from_utf8(&self.prefix) {
47                        if text.ends_with('Z') {
48                            if let Ok(time) = DateTime::parse_from_rfc3339(text) {
49                                let time = time.with_timezone(&Utc);
50                                if time >= self.since && self.latest.is_none_or(|last| time > last)
51                                {
52                                    self.latest = Some(time);
53                                }
54                            }
55                        }
56                    }
57                    self.at_line_start = false;
58                } else if self.prefix.len() < 80 {
59                    self.prefix.push(*byte);
60                } else {
61                    self.at_line_start = false;
62                }
63            }
64        }
65        if self.latest != previous {
66            if let Some(time) = self.latest {
67                let temporary = format!("{}.{}.tmp", self.activity, std::process::id());
68                fs::write(
69                    &temporary,
70                    time.to_rfc3339_opts(SecondsFormat::AutoSi, true),
71                )?;
72                #[cfg(windows)]
73                let _ = fs::remove_file(&self.activity);
74                fs::rename(temporary, &self.activity)?;
75            }
76        }
77        Ok(())
78    }
79}
80
81pub fn run_internal_output(args: &[String]) {
82    let [log_path, number, since] = args else {
83        return;
84    };
85    let Ok(number) = number.parse() else {
86        return;
87    };
88    let Ok(mut observer) = OutputObserver::new(log_path, number, since) else {
89        return;
90    };
91    let mut input = std::io::stdin().lock();
92    let mut buffer = [0u8; 8192];
93    while let Ok(length) = input.read(&mut buffer) {
94        if length == 0 || observer.write(&buffer[..length]).is_err() {
95            break;
96        }
97    }
98}
start_command/detached_output.rs

start_command/
detached_output.rs
