fix(pegboard): include namespace in actor log query (#2712)

<!-- Please make sure there is an issue that this PR is correlated to. -->

## Changes

<!-- If there are frontend changes, please include screenshots. -->

<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

* **Bug Fixes**
  * Improved filtering of actor log exports and reads by restricting results to the current namespace, ensuring more accurate and relevant log data.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: NathanFlurry

docker/dev-full/vector-server/vector.yaml

@@ -111,7 +111,7 @@ sinks:
     compression: gzip
     database: db_pegboard_actor_log
     endpoint: http://clickhouse:8123
-    table: actor_logs2
+    table: actor_logs3
     auth:
       strategy: basic
       user: vector

docker/monolith/vector-server/vector.yaml

@@ -82,7 +82,7 @@ sinks:
     compression: gzip
     endpoint: http://clickhouse:9300
     database: db_pegboard_actor_log
-    table: actor_logs2
+    table: actor_logs3
     auth:
       strategy: basic
       user: vector

packages/core/api/actor/src/route/logs.rs

@@ -96,6 +96,7 @@ pub async fn get_logs(
 			// frequently and should not return a significant amount of logs.
 			let logs_res = ctx
 				.op(pegboard::ops::actor::log::read::Input {
+					env_id,
 					actor_ids: actor_ids_clone.clone(),
 					stream_types: stream_types_clone.clone(),
 					count: 64,
@@ -136,6 +137,7 @@ pub async fn get_logs(
 		// Read most recent logs
 
 		ctx.op(pegboard::ops::actor::log::read::Input {
+			env_id,
 			actor_ids: actor_ids.clone(),
 			stream_types: stream_types.clone(),
 			count: 256,

packages/edge/infra/client/container-runner/Cargo.toml

@@ -16,8 +16,8 @@ rivet-logs.workspace = true
 serde = { version = "1.0.195", features = ["derive"] }
 serde_json = "1.0.111"
 signal-hook = "0.3.17"
+uuid = { version = "1.6.1", features = ["v4"] }
 
 [dev-dependencies]
 portpicker = "0.1.1"
 tempfile = "3.9.0"
-uuid = { version = "1.6.1", features = ["v4"] }

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

@@ -3,6 +3,7 @@ use std::{io::Write, net::TcpStream, sync::mpsc, thread::JoinHandle};
 use anyhow::*;
 use serde::Serialize;
 use serde_json;
+use uuid::Uuid;
 
 #[derive(Copy, Clone, Debug)]
 #[repr(u8)]
@@ -37,6 +38,8 @@ pub struct LogShipper {
 	pub vector_socket_addr: String,
 
 	pub actor_id: String,
+
+	pub env_id: Uuid,
 }
 
 impl LogShipper {
@@ -91,7 +94,7 @@ impl LogShipper {
 		while let Result::Ok(message) = self.msg_rx.recv() {
 			let vector_message = VectorMessage::Actors {
 				actor_id: self.actor_id.as_str(),
-				task: "main", // Backwards compatibility with logs
+				env_id: self.env_id,
 				stream_type: message.stream_type as u8,
 				ts: message.ts,
 				message: message.message.as_str(),
@@ -114,7 +117,7 @@ enum VectorMessage<'a> {
 	#[serde(rename = "actors")]
 	Actors {
 		actor_id: &'a str,
-		task: &'a str,
+		env_id: Uuid,
 		stream_type: u8,
 		ts: u64,
 		message: &'a str,

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

@@ -2,6 +2,7 @@ use std::{fs, path::Path, sync::mpsc, time::Duration};
 
 use anyhow::*;
 use utils::var;
+use uuid::Uuid;
 
 mod container;
 mod log_shipper;
@@ -36,6 +37,7 @@ fn main() -> Result<()> {
 		.transpose()
 		.context("failed to parse vector socket addr")?;
 	let actor_id = var("ACTOR_ID")?;
+	let env_id = Uuid::parse_str(&var("ENVIRONMENT_ID")?)?;
 
 	let (shutdown_tx, shutdown_rx) = mpsc::sync_channel(1);
 
@@ -48,6 +50,7 @@ fn main() -> Result<()> {
 			msg_rx,
 			vector_socket_addr,
 			actor_id,
+			env_id,
 		};
 		let log_shipper_thread = log_shipper.spawn();
 		(Some(msg_tx), Some(log_shipper_thread))

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

@@ -29,17 +29,24 @@ pub struct Actor {
 	actor_id: Uuid,
 	generation: u32,
 	config: protocol::ActorConfig,
+	metadata: protocol::ActorMetadata,
 
 	runner: Mutex<Option<runner::Handle>>,
 	exited: Mutex<bool>,
 }
 
 impl Actor {
-	pub fn new(actor_id: Uuid, generation: u32, config: protocol::ActorConfig) -> Arc<Self> {
+	pub fn new(
+		actor_id: Uuid,
+		generation: u32,
+		config: protocol::ActorConfig,
+		metadata: protocol::ActorMetadata,
+	) -> Arc<Self> {
 		Arc::new(Actor {
 			actor_id,
 			generation,
 			config,
+			metadata,
 
 			runner: Mutex::new(None),
 			exited: Mutex::new(false),
@@ -50,12 +57,14 @@ impl Actor {
 		actor_id: Uuid,
 		generation: u32,
 		config: protocol::ActorConfig,
+		metadata: protocol::ActorMetadata,
 		runner: runner::Handle,
 	) -> Arc<Self> {
 		Arc::new(Actor {
 			actor_id,
 			generation,
 			config,
+			metadata,
 
 			runner: Mutex::new(Some(runner)),
 			exited: Mutex::new(false),
@@ -209,6 +218,10 @@ impl Actor {
 				.to_string(),
 			),
 			("ACTOR_ID", self.actor_id.to_string()),
+			(
+				"ENVIRONMENT_ID",
+				self.metadata.environment.env_id.to_string(),
+			),
 		];
 		if let Some(vector) = &ctx.config().vector {
 			runner_env.push(("VECTOR_SOCKET_ADDR", vector.address.to_string()));

packages/edge/infra/client/manager/src/ctx.rs

@@ -421,6 +421,8 @@ impl Ctx {
 				generation,
 				config,
 			} => {
+				let metadata = config.metadata.deserialize()?;
+
 				let mut actors = self.actors.write().await;
 
 				if actors.contains_key(&(actor_id, generation)) {
@@ -430,7 +432,7 @@ impl Ctx {
 						"actor with this actor id + generation already exists, ignoring start command",
 					);
 				} else {
-					let actor = Actor::new(actor_id, generation, *config);
+					let actor = Actor::new(actor_id, generation, *config, metadata);
 
 					// Insert actor
 					actors.insert((actor_id, generation), actor);
@@ -718,6 +720,7 @@ impl Ctx {
 
 			let config = serde_json::from_slice::<protocol::ActorConfig>(&row.config)?;
 			let generation = row.generation.try_into()?;
+			let metadata = config.metadata.deserialize()?;
 
 			match &isolate_runner {
 				Some(isolate_runner) if pid == isolate_runner.pid().as_raw() => {}
@@ -736,7 +739,7 @@ impl Ctx {
 			}
 
 			// Clean up actor. We run `cleanup_setup` instead of `cleanup` because `cleanup` publishes events.
-			let actor = Actor::new(row.actor_id, generation, config);
+			let actor = Actor::new(row.actor_id, generation, config, metadata);
 			actor.cleanup_setup(self).await;
 		}
 
@@ -878,6 +881,7 @@ impl Ctx {
 
 			let config = serde_json::from_slice::<protocol::ActorConfig>(&row.config)?;
 			let generation = row.generation.try_into()?;
+			let metadata = config.metadata.deserialize()?;
 
 			let runner = match &isolate_runner {
 				// We have to clone the existing isolate runner handle instead of creating a new one so it
@@ -901,7 +905,7 @@ impl Ctx {
 				},
 			};
 
-			let actor = Actor::with_runner(row.actor_id, generation, config, runner);
+			let actor = Actor::with_runner(row.actor_id, generation, config, metadata, runner);
 			let actor = actors_guard
 				.entry((row.actor_id, generation))
 				.or_insert(actor);

packages/edge/services/pegboard/db/actor-log/migrations/20250703191728_drop_actor_logs_metadata.down.sql

@@ -0,0 +1 @@
+DROP VIEW IF EXISTS actor_logs2_with_metadata;