Merge pull request opensandbox-group#752 from Pangjiping/feat/server/windows-k8s

feat(server): windows profile for batchsandbox

Author: ninan-nn

server/opensandbox_server/services/k8s/agent_sandbox_provider.py

@@ -39,6 +39,7 @@
 )
 from opensandbox_server.services.k8s.volume_helper import apply_volumes_to_pod_spec
 from opensandbox_server.services.k8s.workload_provider import WorkloadProvider
+from opensandbox_server.services.k8s.windows_profile import is_windows_profile
 from opensandbox_server.services.runtime_resolver import SecureRuntimeResolver
 
 logger = logging.getLogger(__name__)
@@ -140,12 +141,11 @@ def create_workload(
         egress_mode: str = EGRESS_MODE_DNS,
     ) -> Dict[str, Any]:
         """Create an agent-sandbox Sandbox CRD workload."""
+        if is_windows_profile(platform):
+            raise ValueError("agent-sandbox does not support platform.os=windows.")
+
         if self.runtime_class:
-            logger.info(
-                "Using Kubernetes RuntimeClass '%s' for sandbox %s",
-                self.runtime_class,
-                sandbox_id,
-            )
+            logger.info(f"Using Kubernetes RuntimeClass '{self.runtime_class}' for sandbox {sandbox_id}")
 
         pod_spec = self._build_pod_spec(
             image_spec=image_spec,
@@ -260,7 +260,6 @@ def _build_pod_spec(
             entrypoint=entrypoint,
             env=env,
             resource_limits=resource_limits,
-            include_execd_volume=True,
             has_network_policy=network_policy is not None,
         )
 
@@ -363,7 +362,7 @@ def get_expiration(self, workload: Dict[str, Any]) -> Optional[datetime]:
         try:
             return datetime.fromisoformat(shutdown_time_str.replace("Z", "+00:00"))
         except (ValueError, TypeError) as e:
-            logger.warning("Invalid shutdownTime format: %s, error: %s", shutdown_time_str, e)
+            logger.warning(f"Invalid shutdownTime format: {shutdown_time_str}, error: {e}")
             return None
 
     def get_status(self, workload: Dict[str, Any]) -> Dict[str, Any]:
@@ -515,7 +514,7 @@ def get_endpoint_info(self, workload: Dict[str, Any], port: int, sandbox_id: str
                     if pod.status and pod.status.pod_ip and pod.status.phase == "Running":
                         return Endpoint(endpoint=f"{pod.status.pod_ip}:{port}")
             except Exception as e:
-                logger.warning("Failed to resolve pod endpoint: %s", e)
+                logger.warning(f"Failed to resolve pod endpoint: {e}")
 
         service_fqdn = status.get("serviceFQDN")
         if service_fqdn:

server/opensandbox_server/services/k8s/batchsandbox_provider.py

@@ -46,6 +46,12 @@
     _extract_platform_unschedulable_message_from_pod,
     _workload_platform_constraint_scope,
 )
+from opensandbox_server.services.k8s.windows_profile import (
+    apply_windows_profile_arch_selector,
+    apply_windows_profile_overrides,
+    is_windows_profile,
+    validate_windows_profile_resource_limits,
+)
 from opensandbox_server.services.k8s.volume_helper import apply_volumes_to_pod_spec
 from opensandbox_server.services.k8s.workload_provider import WorkloadProvider
 from opensandbox_server.services.runtime_resolver import SecureRuntimeResolver
@@ -67,7 +73,7 @@ def __init__(
         k8s_config = app_config.kubernetes if app_config else None
         template_file_path = k8s_config.batchsandbox_template_file if k8s_config else None
         if template_file_path:
-            logger.info("Using BatchSandbox template file: %s", template_file_path)
+            logger.info(f"Using BatchSandbox template file: {template_file_path}")
         self.execd_init_resources = k8s_config.execd_init_resources if k8s_config else None
 
         self.resolver = SecureRuntimeResolver(app_config) if app_config else None
@@ -113,13 +119,10 @@ def create_workload(
     ) -> Dict[str, Any]:
         """Create a BatchSandbox in template mode or pool mode."""
         extensions = extensions or {}
+        windows_profile = is_windows_profile(platform)
 
         if self.runtime_class:
-            logger.info(
-                "Using Kubernetes RuntimeClass '%s' for sandbox %s",
-                self.runtime_class,
-                sandbox_id,
-            )
+            logger.info(f"Using Kubernetes RuntimeClass '{self.runtime_class}' for sandbox {sandbox_id}")
 
         if extensions.get("poolRef"):
             if platform is not None:
@@ -145,6 +148,9 @@ def create_workload(
 
         extra_volumes, extra_mounts = self._extract_template_pod_extras()
 
+        if windows_profile:
+            validate_windows_profile_resource_limits(resource_limits)
+
         disable_ipv6_for_egress = (
             network_policy is not None
             and egress_image is not None
@@ -161,13 +167,11 @@ def create_workload(
             entrypoint=entrypoint,
             env=env,
             resource_limits=resource_limits,
-            include_execd_volume=True,
             has_network_policy=network_policy is not None,
         )
 
         containers = [_container_to_dict(main_container)]
-        
-        pod_spec: Dict[str, Any] = {
+        pod_spec = {
             "initContainers": [_container_to_dict(init_container)],
             "containers": containers,
             "volumes": [
@@ -177,8 +181,29 @@ def create_workload(
                 }
             ],
         }
-        self._apply_platform_node_selector(pod_spec, platform)
+        if windows_profile:
+            apply_windows_profile_overrides(
+                pod_spec=pod_spec,
+                entrypoint=entrypoint,
+                env=env,
+                resource_limits=resource_limits,
+                disable_ipv6_for_egress=disable_ipv6_for_egress,
+            )
+            template = self.template_manager.get_base_template()
+            template_spec = (
+                template.get("spec", {})
+                .get("template", {})
+                .get("spec", {})
+            )
+            apply_windows_profile_arch_selector(
+                pod_spec=pod_spec,
+                template_spec=template_spec if isinstance(template_spec, dict) else {},
+                platform=platform,
+            )
+        else:
+            self._apply_platform_node_selector(pod_spec, platform)
 
+        containers = pod_spec.get("containers", [])
         if self.runtime_class:
             pod_spec["runtimeClassName"] = self.runtime_class
 
@@ -222,7 +247,7 @@ def create_workload(
         else:
             batchsandbox["spec"]["expireTime"] = expires_at.isoformat()
         self._merge_pod_spec_extras(batchsandbox, extra_volumes, extra_mounts)
-        if platform is not None:
+        if platform is not None and not windows_profile:
             merged_pod_spec = batchsandbox.get("spec", {}).get("template", {}).get("spec", {})
             WorkloadProvider.ensure_platform_compatible_with_affinity(merged_pod_spec, platform)
 
@@ -245,9 +270,9 @@ def create_workload(
             )
             try:
                 self.k8s_client.create_secret(namespace=namespace, body=secret)
-                logger.info("Created imagePullSecret for sandbox %s", sandbox_id)
+                logger.info(f"Created imagePullSecret for sandbox {sandbox_id}")
             except Exception:
-                logger.warning("Failed to create imagePullSecret for sandbox %s, rolling back BatchSandbox", sandbox_id)
+                logger.warning(f"Failed to create imagePullSecret for sandbox {sandbox_id}, rolling back BatchSandbox")
                 try:
                     self.k8s_client.delete_custom_object(
                         group=self.group,
@@ -258,7 +283,7 @@ def create_workload(
                         grace_period_seconds=0,
                     )
                 except Exception as del_exc:
-                    logger.warning("Failed to rollback BatchSandbox %s: %s", sandbox_id, del_exc)
+                    logger.warning(f"Failed to rollback BatchSandbox {sandbox_id}: {del_exc}")
                 raise
 
         return {
@@ -502,7 +527,7 @@ def get_expiration(self, workload: Dict[str, Any]) -> Optional[datetime]:
         try:
             return datetime.fromisoformat(expire_time_str.replace('Z', '+00:00'))
         except (ValueError, TypeError) as e:
-            logger.warning("Invalid expireTime format: %s, error: %s", expire_time_str, e)
+            logger.warning(f"Invalid expireTime format: {expire_time_str}, error: {e}")
             return None
 
     def _parse_pod_ip(self, workload: Dict[str, Any]) -> Optional[str]:

server/opensandbox_server/services/k8s/client.py

@@ -113,7 +113,7 @@ def _get_informer(self, group: str, version: str, plural: str, namespace: str) -
                 try:
                     informer.start()
                 except Exception as exc:  # pragma: no cover - defensive
-                    logger.warning("Failed to start informer for %s/%s: %s", plural, namespace, exc)
+                    logger.warning(f"Failed to start informer for {plural}/{namespace}: {exc}")
                     self._informers.pop(key, None)
                     return None
         return informer

server/opensandbox_server/services/k8s/informer.py

@@ -142,12 +142,12 @@ def _run(self) -> None:
                     self._resource_version = None
                     self._has_synced = False
                 else:
-                    logger.warning("Informer watch error: %s", exc, exc_info=True)
+                    logger.warning(f"Informer watch error: {exc}", exc_info=True)
                     self._has_synced = False
                     self._stop_event.wait(min(backoff, 30.0))
                     backoff = min(backoff * 2, 30.0)
             except Exception as exc:  # pragma: no cover - defensive
-                logger.warning("Unexpected informer error: %s", exc, exc_info=True)
+                logger.warning(f"Unexpected informer error: {exc}", exc_info=True)
                 self._has_synced = False
                 self._stop_event.wait(min(backoff, 30.0))
                 backoff = min(backoff * 2, 30.0)

server/opensandbox_server/services/k8s/kubernetes_service.py

@@ -316,14 +316,13 @@ def _ensure_pvc_volumes(self, volumes: list) -> None:
             except ApiException as e:
                 if e.status == 403:
                     logger.warning(
-                        "No RBAC permission to read PVC '%s', skipping auto-create. "
-                        "Grant 'get' and 'create' on 'persistentvolumeclaims' to enable.",
-                        claim_name,
+                        f"No RBAC permission to read PVC '{claim_name}', skipping auto-create. "
+                        "Grant 'get' and 'create' on 'persistentvolumeclaims' to enable."
                     )
                     return  # Skip all remaining PVCs — same SA, same permissions
                 raise
             if existing is not None:
-                logger.debug("PVC '%s' already exists in namespace '%s'", claim_name, self.namespace)
+                logger.debug(f"PVC '{claim_name}' already exists in namespace '{self.namespace}'")
                 continue
 
             storage = vol.pvc.storage or default_size
@@ -346,18 +345,17 @@ def _ensure_pvc_volumes(self, volumes: list) -> None:
             try:
                 self.k8s_client.create_pvc(self.namespace, pvc_body)
                 logger.info(
-                    "Auto-created PVC '%s' (size=%s, class=%s) in namespace '%s'",
-                    claim_name, storage, storage_class or "<default>", self.namespace,
+                    f"Auto-created PVC '{claim_name}' (size={storage}, class={storage_class or '<default>'}) "
+                    f"in namespace '{self.namespace}'"
                 )
             except ApiException as e:
                 if e.status == 409:
                     # Race condition: another request created it between our check and create
-                    logger.info("PVC '%s' was created concurrently, proceeding", claim_name)
+                    logger.info(f"PVC '{claim_name}' was created concurrently, proceeding")
                 elif e.status == 403:
                     logger.warning(
-                        "No RBAC permission to create PVC '%s', skipping. "
-                        "The PVC must be pre-created or RBAC must be updated.",
-                        claim_name,
+                        f"No RBAC permission to create PVC '{claim_name}', skipping. "
+                        "The PVC must be pre-created or RBAC must be updated."
                     )
                 elif e.status in (400, 422):
                     # Invalid PVC spec from user-provided hints
@@ -371,7 +369,7 @@ def _ensure_pvc_volumes(self, volumes: list) -> None:
                         },
                     ) from e
                 else:
-                    logger.error("Failed to create PVC '%s': %s", claim_name, e)
+                    logger.error(f"Failed to create PVC '{claim_name}': {e}")
                     raise HTTPException(
                         status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                         detail={

server/opensandbox_server/services/k8s/pool_service.py

@@ -119,7 +119,7 @@ def create_pool(self, request: CreatePoolRequest) -> PoolResponse:
                 plural=_PLURAL,
                 body=manifest,
             )
-            logger.info("Created pool: name=%s, namespace=%s", request.name, self._namespace)
+            logger.info(f"Created pool: name={request.name}, namespace={self._namespace}")
             return self._pool_from_raw(created)
 
         except ApiException as e:
@@ -131,7 +131,7 @@ def create_pool(self, request: CreatePoolRequest) -> PoolResponse:
                         "message": f"Pool '{request.name}' already exists.",
                     },
                 ) from e
-            logger.error("Kubernetes API error creating pool %s: %s", request.name, e)
+            logger.error(f"Kubernetes API error creating pool {request.name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -140,7 +140,7 @@ def create_pool(self, request: CreatePoolRequest) -> PoolResponse:
                 },
             ) from e
         except Exception as e:
-            logger.error("Unexpected error creating pool %s: %s", request.name, e)
+            logger.error(f"Unexpected error creating pool {request.name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -170,7 +170,7 @@ def get_pool(self, pool_name: str) -> PoolResponse:
                         "message": f"Pool '{pool_name}' not found.",
                     },
                 ) from e
-            logger.error("Kubernetes API error getting pool %s: %s", pool_name, e)
+            logger.error(f"Kubernetes API error getting pool {pool_name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -181,7 +181,7 @@ def get_pool(self, pool_name: str) -> PoolResponse:
         except HTTPException:
             raise
         except Exception as e:
-            logger.error("Unexpected error getting pool %s: %s", pool_name, e)
+            logger.error(f"Unexpected error getting pool {pool_name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -209,7 +209,7 @@ def list_pools(self) -> ListPoolsResponse:
                 # CRD not installed — return empty list gracefully
                 logger.warning("Pool CRD not found (404); returning empty list.")
                 return ListPoolsResponse(items=[])
-            logger.error("Kubernetes API error listing pools: %s", e)
+            logger.error(f"Kubernetes API error listing pools: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -218,7 +218,7 @@ def list_pools(self) -> ListPoolsResponse:
                 },
             ) from e
         except Exception as e:
-            logger.error("Unexpected error listing pools: %s", e)
+            logger.error(f"Unexpected error listing pools: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -249,7 +249,7 @@ def update_pool(self, pool_name: str, request: UpdatePoolRequest) -> PoolRespons
                 name=pool_name,
                 body=patch_body,
             )
-            logger.info("Updated pool capacity: name=%s", pool_name)
+            logger.info(f"Updated pool capacity: name={pool_name}")
             return self._pool_from_raw(updated)
 
         except ApiException as e:
@@ -261,7 +261,7 @@ def update_pool(self, pool_name: str, request: UpdatePoolRequest) -> PoolRespons
                         "message": f"Pool '{pool_name}' not found.",
                     },
                 ) from e
-            logger.error("Kubernetes API error updating pool %s: %s", pool_name, e)
+            logger.error(f"Kubernetes API error updating pool {pool_name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -272,7 +272,7 @@ def update_pool(self, pool_name: str, request: UpdatePoolRequest) -> PoolRespons
         except HTTPException:
             raise
         except Exception as e:
-            logger.error("Unexpected error updating pool %s: %s", pool_name, e)
+            logger.error(f"Unexpected error updating pool {pool_name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -292,7 +292,7 @@ def delete_pool(self, pool_name: str) -> None:
                 name=pool_name,
                 grace_period_seconds=0,
             )
-            logger.info("Deleted pool: name=%s, namespace=%s", pool_name, self._namespace)
+            logger.info(f"Deleted pool: name={pool_name}, namespace={self._namespace}")
 
         except ApiException as e:
             if e.status == 404:
@@ -303,7 +303,7 @@ def delete_pool(self, pool_name: str) -> None:
                         "message": f"Pool '{pool_name}' not found.",
                     },
                 ) from e
-            logger.error("Kubernetes API error deleting pool %s: %s", pool_name, e)
+            logger.error(f"Kubernetes API error deleting pool {pool_name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={
@@ -314,7 +314,7 @@ def delete_pool(self, pool_name: str) -> None:
         except HTTPException:
             raise
         except Exception as e:
-            logger.error("Unexpected error deleting pool %s: %s", pool_name, e)
+            logger.error(f"Unexpected error deleting pool {pool_name}: {e}")
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail={