diff --git a/internal/sessions/fleet_client.go b/internal/sessions/fleet_client.go new file mode 100644 index 0000000..da7684a --- /dev/null +++ b/internal/sessions/fleet_client.go @@ -0,0 +1,97 @@ +// Package sessions — FleetClient calls the SandboxFleet controller REST API +// to assign sessions to fleet pods. This is the primary sandbox path; +// the raw-k8s SandboxManager is the fallback when the fleet controller +// is unavailable (e.g. during fleet rollout). +package sessions + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "time" +) + +const ( + defaultFleetControllerURL = "http://tool-definition-controller.poimen.svc.cluster.local:8788" +) + +// FleetClient calls the SandboxFleet controller REST API. +type FleetClient struct { + baseURL string + client *http.Client +} + +// FleetAssignResult is returned by the controller assign endpoint. +type FleetAssignResult struct { + PodSvcDns string `json:"podSvcDns"` + PodName string `json:"podName"` +} + +// NewFleetClient creates a FleetClient from env or default URL. +// FLEET_CONTROLLER_URL — default: defaultFleetControllerURL +func NewFleetClient() *FleetClient { + url := os.Getenv("FLEET_CONTROLLER_URL") + if url == "" { + url = defaultFleetControllerURL + } + return &FleetClient{ + baseURL: url, + client: &http.Client{Timeout: 35 * time.Second}, + } +} + +// Assign asks the fleet controller to assign sessionID to a fleet pod. +// Returns the pod's ClusterIP service DNS and pod name. +func (f *FleetClient) Assign(ctx context.Context, sessionID string) (*FleetAssignResult, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + f.baseURL+"/sessions/"+sessionID+"/assign", nil) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := f.client.Do(req) + if err != nil { + return nil, fmt.Errorf("fleet controller unreachable: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 300 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return nil, fmt.Errorf("fleet controller returned %d: %s", resp.StatusCode, body) + } + + var result FleetAssignResult + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("decode: %w", err) + } + if result.PodSvcDns == "" { + return nil, fmt.Errorf("fleet controller returned empty podSvcDns") + } + return &result, nil +} + +// Release notifies the fleet controller that a session has ended. +// Cleans up the per-session ServiceAccount (audit trail SA). +// Non-fatal: fleet controller handles this async. +func (f *FleetClient) Release(ctx context.Context, sessionID string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, + f.baseURL+"/sessions/"+sessionID, bytes.NewReader(nil)) + if err != nil { + return fmt.Errorf("build request: %w", err) + } + resp, err := f.client.Do(req) + if err != nil { + return fmt.Errorf("fleet controller unreachable: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode >= 300 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("fleet controller returned %d: %s", resp.StatusCode, body) + } + return nil +} diff --git a/internal/sessions/handler.go b/internal/sessions/handler.go index e81044a..3cfe99a 100644 --- a/internal/sessions/handler.go +++ b/internal/sessions/handler.go @@ -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"}) } diff --git a/k8s/network-policy.yaml b/k8s/network-policy.yaml index 8e05a3c..c82ec15 100644 --- a/k8s/network-policy.yaml +++ b/k8s/network-policy.yaml @@ -122,7 +122,7 @@ spec: ports: - protocol: TCP port: 7233 - # Allow to Poimen (memory:8080 + graphql:8090 + redis:6379) + # Allow to Poimen (memory:8080 + graphql:8090 + redis:6379 + fleet-controller:8788) - to: - namespaceSelector: matchLabels: @@ -134,6 +134,8 @@ spec: port: 8090 - protocol: TCP port: 6379 + - protocol: TCP + port: 8788 # Allow to Gotify (push notifications) - to: - namespaceSelector: