Skip to content

Commit de999a9

Browse files
jjamrogajjamroga
committed
PoC: feat(sandbox): add publish_ports ingress - expose sandbox services via K8s Service
Adds a --publish PORT flag to sandbox create that makes services running inside the sandbox network namespace reachable from outside. Changes by layer: proto: add publish_ports to SandboxTemplate and DriverSandboxTemplate server: copy publish_ports through driver_sandbox_template_from_public driver-kubernetes: - Set OPENSHELL_PUBLISH_PORTS env var and containerPort declarations on the agent container so the supervisor knows which ports to expose - Inject openshell.ai/sandbox-id label into every pod template so Services can select pods by a stable, openshell-controlled label - After creating the Sandbox CR, create an <name>-inbound ClusterIP Service exposing the published ports; roll back the Sandbox CR if Service creation fails so nothing is left in a half-created state - Delete the -inbound Service on sandbox teardown sandbox (netns): add publish_ports() which sets up in the pod primary network namespace: - PREROUTING DNAT to sandbox netns IP (10.200.0.x) - OUTPUT DNAT for loopback-originated traffic (kubectl port-forward) - POSTROUTING MASQUERADE so return traffic from the sandbox netns has a valid source address - FORWARD ACCEPT for forwarded packets - ip_forward=1 and route_localnet=1 (required for OUTPUT DNAT from 127.0.0.1 to route through non-loopback interfaces) sandbox (lib): read OPENSHELL_PUBLISH_PORTS at startup and call publish_ports() after the network namespace is configured cli: add --publish/-p PORT (repeatable), --env KEY=VALUE (repeatable), and --expose PORT flags to sandbox create; wire publish_ports into SandboxTemplate helm/role: grant create/delete/get/list on services to the gateway service account (required for -inbound Service lifecycle management) The -inbound Service naming is a workaround for the upstream kubernetes-sigs/agent-sandbox controller creating the primary headless Service without a field for additional ports. The correct long-term fix is to contribute a servicePorts field upstream (NVIDIA#994). Known limitations: - No per-sandbox NetworkPolicy for inbound ports (any pod in the cluster can reach published ports) - publish_ports not yet declarable in policy YAML or provider profiles
1 parent adf18c0 commit de999a9

9 files changed

Lines changed: 378 additions & 12 deletions

File tree

‎crates/openshell-cli/src/main.rs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1212,6 +1212,12 @@ enum SandboxCommands {
12121212
#[arg(long = "env", value_name = "KEY=VALUE")]
12131213
env: Vec<String>,
12141214

1215+
/// Publish a port from inside the sandbox network namespace to the pod.
1216+
/// The supervisor will DNAT traffic on this port into the sandbox namespace,
1217+
/// making it reachable via kubectl port-forward. Repeatable.
1218+
#[arg(long = "publish", short = 'p', value_name = "PORT")]
1219+
publish: Vec<u16>,
1220+
12151221
/// Command to run after "--" (defaults to an interactive shell).
12161222
#[arg(last = true, allow_hyphen_values = true)]
12171223
command: Vec<String>,
@@ -2332,6 +2338,7 @@ async fn main() -> Result<()> {
23322338
no_auto_providers,
23332339
labels,
23342340
env,
2341+
publish,
23352342
command,
23362343
} => {
23372344
// Resolve --tty / --no-tty into an Option<bool> override.
@@ -2439,6 +2446,7 @@ async fn main() -> Result<()> {
24392446
policy.as_deref(),
24402447
forward,
24412448
expose_specs.clone(),
2449+
&publish,
24422450
&command,
24432451
tty_override,
24442452
Some(false),
@@ -2464,6 +2472,7 @@ async fn main() -> Result<()> {
24642472
policy.as_deref(),
24652473
forward,
24662474
expose_specs,
2475+
&publish,
24672476
&command,
24682477
tty_override,
24692478
bootstrap_override,

‎crates/openshell-cli/src/run.rs‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1930,6 +1930,7 @@ pub async fn sandbox_create_with_bootstrap(
19301930
policy: Option<&str>,
19311931
forward: Option<openshell_core::forward::ForwardSpec>,
19321932
expose: Vec<openshell_core::forward::ForwardSpec>,
1933+
publish: &[u16],
19331934
command: &[String],
19341935
tty_override: Option<bool>,
19351936
bootstrap_override: Option<bool>,
@@ -1963,6 +1964,7 @@ pub async fn sandbox_create_with_bootstrap(
19631964
policy,
19641965
forward,
19651966
expose,
1967+
publish,
19661968
command,
19671969
tty_override,
19681970
Some(false),
@@ -2021,6 +2023,7 @@ pub async fn sandbox_create(
20212023
policy: Option<&str>,
20222024
forward: Option<openshell_core::forward::ForwardSpec>,
20232025
expose: Vec<openshell_core::forward::ForwardSpec>,
2026+
publish: &[u16],
20242027
command: &[String],
20252028
tty_override: Option<bool>,
20262029
bootstrap_override: Option<bool>,
@@ -2115,10 +2118,19 @@ pub async fn sandbox_create(
21152118

21162119
let policy = load_sandbox_policy(policy)?;
21172120

2118-
let template = image.map(|img| SandboxTemplate {
2119-
image: img,
2120-
..SandboxTemplate::default()
2121-
});
2121+
let publish_ports: Vec<u32> = publish.iter().map(|&p| p as u32).collect();
2122+
let template = match (image, publish_ports.is_empty()) {
2123+
(Some(img), _) => Some(SandboxTemplate {
2124+
image: img,
2125+
publish_ports: publish_ports.clone(),
2126+
..SandboxTemplate::default()
2127+
}),
2128+
(None, false) => Some(SandboxTemplate {
2129+
publish_ports: publish_ports.clone(),
2130+
..SandboxTemplate::default()
2131+
}),
2132+
(None, true) => None,
2133+
};
21222134

21232135
let request = CreateSandboxRequest {
21242136
spec: Some(SandboxSpec {

‎crates/openshell-driver-kubernetes/src/driver.rs‎

Lines changed: 163 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,8 @@
55
66
use crate::config::KubernetesComputeConfig;
77
use futures::{Stream, StreamExt, TryStreamExt};
8-
use k8s_openapi::api::core::v1::{Event as KubeEventObj, Node};
8+
use k8s_openapi::api::core::v1::{Event as KubeEventObj, Node, Service, ServicePort, ServiceSpec};
9+
use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;
910
use kube::api::{Api, ApiResource, DeleteParams, ListParams, PostParams};
1011
use kube::core::gvk::GroupVersionKind;
1112
use kube::core::{DynamicObject, ObjectMeta};
@@ -331,7 +332,6 @@ impl KubernetesComputeDriver {
331332
sandbox_name = %name,
332333
"Sandbox created in Kubernetes successfully"
333334
);
334-
Ok(())
335335
}
336336
Ok(Err(err)) => {
337337
warn!(
@@ -340,7 +340,7 @@ impl KubernetesComputeDriver {
340340
error = %err,
341341
"Failed to create sandbox in Kubernetes"
342342
);
343-
Err(KubernetesDriverError::from_kube(err))
343+
return Err(KubernetesDriverError::from_kube(err));
344344
}
345345
Err(_elapsed) => {
346346
warn!(
@@ -349,10 +349,124 @@ impl KubernetesComputeDriver {
349349
timeout_secs = KUBE_API_TIMEOUT.as_secs(),
350350
"Timed out creating sandbox in Kubernetes"
351351
);
352-
Err(KubernetesDriverError::Message(format!(
352+
return Err(KubernetesDriverError::Message(format!(
353353
"timed out after {}s waiting for Kubernetes API",
354354
KUBE_API_TIMEOUT.as_secs()
355-
)))
355+
)));
356+
}
357+
}
358+
359+
// If publish_ports are declared, create a ClusterIP Service so the
360+
// sandbox ports are reachable by DNS name within the cluster.
361+
let publish_ports: Vec<u32> = sandbox
362+
.spec
363+
.as_ref()
364+
.and_then(|s| s.template.as_ref())
365+
.map(|t| t.publish_ports.clone())
366+
.unwrap_or_default();
367+
368+
if !publish_ports.is_empty() {
369+
if let Err(err) = self.apply_publish_ports_service(name, sandbox, &publish_ports).await {
370+
// Service creation failed — roll back the Sandbox CR so the
371+
// user is not left with a sandbox that is missing its inbound
372+
// Service. Ignore the rollback result; the original error is
373+
// what matters.
374+
warn!(
375+
sandbox_id = %sandbox.id,
376+
sandbox_name = %name,
377+
error = %err,
378+
"Rolling back sandbox creation due to inbound Service failure"
379+
);
380+
let _ = tokio::time::timeout(
381+
KUBE_API_TIMEOUT,
382+
self.api().delete(name, &DeleteParams::default()),
383+
)
384+
.await;
385+
return Err(KubernetesDriverError::Message(err));
386+
}
387+
}
388+
389+
Ok(())
390+
}
391+
392+
/// Create a ClusterIP Service named `<sandbox-name>-inbound` exposing the
393+
/// published ports. The Service selects pods by `openshell.ai/sandbox-id`
394+
/// which is injected into every sandbox pod template by `sandbox_template_to_k8s`.
395+
///
396+
/// Returns an error string if the Service could not be created, so the caller
397+
/// can roll back the Sandbox CR and surface the failure to the user.
398+
async fn apply_publish_ports_service(
399+
&self,
400+
sandbox_name: &str,
401+
sandbox: &Sandbox,
402+
publish_ports: &[u32],
403+
) -> Result<(), String> {
404+
let svc_name = format!("{sandbox_name}-inbound");
405+
let svc_ports: Vec<ServicePort> = publish_ports
406+
.iter()
407+
.map(|&p| ServicePort {
408+
port: p as i32,
409+
target_port: Some(IntOrString::Int(p as i32)),
410+
protocol: Some("TCP".to_string()),
411+
name: Some(format!("port-{p}")),
412+
..Default::default()
413+
})
414+
.collect();
415+
416+
let mut selector = BTreeMap::new();
417+
selector.insert(SANDBOX_ID_LABEL.to_string(), sandbox.id.clone());
418+
419+
let svc = Service {
420+
metadata: ObjectMeta {
421+
name: Some(svc_name.clone()),
422+
namespace: Some(self.config.namespace.clone()),
423+
labels: Some(sandbox_labels(sandbox)),
424+
..Default::default()
425+
},
426+
spec: Some(ServiceSpec {
427+
type_: Some("ClusterIP".to_string()),
428+
selector: Some(selector),
429+
ports: Some(svc_ports),
430+
..Default::default()
431+
}),
432+
..Default::default()
433+
};
434+
435+
let svc_api: Api<Service> = Api::namespaced(self.client.clone(), &self.config.namespace);
436+
match tokio::time::timeout(KUBE_API_TIMEOUT, svc_api.create(&PostParams::default(), &svc))
437+
.await
438+
{
439+
Ok(Ok(_)) => {
440+
info!(
441+
sandbox_id = %sandbox.id,
442+
sandbox_name = %sandbox_name,
443+
svc_name = %svc_name,
444+
ports = ?publish_ports,
445+
"Created ClusterIP Service for published ports"
446+
);
447+
Ok(())
448+
}
449+
Ok(Err(err)) => {
450+
warn!(
451+
sandbox_id = %sandbox.id,
452+
sandbox_name = %sandbox_name,
453+
svc_name = %svc_name,
454+
error = %err,
455+
"Failed to create inbound Service for published ports"
456+
);
457+
Err(format!("failed to create inbound Service '{svc_name}': {err}"))
458+
}
459+
Err(_) => {
460+
warn!(
461+
sandbox_id = %sandbox.id,
462+
sandbox_name = %sandbox_name,
463+
svc_name = %svc_name,
464+
"Timed out creating inbound Service for published ports"
465+
);
466+
Err(format!(
467+
"timed out creating inbound Service '{svc_name}' after {}s",
468+
KUBE_API_TIMEOUT.as_secs()
469+
))
356470
}
357471
}
358472
}
@@ -364,6 +478,15 @@ impl KubernetesComputeDriver {
364478
"Deleting sandbox from Kubernetes"
365479
);
366480

481+
// Best-effort cleanup of the inbound Service created for publish_ports.
482+
let svc_api: Api<Service> = Api::namespaced(self.client.clone(), &self.config.namespace);
483+
let inbound_svc = format!("{name}-inbound");
484+
let _ = tokio::time::timeout(
485+
KUBE_API_TIMEOUT,
486+
svc_api.delete(&inbound_svc, &DeleteParams::default()),
487+
)
488+
.await;
489+
367490
let api = self.api();
368491
match tokio::time::timeout(KUBE_API_TIMEOUT, api.delete(name, &DeleteParams::default()))
369492
.await
@@ -1057,8 +1180,17 @@ fn sandbox_template_to_k8s(
10571180
// regardless of which sandbox image is used.
10581181

10591182
let mut metadata = serde_json::Map::new();
1060-
if !template.labels.is_empty() {
1061-
metadata.insert("labels".to_string(), serde_json::json!(template.labels));
1183+
{
1184+
// Always inject the sandbox-id label so that Services can select pods
1185+
// by a stable, openshell-controlled label (the controller only sets
1186+
// agents.x-k8s.io/sandbox-name-hash, which we don't control).
1187+
let mut labels: serde_json::Map<String, serde_json::Value> =
1188+
template.labels.iter().map(|(k, v)| (k.clone(), serde_json::json!(v))).collect();
1189+
labels.insert(
1190+
SANDBOX_ID_LABEL.to_string(),
1191+
serde_json::json!(sandbox_id),
1192+
);
1193+
metadata.insert("labels".to_string(), serde_json::Value::Object(labels));
10621194
}
10631195
if let Some(annotations) = platform_config_struct(template, "annotations") {
10641196
metadata.insert("annotations".to_string(), annotations);
@@ -1091,7 +1223,7 @@ fn sandbox_template_to_k8s(
10911223
}
10921224

10931225
// Build environment variables - start with OpenShell-required vars
1094-
let env = build_env_list(
1226+
let mut env = build_env_list(
10951227
None,
10961228
&template.environment,
10971229
spec_environment,
@@ -1104,8 +1236,31 @@ fn sandbox_template_to_k8s(
11041236
!client_tls_secret_name.is_empty(),
11051237
);
11061238

1239+
// Expose publish_ports to the supervisor via env var so it can set up DNAT.
1240+
if !template.publish_ports.is_empty() {
1241+
let ports_str = template
1242+
.publish_ports
1243+
.iter()
1244+
.map(|p| p.to_string())
1245+
.collect::<Vec<_>>()
1246+
.join(",");
1247+
set_env_default(&mut env, "OPENSHELL_PUBLISH_PORTS", &ports_str);
1248+
}
1249+
11071250
container.insert("env".to_string(), serde_json::Value::Array(env));
11081251

1252+
// Declare containerPorts for each published port so the Kubernetes Service
1253+
// can route to them and kubectl port-forward works without specifying an
1254+
// explicit pod IP.
1255+
if !template.publish_ports.is_empty() {
1256+
let ports: Vec<serde_json::Value> = template
1257+
.publish_ports
1258+
.iter()
1259+
.map(|&p| serde_json::json!({"containerPort": p, "protocol": "TCP"}))
1260+
.collect();
1261+
container.insert("ports".to_string(), serde_json::Value::Array(ports));
1262+
}
1263+
11091264
// The sandbox process needs SYS_ADMIN (for seccomp filter installation and
11101265
// network namespace creation), NET_ADMIN (for network namespace veth setup),
11111266
// SYS_PTRACE (for the CONNECT proxy to read /proc/<pid>/fd/ of sandbox-user

‎crates/openshell-sandbox/src/lib.rs‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -399,6 +399,23 @@ pub async fn run_sandbox(
399399
.build()
400400
);
401401
}
402+
403+
// Publish ports declared via OPENSHELL_PUBLISH_PORTS by setting
404+
// up DNAT rules from the primary namespace into the sandbox namespace.
405+
let publish_ports: Vec<u16> = std::env::var("OPENSHELL_PUBLISH_PORTS")
406+
.unwrap_or_default()
407+
.split(',')
408+
.filter(|s| !s.is_empty())
409+
.filter_map(|s| s.trim().parse::<u16>().ok())
410+
.collect();
411+
if !publish_ports.is_empty() {
412+
if let Err(e) = ns.publish_ports(&publish_ports) {
413+
return Err(miette::miette!(
414+
"Failed to publish ports {publish_ports:?}: {e}"
415+
));
416+
}
417+
}
418+
402419
Some(ns)
403420
}
404421
Err(e) => {

0 commit comments

Comments
 (0)