riotpiao-poimen/homelab-frontend
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sessions): SandboxFleet integration — fleet controller primary, SandboxManager fallback

Browse Source 
FleetClient calls tool-definition-controller:8788 /sessions/{id}/assign
as the primary sandbox create path. Falls back to direct k8s SandboxManager
if fleet controller is unreachable.

handleSandboxCreate:
  1. fleet.Assign() → podSvcDns from pre-warmed fleet pod (fast, no wait)
  2. fallback: sandbox.CreateOrResume() → dedicated pod (slow, 30s wait)

handleSandboxDelete: fleet.Release() + sandbox.Delete() both run (non-fatal).
updateSandboxSummary: extracted helper, updates session memory_entity with
  pod_name / svc_dns / last_active_at.

NetworkPolicy: added port 8788 to poimen egress (fleet controller REST API).

Response adds fleet:true field when served by fleet.
This commit is contained in:
Admin Bot
2026-09-29 16:10:43 +08:00
parent 49c9071a59
commit 1e17706367
3 changed files with 147 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/sessions/handler.go
+47 -19
View File
@@ -32,10 +32,13 @@ const (
// Handler handles session REST endpoints via internal service calls.
// All traffic (reads and writes) goes to graphqlURL (poimen-graphql).
// Sandbox creates/deletes pi-session pods via the SandboxFleet controller
// (primary) with SandboxManager as direct-k8s fallback.
type Handler struct {
graphqlURL string
client *http.Client
sandbox *SandboxManager
fleet *FleetClient
}
// NewHandler creates a Handler reading upstream URL from env var.
@@ -49,6 +52,7 @@ func NewHandler() *Handler {
graphqlURL: gql,
client: &http.Client{Timeout: 30 * time.Second, Transport: tracing.NewTransport(nil)},
sandbox: NewSandboxManager(),
fleet: NewFleetClient(),
}
h.startIdleReaper()
return h
@@ -545,39 +549,63 @@ func (h *Handler) handleSandbox(w http.ResponseWriter, r *http.Request) {
}
}
// POST ?session_id= — create or resume pi-session sandbox.
// POST ?session_id= — assign session to SandboxFleet pod (primary) or
// create a dedicated pod via SandboxManager (fallback).
func (h *Handler) handleSandboxCreate(w http.ResponseWriter, r *http.Request, sessionID string) {
// Primary: fleet controller assigns session to a pre-warmed pod.
if assign, err := h.fleet.Assign(r.Context(), sessionID); err == nil {
svcDns := assign.PodSvcDns + fmt.Sprintf(":%d", sandboxPort)
log.Printf("sandbox: fleet assigned session=%s pod=%s dns=%s", sessionID, assign.PodName, svcDns)
go h.updateSandboxSummary(r, sessionID, assign.PodName, "", svcDns)
writeJSON(w, http.StatusOK, map[string]any{
"session_id": sessionID,
"pod_name": assign.PodName,
"svc_dns": svcDns,
"resumed": false,
"fleet": true,
})
return
} else {
log.Printf("sandbox: fleet unavailable for session=%s (%v), falling back to direct k8s", sessionID, err)
}
// Fallback: create a dedicated pod via SandboxManager.
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)
go h.updateSandboxSummary(r, sessionID, result.PodName, result.PVCName, result.SvcDNS)
log.Printf("sandbox: direct pod 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.
// updateSandboxSummary merges pod/svc metadata into the session memory_entity (fire-and-forget).
func (h *Handler) updateSandboxSummary(r *http.Request, sessionID, podName, pvcName, svcDns string) {
current, _ := h.fetchEntitySummary(r, "session-"+sessionID, sessionID, "SESSION")
current["pod_name"] = podName
current["svc_dns"] = svcDns
current["last_active_at"] = time.Now().UTC().Format(time.RFC3339)
if pvcName != "" {
current["pvc_name"] = pvcName
}
if b, err := json.Marshal(current); err == nil {
h.upsertEntity(r, "session-"+sessionID, sessionID, "SESSION", string(b)) //nolint:errcheck
}
}
// DELETE ?session_id= — release from fleet + delete dedicated pod+svc+pvc if any.
func (h *Handler) handleSandboxDelete(w http.ResponseWriter, r *http.Request, sessionID string) {
// Release from fleet (cleans up per-session SA; non-fatal if session was fleet-based).
if err := h.fleet.Release(r.Context(), sessionID); err != nil {
log.Printf("sandbox: fleet release session=%s: %v (continuing)", sessionID, err)
}
// Also attempt direct k8s cleanup (no-ops if resources don't exist).
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)
log.Printf("sandbox: released session=%s", sessionID)
writeJSON(w, http.StatusOK, map[string]string{"session_id": sessionID, "status": "deleted"})
}