feat(toolchain): remote build logs

cloud/packages/ci-manager/src/executors/docker.ts

@@ -38,6 +38,11 @@ export async function runDockerBuild(
 		`Starting kaniko with args: docker ${kanikoArgs.join(" ")}`,
 	);
 
+	buildStore.updateStatus(buildId, {
+		type: "running",
+		data: { docker: {} }
+	});
+
 	return new Promise<void>((resolve, reject) => {
 		const dockerProcess = spawn("docker", kanikoArgs, {
 			stdio: ["pipe", "pipe", "pipe"]

cloud/packages/ci-manager/src/executors/rivet.ts

@@ -78,6 +78,13 @@ export async function runRivetBuild(
 		const actorId = createResponse.actor.id;
 		buildStore.addLog(buildId, `Created Rivet actor: ${actorId}`);
 
+		buildStore.updateStatus(buildId, {
+			type: "running",
+			data: {
+				rivet: { actorId }
+			}
+		});
+
 		await pollActorStatus(
 			buildStore,
 			client,

cloud/packages/ci-manager/src/kaniko-runner.ts

@@ -17,7 +17,12 @@ export async function runKanikoBuild(
 
 	await mkdir(dirname(build.contextPath!), { recursive: true });
 
-	buildStore.updateStatus(buildId, { type: "running", data: {} });
+	buildStore.updateStatus(buildId, {
+		type: "running",
+		data: {
+			noRunner: {}
+		}
+	});
 
 	const executionMode = process.env.KANIKO_EXECUTION_MODE || "docker";
 	buildStore.addLog(buildId, `Using execution mode: ${executionMode}`);

cloud/packages/ci-manager/src/types.ts

@@ -1,9 +1,25 @@
 import { z } from "zod";
 import { NO_SEP_CHAR_REGEX, UNIT_SEP_CHAR } from "./common";
 
+export const RunnerSchema = z.union([
+    z.object({
+        rivet: z.object({
+            actorId: z.string(),
+        })
+    }),
+    z.object({
+        docker: z.object({})
+    }),
+    z.object({
+        noRunner: z.object({})
+    })
+]);
+
+export type Runner = z.infer<typeof RunnerSchema>;
+
 export const StatusSchema = z.discriminatedUnion("type", [
     z.object({ type: z.literal("starting"), data: z.object({}) }),
-    z.object({ type: z.literal("running"), data: z.object({}) }),
+    z.object({ type: z.literal("running"), data: RunnerSchema }),
     z.object({ type: z.literal("finishing"), data: z.object({}) }),
     z.object({ type: z.literal("converting"), data: z.object({}) }),
     z.object({ type: z.literal("uploading"), data: z.object({}) }),

packages/toolchain/cli/Cargo.toml

@@ -16,7 +16,6 @@ path = "src/main.rs"
 [dependencies]
 anyhow = "1.0"
 async-posthog.workspace = true
-base64 = "0.22.1"
 clap = { version = "4.5.9", features = ["derive"] }
 ctrlc = "3.4.5"
 deno-embed.workspace = true

packages/toolchain/cli/src/commands/actor/create.rs

@@ -103,7 +103,7 @@ pub struct Opts {
 
 	/// Specify which log stream to display
 	#[clap(long)]
-	log_stream: Option<crate::util::actor::logs::LogStream>,
+	log_stream: Option<toolchain::util::actor::logs::LogStream>,
 
 	/// Deploy the build before creating the actor
 	#[clap(long)]
@@ -299,17 +299,18 @@ impl Opts {
 
 		// Tail logs
 		if self.logs {
-			crate::util::actor::logs::tail(
+			toolchain::util::actor::logs::tail(
 				&ctx,
-				crate::util::actor::logs::TailOpts {
+				toolchain::util::actor::logs::TailOpts {
 					environment: &env,
 					actor_id: response.actor.id,
 					stream: self
 						.log_stream
 						.clone()
-						.unwrap_or(crate::util::actor::logs::LogStream::All),
+						.unwrap_or(toolchain::util::actor::logs::LogStream::All),
 					follow: true,
-					timestamps: true,
+					print_type: toolchain::util::actor::logs::PrintType::PrintWithTime,
+					exit_on_ctrl_c: true,
 				},
 			)
 			.await?;

packages/toolchain/cli/src/commands/actor/logs.rs

@@ -16,7 +16,7 @@ pub struct Opts {
 
 	/// Specify which log stream to display (stdout, stderr, or all)
 	#[clap(long, short = 's')]
-	stream: Option<crate::util::actor::logs::LogStream>,
+	stream: Option<toolchain::util::actor::logs::LogStream>,
 
 	/// Disable timestamp display in logs
 	#[clap(long)]
@@ -36,17 +36,23 @@ impl Opts {
 		let actor_id =
 			Uuid::parse_str(&self.id).map_err(|_| errors::UserError::new("invalid id uuid"))?;
 
-		crate::util::actor::logs::tail(
+		let print_type = if self.no_timestamps {
+			toolchain::util::actor::logs::PrintType::Print
+		} else {
+			toolchain::util::actor::logs::PrintType::PrintWithTime
+		};
+		toolchain::util::actor::logs::tail(
 			&ctx,
-			crate::util::actor::logs::TailOpts {
+			toolchain::util::actor::logs::TailOpts {
 				environment: &env,
 				actor_id,
 				stream: self
 					.stream
 					.clone()
-					.unwrap_or(crate::util::actor::logs::LogStream::All),
+					.unwrap_or(toolchain::util::actor::logs::LogStream::All),
 				follow: !self.no_follow,
-				timestamps: !self.no_timestamps,
+				print_type,
+				exit_on_ctrl_c: true
 			},
 		)
 		.await?;

packages/toolchain/cli/src/util/mod.rs

@@ -1,4 +1,3 @@
-pub mod actor;
 pub mod deploy;
 pub mod env;
 pub mod login;

packages/toolchain/toolchain/Cargo.toml

@@ -9,6 +9,8 @@ edition.workspace = true
 [dependencies]
 anyhow = "1.0"
 async-stream = "0.3.3"
+base64 = "0.22.1"
+chrono = "0.4"
 clap = { version = "4.5", features = ["derive"] }
 console = "0.15"
 const_format = "0.2.32"
@@ -34,6 +36,7 @@ schemars = "0.8.21"
 serde = { version = "1.0", features = ["derive", "rc"] }
 serde_json = { version = "1.0", features = ["raw_value"] }
 sha1 = "0.10.6"
+strip-ansi-escapes = "0.2.1"
 strum = { version = "0.24", features = ["derive"] }
 tar = "0.4.40"
 tempfile = "3.13.0"

packages/toolchain/cli/src/util/actor/logs.rs → packages/toolchain/toolchain/src/util/actor/logs.rs

@@ -1,10 +1,14 @@
 use anyhow::*;
 use base64::{engine::general_purpose::STANDARD, Engine};
 use clap::ValueEnum;
+use chrono::{DateTime, Utc};
 use std::time::Duration;
 use tokio::signal;
 use tokio::sync::watch;
-use toolchain::rivet_api::{apis, models};
+use crate::{
+	rivet_api::{apis, models},
+	ToolchainCtx
+};
 use uuid::Uuid;
 
 #[derive(ValueEnum, Clone)]
@@ -17,31 +21,44 @@ pub enum LogStream {
 	StdErr,
 }
 
+pub enum PrintType {
+	/// Callback that is called when a new line is fetched.
+	/// The first argument is the timestamp of the line, the second argument is the decoded line.
+	Custom(fn(DateTime<Utc>, String)),
+	/// Prints the line to stdout.
+	Print,
+	/// Prints with timestamp
+	PrintWithTime,
+}
+
 pub struct TailOpts<'a> {
+	pub print_type: PrintType,
 	pub environment: &'a str,
 	pub actor_id: Uuid,
 	pub stream: LogStream,
 	pub follow: bool,
-	pub timestamps: bool,
+	pub exit_on_ctrl_c: bool,
 }
 
 /// Reads the logs of an actor.
-pub async fn tail(ctx: &toolchain::ToolchainCtx, opts: TailOpts<'_>) -> Result<()> {
+pub async fn tail(ctx: &ToolchainCtx, opts: TailOpts<'_>) -> Result<()> {
 	let (stdout_fetched_tx, stdout_fetched_rx) = watch::channel(false);
 	let (stderr_fetched_tx, stderr_fetched_rx) = watch::channel(false);
 
+	let exit_on_ctrl_c = opts.exit_on_ctrl_c;
+
 	tokio::select! {
 		result = tail_streams(ctx, &opts, stdout_fetched_tx, stderr_fetched_tx) => result,
 		result = poll_actor_state(ctx, &opts, stdout_fetched_rx, stderr_fetched_rx) => result,
-		_ = signal::ctrl_c() => {
+		_ = signal::ctrl_c(), if exit_on_ctrl_c => {
 			Ok(())
 		}
 	}
 }
 
 /// Reads the streams of an actor's logs.
 async fn tail_streams(
-	ctx: &toolchain::ToolchainCtx,
+	ctx: &ToolchainCtx,
 	opts: &TailOpts<'_>,
 	stdout_fetched_tx: watch::Sender<bool>,
 	stderr_fetched_tx: watch::Sender<bool>,
@@ -66,7 +83,7 @@ async fn tail_streams(
 
 /// Reads a specific stream of an actor's log.
 async fn tail_stream(
-	ctx: &toolchain::ToolchainCtx,
+	ctx: &ToolchainCtx,
 	opts: &TailOpts<'_>,
 	stream: models::ActorsQueryLogStream,
 	log_fetched_tx: watch::Sender<bool>,
@@ -111,6 +128,10 @@ async fn tail_stream(
 		}
 
 		for (ts, line) in res.timestamps.iter().zip(res.lines.iter()) {
+			let Result::Ok(ts) = ts.parse::<DateTime<Utc>>() else {
+				eprintln!("Failed to parse timestamp: {ts} for line {line}");
+				continue;
+			};
 			let decoded_line = match STANDARD.decode(line) {
 				Result::Ok(bytes) => String::from_utf8_lossy(&bytes).to_string(),
 				Err(_) => {
@@ -119,10 +140,17 @@ async fn tail_stream(
 				}
 			};
 
-			if opts.timestamps {
-				println!("{ts} {decoded_line}");
-			} else {
-				println!("{decoded_line}");
+
+			match &opts.print_type {
+				PrintType::Custom(callback) => {
+					(callback)(ts, decoded_line);
+				}
+				PrintType::Print => {
+					println!("{decoded_line}");
+				}
+				PrintType::PrintWithTime => {
+					println!("{ts} {decoded_line}");
+				}
 			}
 		}
 
@@ -138,7 +166,7 @@ async fn tail_stream(
 ///
 /// Using this in a `tokio::select` will make all other tasks cancel when the actor finishes.
 async fn poll_actor_state(
-	ctx: &toolchain::ToolchainCtx,
+	ctx: &ToolchainCtx,
 	opts: &TailOpts<'_>,
 	mut stdout_fetched_rx: watch::Receiver<bool>,
 	mut stderr_fetched_rx: watch::Receiver<bool>,
@@ -171,7 +199,12 @@ async fn poll_actor_state(
 		.map_err(|err| anyhow!("Failed to poll actor: {err}"))?;
 
 		if res.actor.destroyed_at.is_some() {
-			println!("Actor finished");
+			match opts.print_type {
+				PrintType::Custom(_cb) => {}
+				_ => {
+					println!("Actor finished");
+				}
+			}
 			return Ok(());
 		}
 	}