chii
eb972dfa93
单二进制三角色 + Dock 适配器 + Swarm stack 达到可部署态；mgr1 实测订阅经 central-proxy bootstrap，探活 alive=41/52。 Co-authored-by: Cursor <cursoragent@cursor.com>
265 lines
9.7 KiB
Go
265 lines
9.7 KiB
Go
// server.go：scheduler HTTP 面（:8641，仅 overlay）。
|
||
//
|
||
// 路由（A3.1/A3.6）：
|
||
//
|
||
// POST /enqueue → JobEnvelope → 落盘成功才 ACK {job_id, request_id}
|
||
// 超限 429/503 + Retry-After + {running,queued} 现状
|
||
// GET /result/{req_id} → done 信封 / 202+{status} / 404
|
||
// GET /pressure → {cpu,memory_pct,running,queued,recently_rejected,
|
||
// is_available,reason}（design §7.1 Browserless 形状）
|
||
// GET /metrics → 文本格式（design §7.2 清单）
|
||
// GET /healthz → liveness
|
||
package scheduler
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"onesvm.com/onesvm/browser-server/internal/contract"
|
||
)
|
||
|
||
// enqueueACK /enqueue 成功响应形状（W2 gateway 对齐面）。
|
||
type enqueueACK struct {
|
||
JobID int64 `json:"job_id"`
|
||
RequestID string `json:"request_id"`
|
||
Status string `json:"status"`
|
||
}
|
||
|
||
// enqueueError 拒绝响应形状（携带现状与原因，mcp-usage §3 503 纪律）。
|
||
type enqueueRejectBody struct {
|
||
Code string `json:"code"` // rate_limited | unavailable
|
||
Message string `json:"message"`
|
||
Reason string `json:"reason"`
|
||
Running int `json:"running"`
|
||
Queued int `json:"queued"`
|
||
AdmitMax int `json:"admit_max"`
|
||
RetryAfterS int `json:"retry_after_s"`
|
||
}
|
||
|
||
// Handler 构建 scheduler HTTP mux。
|
||
func (c *Core) Handler() *http.ServeMux {
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("POST /enqueue", c.handleEnqueue)
|
||
mux.HandleFunc("GET /result/", c.handleResult)
|
||
mux.HandleFunc("GET /pressure", c.handlePressure)
|
||
mux.HandleFunc("GET /metrics", c.handleMetrics)
|
||
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
|
||
writeJSON(w, http.StatusOK, map[string]bool{"ok": true})
|
||
})
|
||
return mux
|
||
}
|
||
|
||
// handleEnqueue POST /enqueue：JobEnvelope → ADMIT_MAX/互斥检查 → WAL 落盘 → ACK。
|
||
func (c *Core) handleEnqueue(w http.ResponseWriter, r *http.Request) {
|
||
defer r.Body.Close()
|
||
dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)) // 1MB 入参硬限（T7）
|
||
var job contract.JobEnvelope
|
||
if err := dec.Decode(&job); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"code": "bad_json", "message": err.Error()})
|
||
return
|
||
}
|
||
// 入参校验（T7：长度/范围边界）。
|
||
if rej := validateJob(job); rej != nil {
|
||
writeJSON(w, http.StatusBadRequest, rej)
|
||
return
|
||
}
|
||
if job.RequestID == "" {
|
||
job.RequestID = newRequestID()
|
||
}
|
||
if job.SubmittedAt.IsZero() {
|
||
job.SubmittedAt = contract.NowTime()
|
||
}
|
||
if job.Status == "" {
|
||
job.Status = contract.JobQueued
|
||
}
|
||
if job.Priority == 0 {
|
||
job.Priority = 100
|
||
}
|
||
id, rej := c.Enqueue(job)
|
||
if rej != nil {
|
||
w.Header().Set("Retry-After", strconv.Itoa(rej.RetryAfterS))
|
||
body := enqueueRejectBody{
|
||
Code: contract.CodeRateLimited,
|
||
Message: firstNonEmptyStr(rej.Message, "队列接纳上限，稍后重试"),
|
||
Reason: rej.Reason,
|
||
Running: rej.Running,
|
||
Queued: rej.Queued,
|
||
AdmitMax: c.admitMax,
|
||
RetryAfterS: rej.RetryAfterS,
|
||
}
|
||
if rej.HTTPStatus == 503 {
|
||
body.Code = contract.CodeUnavailable
|
||
}
|
||
if rej.Reason == "shell_active_panda_hold" {
|
||
c.st.ShellAct.Add(1)
|
||
c.st.Rejected.Add(1)
|
||
}
|
||
writeJSON(w, rej.HTTPStatus, body)
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, enqueueACK{JobID: id, RequestID: job.RequestID, Status: contract.JobQueued})
|
||
}
|
||
|
||
// validateJob 入参边界（T7）。
|
||
func validateJob(job contract.JobEnvelope) map[string]string {
|
||
if job.Intent != "search" && job.Intent != "read" {
|
||
return map[string]string{"code": "bad_intent", "message": "intent 必须为 search|read"}
|
||
}
|
||
switch job.Intent {
|
||
case "search":
|
||
if job.Search == nil {
|
||
return map[string]string{"code": "bad_payload", "message": "search 任务缺 search 字段"}
|
||
}
|
||
q := strings.TrimSpace(job.Search.Query)
|
||
if q == "" || len([]rune(q)) > 512 {
|
||
return map[string]string{"code": "bad_query", "message": "query 必填且 ≤512 字符"}
|
||
}
|
||
if job.Search.MaxResults < 0 || job.Search.MaxResults > 20 {
|
||
return map[string]string{"code": "bad_max_results", "message": "max_results ∈ [0,20]"}
|
||
}
|
||
case "read":
|
||
if job.Read == nil {
|
||
return map[string]string{"code": "bad_payload", "message": "read 任务缺 read 字段"}
|
||
}
|
||
if !strings.HasPrefix(job.Read.URL, "http://") && !strings.HasPrefix(job.Read.URL, "https://") {
|
||
return map[string]string{"code": "bad_url", "message": "url 必须为 http/https"}
|
||
}
|
||
if len(job.Read.URL) > 2048 {
|
||
return map[string]string{"code": "bad_url", "message": "URL ≤2048 字符"}
|
||
}
|
||
if job.Read.MaxChars < 0 || job.Read.MaxChars > 200000 {
|
||
return map[string]string{"code": "bad_max_chars", "message": "max_chars ∈ [0,200000]"}
|
||
}
|
||
if _, err := contract.ValidRegion(job.Read.Region); err != nil {
|
||
return map[string]string{"code": "bad_region", "message": err.Error()}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// handleResult GET /result/{request_id}。
|
||
func (c *Core) handleResult(w http.ResponseWriter, r *http.Request) {
|
||
reqID := strings.TrimPrefix(r.URL.Path, "/result/")
|
||
if reqID == "" {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"code": "not_found"})
|
||
return
|
||
}
|
||
// 先查进程内缓存（done 信封）。
|
||
if env, ok := c.results.get(reqID); ok {
|
||
// ITER-1 F1（fail-w5-smoke-iter1 方案 A）：统一线形状
|
||
// {request_id, status, envelope}——终态 200 + 信封嵌套（golden 锁形：
|
||
// contract.ResultShapeDone，gateway/scheduler 测试双端同源，杜绝 T1 漂移）。
|
||
writeJSON(w, http.StatusOK, resultResponse{
|
||
RequestID: reqID, Status: contract.JobDone, Envelope: env,
|
||
})
|
||
return
|
||
}
|
||
job, err := c.db.JobByRequestID(reqID)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusNotFound, map[string]string{"code": "not_found", "message": "无此 request_id"})
|
||
return
|
||
}
|
||
switch job.Status {
|
||
case "done":
|
||
// 缓存淘汰：按统一形状给引导响应（envelope 缺位时仅 status，gateway 继续轮询
|
||
// 或读自身缓存；不再裸返信封本体——T1 契约漂移根因，ITER-1 修复）。
|
||
writeJSON(w, http.StatusAccepted, resultResponse{
|
||
RequestID: reqID, Status: contract.JobDone, JobID: job.ID,
|
||
})
|
||
case "failed", "dead":
|
||
writeJSON(w, http.StatusOK, resultResponse{
|
||
RequestID: reqID, Status: job.Status, JobID: job.ID,
|
||
Envelope: &contract.Envelope{
|
||
OK: false, Kind: job.Intent, RequestID: reqID,
|
||
Error: &contract.ErrBody{Code: contract.CodeUpstream, Message: job.Error},
|
||
},
|
||
})
|
||
default: // queued / running
|
||
writeJSON(w, http.StatusAccepted, resultResponse{
|
||
RequestID: reqID, Status: job.Status, JobID: job.ID,
|
||
})
|
||
}
|
||
}
|
||
|
||
// resultResponse /result 统一线形状（ITER-1 F1：终态 200+envelope 嵌套，
|
||
// 非终态 202+{status,position?}；golden 锁形 contract.ResultShapeDone/Accepted/Failed）。
|
||
type resultResponse struct {
|
||
RequestID string `json:"request_id"`
|
||
Status string `json:"status"`
|
||
Position int `json:"position,omitempty"`
|
||
JobID int64 `json:"job_id,omitempty"`
|
||
Envelope *contract.Envelope `json:"envelope,omitempty"`
|
||
}
|
||
|
||
// handlePressure GET /pressure（design §7.1 Browserless 形状）。
|
||
func (c *Core) handlePressure(w http.ResponseWriter, _ *http.Request) {
|
||
writeJSON(w, http.StatusOK, c.PressureData())
|
||
}
|
||
|
||
// handleMetrics GET /metrics 文本格式（design §7.2 清单）。
|
||
func (c *Core) handleMetrics(w http.ResponseWriter, _ *http.Request) {
|
||
running, _ := c.db.CountRunning()
|
||
queued, _ := c.db.CountQueued()
|
||
waitAvg, _ := c.db.(interface{ QueueWaitAvgMs() (float64, error) }).QueueWaitAvgMs()
|
||
dead := 0
|
||
if dn, ok := c.db.(interface{ CountDead() (int, error) }); ok {
|
||
dead, _ = dn.CountDead()
|
||
}
|
||
var b strings.Builder
|
||
b.WriteString("# HELP queue_depth 排队深度\n# TYPE queue_depth gauge\n")
|
||
fmt.Fprintf(&b, "queue_depth %d\n", queued)
|
||
b.WriteString("# TYPE admitted_total counter\n")
|
||
fmt.Fprintf(&b, "admitted_total %d\n", c.st.Admitted.Load())
|
||
b.WriteString("# TYPE rejected_total counter\n")
|
||
fmt.Fprintf(&b, "rejected_total{reason} %d\n", c.st.Rejected.Load())
|
||
fmt.Fprintf(&b, "rejected_total{reason=\"shell_active_panda_hold\"} %d\n", c.st.ShellAct.Load())
|
||
b.WriteString("# TYPE sessions_running gauge\n")
|
||
fmt.Fprintf(&b, "sessions_running{adapter=\"all\"} %d\n", running)
|
||
fmt.Fprintf(&b, "sessions_running{adapter=\"in_process\"} %d\n", c.st.RunningG.Load())
|
||
b.WriteString("# TYPE recycles_total counter\n")
|
||
fmt.Fprintf(&b, "recycles_total %d\n", c.st.Recycles.Load())
|
||
b.WriteString("# TYPE queue_wait_seconds_avg gauge\n")
|
||
fmt.Fprintf(&b, "queue_wait_seconds_avg %.3f\n", waitAvg/1000)
|
||
b.WriteString("# TYPE dead_letters_total counter\n")
|
||
fmt.Fprintf(&b, "dead_letters_total %d\n", dead)
|
||
// 合规：denied_total{rule_id}（audit 表聚合）。
|
||
if db, ok := c.db.(interface {
|
||
AuditCountByRule(string) (int, error)
|
||
}); ok {
|
||
for _, rule := range []string{"safetyscan_high_risk", "domain_deny", "ssrf"} {
|
||
n, _ := db.AuditCountByRule(rule)
|
||
fmt.Fprintf(&b, "denied_total{rule_id=%q} %d\n", rule, n)
|
||
}
|
||
}
|
||
// 适配器健康聚合。
|
||
for name, h := range c.reg.HealthAll() {
|
||
fmt.Fprintf(&b, "adapter_health{adapter=%q} %d\n", name, boolToInt(h.OK))
|
||
fmt.Fprintf(&b, "adapter_slots_free{adapter=%q} %d\n", name, h.SlotsFree)
|
||
}
|
||
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
|
||
_, _ = w.Write([]byte(b.String()))
|
||
}
|
||
|
||
// writeJSON 统一 JSON 输出。
|
||
func writeJSON(w http.ResponseWriter, code int, v any) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(code)
|
||
_ = json.NewEncoder(w).Encode(v)
|
||
}
|
||
|
||
// boolToInt bool → 0/1。
|
||
func boolToInt(b bool) int {
|
||
if b {
|
||
return 1
|
||
}
|
||
return 0
|
||
}
|
||
|
||
// newRequestID 兜底 request_id（gateway 正常生成 ULID；直连 enqueue 时兜底）。
|
||
func newRequestID() string {
|
||
return "req-" + strconv.FormatInt(time.Now().UnixNano(), 36)
|
||
}
|