From 16ddd6d4addd14dd00dd73a6b3a66de5c0aea06e Mon Sep 17 00:00:00 2001 From: Admin Bot Date: Mon, 28 Sep 2026 19:53:37 +0800 Subject: [PATCH] =?UTF-8?q?feat(sessions):=20sandbox=20resource=20?= =?UTF-8?q?=E2=80=94=20per-session=20k8s=20Pod+PVC+Service=20(=C2=A74.15.4?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New X-Resource: sandbox on X-Service: sessions POST ?session_id= → CreateOrResume 1. Ensure PVC workspace-session- (Longhorn 1Gi) 2. If pod already Running → return DNS (resume path, idempotent) 3. Ensure pi-session-sandbox NetworkPolicy (shared, created once) 4. Create Pod pi-session- + Service pi-session-:8787 5. Wait up to 30s for Pod Ready 6. Fire-and-forget: merge pod_name/pvc_name/svc_dns/last_active_at into session memory_entity summary DELETE ?session_id= → delete Pod + Service + PVC Idle reaper goroutine (5min interval, configurable via SANDBOX_REAPER_INTERVAL_MINUTES): lists pi-session-* pods, deletes Succeeded/Failed ones, preserves PVCs for resume. Security: runAsNonRoot/1000, readOnlyRootFilesystem, drop ALL caps, RuntimeDefault seccomp, automountServiceAccountToken:false. Resources: 500m/512Mi req, 1/1Gi limit. NetworkPolicy: egress → DNS + api:8080 only; ingress ← poimen-frontend. RBAC: Role session-sandbox-manager + RoleBinding in poimen ns. 7 unit tests — all pass (mock k8s server). Image: forgejo.riotpiao.com/riotpiao-poimen/poimen-pi-session-runtime:latest --- internal/serviceadapter/session_adapter.go | 16 + internal/sessions/handler.go | 85 +++- internal/sessions/sandbox.go | 492 +++++++++++++++++++++ internal/sessions/sandbox_test.go | 271 ++++++++++++ k8s/rbac.yaml | 34 ++ 5 files changed, 897 insertions(+), 1 deletion(-) create mode 100644 internal/sessions/sandbox.go create mode 100644 internal/sessions/sandbox_test.go diff --git a/internal/serviceadapter/session_adapter.go b/internal/serviceadapter/session_adapter.go index e60ac49..b408a11 100644 --- a/internal/serviceadapter/session_adapter.go +++ b/internal/serviceadapter/session_adapter.go @@ -102,6 +102,22 @@ func GetSessionSpec() *Spec { }, }, }, + // sandbox — per-session k8s Pod+PVC+Service lifecycle (§4.15.4) + { + Name: "sandbox", + Methods: []Method{ + { + Verb: "POST", + UpstreamPath: "/v1/sessions/sandbox", + Auth: &Auth{Required: true, Capability: "sessions:write"}, + }, + { + Verb: "DELETE", + UpstreamPath: "/v1/sessions/sandbox", + Auth: &Auth{Required: true, Capability: "sessions:write"}, + }, + }, + }, }, } } diff --git a/internal/sessions/handler.go b/internal/sessions/handler.go index b51d346..e81044a 100644 --- a/internal/sessions/handler.go +++ b/internal/sessions/handler.go @@ -12,9 +12,11 @@ package sessions import ( "bytes" + "context" "crypto/sha256" "encoding/json" "fmt" + "log" "net/http" "os" "time" @@ -33,6 +35,7 @@ const ( type Handler struct { graphqlURL string client *http.Client + sandbox *SandboxManager } // NewHandler creates a Handler reading upstream URL from env var. @@ -42,10 +45,33 @@ func NewHandler() *Handler { if gql == "" { gql = defaultGraphQLURL } - return &Handler{ + h := &Handler{ graphqlURL: gql, client: &http.Client{Timeout: 30 * time.Second, Transport: tracing.NewTransport(nil)}, + sandbox: NewSandboxManager(), } + h.startIdleReaper() + return h +} + +// startIdleReaper launches a background goroutine that deletes exited pi-session +// pods (bridge.js idle-timeout triggered) while preserving their PVCs. +func (h *Handler) startIdleReaper() { + intervalStr := os.Getenv("SANDBOX_REAPER_INTERVAL_MINUTES") + interval := defaultReaperInterval + if intervalStr != "" { + var min int + if n, _ := fmt.Sscanf(intervalStr, "%d", &min); n == 1 && min > 0 { + interval = time.Duration(min) * time.Minute + } + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for range ticker.C { + h.sandbox.ReapIdlePods(context.Background()) + } + }() } // ServeHTTP routes by X-Upstream-Path set by the dispatcher. @@ -66,6 +92,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.handleListEpisodes(w, r) case "/v1/sessions/episodes/get": h.handleGetEpisode(w, r) + case "/v1/sessions/sandbox": + h.handleSandbox(w, r) default: writeJSON(w, http.StatusNotFound, map[string]string{"error": "sessions endpoint not found"}) } @@ -498,6 +526,61 @@ func (h *Handler) gql(r *http.Request, query string, variables map[string]any) ( return result, nil } +// handleSandbox dispatches POST (create/resume) and DELETE (teardown). +// Both take ?session_id= as a query param. +func (h *Handler) handleSandbox(w http.ResponseWriter, r *http.Request) { + sessionID := r.URL.Query().Get("session_id") + if sessionID == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"}) + return + } + + switch r.Method { + case http.MethodPost: + h.handleSandboxCreate(w, r, sessionID) + case http.MethodDelete: + h.handleSandboxDelete(w, r, sessionID) + default: + writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"}) + } +} + +// POST ?session_id= — create or resume pi-session sandbox. +func (h *Handler) handleSandboxCreate(w http.ResponseWriter, r *http.Request, sessionID string) { + result, err := h.sandbox.CreateOrResume(r.Context(), sessionID) + if err != nil { + log.Printf("sandbox create/resume session=%s: %v", sessionID, err) + writeJSON(w, http.StatusBadGateway, map[string]string{"error": "sandbox error: " + err.Error()}) + return + } + + // Update session memory_entity with pod/pvc metadata (fire-and-forget — don't fail the response) + go func() { + current, _ := h.fetchEntitySummary(r, "session-"+sessionID, sessionID, "SESSION") + current["pod_name"] = result.PodName + current["pvc_name"] = result.PVCName + current["svc_dns"] = result.SvcDNS + current["last_active_at"] = time.Now().UTC().Format(time.RFC3339) + if b, err := json.Marshal(current); err == nil { + h.upsertEntity(r, "session-"+sessionID, sessionID, "SESSION", string(b)) //nolint:errcheck + } + }() + + log.Printf("sandbox: session=%s pod=%s resumed=%v dns=%s", sessionID, result.PodName, result.Resumed, result.SvcDNS) + writeJSON(w, http.StatusOK, result) +} + +// DELETE ?session_id= — delete Pod + Service + PVC. +func (h *Handler) handleSandboxDelete(w http.ResponseWriter, r *http.Request, sessionID string) { + if err := h.sandbox.Delete(r.Context(), sessionID); err != nil { + log.Printf("sandbox delete session=%s: %v", sessionID, err) + writeJSON(w, http.StatusBadGateway, map[string]string{"error": "sandbox delete error: " + err.Error()}) + return + } + log.Printf("sandbox: deleted pod+svc+pvc for session=%s", sessionID) + writeJSON(w, http.StatusOK, map[string]string{"session_id": sessionID, "status": "deleted"}) +} + // EpisodeID computes the deterministic episode ID per §4.14 Q6: // sha256(session_id + milestone_id + attempt). // Exported so callers can verify the value the server would produce. diff --git a/internal/sessions/sandbox.go b/internal/sessions/sandbox.go new file mode 100644 index 0000000..3368820 --- /dev/null +++ b/internal/sessions/sandbox.go @@ -0,0 +1,492 @@ +// Package sessions — SandboxManager provisions per-session k8s resources: +// +// PVC workspace-session- Longhorn 1Gi +// Pod pi-session- single container, /workspace <- PVC +// Svc pi-session- ClusterIP :8787 +// NetPol pi-session-sandbox shared for all pi-session pods (created once) +// +// All k8s API calls use raw HTTP + in-cluster SA token (no client-go). +// Same pattern as internal/webhook/promoter.go. +package sessions + +import ( + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "os" + "strings" + "time" +) + +const ( + sandboxNamespace = "poimen" + sandboxImage = "forgejo.riotpiao.com/riotpiao-poimen/poimen-pi-session-runtime:latest" + sandboxPort = 8787 + defaultPVCSize = "1Gi" + defaultStorageClass = "longhorn" + defaultReadyTimeout = 30 * time.Second + defaultReaperInterval = 5 * time.Minute + sandboxSATokenPath = "/var/run/secrets/kubernetes.io/serviceaccount/token" + sandboxCACertPath = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt" +) + +// SandboxManager manages the k8s lifecycle of pi-session Pods, PVCs, Services. +type SandboxManager struct { + k8sHost string + k8sClient *http.Client + pvcSize string + readyTimeout time.Duration +} + +// SandboxResult is returned by CreateOrResume. +type SandboxResult struct { + SessionID string `json:"session_id"` + PodName string `json:"pod_name"` + PVCName string `json:"pvc_name"` + SvcDNS string `json:"svc_dns"` + Resumed bool `json:"resumed"` // true = pod already existed +} + +// NewSandboxManager creates a SandboxManager from env + in-cluster SA files. +func NewSandboxManager() *SandboxManager { + host := os.Getenv("K8S_API_HOST") + if host == "" { + host = "https://kubernetes.default.svc" + } + pvcSize := os.Getenv("SANDBOX_PVC_SIZE") + if pvcSize == "" { + pvcSize = defaultPVCSize + } + caCert, _ := os.ReadFile(sandboxCACertPath) + return &SandboxManager{ + k8sHost: host, + k8sClient: buildSandboxK8sClient(caCert), + pvcSize: pvcSize, + readyTimeout: defaultReadyTimeout, + } +} + +func buildSandboxK8sClient(caCert []byte) *http.Client { + transport := &http.Transport{} + if len(caCert) > 0 { + pool := x509.NewCertPool() + pool.AppendCertsFromPEM(caCert) + transport.TLSClientConfig = &tls.Config{RootCAs: pool} + } else { + log.Printf("sandbox: WARNING no SA CA cert at %s — using InsecureSkipVerify", sandboxCACertPath) + transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec + } + return &http.Client{Timeout: 30 * time.Second, Transport: transport} +} + +func sandboxSAToken() string { + b, err := os.ReadFile(sandboxSATokenPath) + if err != nil { + log.Printf("sandbox: WARNING failed to read SA token %s: %v", sandboxSATokenPath, err) + return "" + } + return strings.TrimSpace(string(b)) +} + +func pvcName(sessionID string) string { return "workspace-session-" + sessionID } +func podName(sessionID string) string { return "pi-session-" + sessionID } +func svcName(sessionID string) string { return "pi-session-" + sessionID } +func svcDNS(sessionID string) string { + return fmt.Sprintf("%s.%s.svc.cluster.local:%d", svcName(sessionID), sandboxNamespace, sandboxPort) +} + +// ── public API ──────────────────────────────────────────────────────────── + +// CreateOrResume provisions or resumes a sandbox for sessionID. +// Idempotent: if the pod is already Running, returns its DNS immediately. +func (s *SandboxManager) CreateOrResume(ctx context.Context, sessionID string) (*SandboxResult, error) { + res := &SandboxResult{ + SessionID: sessionID, + PodName: podName(sessionID), + PVCName: pvcName(sessionID), + SvcDNS: svcDNS(sessionID), + } + + // 1. Ensure PVC exists + if err := s.ensurePVC(ctx, sessionID); err != nil { + return nil, fmt.Errorf("ensure PVC: %w", err) + } + + // 2. Check if pod already running (resume case) + phase, err := s.getPodPhase(ctx, sessionID) + if err != nil { + return nil, fmt.Errorf("check pod: %w", err) + } + if phase == "Running" { + res.Resumed = true + return res, nil + } + + // 3. Ensure NetworkPolicy for all pi-session pods exists + if err := s.ensureNetworkPolicy(ctx); err != nil { + log.Printf("sandbox: WARNING failed to ensure network policy: %v", err) + } + + // 4. Create Pod + Service + if err := s.createPod(ctx, sessionID); err != nil { + return nil, fmt.Errorf("create pod: %w", err) + } + if err := s.ensureService(ctx, sessionID); err != nil { + return nil, fmt.Errorf("ensure service: %w", err) + } + + // 5. Wait for Ready + if err := s.waitPodReady(ctx, sessionID); err != nil { + return nil, fmt.Errorf("wait ready: %w", err) + } + + return res, nil +} + +// Delete removes the Pod, Service, and PVC for sessionID. +func (s *SandboxManager) Delete(ctx context.Context, sessionID string) error { + var errs []string + for _, fn := range []func(context.Context, string) error{ + s.deletePod, s.deleteService, s.deletePVC, + } { + if err := fn(ctx, sessionID); err != nil && !isNotFound(err) { + errs = append(errs, err.Error()) + } + } + if len(errs) > 0 { + return fmt.Errorf("delete sandbox: %s", strings.Join(errs, "; ")) + } + return nil +} + +// ReapIdlePods lists all pi-session-* pods and deletes those that have exited +// (Succeeded/Failed) — bridge.js self-exits after idle timeout. +// The PVC is preserved so the next CreateOrResume reattaches the same workspace. +func (s *SandboxManager) ReapIdlePods(ctx context.Context) { + pods, err := s.listSandboxPods(ctx) + if err != nil { + log.Printf("sandbox reaper: list pods: %v", err) + return + } + for _, podAny := range pods { + pod, _ := podAny.(map[string]any) + phase, _ := pod["status"].(map[string]any)["phase"].(string) + name, _ := pod["metadata"].(map[string]any)["name"].(string) + if phase == "Succeeded" || phase == "Failed" { + log.Printf("sandbox reaper: deleting exited pod %s (phase=%s)", name, phase) + if err := s.k8sDelete(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/pods/"+name); err != nil && !isNotFound(err) { + log.Printf("sandbox reaper: delete pod %s: %v", name, err) + } + } + } +} + +// ── k8s resource helpers ───────────────────────────────────────────────── + +func (s *SandboxManager) ensurePVC(ctx context.Context, sessionID string) error { + name := pvcName(sessionID) + _, err := s.k8sGet(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/persistentvolumeclaims/"+name) + if err == nil { + return nil // already exists + } + if !isNotFound(err) { + return err + } + // Create + body := map[string]any{ + "apiVersion": "v1", + "kind": "PersistentVolumeClaim", + "metadata": map[string]any{ + "name": name, "namespace": sandboxNamespace, + "labels": map[string]string{"app": "pi-session", "session-id": sessionID}, + }, + "spec": map[string]any{ + "accessModes": []string{"ReadWriteOnce"}, + "storageClassName": defaultStorageClass, + "resources": map[string]any{ + "requests": map[string]string{"storage": s.pvcSize}, + }, + }, + } + return s.k8sCreate(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/persistentvolumeclaims", body) +} + +func (s *SandboxManager) getPodPhase(ctx context.Context, sessionID string) (string, error) { + pod, err := s.k8sGet(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/pods/"+podName(sessionID)) + if isNotFound(err) { + return "", nil + } + if err != nil { + return "", err + } + phase, _ := pod["status"].(map[string]any)["phase"].(string) + return phase, nil +} + +func (s *SandboxManager) createPod(ctx context.Context, sessionID string) error { + body := map[string]any{ + "apiVersion": "v1", + "kind": "Pod", + "metadata": map[string]any{ + "name": podName(sessionID), "namespace": sandboxNamespace, + "labels": map[string]string{"app": "pi-session", "session-id": sessionID}, + }, + "spec": map[string]any{ + "restartPolicy": "Never", + "automountServiceAccountToken": false, + "securityContext": map[string]any{ + "runAsNonRoot": true, + "runAsUser": int64(1000), + "seccompProfile": map[string]string{"type": "RuntimeDefault"}, + }, + "containers": []map[string]any{{ + "name": "pi-session", + "image": sandboxImage, + "ports": []map[string]any{{"containerPort": sandboxPort}}, + "securityContext": map[string]any{ + "readOnlyRootFilesystem": true, + "allowPrivilegeEscalation": false, + "capabilities": map[string]any{"drop": []string{"ALL"}}, + }, + "resources": map[string]any{ + "requests": map[string]string{"cpu": "500m", "memory": "512Mi"}, + "limits": map[string]string{"cpu": "1", "memory": "1Gi"}, + }, + "readinessProbe": map[string]any{ + "tcpSocket": map[string]any{"port": sandboxPort}, + "initialDelaySeconds": 3, + "periodSeconds": 2, + "failureThreshold": 15, + }, + "volumeMounts": []map[string]any{ + {"name": "workspace", "mountPath": "/workspace"}, + {"name": "tmp", "mountPath": "/tmp"}, + }, + }}, + "volumes": []map[string]any{ + {"name": "workspace", "persistentVolumeClaim": map[string]string{"claimName": pvcName(sessionID)}}, + {"name": "tmp", "emptyDir": map[string]any{}}, + }, + }, + } + return s.k8sCreate(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/pods", body) +} + +func (s *SandboxManager) ensureService(ctx context.Context, sessionID string) error { + name := svcName(sessionID) + _, err := s.k8sGet(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/services/"+name) + if err == nil { + return nil + } + if !isNotFound(err) { + return err + } + body := map[string]any{ + "apiVersion": "v1", + "kind": "Service", + "metadata": map[string]any{ + "name": name, "namespace": sandboxNamespace, + "labels": map[string]string{"app": "pi-session", "session-id": sessionID}, + }, + "spec": map[string]any{ + "type": "ClusterIP", + "selector": map[string]string{"app": "pi-session", "session-id": sessionID}, + "ports": []map[string]any{{ + "port": sandboxPort, "targetPort": sandboxPort, "protocol": "TCP", + }}, + }, + } + return s.k8sCreate(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/services", body) +} + +func (s *SandboxManager) ensureNetworkPolicy(ctx context.Context) error { + const npName = "pi-session-sandbox" + _, err := s.k8sGet(ctx, "/apis/networking.k8s.io/v1/namespaces/"+sandboxNamespace+"/networkpolicies/"+npName) + if err == nil { + return nil // already exists + } + if !isNotFound(err) { + return err + } + body := map[string]any{ + "apiVersion": "networking.k8s.io/v1", + "kind": "NetworkPolicy", + "metadata": map[string]any{ + "name": npName, "namespace": sandboxNamespace, + "labels": map[string]string{"app": "pi-session"}, + }, + "spec": map[string]any{ + "podSelector": map[string]any{"matchLabels": map[string]string{"app": "pi-session"}}, + "policyTypes": []string{"Ingress", "Egress"}, + "ingress": []map[string]any{{ + "from": []map[string]any{{ + "podSelector": map[string]any{"matchLabels": map[string]string{"app": "poimen-frontend"}}, + }}, + "ports": []map[string]any{{"protocol": "TCP", "port": sandboxPort}}, + }}, + "egress": []map[string]any{ + // DNS + { + "to": []map[string]any{{"namespaceSelector": map[string]any{"matchLabels": map[string]string{"kubernetes.io/metadata.name": "kube-system"}}}}, + "ports": []map[string]any{{"protocol": "UDP", "port": 53}, {"protocol": "TCP", "port": 53}}, + }, + // api-gateway (LLM inference) + { + "to": []map[string]any{{"namespaceSelector": map[string]any{"matchLabels": map[string]string{"kubernetes.io/metadata.name": "api"}}}}, + "ports": []map[string]any{{"protocol": "TCP", "port": 8080}}, + }, + }, + }, + } + return s.k8sCreate(ctx, "/apis/networking.k8s.io/v1/namespaces/"+sandboxNamespace+"/networkpolicies", body) +} + +func (s *SandboxManager) waitPodReady(ctx context.Context, sessionID string) error { + deadline := time.Now().Add(s.readyTimeout) + for time.Now().Before(deadline) { + pod, err := s.k8sGet(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/pods/"+podName(sessionID)) + if err != nil { + return fmt.Errorf("poll pod: %w", err) + } + status, _ := pod["status"].(map[string]any) + phase, _ := status["phase"].(string) + if phase == "Failed" || phase == "Succeeded" { + return fmt.Errorf("pod entered terminal phase %s before ready", phase) + } + if phase == "Running" { + for _, cond := range toSlice(status["conditions"]) { + if c, ok := cond.(map[string]any); ok { + if c["type"] == "Ready" && c["status"] == "True" { + return nil + } + } + } + } + select { + case <-time.After(2 * time.Second): + case <-ctx.Done(): + return ctx.Err() + } + } + return fmt.Errorf("pod not ready after %v", s.readyTimeout) +} + +func (s *SandboxManager) deletePod(ctx context.Context, sessionID string) error { + return s.k8sDelete(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/pods/"+podName(sessionID)) +} +func (s *SandboxManager) deleteService(ctx context.Context, sessionID string) error { + return s.k8sDelete(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/services/"+svcName(sessionID)) +} +func (s *SandboxManager) deletePVC(ctx context.Context, sessionID string) error { + return s.k8sDelete(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/persistentvolumeclaims/"+pvcName(sessionID)) +} + +func (s *SandboxManager) listSandboxPods(ctx context.Context) ([]any, error) { + body, err := s.k8sGet(ctx, "/api/v1/namespaces/"+sandboxNamespace+"/pods?labelSelector=app%3Dpi-session") + if err != nil { + return nil, err + } + items, _ := body["items"].([]any) + return items, nil +} + +// ── raw k8s HTTP ───────────────────────────────────────────────────────── + +func (s *SandboxManager) k8sGet(ctx context.Context, path string) (map[string]any, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.k8sHost+path, nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+sandboxSAToken()) + resp, err := s.k8sClient.Do(req) + if err != nil { + return nil, fmt.Errorf("k8s GET %s: %w", path, err) + } + defer resp.Body.Close() + if resp.StatusCode == 404 { + return nil, &k8sNotFound{path: path} + } + if resp.StatusCode >= 400 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return nil, fmt.Errorf("k8s GET %s returned %d: %s", path, resp.StatusCode, b) + } + var result map[string]any + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("decode: %w", err) + } + return result, nil +} + +func (s *SandboxManager) k8sCreate(ctx context.Context, path string, body any) error { + payload, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshal: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.k8sHost+path, bytes.NewReader(payload)) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+sandboxSAToken()) + req.Header.Set("Content-Type", "application/json") + resp, err := s.k8sClient.Do(req) + if err != nil { + return fmt.Errorf("k8s POST %s: %w", path, err) + } + defer resp.Body.Close() + if resp.StatusCode == 409 { // AlreadyExists — idempotent + return nil + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("k8s POST %s returned %d: %s", path, resp.StatusCode, b) + } + return nil +} + +func (s *SandboxManager) k8sDelete(ctx context.Context, path string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, s.k8sHost+path, nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+sandboxSAToken()) + resp, err := s.k8sClient.Do(req) + if err != nil { + return fmt.Errorf("k8s DELETE %s: %w", path, err) + } + defer resp.Body.Close() + if resp.StatusCode == 404 { + return &k8sNotFound{path: path} + } + if resp.StatusCode >= 400 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("k8s DELETE %s returned %d: %s", path, resp.StatusCode, b) + } + return nil +} + +// ── error helpers ───────────────────────────────────────────────────────── + +type k8sNotFound struct{ path string } + +func (e *k8sNotFound) Error() string { return "not found: " + e.path } + +func isNotFound(err error) bool { + if err == nil { + return false + } + _, ok := err.(*k8sNotFound) + return ok +} + +func toSlice(v any) []any { + if s, ok := v.([]any); ok { + return s + } + return nil +} diff --git a/internal/sessions/sandbox_test.go b/internal/sessions/sandbox_test.go new file mode 100644 index 0000000..aef070b --- /dev/null +++ b/internal/sessions/sandbox_test.go @@ -0,0 +1,271 @@ +package sessions + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +// ── helpers ─────────────────────────────────────────────────────────────────── + +// runningReadyPod returns minimal k8s pod JSON: phase Running + Ready condition. +func runningReadyPod(name string) map[string]any { + return map[string]any{ + "metadata": map[string]any{"name": name}, + "status": map[string]any{ + "phase": "Running", + "conditions": []any{ + map[string]any{"type": "Ready", "status": "True"}, + }, + }, + } +} + +// respondK8sJSON writes a JSON body with the given status code. +func respondK8sJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(body) //nolint:errcheck +} + +// newTestSM builds a SandboxManager pointed at the given test server URL. +func newTestSM(url string) *SandboxManager { + return &SandboxManager{ + k8sHost: url, + k8sClient: &http.Client{Timeout: 5 * time.Second}, + pvcSize: "1Gi", + readyTimeout: 3 * time.Second, + } +} + +// ── TestSandboxCreateOrResume_CreateNew ─────────────────────────────────────── + +// No existing resources → creates PVC, NetworkPolicy, Pod, Service; Resumed=false. +func TestSandboxCreateOrResume_CreateNew(t *testing.T) { + id := "newid" + var podGetCount int32 + + mux := http.NewServeMux() + + // PVC: GET → 404, POST → 201 + mux.HandleFunc("GET /api/v1/namespaces/poimen/persistentvolumeclaims/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + }) + mux.HandleFunc("POST /api/v1/namespaces/poimen/persistentvolumeclaims", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + }) + + // NetworkPolicy (fixed name "pi-session-sandbox"): GET → 404, POST → 201 + mux.HandleFunc("GET /apis/networking.k8s.io/v1/namespaces/poimen/networkpolicies/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + }) + mux.HandleFunc("POST /apis/networking.k8s.io/v1/namespaces/poimen/networkpolicies", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + }) + + // Pod: first GET (getPodPhase) → 404; second GET (waitPodReady) → Running+Ready; POST → 201 + mux.HandleFunc("GET /api/v1/namespaces/poimen/pods/{name}", func(w http.ResponseWriter, _ *http.Request) { + if atomic.AddInt32(&podGetCount, 1) == 1 { + w.WriteHeader(http.StatusNotFound) + return + } + respondK8sJSON(w, http.StatusOK, runningReadyPod(podName(id))) + }) + mux.HandleFunc("POST /api/v1/namespaces/poimen/pods", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + }) + + // Service: GET → 404, POST → 201 + mux.HandleFunc("GET /api/v1/namespaces/poimen/services/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + }) + mux.HandleFunc("POST /api/v1/namespaces/poimen/services", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + }) + + srv := httptest.NewServer(mux) + defer srv.Close() + + result, err := newTestSM(srv.URL).CreateOrResume(context.Background(), id) + if err != nil { + t.Fatalf("CreateOrResume: %v", err) + } + if want := svcDNS(id); result.SvcDNS != want { + t.Errorf("SvcDNS: want %q, got %q", want, result.SvcDNS) + } + if result.Resumed { + t.Error("want Resumed=false, got true") + } +} + +// ── TestSandboxCreateOrResume_Resume ───────────────────────────────────────── + +// PVC exists + pod Running → Resumed=true, no POST calls. +func TestSandboxCreateOrResume_Resume(t *testing.T) { + id := "resumeid" + + mux := http.NewServeMux() + + // PVC exists + mux.HandleFunc("GET /api/v1/namespaces/poimen/persistentvolumeclaims/{name}", func(w http.ResponseWriter, _ *http.Request) { + respondK8sJSON(w, http.StatusOK, map[string]any{"metadata": map[string]any{"name": pvcName(id)}}) + }) + + // Pod exists and Running (getPodPhase uses GET /{name}) + mux.HandleFunc("GET /api/v1/namespaces/poimen/pods/{name}", func(w http.ResponseWriter, _ *http.Request) { + respondK8sJSON(w, http.StatusOK, map[string]any{ + "metadata": map[string]any{"name": podName(id)}, + "status": map[string]any{"phase": "Running"}, + }) + }) + + // Any POST → 500 to assert no creates are issued + mux.HandleFunc("POST /", func(w http.ResponseWriter, r *http.Request) { + t.Errorf("unexpected POST to %s", r.URL.Path) + w.WriteHeader(http.StatusInternalServerError) + }) + + srv := httptest.NewServer(mux) + defer srv.Close() + + result, err := newTestSM(srv.URL).CreateOrResume(context.Background(), id) + if err != nil { + t.Fatalf("CreateOrResume: %v", err) + } + if !result.Resumed { + t.Error("want Resumed=true, got false") + } + if want := svcDNS(id); result.SvcDNS != want { + t.Errorf("SvcDNS: want %q, got %q", want, result.SvcDNS) + } +} + +// ── TestSandboxDelete ───────────────────────────────────────────────────────── + +// All DELETEs succeed → no error returned. +func TestSandboxDelete(t *testing.T) { + id := "deleteid" + + mux := http.NewServeMux() + mux.HandleFunc("DELETE /api/v1/namespaces/poimen/pods/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("DELETE /api/v1/namespaces/poimen/services/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("DELETE /api/v1/namespaces/poimen/persistentvolumeclaims/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + + srv := httptest.NewServer(mux) + defer srv.Close() + + if err := newTestSM(srv.URL).Delete(context.Background(), id); err != nil { + t.Errorf("Delete: %v", err) + } +} + +// ── TestSandboxDelete_NotFound ──────────────────────────────────────────────── + +// All DELETEs return 404 → isNotFound errors swallowed → no error returned. +func TestSandboxDelete_NotFound(t *testing.T) { + id := "ghostid" + + mux := http.NewServeMux() + mux.HandleFunc("DELETE /api/v1/namespaces/poimen/pods/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + }) + mux.HandleFunc("DELETE /api/v1/namespaces/poimen/services/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + }) + mux.HandleFunc("DELETE /api/v1/namespaces/poimen/persistentvolumeclaims/{name}", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + }) + + srv := httptest.NewServer(mux) + defer srv.Close() + + if err := newTestSM(srv.URL).Delete(context.Background(), id); err != nil { + t.Errorf("Delete_NotFound: want nil, got %v", err) + } +} + +// ── TestSandboxReapIdlePods ─────────────────────────────────────────────────── + +// pod-a=Succeeded → deleted; pod-b=Running → skipped. +func TestSandboxReapIdlePods(t *testing.T) { + var deletedA int32 + + mux := http.NewServeMux() + + // List all pi-session pods (query string not part of mux path matching) + mux.HandleFunc("GET /api/v1/namespaces/poimen/pods", func(w http.ResponseWriter, _ *http.Request) { + respondK8sJSON(w, http.StatusOK, map[string]any{ + "items": []any{ + map[string]any{ + "metadata": map[string]any{"name": "pod-a"}, + "status": map[string]any{"phase": "Succeeded"}, + }, + map[string]any{ + "metadata": map[string]any{"name": "pod-b"}, + "status": map[string]any{"phase": "Running"}, + }, + }, + }) + }) + + mux.HandleFunc("DELETE /api/v1/namespaces/poimen/pods/{name}", func(w http.ResponseWriter, r *http.Request) { + name := r.PathValue("name") + if name == "pod-b" { + t.Errorf("pod-b (Running) must NOT be deleted") + } + if name == "pod-a" { + atomic.AddInt32(&deletedA, 1) + } + w.WriteHeader(http.StatusOK) + }) + + srv := httptest.NewServer(mux) + defer srv.Close() + + newTestSM(srv.URL).ReapIdlePods(context.Background()) + + if got := atomic.LoadInt32(&deletedA); got != 1 { + t.Errorf("pod-a delete count: want 1, got %d", got) + } +} + +// ── TestSandboxNameHelpers ──────────────────────────────────────────────────── + +func TestSandboxNameHelpers(t *testing.T) { + cases := []struct{ fn, want, got string }{ + {"pvcName", "workspace-session-abc", pvcName("abc")}, + {"podName", "pi-session-abc", podName("abc")}, + {"svcName", "pi-session-abc", svcName("abc")}, + {"svcDNS", "pi-session-abc.poimen.svc.cluster.local:8787", svcDNS("abc")}, + } + for _, c := range cases { + if c.got != c.want { + t.Errorf("%s: want %q, got %q", c.fn, c.want, c.got) + } + } +} + +// ── TestSandboxIsNotFound ───────────────────────────────────────────────────── + +func TestSandboxIsNotFound(t *testing.T) { + if !isNotFound(&k8sNotFound{}) { + t.Error("isNotFound(&k8sNotFound{}) must be true") + } + if isNotFound(fmt.Errorf("other")) { + t.Error("isNotFound(fmt.Errorf) must be false") + } + if isNotFound(nil) { + t.Error("isNotFound(nil) must be false") + } +} diff --git a/k8s/rbac.yaml b/k8s/rbac.yaml index fe55b43..9a043db 100644 --- a/k8s/rbac.yaml +++ b/k8s/rbac.yaml @@ -26,3 +26,37 @@ roleRef: kind: ClusterRole name: image-promoter apiGroup: rbac.authorization.k8s.io +--- +# Role + RoleBinding for pi-session sandbox manager (§4.15.4). +# api-gateway SA needs get/create/delete for pods, pvcs, services, +# and networkpolicies in the poimen namespace. +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: session-sandbox-manager + namespace: poimen + labels: + app: api-gateway +rules: +- apiGroups: [""] + resources: ["pods", "persistentvolumeclaims", "services"] + verbs: ["get", "list", "create", "delete"] +- apiGroups: ["networking.k8s.io"] + resources: ["networkpolicies"] + verbs: ["get", "create", "delete"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: session-sandbox-manager-api-gateway + namespace: poimen + labels: + app: api-gateway +subjects: +- kind: ServiceAccount + name: api-gateway + namespace: api +roleRef: + kind: Role + name: session-sandbox-manager + apiGroup: rbac.authorization.k8s.io