Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Process logs before waiting container #6

Merged
merged 4 commits into from
Sep 5, 2024
Merged

Process logs before waiting container #6

Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
256a687
Process logs before waiting container
jrm5100 Sep 5, 2024
f601be0
fix fmt
jrm5100 Sep 5, 2024
41115b3
Initialize stdout and stderr with 256 bytes
jrm5100 Sep 5, 2024
00d3b59
Merge branch 'main' into jm-fix-race-condition
jrm5100 Sep 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 34 additions & 27 deletions src/engine/service/runner/backend/docker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,8 @@ use bollard::container::WaitContainerOptions;
use bollard::errors::Error;
use bollard::secret::ContainerWaitResponse;
use bollard::Docker;
use futures::select;
use futures::FutureExt;
use futures::StreamExt;
use futures::TryStreamExt;
use nonempty::NonEmpty;
use random_word::Lang;
use tokio::sync::oneshot::Sender;
Expand Down Expand Up @@ -62,35 +61,43 @@ impl Runner {
container_create(&name, execution, &mut client).await;
container_start(&name, &mut client).await;

let mut logs = configure_logs(&name, execution, &mut client);
let logs = configure_logs(&name, execution, &mut client);
let mut wait = configure_wait(&name, &mut client);

let status;
let mut stdout = String::with_capacity(1 >> 8);
let mut stderr = String::with_capacity(1 >> 8);

loop {
select! {
result = logs.next().fuse() => match result {
Some(Ok(LogOutput::StdOut { message })) => {
stdout.push_str(&String::from_utf8_lossy(message.as_ref()))
}
Some(Ok(LogOutput::StdErr { message })) => {
stderr.push_str(&String::from_utf8_lossy(message.as_ref()))
// Process logs until they stop when container stops
let (stdout, stderr) = logs
.try_fold(
(String::new(), String::new()),
|(mut stdout, mut stderr), log| async move {
match log {
LogOutput::StdOut { message } => {
stdout.push_str(&String::from_utf8_lossy(&message));
}
LogOutput::StdErr { message } => {
stderr.push_str(&String::from_utf8_lossy(&message));
}
_ => {}
}
Some(Err(e)) => eprintln!("error reading log: {:?}", e),
Some(Ok(_)) | None => {}
Ok((stdout, stderr))
},
result = wait.next().fuse() => match result {
Some(Ok(response)) => {
status = response.status_code;
break;
}
Some(Err(e)) => eprintln!("error waiting for container: {e:?}"),
None => unreachable!(),
}
}
}
)
.await
.unwrap_or_else(|e| {
eprintln!("Error collecting logs: {:?}", e);
(String::new(), String::new())
});

// Process container stop
let status = wait
.next()
.await
.transpose()
.unwrap_or_else(|e| {
eprintln!("Error waiting for container: {:?}", e);
None
})
.map(|response| response.status_code)
.unwrap_or(-1);

client.remove_container(&name, None).await.unwrap();

Expand Down
Loading