Skip to content
Sign up

flagon-io/g1t

public
0
455 linesCodeBlameRaw
1//! Running one process for a step: its output streamed to the log as it
2//! comes, with GitHub's workflow commands (`::error::`, `::group::`,
3//! `::add-mask::`…) read out of it.
4
5use std::collections::BTreeMap;
6use std::io::{BufRead, BufReader, Read};
7use std::process::{Command, Stdio};
8use std::sync::mpsc;
9use std::time::{Duration, Instant};
10
11use serde_json::{Map, Value};
12
13use super::report::Log;
14
15/// What a step's workflow commands left behind.
16#[derive(Default)]
17pub(crate) struct Commands {
18 /// `::set-output` (old, still honoured).
19 pub(crate) outputs: BTreeMap<String, String>,
20 /// `::save-state`, for the action's post step.
21 pub(crate) state: BTreeMap<String, String>,
22 /// Set by `::stop-commands::token` until `::token::`.
23 pub(crate) stopped: Option<String>,
24 /// Whether `::debug::` lines are shown (`ACTIONS_STEP_DEBUG`).
25 pub(crate) debug: bool,
26 /// `::echo::on` or `::echo::off`: whether command lines are shown as
27 /// well as acted on. Until a step says, they are with debug logging.
28 pub(crate) echo: Option<bool>,
29}
30
31/// `%25`, `%0D`, `%0A`, and in properties `%3A` and `%2C`, as the
32/// toolkit escapes them.
33fn unescape(text: &str, property: bool) -> String {
34 let mut out = text.replace("%0D", "\r").replace("%0A", "\n");
35 if property {
36 out = out.replace("%3A", ":").replace("%2C", ",");
37 }
38 out.replace("%25", "%")
39}
40
41/// `::name key=value,key=value::message`, if the line is a command.
42pub(crate) fn parse_command(line: &str) -> Option<(String, Map<String, Value>, String)> {
43 let rest = line.trim_start().strip_prefix("::")?;
44 let end = rest.find("::")?;
45 let (head, data) = (&rest[..end], &rest[end + 2..]);
46 let (name, properties) = match head.split_once(' ') {
47 Some((name, properties)) => (name, properties),
48 None => (head, ""),
49 };
50 if name.is_empty() || !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
51 return None;
52 }
53 let mut map = Map::new();
54 for pair in properties.split(',').filter(|p| !p.trim().is_empty()) {
55 if let Some((key, value)) = pair.split_once('=') {
56 map.insert(key.trim().to_owned(), Value::String(unescape(value, true)));
57 }
58 }
59 Some((name.to_owned(), map, unescape(data, false)))
60}
61
62impl Commands {
63 /// Handles one line of output: a command is acted on, and the line to
64 /// show (if any) is returned.
65 pub(crate) fn handle(&mut self, line: &str, log: &mut Log) -> Option<String> {
66 if let Some(token) = &self.stopped {
67 if line.trim() == format!("::{token}::") {
68 self.stopped = None;
69 return None;
70 }
71 return Some(output(line, log));
72 }
73 let Some((name, properties, data)) = parse_command(line) else {
74 return Some(output(line, log));
75 };
76 // Echoed as it reads, before it is acted on; `::add-mask::` never is.
77 let known = matches!(
78 name.as_str(),
79 "add-mask" | "error" | "warning" | "notice" | "group" | "endgroup" | "debug" | "set-output" | "save-state" | "stop-commands" | "echo" | "add-matcher" | "remove-matcher" | "add-path" | "set-env"
80 );
81 if known && name != "add-mask" && self.echo.unwrap_or(self.debug) {
82 log.line(line);
83 }
84 match name.as_str() {
85 "add-mask" => {
86 if !data.trim().is_empty() {
87 log.add_mask(data.trim());
88 }
89 None
90 }
91 "error" | "warning" | "notice" => {
92 log.annotation(&name, &data, &properties);
93 Some(issue_line(&name, &data))
94 }
95 "group" => Some(format!("##[group]{data}")),
96 "endgroup" => Some("##[endgroup]".to_owned()),
97 "debug" => self.debug.then(|| format!("##[debug]{data}")),
98 "set-output" => {
99 if let Some(name) = properties.get("name").and_then(Value::as_str) {
100 self.outputs.insert(name.to_owned(), data);
101 }
102 None
103 }
104 "save-state" => {
105 if let Some(name) = properties.get("name").and_then(Value::as_str) {
106 self.state.insert(name.to_owned(), data);
107 }
108 None
109 }
110 "stop-commands" => {
111 self.stopped = Some(data);
112 None
113 }
114 "echo" => {
115 match data.trim().to_ascii_lowercase().as_str() {
116 "on" => self.echo = Some(true),
117 "off" => self.echo = Some(false),
118 _ => return Some(format!("##[error]Invalid echo command value. Possible values can be: 'on', 'off'. Current value is: '{}'.", data.trim())),
119 }
120 None
121 }
122 "add-matcher" => {
123 let path = matcher_path(data.trim(), &log.matchers.workspace);
124 let read = std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display())).and_then(|text| super::matchers::parse(&text));
125 match read {
126 Ok(matchers) => {
127 log.matchers.add(matchers);
128 None
129 }
130 Err(problem) => Some(format!("##[error]Unable to process command '{line}' successfully.\n##[error]{problem}")),
131 }
132 }
133 "remove-matcher" => {
134 match properties.get("owner").and_then(Value::as_str).filter(|o| !o.trim().is_empty()) {
135 Some(owner) => log.matchers.remove(owner.trim()),
136 // No owner: every matcher in the file named.
137 None => {
138 let path = matcher_path(data.trim(), &log.matchers.workspace);
139 match std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display())).and_then(|text| super::matchers::parse(&text)) {
140 Ok(matchers) => matchers.iter().for_each(|m| log.matchers.remove(&m.owner)),
141 Err(problem) => return Some(format!("##[error]Unable to process command '{line}' successfully.\n##[error]{problem}")),
142 }
143 }
144 }
145 None
146 }
147 "add-path" | "set-env" => Some(format!(
148 "##[error]The `{name}` command is disabled, as on GitHub. Write to the file in $GITHUB_{} instead.",
149 if name == "add-path" { "PATH" } else { "ENV" }
150 )),
151 _ => Some(output(line, log)),
152 }
153 }
154}
155
156/// How an annotation shows in the log.
157fn issue_line(level: &str, message: &str) -> String {
158 let label = match level {
159 "error" => "Error",
160 "warning" => "Warning",
161 _ => "Notice",
162 };
163 format!("##[{level}]{label}: {message}")
164}
165
166/// A line of output, as the log shows it: as it is, or, when a problem
167/// matcher reads an issue in it, the issue, annotated.
168fn output(line: &str, log: &mut Log) -> String {
169 match log.matchers.issue(line) {
170 Some(issue) => {
171 log.annotation(issue.level, &issue.message, &issue.properties);
172 issue_line(issue.level, &issue.message)
173 }
174 None => line.to_owned(),
175 }
176}
177
178/// A matcher file's path: relative to the workspace, and a Docker action's
179/// `/github/workspace` read as the workspace it is.
180fn matcher_path(given: &str, workspace: &std::path::Path) -> std::path::PathBuf {
181 if let Some(rest) = given.strip_prefix("/github/workspace") {
182 return workspace.join(rest.trim_start_matches('/'));
183 }
184 let path = std::path::Path::new(given);
185 if path.is_absolute() { path.to_path_buf() } else { workspace.join(path) }
186}
187
188/// How a process ended.
189pub(crate) enum Ended {
190 Exited(i32),
191 TimedOut,
192 /// The run was cancelled: it was interrupted, then stopped.
193 Cancelled,
194}
195
196/// After a cancellation, how long a step has after SIGINT before SIGTERM,
197/// and after SIGTERM before it is killed, as GitHub's runner waits.
198const INTERRUPT_GRACE: Duration = Duration::from_millis(7500);
199const TERMINATE_GRACE: Duration = Duration::from_millis(2500);
200
201/// Sends `signal` to the process's group (it leads its own), so what the
202/// step started hears it too. Windows has no signals: it is left to `kill`.
203/// Sent with kill(2) itself: a machine without the `kill` program (procps
204/// is not in every image) would otherwise give the step no SIGINT at all,
205/// and leave what it started running after the step was killed.
206#[cfg(unix)]
207fn signal(child: &std::process::Child, signal: &str) {
208 let number = match signal {
209 "INT" => libc::SIGINT,
210 "TERM" => libc::SIGTERM,
211 _ => libc::SIGKILL,
212 };
213 let Ok(group) = libc::pid_t::try_from(child.id()) else {
214 return;
215 };
216 // SAFETY: kill(2) with a negative pid signals that process group; it
217 // reads no memory of ours.
218 unsafe {
219 libc::kill(-group, number);
220 }
221}
222
223#[cfg(not(unix))]
224fn signal(_child: &std::process::Child, _signal: &str) {}
225
226/// Runs the command, sending its output (stdout and stderr together, a
227/// line at a time) through `commands` to the log, until it ends or
228/// `timeout` passes.
229pub(crate) fn run(command: Command, timeout: Duration, log: &mut Log, commands: &mut Commands) -> std::io::Result<Ended> {
230 run_until(command, timeout, log, commands, &super::report::interrupt)
231}
232
233/// `run`, stopping the process gracefully once `interrupt` says so.
234fn run_until(
235 mut command: Command,
236 timeout: Duration,
237 log: &mut Log,
238 commands: &mut Commands,
239 interrupt: &dyn Fn() -> bool,
240) -> std::io::Result<Ended> {
241 command.stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped());
242 // A group of its own, so a cancellation reaches what the step started.
243 #[cfg(unix)]
244 {
245 use std::os::unix::process::CommandExt;
246 command.process_group(0);
247 // A shell cannot trap a signal it was started with ignored, and a
248 // runner started in the background (as a service, or under `&`)
249 // passes SIGINT on ignored. Back to the defaults, so a cancelled
250 // step hears SIGINT and its own trap runs.
251 // SAFETY: only signal(), which is async-signal-safe, between fork and exec.
252 unsafe {
253 command.pre_exec(|| {
254 for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGQUIT] {
255 libc::signal(signal, libc::SIG_DFL);
256 }
257 Ok(())
258 });
259 }
260 }
261 let mut child = command.spawn()?;
262 let (sender, lines) = mpsc::channel::<String>();
263 let mut readers = Vec::new();
264 let pipes: Vec<Box<dyn Read + Send>> = vec![
265 Box::new(child.stdout.take().expect("piped")),
266 Box::new(child.stderr.take().expect("piped")),
267 ];
268 for pipe in pipes {
269 let sender = sender.clone();
270 readers.push(std::thread::spawn(move || {
271 let mut reader = BufReader::new(pipe);
272 let mut buffer = Vec::new();
273 loop {
274 buffer.clear();
275 match reader.read_until(b'\n', &mut buffer) {
276 Ok(0) | Err(_) => break,
277 Ok(_) => {
278 let text = String::from_utf8_lossy(&buffer);
279 let text = text.trim_end_matches(['\n', '\r']);
280 // A progress bar redraws with \r; keep its last state.
281 let text = text.rsplit('\r').next().unwrap_or(text);
282 if sender.send(text.to_owned()).is_err() {
283 break;
284 }
285 }
286 }
287 }
288 }));
289 }
290 drop(sender);
291 let deadline = Instant::now() + timeout;
292 let mut timed_out = false;
293 // When the cancellation reached it, and which signal it has had.
294 let mut interrupted: Option<(Instant, u8)> = None;
295 loop {
296 match lines.recv_timeout(Duration::from_millis(250)) {
297 Ok(line) => {
298 if let Some(shown) = commands.handle(&line, log) {
299 log.line(&shown);
300 }
301 }
302 Err(mpsc::RecvTimeoutError::Timeout) => {
303 // The job's Docker Engine starting, from its own thread.
304 for note in crate::docker::take_notes() {
305 log.line(&note);
306 }
307 log.tick();
308 }
309 Err(mpsc::RecvTimeoutError::Disconnected) => break,
310 }
311 if Instant::now() >= deadline {
312 timed_out = true;
313 signal(&child, "KILL");
314 let _ = child.kill();
315 break;
316 }
317 // Cancelled: SIGINT, then SIGTERM, then killed, as on GitHub.
318 match interrupted {
319 None if interrupt() => {
320 log.line("##[error]The operation was canceled.");
321 signal(&child, "INT");
322 interrupted = Some((Instant::now(), 1));
323 }
324 Some((at, 1)) if at.elapsed() >= INTERRUPT_GRACE => {
325 signal(&child, "TERM");
326 interrupted = Some((Instant::now(), 2));
327 }
328 Some((at, 2)) if at.elapsed() >= TERMINATE_GRACE => {
329 signal(&child, "KILL");
330 let _ = child.kill();
331 break;
332 }
333 _ => {}
334 }
335 if interrupted.is_some() && matches!(child.try_wait(), Ok(Some(_))) {
336 // The step is gone; what it started may still hold its output.
337 signal(&child, "KILL");
338 break;
339 }
340 }
341 let status = child.wait()?;
342 for reader in readers {
343 let _ = reader.join();
344 }
345 // Whatever arrived after the readers finished.
346 while let Ok(line) = lines.try_recv() {
347 if let Some(shown) = commands.handle(&line, log) {
348 log.line(&shown);
349 }
350 }
351 if timed_out {
352 return Ok(Ended::TimedOut);
353 }
354 if interrupted.is_some() {
355 return Ok(Ended::Cancelled);
356 }
357 Ok(Ended::Exited(status.code().unwrap_or(1)))
358}
359
360#[cfg(test)]
361mod tests {
362 use super::parse_command;
363
364 /// A cancelled step hears SIGINT, and its own trap runs.
365 #[cfg(unix)]
366 #[test]
367 fn a_cancelled_step_is_interrupted_and_may_clean_up() {
368 use super::{Commands, Ended, run_until};
369 use crate::actions::report::{Api, Log};
370 use std::process::Command;
371 use std::time::{Duration, Instant};
372
373 // Offline: on a busy machine the step can outlast a batch of the log,
374 // and sending it to an address that is not there held the loop for
375 // seconds, past the grace this test measures.
376 let mut log = Log::new(Api::offline(), Vec::new());
377 let mut command = Command::new("sh");
378 command.args(["-c", "trap 'echo cleaned up; exit 3' INT; echo started; while true; do sleep 0.1; done"]);
379 let began = Instant::now();
380 let ask = move || began.elapsed() >= Duration::from_millis(600);
381 let ended = run_until(command, Duration::from_secs(60), &mut log, &mut Commands::default(), &ask).unwrap();
382 assert!(matches!(ended, Ended::Cancelled));
383 assert!(began.elapsed() < Duration::from_secs(8), "SIGINT ended it, not the kill after the grace period");
384 let text = log.buffered();
385 assert!(text.contains("The operation was canceled."), "{text}");
386 assert!(text.contains("cleaned up"), "{text}");
387 }
388
389 /// `::echo::on` shows command lines until `::echo::off`, which is
390 /// itself shown; `::add-mask::` never is.
391 #[cfg(unix)]
392 #[test]
393 fn echo_shows_command_lines_while_on() {
394 use super::Commands;
395 use crate::actions::report::{Api, Log};
396
397 let mut log = Log::new(Api::offline(), Vec::new());
398 let mut commands = Commands::default();
399 for line in ["::set-output name=a::1", "::echo::on", "::save-state name=b::2", "::add-mask::s3cret", "::echo::off", "::save-state name=c::3"] {
400 if let Some(shown) = commands.handle(line, &mut log) {
401 log.line(&shown);
402 }
403 }
404 let text = log.buffered();
405 assert_eq!(text, "::save-state name=b::2\n::echo::off\n");
406 assert_eq!((commands.outputs.len(), commands.state.len()), (1, 2));
407 // Debug logging echoes until a step says otherwise.
408 let mut log = Log::new(Api::offline(), Vec::new());
409 let mut commands = Commands { debug: true, ..Commands::default() };
410 commands.handle("::set-output name=a::1", &mut log);
411 assert!(log.buffered().contains("::set-output name=a::1"));
412 assert!(commands.handle("::echo::maybe", &mut log).unwrap().contains("Invalid echo command value"));
413 }
414
415 /// A matcher added by one line reads the lines after it, until removed.
416 #[cfg(unix)]
417 #[test]
418 fn matchers_are_added_and_removed_by_command() {
419 use super::Commands;
420 use crate::actions::report::{Api, Log};
421
422 let dir = std::env::temp_dir().join(format!("g1t-add-matcher-{}", std::process::id()));
423 std::fs::create_dir_all(&dir).unwrap();
424 std::fs::write(
425 dir.join("go.json"),
426 r#"{"problemMatcher": [{"owner": "go", "pattern": [{"regexp": "^\\s*(.+\\.go):(?:(\\d+):(\\d+):)? (.*)", "file": 1, "line": 2, "column": 3, "message": 4}]}]}"#,
427 )
428 .unwrap();
429 let mut log = Log::new(Api::offline(), Vec::new());
430 log.matchers.workspace = dir.clone();
431 let mut commands = Commands::default();
432 assert_eq!(commands.handle("main.go:3:1: undefined: x", &mut log).unwrap(), "main.go:3:1: undefined: x");
433 assert!(commands.handle("::add-matcher::go.json", &mut log).is_none());
434 assert_eq!(commands.handle("main.go:3:1: undefined: x", &mut log).unwrap(), "##[error]Error: undefined: x");
435 assert!(commands.handle("::remove-matcher owner=go::", &mut log).is_none());
436 assert_eq!(commands.handle("main.go:3:1: undefined: x", &mut log).unwrap(), "main.go:3:1: undefined: x");
437 assert!(commands.handle("::add-matcher::missing.json", &mut log).unwrap().contains("Unable to process command"));
438 let _ = std::fs::remove_dir_all(&dir);
439 }
440
441 #[test]
442 fn commands_are_read_with_their_properties() {
443 let (name, properties, data) = parse_command("::error file=app.js,line=10,title=Bad%3A thing::Something%0Abroke").unwrap();
444 assert_eq!(name, "error");
445 assert_eq!(properties["file"], "app.js");
446 assert_eq!(properties["line"], "10");
447 assert_eq!(properties["title"], "Bad: thing");
448 assert_eq!(data, "Something\nbroke");
449 let (name, properties, data) = parse_command("::group::Install").unwrap();
450 assert_eq!((name.as_str(), properties.len(), data.as_str()), ("group", 0, "Install"));
451 assert_eq!(parse_command("::set-output name=version::1.2.3").unwrap().1["name"], "version");
452 assert!(parse_command("plain text").is_none());
453 assert!(parse_command(":: not a command").is_none());
454 }
455}