Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 5 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -7,3 +7,8 @@ target/
.spike-*-id
.control-id
.DS_Store

.openshell-separated/
.control-separated-id
.builder-separated-id
__pycache__/
188 changes: 135 additions & 53 deletions README.md

Large diffs are not rendered by default.

5 changes: 3 additions & 2 deletions driver/e2b-helper.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
// extend {sandboxId, timeoutMs} → {}
//
// Every sandbox is created private (allowPublicTraffic: false): its URLs
// require the traffic access token, which only the driver holds.
// require the traffic access token, which the driver delegates only to the paired supervisor.
import { Sandbox, CommandExitError } from 'e2b'

const TUNNEL_PORT = 9000
Expand All @@ -32,10 +32,11 @@ async function readStdin() {
}

const ops = {
async create({ template, metadata = {}, timeoutMs = 60 * 60_000 }) {
async create({ template, metadata = {}, timeoutMs = 60 * 60_000, allowInternetAccess = false }) {
const sbx = await Sandbox.create(template, {
metadata,
timeoutMs,
allowInternetAccess,
network: { allowPublicTraffic: false },
})
return { sandboxId: sbx.sandboxId, trafficAccessToken: sbx.trafficAccessToken, host9000: sbx.getHost(TUNNEL_PORT) }
Expand Down
6 changes: 4 additions & 2 deletions driver/src/boundary.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
//! ┌───────────────────────┴───────────────────────┐
//! ▼ ▼
//! BoundaryConfig → bootstrap.json SandboxRuntimeDescriptor → runtime-descriptor.json
//! read by openshell-sandbox (agent box) read by openshell-supervisor (control box)
//! read by openshell-sandbox (agent box) read by openshell-supervisor (separate supervisor box)
//! "listen on control.sock with this TLS cert, "connect to 127.0.0.1:<port>, trust only
//! accept only this gateway's keys" this session's CA, expect this fence"
//!
Expand Down Expand Up @@ -43,6 +43,7 @@ struct E2bOuterFenceEvidence<'a> {
network_namespace: &'static str,
fenced_interfaces: &'a [String],
public_traffic: bool,
host_egress: &'static str,
}

impl E2bOuterFenceEvidence<'_> {
Expand Down Expand Up @@ -85,7 +86,7 @@ pub struct E2bBoundarySpec {
pub e2b_sandbox_id: String,
pub template: String,
pub fenced_interfaces: Vec<String>,
/// Local end of tunnel 2 in the control box; the supervisor connects here.
/// Local end of tunnel 2 in the supervisor box; the supervisor connects here.
pub supervisor_connect: SocketAddr,
pub supervisor_tls: SandboxTlsClientConfig,
pub workload_identity: ResolvedWorkloadIdentity,
Expand All @@ -109,6 +110,7 @@ impl E2bBoundarySpec {
network_namespace: "os",
fenced_interfaces: &self.fenced_interfaces,
public_traffic: false,
host_egress: "E2B allowInternetAccess=false",
}
.project(&self.generation)?;

Expand Down
11 changes: 8 additions & 3 deletions driver/src/e2b.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ pub struct E2b {
pub helper: PathBuf,
}

#[derive(Debug, Deserialize)]
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Created {
pub sandbox_id: String,
Expand Down Expand Up @@ -69,10 +69,15 @@ impl E2b {
}

/// Create a private sandbox from `template`, tagged with `metadata`.
pub async fn create(&self, template: &str, metadata: Value) -> Result<Created, String> {
pub async fn create(
&self,
template: &str,
metadata: Value,
allow_internet: bool,
) -> Result<Created, String> {
self.call(
"create",
json!({ "template": template, "metadata": metadata, "timeoutMs": 3_600_000 }),
json!({ "template": template, "metadata": metadata, "timeoutMs": 3_600_000, "allowInternetAccess": allow_internet }),
)
.await
}
Expand Down
166 changes: 56 additions & 110 deletions driver/src/health.rs
Original file line number Diff line number Diff line change
@@ -1,126 +1,72 @@
//! Health monitor: notices when a running sandbox breaks and tells the gateway.
//!
//! every 5 s did the supervisor or the tunnel client exit? (local, cheap)
//! every 30 s does the E2B box still exist? (one E2B API call for all)
//! retry killing boxes whose create rollback failed
//!
//! A problem flips the sandbox's Ready condition to False with a reason, and
//! the change goes out on the watch stream. Without this, a sandbox whose
//! tunnel died would keep reporting Ready.

use crate::lifecycle::{self, Running};
//! Monitor both task microVMs and the remote supervisor process tree.
//! Snapshot before network calls; never hold the sandbox table lock over I/O.
use crate::lifecycle;
use crate::service::Shared;
use openshell_core::proto::compute::v1::{
WatchSandboxesEvent, WatchSandboxesSandboxEvent, watch_sandboxes_event::Payload,
};
use std::collections::HashSet;
use std::time::Duration;
use tracing::{info, warn};

const PROCESS_CHECK: Duration = Duration::from_secs(5);
/// Check E2B every Nth process check (6 × 5 s = 30 s).
const E2B_CHECK_EVERY: u32 = 6;
/// Names for `Running::children`, in the order lifecycle.rs starts them.
const CHILD_NAMES: [&str; 2] = ["tunnel", "supervisor"];

pub fn spawn(shared: Shared) {
tokio::spawn(async move {
let mut tick: u32 = 0;
loop {
tokio::time::sleep(PROCESS_CHECK).await;
tick = tick.wrapping_add(1);
let box_check = if tick.is_multiple_of(E2B_CHECK_EVERY) {
retry_leaked(&shared).await;
check_boxes(&shared).await
} else {
None
};
let mut table = shared.sandboxes.lock().await;
for running in table.values_mut() {
if running.deleting {
continue; // the delete task owns this one now
tokio::time::sleep(Duration::from_secs(10)).await;
let leaked: Vec<_> = shared.leaked.lock().await.iter().cloned().collect();
for id in leaked {
if shared.cfg.e2b.kill(&id).await.is_ok() {
shared.leaked.lock().await.remove(&id);
info!(e2b = %id, "health: rollback cleaned up");
}
}
let snapshot: Vec<_> = shared
.sandboxes
.lock()
.await
.iter()
.filter(|(_, r)| !r.deleting)
.map(|(id, r)| (id.clone(), r.e2b_id.clone(), r.supervisor_id.clone()))
.collect();
if snapshot.is_empty() {
continue;
}
let alive = match shared.cfg.e2b.list_ids(&shared.cfg.owner).await {
Ok(alive) => alive,
Err(error) => {
warn!(%error, "health: E2B list unavailable");
continue;
}
if let Some((reason, message)) = problem(running, box_check.as_ref())
&& lifecycle::mark_not_ready(running, reason, &message)
{
warn!(sandbox = %running.sandbox.name, reason, %message, "health: sandbox not ready");
let _ = shared.events.send(WatchSandboxesEvent {
payload: Some(Payload::Sandbox(WatchSandboxesSandboxEvent {
sandbox: Some(running.sandbox.clone()),
})),
});
};
for (id, worker, supervisor) in snapshot {
let problem = if !alive.contains(&worker) || !alive.contains(&supervisor) {
Some((
"E2bSandboxGone",
"A member of the task pair expired or was killed",
))
} else {
match shared.cfg.e2b.run(&supervisor,
"test -f /run/openshell-task/ready && test ! -f /run/openshell-task/failed && kill -0 $(cat /run/openshell-task/launcher.pid) 2>/dev/null", "root").await {
Ok(result) if result.exit_code != 0 => Some(("SupervisorExited", "Supervisor or authenticated transport exited")),
// A transient SDK transport failure is not proof of a dead process.
_ => None,
}
};
if let Some((reason, message)) = problem {
let mut table = shared.sandboxes.lock().await;
if let Some(running) = table.get_mut(&id)
&& !running.deleting
&& running.supervisor_id == supervisor
&& lifecycle::mark_not_ready(running, reason, message)
{
warn!(sandbox = %running.sandbox.name, reason, message, "health: sandbox not ready");
let _ = shared.events.send(WatchSandboxesEvent {
payload: Some(Payload::Sandbox(WatchSandboxesSandboxEvent {
sandbox: Some(running.sandbox.clone()),
})),
});
}
}
}
drop(table);
}
});
}

/// Boxes we knew about BEFORE asking E2B, and the boxes E2B reports alive.
/// Only boxes in the first set can be judged, so a sandbox created while the
/// query was in flight is never mistaken for a dead one.
struct BoxCheck {
known_before: HashSet<String>,
alive: HashSet<String>,
}

async fn check_boxes(shared: &Shared) -> Option<BoxCheck> {
let known_before: HashSet<String> = shared
.sandboxes
.lock()
.await
.values()
.map(|r| r.e2b_id.clone())
.collect();
if known_before.is_empty() {
return None;
}
// The network call happens outside the lock.
match shared.cfg.e2b.list_ids(&shared.cfg.owner).await {
Ok(alive) => Some(BoxCheck {
known_before,
alive,
}),
Err(e) => {
warn!(error = %e, "health: E2B list failed; skipping box check");
None
}
}
}

async fn retry_leaked(shared: &Shared) {
let leaked: Vec<String> = shared.leaked.lock().await.iter().cloned().collect();
for box_id in leaked {
if shared.cfg.e2b.kill(&box_id).await.is_ok() {
shared.leaked.lock().await.remove(&box_id);
info!(e2b = %box_id, "health: leaked E2B box killed");
}
}
}

/// The first thing wrong with this sandbox, if anything.
fn problem(running: &mut Running, box_check: Option<&BoxCheck>) -> Option<(&'static str, String)> {
for (child, name) in running.children.iter_mut().zip(CHILD_NAMES) {
if let Ok(Some(status)) = child.try_wait() {
let reason = if name == "supervisor" {
"SupervisorExited"
} else {
"TunnelExited"
};
return Some((reason, format!("{name} process exited ({status})")));
}
}
if let Some(check) = box_check
&& check.known_before.contains(&running.e2b_id)
&& !check.alive.contains(&running.e2b_id)
{
return Some((
"E2bSandboxGone",
format!(
"E2B box {} no longer exists (expired or killed)",
running.e2b_id
),
));
}
None
}
Loading
Loading