feat(sessions): sandbox resource — per-session k8s Pod+PVC+Service (§4.15.4)
CI / CI (pull_request) Successful in 2m8s
CI / CI (pull_request) Successful in 2m8s
New X-Resource: sandbox on X-Service: sessions
POST ?session_id= → CreateOrResume
1. Ensure PVC workspace-session-<id> (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-<id> + Service pi-session-<id>: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
This commit is contained in:
Admin Bot
@@ -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"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -0,0 +1,492 @@
|
||||
// Package sessions — SandboxManager provisions per-session k8s resources:
|
||||
//
|
||||
// PVC workspace-session-<uuid> Longhorn 1Gi
|
||||
// Pod pi-session-<uuid> single container, /workspace <- PVC
|
||||
// Svc pi-session-<uuid> 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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user