feat(pb): add runner logs

Author: MasterPtato

packages/edge/infra/client/container-runner/src/log_shipper.rs

@@ -37,6 +37,7 @@ pub struct LogShipper {
 	pub vector_socket_addr: String,
 
 	pub runner_id: String,
+	pub actor_id: Option<String>,
 }
 
 impl LogShipper {
@@ -91,7 +92,7 @@ impl LogShipper {
 		while let Result::Ok(message) = self.msg_rx.recv() {
 			let vector_message = VectorMessage::Runners {
 				runner_id: self.runner_id.as_str(),
-				task: "main", // Backwards compatibility with logs
+				actor_id: self.actor_id.as_ref().map(|x| x.as_str()),
 				stream_type: message.stream_type as u8,
 				ts: message.ts,
 				message: message.message.as_str(),
@@ -114,7 +115,7 @@ enum VectorMessage<'a> {
 	#[serde(rename = "runners")]
 	Runners {
 		runner_id: &'a str,
-		task: &'a str,
+		actor_id: Option<&'a str>,
 		stream_type: u8,
 		ts: u64,
 		message: &'a str,

packages/edge/infra/client/container-runner/src/main.rs

@@ -37,6 +37,8 @@ fn main() -> Result<()> {
 		.transpose()
 		.context("failed to parse vector socket addr")?;
 	let runner_id = var("RUNNER_ID")?;
+	// Only set if this is a single allocation runner (one actor running on it)
+	let actor_id = var("ACTOR_ID").ok();
 
 	let (shutdown_tx, shutdown_rx) = mpsc::sync_channel(1);
 
@@ -49,6 +51,7 @@ fn main() -> Result<()> {
 			msg_rx,
 			vector_socket_addr,
 			runner_id,
+			actor_id,
 		};
 		let log_shipper_thread = log_shipper.spawn();
 		(Some(msg_tx), Some(log_shipper_thread))

packages/edge/infra/client/manager/src/actor/mod.rs

@@ -114,9 +114,15 @@ impl Actor {
 			.context("should have runner config")?
 		{
 			protocol::ActorRunner::New { .. } => {
+				let actor_id = matches!(
+					self.runner.config().image.allocation_type,
+					protocol::ImageAllocationType::Single
+				)
+				.then_some(self.actor_id);
+
 				// Because the runner is not already started we can get the ports here instead of reading from
 				// sqlite
-				let ports = self.runner.start(ctx).await?;
+				let ports = self.runner.start(ctx, actor_id).await?;
 
 				let pid = self.runner.pid().await?;
 

packages/edge/infra/client/manager/src/runner/mod.rs

@@ -243,9 +243,12 @@ impl Runner {
 		Ok(())
 	}
 
+	// `actor_id` is set if this runner has a single allocation type which means there is only one actor
+	// runner on it
 	pub async fn start(
 		self: &Arc<Self>,
 		ctx: &Arc<Ctx>,
+		actor_id: Option<Uuid>,
 	) -> Result<protocol::HashableMap<String, protocol::ProxiedPort>> {
 		tracing::info!(runner_id=?self.runner_id, "starting");
 
@@ -298,7 +301,7 @@ impl Runner {
 		let self2 = self.clone();
 		let ctx2 = ctx.clone();
 		tokio::spawn(async move {
-			match self2.run(&ctx2).await {
+			match self2.run(&ctx2, actor_id).await {
 				Ok(_) => {
 					if let Err(err) = self2.observe(&ctx2, false).await {
 						tracing::error!(runner_id=?self2.runner_id, ?err, "observe failed");
@@ -318,7 +321,7 @@ impl Runner {
 		Ok(proxied_ports)
 	}
 
-	async fn run(&self, ctx: &Ctx) -> Result<()> {
+	async fn run(&self, ctx: &Ctx, actor_id: Option<Uuid>) -> Result<()> {
 		// NOTE: This is the env that goes to the container-runner process, NOT the env that in inserted into
 		// the container.
 		let mut runner_env = vec![
@@ -328,6 +331,11 @@ impl Runner {
 			),
 			("RUNNER_ID", self.runner_id.to_string()),
 		];
+
+		if let Some(actor_id) = actor_id {
+			runner_env.push(("ACTOR_ID", actor_id.to_string()));
+		}
+
 		if let Some(vector) = &ctx.config().vector {
 			runner_env.push(("VECTOR_SOCKET_ADDR", vector.address.to_string()));
 		}

packages/edge/services/pegboard/db/runner-log/migrations/20200101000000_init.down.sql

@@ -0,0 +1,17 @@
+
+CREATE TABLE IF NOT EXISTS runner_logs (
+    runner_id UUID,
+    actor_id UUID,      -- When not set will be the NIL UUID (all zeros)
+    stream_type UInt8,  -- pegboard::types::LogsStreamType
+    ts DateTime64 (9),
+    message String
+) ENGINE = ReplicatedMergeTree ()
+PARTITION BY
+    toStartOfHour (ts)
+ORDER BY (
+    runner_id,
+    toUnixTimestamp (ts),
+    stream_type
+)
+TTL toDate (ts + toIntervalDay (3))
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;