fix(snapshot): harden Kubernetes public snapshot runtime

Author: fengcone

kubernetes/charts/opensandbox-server/values.yaml

@@ -91,10 +91,10 @@ configToml: |
   informer_enabled = true
   informer_resync_seconds = 300
   informer_watch_timeout_seconds = 60
+  snapshot_create_timeout_seconds = 900
   workload_provider = "batchsandbox"
   batchsandbox_template_file = "/etc/opensandbox/example.batchsandbox-template.yaml"
   
   [egress]
   image = "sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/egress:v1.0.10"
   mode = "dns+nft"
-

server/configuration.md

@@ -120,6 +120,7 @@ If `runtime.type = "kubernetes"` and the `[kubernetes]` table is absent, the ser
 | `image_pull_policy` | string \| omitted | `"IfNotPresent"` | Image pull policy for the BatchSandbox main container. Values: **`Always`**, **`IfNotPresent`**, **`Never`**. |
 | `sandbox_create_timeout_seconds` | integer | `60` | Max time to wait for a new sandbox to become ready (e.g. IP assigned), in seconds. |
 | `sandbox_create_poll_interval_seconds` | float | `1.0` | Poll interval while waiting for readiness. |
+| `snapshot_create_timeout_seconds` | integer | `900` | Max time to wait for a Kubernetes public snapshot to become ready, in seconds. Set this greater than the controller snapshot `commitJobTimeout` / `--commit-job-timeout`. |
 | `informer_enabled` | boolean | `true` | **[Beta]** Use informer/watch cache for reads to reduce API load. |
 | `informer_resync_seconds` | integer | `300` | **[Beta]** Full resync period for the informer cache. |
 | `informer_watch_timeout_seconds` | integer | `60` | **[Beta]** Watch stream restart interval. |

server/opensandbox_server/config.py

@@ -558,6 +558,14 @@ class KubernetesRuntimeConfig(BaseModel):
         gt=0,
         description="Polling interval in seconds when waiting for a sandbox to become ready after creation.",
     )
