Skip to content

Commit 3fa852f

Browse files
rhussrhuss
committed
fix(driver-k8s): add label selector to sandbox watch stream and list (#2211)
The Kubernetes driver's watch_sandboxes() crashed with a file descriptor leak when non-gateway Sandbox resources (e.g. from SandboxWarmPool) existed in the namespace. The unfiltered watch received events for all Sandbox objects, failed to extract IDs from warm pool objects, and propagated errors that triggered a 2-second reconnect loop leaking HTTP/2 connections until FD exhaustion. Two fixes: - Add openshell.ai/managed-by=openshell label selector to the watcher config and list params so only gateway-managed objects are received. - Defensively skip unrecognized objects with a debug log instead of sending errors to the channel, using let-else for flat control flow. Signed-off-by: Roland Huß <rhuss@redhat.com> Signed-off-by: Roland Huß <rhuss@redhat.com>
1 parent 614c8c1 commit 3fa852f

1 file changed

Lines changed: 113 additions & 53 deletions

File tree

  • crates/openshell-driver-kubernetes/src

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

Lines changed: 113 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -733,7 +733,9 @@ impl KubernetesComputeDriver {
733733
.await?;
734734
match tokio::time::timeout(
735735
KUBE_API_TIMEOUT,
736-
agent_sandbox_api.api.list(&ListParams::default()),
736+
agent_sandbox_api
737+
.api
738+
.list(&ListParams::default().labels(&format!("{LABEL_MANAGED_BY}={LABEL_MANAGED_BY_VALUE}"))),
737739
)
738740
.await
739741
{
@@ -969,8 +971,10 @@ impl KubernetesComputeDriver {
969971
.supported_agent_sandbox_api(self.watch_client.clone())
970972
.await?;
971973
let event_api: Api<KubeEventObj> = Api::namespaced(self.watch_client.clone(), &namespace);
974+
let sandbox_watcher_config = watcher::Config::default()
975+
.labels(&format!("{LABEL_MANAGED_BY}={LABEL_MANAGED_BY_VALUE}"));
972976
let mut sandbox_stream =
973-
watcher::watcher(agent_sandbox_api.api, watcher::Config::default()).boxed();
977+
watcher::watcher(agent_sandbox_api.api, sandbox_watcher_config).boxed();
974978
let mut event_stream = watcher::watcher(event_api, watcher::Config::default()).boxed();
975979
let (tx, rx) = mpsc::channel(256);
976980

@@ -982,64 +986,52 @@ impl KubernetesComputeDriver {
982986
tokio::select! {
983987
result = sandbox_stream.try_next() => match result {
984988
Ok(Some(Event::Applied(obj))) => {
985-
match sandbox_from_object(&namespace, obj) {
986-
Ok(sandbox) => {
987-
update_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &sandbox);
988-
let event = WatchSandboxesEvent {
989-
payload: Some(watch_sandboxes_event::Payload::Sandbox(
990-
WatchSandboxesSandboxEvent { sandbox: Some(sandbox) }
991-
)),
992-
};
993-
if tx.send(Ok(event)).await.is_err() {
994-
break;
995-
}
996-
}
997-
Err(err) => {
998-
if tx.send(Err(KubernetesDriverError::Message(err))).await.is_err() {
999-
break;
1000-
}
1001-
}
989+
let obj_name = obj.metadata.name.clone().unwrap_or_default();
990+
let Ok(sandbox) = sandbox_from_object(&namespace, obj) else {
991+
debug!(object_name = %obj_name, "skipping non-gateway Sandbox in Applied event");
992+
continue;
993+
};
994+
update_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &sandbox);
995+
let event = WatchSandboxesEvent {
996+
payload: Some(watch_sandboxes_event::Payload::Sandbox(
997+
WatchSandboxesSandboxEvent { sandbox: Some(sandbox) }
998+
)),
999+
};
1000+
if tx.send(Ok(event)).await.is_err() {
1001+
break;
10021002
}
10031003
}
10041004
Ok(Some(Event::Deleted(obj))) => {
1005-
match sandbox_id_from_object(&obj) {
1006-
Ok(sandbox_id) => {
1007-
remove_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &sandbox_id);
1008-
let event = WatchSandboxesEvent {
1009-
payload: Some(watch_sandboxes_event::Payload::Deleted(
1010-
WatchSandboxesDeletedEvent { sandbox_id }
1011-
)),
1012-
};
1013-
if tx.send(Ok(event)).await.is_err() {
1014-
break;
1015-
}
1016-
}
1017-
Err(err) => {
1018-
if tx.send(Err(KubernetesDriverError::Message(err))).await.is_err() {
1019-
break;
1020-
}
1021-
}
1005+
let obj_name = obj.metadata.name.clone().unwrap_or_default();
1006+
let Ok(sandbox_id) = sandbox_id_from_object(&obj) else {
1007+
debug!(object_name = %obj_name, "skipping non-gateway Sandbox in Deleted event");
1008+
continue;
1009+
};
1010+
remove_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &sandbox_id);
1011+
let event = WatchSandboxesEvent {
1012+
payload: Some(watch_sandboxes_event::Payload::Deleted(
1013+
WatchSandboxesDeletedEvent { sandbox_id }
1014+
)),
1015+
};
1016+
if tx.send(Ok(event)).await.is_err() {
1017+
break;
10221018
}
10231019
}
10241020
Ok(Some(Event::Restarted(objs))) => {
10251021
for obj in objs {
1026-
match sandbox_from_object(&namespace, obj) {
1027-
Ok(sandbox) => {
1028-
update_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &sandbox);
1029-
let event = WatchSandboxesEvent {
1030-
payload: Some(watch_sandboxes_event::Payload::Sandbox(
1031-
WatchSandboxesSandboxEvent { sandbox: Some(sandbox) }
1032-
)),
1033-
};
1034-
if tx.send(Ok(event)).await.is_err() {
1035-
return;
1036-
}
1037-
}
1038-
Err(err) => {
1039-
if tx.send(Err(KubernetesDriverError::Message(err))).await.is_err() {
1040-
return;
1041-
}
1042-
}
1022+
let obj_name = obj.metadata.name.clone().unwrap_or_default();
1023+
let Ok(sandbox) = sandbox_from_object(&namespace, obj) else {
1024+
debug!(object_name = %obj_name, "skipping non-gateway Sandbox in Restarted event");
1025+
continue;
1026+
};
1027+
update_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &sandbox);
1028+
let event = WatchSandboxesEvent {
1029+
payload: Some(watch_sandboxes_event::Payload::Sandbox(
1030+
WatchSandboxesSandboxEvent { sandbox: Some(sandbox) }
1031+
)),
1032+
};
1033+
if tx.send(Ok(event)).await.is_err() {
1034+
return;
10431035
}
10441036
}
10451037
}
@@ -5625,4 +5617,72 @@ mod tests {
56255617
let storage = &vct[0]["spec"]["resources"]["requests"]["storage"];
56265618
assert_eq!(storage, DEFAULT_WORKSPACE_STORAGE_SIZE);
56275619
}
5620+
5621+
fn make_dynamic_object(
5622+
name: &str,
5623+
labels: Option<BTreeMap<String, String>>,
5624+
) -> DynamicObject {
5625+
DynamicObject {
5626+
metadata: ObjectMeta {
5627+
name: Some(name.to_string()),
5628+
namespace: Some("test-ns".to_string()),
5629+
labels,
5630+
..Default::default()
5631+
},
5632+
types: None,
5633+
data: serde_json::json!({}),
5634+
}
5635+
}
5636+
5637+
#[test]
5638+
fn test_sandbox_id_from_object_with_label() {
5639+
let mut labels = BTreeMap::new();
5640+
labels.insert(LABEL_SANDBOX_ID.to_string(), "my-sandbox-id".to_string());
5641+
let obj = make_dynamic_object("sandbox-obj", Some(labels));
5642+
let result = sandbox_id_from_object(&obj);
5643+
assert_eq!(result, Ok("my-sandbox-id".to_string()));
5644+
}
5645+
5646+
#[test]
5647+
fn test_sandbox_id_from_object_with_name_prefix() {
5648+
let obj = make_dynamic_object("sandbox-abc123", None);
5649+
let result = sandbox_id_from_object(&obj);
5650+
assert_eq!(result, Ok("abc123".to_string()));
5651+
}
5652+
5653+
#[test]
5654+
fn test_sandbox_id_from_object_unknown() {
5655+
let obj = make_dynamic_object("openshell-warm-pool-xyz", None);
5656+
let result = sandbox_id_from_object(&obj);
5657+
assert!(result.is_err());
5658+
}
5659+
5660+
#[test]
5661+
fn test_sandbox_id_from_object_managed_by_but_no_sandbox_id() {
5662+
let mut labels = BTreeMap::new();
5663+
labels.insert(LABEL_MANAGED_BY.to_string(), LABEL_MANAGED_BY_VALUE.to_string());
5664+
let obj = make_dynamic_object("openshell-warm-pool-xyz", Some(labels));
5665+
let result = sandbox_id_from_object(&obj);
5666+
assert!(result.is_err());
5667+
}
5668+
5669+
#[test]
5670+
fn test_sandbox_from_object_success() {
5671+
let mut labels = BTreeMap::new();
5672+
labels.insert(LABEL_SANDBOX_ID.to_string(), "test-id".to_string());
5673+
let obj = make_dynamic_object("sandbox-test-id", Some(labels));
5674+
let result = sandbox_from_object("test-ns", obj);
5675+
assert!(result.is_ok());
5676+
let sandbox = result.unwrap();
5677+
assert_eq!(sandbox.id, "test-id");
5678+
assert_eq!(sandbox.name, "sandbox-test-id");
5679+
assert_eq!(sandbox.namespace, "test-ns");
5680+
}
5681+
5682+
#[test]
5683+
fn test_sandbox_from_object_unknown() {
5684+
let obj = make_dynamic_object("openshell-warm-pool-abc", None);
5685+
let result = sandbox_from_object("test-ns", obj);
5686+
assert!(result.is_err());
5687+
}
56285688
}

0 commit comments

Comments
 (0)