mirror of
https://github.com/R0m1k3/Loki.git
synced 2026-10-15 11:19:07 +02:00
Lot 5 (Go) : décisions de la surface /v1 exposées à la façade
La façade Rust va relayer /v1 directement au moteur. Les décisions restent ici, parce qu'elles dépendent de l'état de la conversation que ce process tient en mémoire : route (port, Strata, clé des complétions, second slot en service), begin (requête « en vol » : le slot de la discussion n'est pas effacé pendant ce temps, préchauffage annulé si besoin ; refus pendant un benchmark ou une optimisation ; trace de perte de cache), end. Mêmes étapes, même ordre qu'oaiHandler. Un ticket jamais rendu (façade tuée en plein flux) expire au bout de six heures, sans quoi le compteur de requêtes en vol resterait bloqué. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
7017b66c08
commit
451770bfcf
3 files changed
+245
No files matched your search
@@ -63,6 +63,12 @@ func withInternalAPI(next http.Handler) http.Handler {
|
||||
handleInternalStore(w, r)
|
||||
case "/internal/v1/store/all":
|
||||
handleInternalStoreAll(w, r)
|
||||
case "/internal/v1/oai/route":
|
||||
handleInternalOAIRoute(w, r)
|
||||
case "/internal/v1/oai/begin":
|
||||
handleInternalOAIBegin(w, r)
|
||||
case "/internal/v1/oai/end":
|
||||
handleInternalOAIEnd(w, r)
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
package loki
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// API interne de la surface /v1 servie par la façade Rust (lot 5).
|
||||
//
|
||||
// Rust relaie désormais /v1 directement au moteur, mais les DÉCISIONS restent
|
||||
// ici, parce qu'elles dépendent de l'état de la conversation que ce process
|
||||
// tient en mémoire :
|
||||
//
|
||||
// - route : ce qu'il faut pour router (port du moteur, Strata, clé des
|
||||
// complétions, second slot en service) ;
|
||||
// - begin : la requête entre « en vol » (engineRequestBeginSide : le slot de
|
||||
// la discussion n'est pas effacé pendant ce temps, un préchauffage est
|
||||
// annulé si besoin), puis les refus benchmark / optimisation, puis la trace
|
||||
// de perte de cache (perfNoteForeign) — dans l'ordre d'oaiHandler ;
|
||||
// - end : la requête est terminée (fin du flux, ou client parti).
|
||||
//
|
||||
// Un ticket jamais rendu (façade tuée en plein flux) expire au bout de
|
||||
// oaiTicketMaxAge : sans ça, le compteur de requêtes en vol resterait bloqué
|
||||
// et le slot ne serait plus jamais isolé.
|
||||
|
||||
const oaiTicketMaxAge = 6 * time.Hour
|
||||
|
||||
type oaiTicket struct {
|
||||
end func()
|
||||
when time.Time
|
||||
}
|
||||
|
||||
var oaiTickets = struct {
|
||||
mu sync.Mutex
|
||||
next uint64
|
||||
m map[uint64]oaiTicket
|
||||
}{m: map[uint64]oaiTicket{}}
|
||||
|
||||
func handleInternalOAIRoute(w http.ResponseWriter, r *http.Request) {
|
||||
q := r.URL.Query()
|
||||
method, path := q.Get("method"), q.Get("path")
|
||||
key, err := effectiveAPIKeyErr()
|
||||
out := map[string]any{
|
||||
"port": LLMPort(),
|
||||
"strata": strataActive(),
|
||||
"key": key,
|
||||
"key_ok": err == nil,
|
||||
"max_body": sideSlotProxyMaxBody,
|
||||
}
|
||||
// La sonde du second slot interroge le moteur (avec cache) : seulement
|
||||
// pour une complétion, comme sideSlotProxyRewrite.
|
||||
out["side_live"] = method == http.MethodPost && strings.HasPrefix(path, "/v1/") &&
|
||||
sideSlotLive(resolveChatEndpoint())
|
||||
sendJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
func handleInternalOAIBegin(w http.ResponseWriter, r *http.Request) {
|
||||
var in struct {
|
||||
Read bool `json:"read"` // GET/HEAD : ne calcule rien dans un slot
|
||||
Side bool `json:"side"` // corps routé vers le second slot
|
||||
PostV1 bool `json:"post_v1"` // complétion d'un client externe
|
||||
}
|
||||
if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&in) != nil {
|
||||
sendJSON(w, http.StatusBadRequest, map[string]any{"error": "POST JSON attendu"})
|
||||
return
|
||||
}
|
||||
var end func()
|
||||
if in.Read {
|
||||
end, _ = engineRequestBeginSide(func(*prewarmRun) bool { return true }, true)
|
||||
} else {
|
||||
end, _ = engineRequestBeginSide(nil, in.Side)
|
||||
}
|
||||
if in.PostV1 {
|
||||
refuse := func(retry, msg, code string) {
|
||||
end()
|
||||
sendJSON(w, http.StatusOK, map[string]any{"refuse": map[string]any{
|
||||
"status": http.StatusServiceUnavailable, "message": msg,
|
||||
"type": "server_error", "code": code, "retry_after": retry,
|
||||
}})
|
||||
}
|
||||
if benchRunning() {
|
||||
refuse("30", "benchmark en cours sur ce moteur — réessaie dans un moment", "benchmark_running")
|
||||
return
|
||||
}
|
||||
if tuneGuard() != nil {
|
||||
refuse("60", "optimisation du moteur en cours — réessaie dans quelques minutes", "tuning_running")
|
||||
return
|
||||
}
|
||||
perfNoteForeign()
|
||||
}
|
||||
now := time.Now()
|
||||
oaiTickets.mu.Lock()
|
||||
for id, t := range oaiTickets.m {
|
||||
if now.Sub(t.when) > oaiTicketMaxAge {
|
||||
t.end()
|
||||
delete(oaiTickets.m, id)
|
||||
}
|
||||
}
|
||||
oaiTickets.next++
|
||||
id := oaiTickets.next
|
||||
oaiTickets.m[id] = oaiTicket{end: end, when: now}
|
||||
oaiTickets.mu.Unlock()
|
||||
sendJSON(w, http.StatusOK, map[string]any{"ticket": id})
|
||||
}
|
||||
|
||||
func handleInternalOAIEnd(w http.ResponseWriter, r *http.Request) {
|
||||
var in struct {
|
||||
Ticket uint64 `json:"ticket"`
|
||||
}
|
||||
if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&in) != nil {
|
||||
sendJSON(w, http.StatusBadRequest, map[string]any{"error": "POST JSON attendu"})
|
||||
return
|
||||
}
|
||||
oaiTickets.mu.Lock()
|
||||
t, ok := oaiTickets.m[in.Ticket]
|
||||
delete(oaiTickets.m, in.Ticket)
|
||||
oaiTickets.mu.Unlock()
|
||||
if !ok {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
t.end()
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// oaiInflight : nombre de tickets ouverts (tests).
|
||||
func oaiInflight() int {
|
||||
oaiTickets.mu.Lock()
|
||||
defer oaiTickets.mu.Unlock()
|
||||
return len(oaiTickets.m)
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
package loki
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func appelOAIInterne(t *testing.T, method, path, body string) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
r := httptest.NewRequest(method, path, strings.NewReader(body))
|
||||
r.Header.Set(internalTokenHeader, jetonTest)
|
||||
w := httptest.NewRecorder()
|
||||
withInternalAPI(http.NotFoundHandler()).ServeHTTP(w, r)
|
||||
return w
|
||||
}
|
||||
|
||||
func enVol() int {
|
||||
engineGate.mu.Lock()
|
||||
defer engineGate.mu.Unlock()
|
||||
return engineGate.inflight
|
||||
}
|
||||
|
||||
// Une requête /v1 relayée par Rust compte « en vol » de begin à end, comme
|
||||
// le defer engineProxyBegin(r, side)() d'oaiHandler.
|
||||
func TestOAIInterneTicket(t *testing.T) {
|
||||
testHome(t)
|
||||
t.Setenv(internalTokenEnv, jetonTest)
|
||||
avant := enVol()
|
||||
w := appelOAIInterne(t, "POST", "/internal/v1/oai/begin", `{"read":false,"side":false,"post_v1":true}`)
|
||||
var got struct {
|
||||
Ticket uint64 `json:"ticket"`
|
||||
Refuse map[string]any `json:"refuse"`
|
||||
}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || got.Ticket == 0 || got.Refuse != nil {
|
||||
t.Fatalf("begin : %d %s", w.Code, w.Body.String())
|
||||
}
|
||||
if enVol() != avant+1 || oaiInflight() != 1 {
|
||||
t.Fatalf("en vol : %d (avant %d), tickets %d", enVol(), avant, oaiInflight())
|
||||
}
|
||||
body, _ := json.Marshal(map[string]any{"ticket": got.Ticket})
|
||||
if w := appelOAIInterne(t, "POST", "/internal/v1/oai/end", string(body)); w.Code != 204 {
|
||||
t.Fatalf("end : %d", w.Code)
|
||||
}
|
||||
if enVol() != avant || oaiInflight() != 0 {
|
||||
t.Fatalf("après end : en vol %d, tickets %d", enVol(), oaiInflight())
|
||||
}
|
||||
if w := appelOAIInterne(t, "POST", "/internal/v1/oai/end", string(body)); w.Code != 404 {
|
||||
t.Fatalf("second end : %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// Pendant un benchmark, une complétion externe est refusée — et ne reste pas
|
||||
// comptée en vol.
|
||||
func TestOAIInterneRefusBenchmark(t *testing.T) {
|
||||
testHome(t)
|
||||
t.Setenv(internalTokenEnv, jetonTest)
|
||||
benchJob.mu.Lock()
|
||||
benchJob.running = true
|
||||
benchJob.mu.Unlock()
|
||||
t.Cleanup(func() {
|
||||
benchJob.mu.Lock()
|
||||
benchJob.running = false
|
||||
benchJob.mu.Unlock()
|
||||
})
|
||||
avant := enVol()
|
||||
w := appelOAIInterne(t, "POST", "/internal/v1/oai/begin", `{"post_v1":true}`)
|
||||
var got struct {
|
||||
Refuse struct {
|
||||
Status int `json:"status"`
|
||||
Code string `json:"code"`
|
||||
RetryAfter string `json:"retry_after"`
|
||||
} `json:"refuse"`
|
||||
}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Refuse.Status != 503 || got.Refuse.Code != "benchmark_running" || got.Refuse.RetryAfter != "30" {
|
||||
t.Fatalf("refus : %s", w.Body.String())
|
||||
}
|
||||
if enVol() != avant || oaiInflight() != 0 {
|
||||
t.Fatalf("refus compté en vol : %d", enVol())
|
||||
}
|
||||
}
|
||||
|
||||
func TestOAIInterneRoute(t *testing.T) {
|
||||
testHome(t)
|
||||
t.Setenv(internalTokenEnv, jetonTest)
|
||||
if err := putStr(bkConfig, "PORT", "8099"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := putStr(bkState, "api_key", "cle-oai"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w := appelOAIInterne(t, "GET", "/internal/v1/oai/route?method=GET&path=/v1/models", "")
|
||||
var got map[string]any
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got["port"] != float64(8099) || got["key"] != "cle-oai" || got["key_ok"] != true || got["side_live"] != false {
|
||||
t.Fatalf("route : %s", w.Body.String())
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user