+    snapshot_create_timeout_seconds: int = Field(
+        default=15 * 60,
+        ge=1,
+        description=(
+            "Timeout in seconds to wait for a Kubernetes public snapshot to become ready. "
+            "Set this greater than the controller snapshot commit-job-timeout."
+        ),
+    )
     execd_init_resources: Optional["ExecdInitResources"] = Field(
         default=None,
         description=(

server/opensandbox_server/examples/example.config.k8s.toml

@@ -64,6 +64,10 @@ workload_provider = "batchsandbox"
 # Values: "Always", "IfNotPresent", "Never".
 image_pull_policy = "IfNotPresent"
 
+# Public snapshot wait timeout. Keep this greater than the Kubernetes
+# controller snapshot commit-job-timeout.
+snapshot_create_timeout_seconds = 900
+
 # Path to the BatchSandbox template file
 batchsandbox_template_file = "~/batchsandbox-template.yaml"
 

server/opensandbox_server/examples/example.config.k8s.zh.toml

@@ -64,6 +64,10 @@ workload_provider = "batchsandbox"
 # 可选值："Always"、"IfNotPresent"、"Never"。
 image_pull_policy = "IfNotPresent"
 
+# public snapshot 等待超时时间。应大于 Kubernetes controller 的
+# snapshot commit-job-timeout。
+snapshot_create_timeout_seconds = 900
+
 # Path to the BatchSandbox template file
 # Replace with your path
 batchsandbox_template_file = "~/batchsandbox-template.yaml"

server/opensandbox_server/services/k8s/snapshot_runtime.py

@@ -88,6 +88,7 @@ def create_snapshot(
     ) -> Optional[SnapshotRuntimeStatus]:
         snapshot_name = build_public_snapshot_name(snapshot_id)
         body = self._build_snapshot_body(snapshot_id, sandbox_id, snapshot_name)
+        should_validate_existing_source = False
 
         try:
             self._k8s_client.create_custom_object(
@@ -105,6 +106,7 @@ def create_snapshot(
                     message=f"Failed to create Kubernetes SandboxSnapshot {snapshot_name}: {exc}",
                 )
             logger.info("Kubernetes SandboxSnapshot %s already exists; continuing", snapshot_name)
+            should_validate_existing_source = True
         except Exception as exc:  # noqa: BLE001
             logger.exception("Failed to create Kubernetes SandboxSnapshot %s: %s", snapshot_name, exc)
             return SnapshotRuntimeStatus(
@@ -113,10 +115,19 @@ def create_snapshot(
                 message=f"Failed to create Kubernetes SandboxSnapshot {snapshot_name}: {exc}",
             )
 
-        current = self._get_snapshot_cr(snapshot_name)
-        conflict = self._validate_existing_source(current, sandbox_id)
-        if conflict is not None:
-            return conflict
+        if should_validate_existing_source:
+            try:
+                current = self._get_snapshot_cr(snapshot_name)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning(
+                    "Failed to inspect existing Kubernetes SandboxSnapshot %s after create conflict: %s",
+                    snapshot_name,
+                    exc,
+                )
+            else:
+                conflict = self._validate_existing_source(current, sandbox_id)
+                if conflict is not None:
+                    return conflict
 
         return self._wait_for_terminal_snapshot(snapshot_id)
 
@@ -144,10 +155,10 @@ def inspect_snapshot(self, snapshot_id: str, image: Optional[str] = None) -> Sna
         try:
             snapshot = self._get_snapshot_cr(snapshot_name)
         except Exception as exc:  # noqa: BLE001
-            logger.exception("Failed to inspect Kubernetes SandboxSnapshot %s: %s", snapshot_name, exc)
+            logger.warning("Failed to inspect Kubernetes SandboxSnapshot %s: %s", snapshot_name, exc)
             return SnapshotRuntimeStatus(
-                state=SnapshotState.FAILED,
-                reason="snapshot_recovery_inspect_failed",
+                state=SnapshotState.CREATING,
+                reason="snapshot_runtime_inspect_failed",
                 message=f"Failed to inspect Kubernetes SandboxSnapshot {snapshot_name}: {exc}",
             )
 

server/opensandbox_server/services/snapshot_runtime_factory.py

@@ -20,7 +20,7 @@
 
 from typing import Optional
 
-from opensandbox_server.config import AppConfig, get_config
+from opensandbox_server.config import AppConfig, KubernetesRuntimeConfig, get_config
 from opensandbox_server.services.snapshot_runtime import SnapshotRuntime
 
 
@@ -44,14 +44,16 @@ def create_snapshot_runtime(
         from opensandbox_server.services.k8s.client import K8sClient
         from opensandbox_server.services.k8s.snapshot_runtime import KubernetesSnapshotRuntime
 
-        kubernetes_config = getattr(active_config, "kubernetes", None)
+        kubernetes_config = getattr(active_config, "kubernetes", None) or KubernetesRuntimeConfig()
         if k8s_client is None:
-            if kubernetes_config is None:
-                raise ValueError("kubernetes config is required when runtime.type = 'kubernetes'.")
             k8s_client = K8sClient(kubernetes_config)
 
-        namespace = getattr(kubernetes_config, "namespace", None) or "default"
-        return KubernetesSnapshotRuntime(k8s_client, namespace=namespace)
+        namespace = kubernetes_config.namespace or "default"
+        return KubernetesSnapshotRuntime(
+            k8s_client,
+            namespace=namespace,
+            wait_timeout_seconds=kubernetes_config.snapshot_create_timeout_seconds,
+        )
 
     raise ValueError(f"Unsupported snapshot runtime type: {runtime_type}")
 

server/tests/test_k8s_snapshot_runtime.py

@@ -40,9 +40,9 @@ def __init__(self) -> None:
     def create_custom_object(self, *, group: str, version: str, namespace: str, plural: str, body: dict):
         self.created.append(deepcopy(body))
         name = body["metadata"]["name"]
-        stored = deepcopy(body)
         if name in self.objects:
-            stored = self.objects[name]
+            raise ApiException(status=409, reason="Already Exists")
+        stored = deepcopy(body)
         self.objects[name] = stored
         return stored
 
@@ -57,6 +57,44 @@ def delete_custom_object(self, *, group: str, version: str, namespace: str, plur
         del self.objects[name]
 
 
+class TransientGetK8sClient(FakeK8sClient):
+    def __init__(self, *, failures: int) -> None:
+        super().__init__()
+        self.failures = failures
+
+    def get_custom_object(self, *, group: str, version: str, namespace: str, plural: str, name: str):
+        if self.failures > 0:
+            self.failures -= 1
+            raise ApiException(status=500, reason="temporary apiserver error")
+        return super().get_custom_object(
+            group=group,
+            version=version,
+            namespace=namespace,
+            plural=plural,
+            name=name,
+        )
+
+
+class TransientThenReadyK8sClient(TransientGetK8sClient):
+    def get_custom_object(self, *, group: str, version: str, namespace: str, plural: str, name: str):
+        obj = super().get_custom_object(
+            group=group,
+            version=version,
+            namespace=namespace,
+            plural=plural,
+            name=name,
+        )
+        if obj is not None:
+            obj["status"] = {
+                "phase": "Succeed",
+                "containers": [
+                    {"containerName": "sandbox", "imageUri": "registry/sandbox:snap"},
+                ],
+            }
+            self.objects[name] = deepcopy(obj)
+        return obj
+
+
 def _snapshot_cr(*, phase: str, containers: list[dict] | None = None, sandbox_id: str = SANDBOX_ID) -> dict:
     name = build_public_snapshot_name(SNAPSHOT_ID)
     return {
@@ -160,6 +198,32 @@ def test_inspect_snapshot_maps_failed_condition() -> None:
     assert status.message == "commit job failed"
 
 
+def test_inspect_snapshot_keeps_transient_read_error_creating() -> None:
+    k8s_client = TransientGetK8sClient(failures=1)
+    runtime = KubernetesSnapshotRuntime(k8s_client, namespace="default")
+
+    status = runtime.inspect_snapshot(SNAPSHOT_ID)
+
+    assert status.state == SnapshotState.CREATING
+    assert status.reason == "snapshot_runtime_inspect_failed"
+    assert "temporary apiserver error" in (status.message or "")
+
+
+def test_create_snapshot_retries_transient_inspect_error_until_controller_ready() -> None:
+    k8s_client = TransientThenReadyK8sClient(failures=1)
+    runtime = KubernetesSnapshotRuntime(
+        k8s_client,
+        namespace="default",
+        wait_timeout_seconds=1,
+        poll_interval_seconds=0,
+    )
+
+    status = runtime.create_snapshot(SNAPSHOT_ID, SANDBOX_ID)
+
+    assert status.state == SnapshotState.READY
+    assert status.image == "registry/sandbox:snap"
+
+
 def test_delete_snapshot_deletes_cr_and_ignores_missing_cr() -> None:
     k8s_client = FakeK8sClient()
     runtime = KubernetesSnapshotRuntime(k8s_client, namespace="default")

server/tests/test_routes_snapshots.py

@@ -207,7 +207,7 @@ def test_snapshot_routes_can_use_persisted_service(
     class StubSandboxService:
         @staticmethod
         def get_sandbox(sandbox_id: str):
-            return {"id": sandbox_id}
+            return {"id": sandbox_id, "status": {"state": "Running"}}
 
     class StubSnapshotRuntime:
         @staticmethod
@@ -254,7 +254,7 @@ def test_create_snapshot_returns_501_when_runtime_is_not_supported(
     class StubSandboxService:
         @staticmethod
         def get_sandbox(sandbox_id: str):
-            return {"id": sandbox_id}
+            return {"id": sandbox_id, "status": {"state": "Running"}}
 
     service = PersistedSnapshotService(
         SQLiteSnapshotRepository(tmp_path / "snapshots.db"),

server/tests/test_snapshot_runtime_factory.py

@@ -14,17 +14,16 @@
 
 from __future__ import annotations
 
-from types import SimpleNamespace
-
 import pytest
 
+from opensandbox_server.config import AppConfig, KubernetesRuntimeConfig, RuntimeConfig
 from opensandbox_server.services.docker.snapshot_runtime import DockerSnapshotRuntime
 from opensandbox_server.services.k8s.snapshot_runtime import KubernetesSnapshotRuntime
 from opensandbox_server.services.snapshot_runtime_factory import create_snapshot_runtime
 
 
 def test_create_snapshot_runtime_selects_docker_runtime() -> None:
-    config = SimpleNamespace(runtime=SimpleNamespace(type="docker"))
+    config = AppConfig(runtime=RuntimeConfig(type="docker", execd_image="opensandbox/execd:test"))
     docker_client = object()
 
     runtime = create_snapshot_runtime(config, docker_client=docker_client)
@@ -33,19 +32,35 @@ def test_create_snapshot_runtime_selects_docker_runtime() -> None:
 
 
 def test_create_snapshot_runtime_selects_kubernetes_runtime() -> None:
-    config = SimpleNamespace(
-        runtime=SimpleNamespace(type="kubernetes"),
-        kubernetes=SimpleNamespace(namespace="default"),
+    config = AppConfig(
+        runtime=RuntimeConfig(type="kubernetes", execd_image="opensandbox/execd:test"),
+        kubernetes=KubernetesRuntimeConfig(namespace="default"),
+    )
+    k8s_client = object()
+
+    runtime = create_snapshot_runtime(config, k8s_client=k8s_client)
+
+    assert isinstance(runtime, KubernetesSnapshotRuntime)
+
+
+def test_create_snapshot_runtime_uses_kubernetes_snapshot_create_timeout() -> None:
+    config = AppConfig(
+        runtime=RuntimeConfig(type="kubernetes", execd_image="opensandbox/execd:test"),
+        kubernetes=KubernetesRuntimeConfig(
+            namespace="default",
+            snapshot_create_timeout_seconds=1234,
+        ),
     )
     k8s_client = object()
 
     runtime = create_snapshot_runtime(config, k8s_client=k8s_client)
 
     assert isinstance(runtime, KubernetesSnapshotRuntime)
+    assert runtime._wait_timeout_seconds == 1234
 
 
 def test_create_snapshot_runtime_requires_docker_client_for_docker() -> None:
-    config = SimpleNamespace(runtime=SimpleNamespace(type="docker"))
+    config = AppConfig(runtime=RuntimeConfig(type="docker", execd_image="opensandbox/execd:test"))
 
     with pytest.raises(ValueError, match="docker_client is required"):
         create_snapshot_runtime(config)