From 5f3f4189e57e16489a253d2aa913864dbbfe3146 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Wed, 17 Jun 2026 09:20:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(scheduler):=20lifecycle=20=E7=9C=9F?= =?UTF-8?q?=E5=AE=9E=E6=8E=A5=E7=BA=BF=E9=80=82=E9=85=8D=E5=99=A8(2/2)=20+?= =?UTF-8?q?=20=E8=A3=85=E9=85=8D=20BuildRealConfig?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把 scheduler 的 stub lifecycle 换成 SQL 后端真实实现,并接入 server: - SQLLifecycle：MySQL 后端,同时满足 detect.LifecycleService 与 orchestrate.LifecycleService(两个 thin adapter 包装共享核心)。 - TransitionStatus：乐观锁 UPDATE nodes SET status WHERE id AND status=from → rows affected(1=成功/0=冲突 no-op)+ 写 node_events(按状态映射 enum, eventForStatus 纯函数已单测)+ bump directory_version,全在一个事务内。 - ListNodes/GetNode/SetWeight/BumpVersion/GetLoad(读 LoadCache)/WriteAuditLog。 - 已知限制:无历史 load 表 → GetLoadHistory 返回当前点(掉量规则安全降级); destroy 的 cert 撤销留 TODO(provision 已拆 VM,撤销为纵深防御,待机群再接)。 - BuildRealConfig：用真实 lifecycle + provisionAdapter 装配三循环。 - main.go：SCHED_ENABLED 且有 DB 时走 BuildRealConfig(provision 无厂商凭证则 CreateNode 优雅失败、替换保持 pending),否则回退 stub。 - 全量 server 23 包测试通过;e2e(判封→drain→换机→置备)待机群+厂商凭证验证。 Co-Authored-By: Claude Opus 4.8 --- server/cmd/server/main.go | 30 +- server/internal/scheduler/wiring.go | 57 +++- server/internal/scheduler/wiring_lifecycle.go | 319 ++++++++++++++++++ .../scheduler/wiring_lifecycle_test.go | 22 ++ 4 files changed, 420 insertions(+), 8 deletions(-) create mode 100644 server/internal/scheduler/wiring_lifecycle.go create mode 100644 server/internal/scheduler/wiring_lifecycle_test.go diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 4d3314e..a015941 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -34,6 +34,8 @@ import ( "github.com/wangjia/pangolin/server/internal/mtls" "github.com/wangjia/pangolin/server/internal/nodes" agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" + "github.com/wangjia/pangolin/server/internal/provision" + "github.com/wangjia/pangolin/server/internal/provision/providers" "github.com/wangjia/pangolin/server/internal/redisutil" "github.com/wangjia/pangolin/server/internal/scheduler" "github.com/wangjia/pangolin/server/internal/scheduler/probe" @@ -164,9 +166,29 @@ func main() { if ps == nil { ps = probe.NewStore(schedRDB) } - cfg := scheduler.BuildStubConfig(schedRDB, ps) - // TODO(#5): replace stub lifecycle with real LifecycleService - // TODO(#14): replace stub provision with real ProvisionService + // Prefer real lifecycle (SQL) + provision (#14) wiring when a DB is + // available; fall back to no-op stubs otherwise. Without vendor + // credentials the provision service's CreateNode fails gracefully + // (pending replacements stay pending) while lifecycle reads/transitions + // still operate on real node rows. + var cfg scheduler.Config + if sqlDB != nil { + provStore := provision.NewMySQLStore(sqlDB) + provSvc, perr := provision.NewService(provision.Config{ + Store: provStore, + Adapters: providers.NewRegistry(), + }) + if perr != nil { + log.Printf("scheduler: provision service init failed (%v) — using stub config", perr) + cfg = scheduler.BuildStubConfig(schedRDB, ps) + } else { + cfg = scheduler.BuildRealConfig(schedRDB, ps, sqlDB, + nodes.NewLoadCache(schedRDB), provSvc, provStore) + log.Printf("scheduler: real lifecycle + provision wired") + } + } else { + cfg = scheduler.BuildStubConfig(schedRDB, ps) + } sched := scheduler.New(cfg) // sigCtx cancels on SIGINT/SIGTERM, giving the scheduler up to @@ -179,7 +201,7 @@ func main() { slog.Error("scheduler: Start error", "error", err) } }() - log.Printf("scheduler started (SCHED_ENABLED=true, stub lifecycle/provision)") + log.Printf("scheduler started (SCHED_ENABLED=true)") } } diff --git a/server/internal/scheduler/wiring.go b/server/internal/scheduler/wiring.go index cdf5444..b53e924 100644 --- a/server/internal/scheduler/wiring.go +++ b/server/internal/scheduler/wiring.go @@ -11,14 +11,63 @@ package scheduler import ( "context" + "database/sql" "github.com/redis/go-redis/v9" + "github.com/wangjia/pangolin/server/internal/nodes" "github.com/wangjia/pangolin/server/internal/scheduler/detect" "github.com/wangjia/pangolin/server/internal/scheduler/orchestrate" "github.com/wangjia/pangolin/server/internal/scheduler/probe" ) +// BuildRealConfig wires the scheduler against the real lifecycle (SQL-backed) +// and provision (#14) services, replacing the no-op stubs of BuildStubConfig. +// +// - detect/orchestrate lifecycle → SQLLifecycle over the nodes table + LoadCache. +// - provision → provisionAdapter over the real provision.Service. +// +// Without vendor credentials the provision service's CreateNode fails (a pending +// replacement simply stays pending), but all lifecycle reads/transitions operate +// on real node rows. Capacity monitoring is left disabled (nil) as in the stub. +// +// e2e behaviour (auto-detect → drain → replace → provision) is verifiable only +// against a real node fleet with vendor provisioning configured. +func BuildRealConfig( + rdb *redis.Client, + probeStore *probe.Store, + db *sql.DB, + loads *nodes.LoadCache, + provSvc provisionSvc, + provStore provisionResolver, +) Config { + sqlLC := NewSQLLifecycle(db, loads) + detectLC := NewDetectLifecycle(sqlLC) + orchLC := NewOrchestrateLifecycle(sqlLC) + prov := NewProvisionAdapter(provSvc, provStore) + + streaks := detect.NewStreakStore(rdb) + engine := detect.NewEngine(probeStore, detectLC, streaks, rdb, nil, nil) + + replacer := orchestrate.NewReplacer(orchestrate.Config{ + RDB: rdb, + Prov: prov, + LC: orchLC, + Snaps: probeStore, + Breaker: nil, // StubBreaker default until 15F policy tuned + Notifier: nil, // LogNotifier default + Clock: nil, + }) + grayscale := orchestrate.NewGrayscale(rdb, orchLC, nil) + + return Config{ + RDB: rdb, + Engine: engine, + Replacer: replacer, + Grayscale: grayscale, + } +} + // BuildStubConfig constructs a scheduler Config wired entirely with in-process // stubs. Useful for: // - Starting the scheduler before tasks #5 / #14 are delivered (no-op ticks). @@ -41,8 +90,8 @@ func BuildStubConfig(rdb *redis.Client, probeStore *probe.Store) Config { streaks := detect.NewStreakStore(rdb) engine := detect.NewEngine( - probeStore, // 15A probe snapshot reader - stubDetectLC, // 15D lifecycle (stub until #5) + probeStore, // 15A probe snapshot reader + stubDetectLC, // 15D lifecycle (stub until #5) streaks, rdb, nil, // notifier — LogNotifier used by default @@ -91,8 +140,8 @@ func (stubOrchLC) TransitionStatus(_ context.Context, _, _, _ string, _ map[stri return 1, nil } -func (stubOrchLC) SetWeight(_ context.Context, _ string, _ int) error { return nil } -func (stubOrchLC) BumpVersion(_ context.Context) error { return nil } +func (stubOrchLC) SetWeight(_ context.Context, _ string, _ int) error { return nil } +func (stubOrchLC) BumpVersion(_ context.Context) error { return nil } func (stubOrchLC) WriteAuditLog(_ context.Context, _, _, _, _ string) error { return nil } // stubProvision implements orchestrate.ProvisionService as a no-op. diff --git a/server/internal/scheduler/wiring_lifecycle.go b/server/internal/scheduler/wiring_lifecycle.go new file mode 100644 index 0000000..818355a --- /dev/null +++ b/server/internal/scheduler/wiring_lifecycle.go @@ -0,0 +1,319 @@ +package scheduler + +import ( + "context" + "database/sql" + "encoding/json" + "strconv" + "strings" + "time" + + "github.com/wangjia/pangolin/server/internal/nodes" + "github.com/wangjia/pangolin/server/internal/scheduler/detect" + "github.com/wangjia/pangolin/server/internal/scheduler/orchestrate" +) + +// SQLLifecycle is a MySQL-backed lifecycle store. The detect and orchestrate +// LifecycleService interfaces are satisfied by the two thin adapter wrappers at +// the bottom of this file (detectLifecycle / orchestrateLifecycle), both of +// which delegate to this shared core. +// +// TransitionStatus is the documented optimistic-lock contract: +// +// UPDATE nodes SET status=to WHERE id=nodeID AND status=from +// +// returning rows affected (1 = applied, 0 = lock conflict / no-op). On a 1-row +// transition it best-effort writes a node_events row and bumps the global +// directory_version so clients re-fetch. +// +// NOTE: cert revocation on a destroyed transition is intentionally NOT done here +// — provision.DestroyNode tears down the VM, and the agent's mTLS cert can be +// revoked separately. Wire CRL revocation when the real fleet exists (TODO). +type SQLLifecycle struct { + db *sql.DB + loads *nodes.LoadCache +} + +// NewSQLLifecycle wires the store. loads may be nil (load reads then return zero). +func NewSQLLifecycle(db *sql.DB, loads *nodes.LoadCache) *SQLLifecycle { + return &SQLLifecycle{db: db, loads: loads} +} + +const nodeSelectCols = `id, uuid, status, region, tier, role, reality_sni, reality_pbk, hy2_port, provider_id, weight, name_zh, name_en` + +type lcRow struct { + id int64 + uuid string + status string + region string + tier string + role string + sni string + pbk string + hy2Port int + providerID int64 + weight int + nameZH string + nameEn string +} + +func scanNode(rows interface{ Scan(...any) error }) (lcRow, error) { + var r lcRow + var hy2 sql.NullInt32 + err := rows.Scan(&r.id, &r.uuid, &r.status, &r.region, &r.tier, &r.role, + &r.sni, &r.pbk, &hy2, &r.providerID, &r.weight, &r.nameZH, &r.nameEn) + if hy2.Valid { + r.hy2Port = int(hy2.Int32) + } + return r, err +} + +func (l *SQLLifecycle) list(ctx context.Context, statuses []string) ([]lcRow, int64, error) { + q := `SELECT ` + nodeSelectCols + ` FROM nodes` + var args []any + if len(statuses) > 0 { + ph := make([]string, len(statuses)) + for i, s := range statuses { + ph[i] = "?" + args = append(args, s) + } + q += ` WHERE status IN (` + strings.Join(ph, ",") + `)` + } + rows, err := l.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + var out []lcRow + for rows.Next() { + r, err := scanNode(rows) + if err != nil { + return nil, 0, err + } + out = append(out, r) + } + return out, l.version(ctx), rows.Err() +} + +func (l *SQLLifecycle) get(ctx context.Context, nodeID string) (*lcRow, error) { + row := l.db.QueryRowContext(ctx, `SELECT `+nodeSelectCols+` FROM nodes WHERE id = ?`, nodeID) + r, err := scanNode(row) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + return &r, nil +} + +func (l *SQLLifecycle) transition(ctx context.Context, nodeID, from, to string, detail map[string]any) (int, error) { + tx, err := l.db.BeginTx(ctx, nil) + if err != nil { + return 0, err + } + defer func() { _ = tx.Rollback() }() + + res, err := tx.ExecContext(ctx, + `UPDATE nodes SET status = ? WHERE id = ? AND status = ?`, to, nodeID, from) + if err != nil { + return 0, err + } + n, _ := res.RowsAffected() + if n == 0 { + return 0, nil // optimistic-lock conflict: node already moved → no-op + } + + if ev := eventForStatus(to); ev != "" { + var dj any + if detail != nil { + if b, err := json.Marshal(detail); err == nil { + dj = string(b) + } + } + _, _ = tx.ExecContext(ctx, + `INSERT INTO node_events (node_id, event, detail) VALUES (?, ?, ?)`, nodeID, ev, dj) + } + if _, err := tx.ExecContext(ctx, + `INSERT INTO directory_version (id, version) VALUES (1, 1) + ON DUPLICATE KEY UPDATE version = version + 1`); err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, err + } + return int(n), nil +} + +func (l *SQLLifecycle) setWeight(ctx context.Context, nodeID string, weight int) error { + _, err := l.db.ExecContext(ctx, `UPDATE nodes SET weight = ? WHERE id = ?`, weight, nodeID) + return err +} + +func (l *SQLLifecycle) bumpVersion(ctx context.Context) error { + _, err := l.db.ExecContext(ctx, + `INSERT INTO directory_version (id, version) VALUES (1, 1) + ON DUPLICATE KEY UPDATE version = version + 1`) + return err +} + +func (l *SQLLifecycle) version(ctx context.Context) int64 { + var v int64 + _ = l.db.QueryRowContext(ctx, `SELECT version FROM directory_version WHERE id = 1`).Scan(&v) + return v +} + +// loadFor resolves a node id → uuid → its current load sample (online, Mbps). +func (l *SQLLifecycle) loadFor(ctx context.Context, nodeID string) (int, float64) { + if l.loads == nil { + return 0, 0 + } + var uuid string + if err := l.db.QueryRowContext(ctx, `SELECT uuid FROM nodes WHERE id = ?`, nodeID).Scan(&uuid); err != nil { + return 0, 0 + } + nl, ok, err := l.loads.Get(ctx, uuid) + if err != nil || !ok || nl == nil { + return 0, 0 + } + mbps := float64(nl.BandwidthUpBps+nl.BandwidthDownBps) / 1_000_000.0 + return int(nl.OnlineCount), mbps +} + +func (l *SQLLifecycle) writeAudit(ctx context.Context, actor, action, target, meta string) error { + var metaArg any + if meta != "" { + metaArg = meta + } + _, err := l.db.ExecContext(ctx, + `INSERT INTO audit_log (actor, action, target, meta) VALUES (?, ?, ?, ?)`, + actor, action, target, metaArg) + return err +} + +// eventForStatus maps a target node status to the node_events ENUM value, or "" +// when the status has no corresponding event (then no event row is written). +func eventForStatus(status string) string { + switch status { + case "provisioning", "probing": + return "provisioned" + case "up": + return "marked_up" + case "draining": + return "draining" + case "blocked_suspect": + return "blocked_suspect" + case "blocked_confirmed": + return "blocked_confirmed" + case "destroyed": + return "destroyed" + default: + return "" // e.g. "down" has no event enum + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// detect.LifecycleService adapter +// ───────────────────────────────────────────────────────────────────────────── + +type detectLifecycle struct{ *SQLLifecycle } + +// NewDetectLifecycle adapts SQLLifecycle to detect.LifecycleService. +func NewDetectLifecycle(l *SQLLifecycle) detect.LifecycleService { return detectLifecycle{l} } + +func (a detectLifecycle) ListNodes(ctx context.Context, f detect.NodeFilter) ([]detect.NodeInfo, error) { + statuses := make([]string, len(f.Statuses)) + for i, s := range f.Statuses { + statuses[i] = string(s) + } + rows, ver, err := a.list(ctx, statuses) + if err != nil { + return nil, err + } + out := make([]detect.NodeInfo, 0, len(rows)) + for _, r := range rows { + out = append(out, detect.NodeInfo{ + ID: strconv.FormatInt(r.id, 10), + UUID: r.uuid, + Status: detect.NodeStatus(r.status), + Weight: r.weight, + Version: ver, + }) + } + return out, nil +} + +func (a detectLifecycle) TransitionStatus(ctx context.Context, nodeID string, from, to detect.NodeStatus, detail map[string]any) (int, error) { + return a.transition(ctx, nodeID, string(from), string(to), detail) +} + +func (a detectLifecycle) SetWeight(ctx context.Context, nodeID string, weight int) error { + return a.setWeight(ctx, nodeID, weight) +} + +func (a detectLifecycle) BumpVersion(ctx context.Context) error { return a.bumpVersion(ctx) } + +func (a detectLifecycle) GetLoad(ctx context.Context, nodeID string) (detect.LoadInfo, error) { + online, mbps := a.loadFor(ctx, nodeID) + return detect.LoadInfo{Online: online, BandwidthMbps: mbps, Timestamp: time.Now().Unix()}, nil +} + +func (a detectLifecycle) GetLoadHistory(ctx context.Context, nodeID string, _ time.Duration) ([]detect.LoadPoint, error) { + // No historical load store yet (only the latest sample lives in Redis), so + // return the current point. The drop-percentage rule degrades safely (no fire). + online, mbps := a.loadFor(ctx, nodeID) + if online == 0 && mbps == 0 { + return nil, nil + } + return []detect.LoadPoint{{Timestamp: time.Now().Unix(), Online: online, BandwidthMbps: mbps}}, nil +} + +// ───────────────────────────────────────────────────────────────────────────── +// orchestrate.LifecycleService adapter +// ───────────────────────────────────────────────────────────────────────────── + +type orchestrateLifecycle struct{ *SQLLifecycle } + +// NewOrchestrateLifecycle adapts SQLLifecycle to orchestrate.LifecycleService. +func NewOrchestrateLifecycle(l *SQLLifecycle) orchestrate.LifecycleService { + return orchestrateLifecycle{l} +} + +func (a orchestrateLifecycle) GetNode(ctx context.Context, nodeID string) (*orchestrate.NodeInfo, error) { + r, err := a.get(ctx, nodeID) + if err != nil || r == nil { + return nil, err + } + return &orchestrate.NodeInfo{ + ID: strconv.FormatInt(r.id, 10), + Tier: r.tier, + Region: r.region, + Role: r.role, + ProviderID: strconv.FormatInt(r.providerID, 10), + RealitySNI: r.sni, + RealityPBK: r.pbk, + HY2Port: r.hy2Port, + NameZH: r.nameZH, + NameEn: r.nameEn, + }, nil +} + +func (a orchestrateLifecycle) TransitionStatus(ctx context.Context, nodeID string, from, to string, detail map[string]any) (int, error) { + return a.transition(ctx, nodeID, from, to, detail) +} + +func (a orchestrateLifecycle) SetWeight(ctx context.Context, nodeID string, weight int) error { + return a.setWeight(ctx, nodeID, weight) +} + +func (a orchestrateLifecycle) BumpVersion(ctx context.Context) error { return a.bumpVersion(ctx) } + +func (a orchestrateLifecycle) WriteAuditLog(ctx context.Context, actor, action, target, meta string) error { + return a.writeAudit(ctx, actor, action, target, meta) +} + +// assert at compile time that the adapters satisfy their interfaces. +var ( + _ detect.LifecycleService = detectLifecycle{} + _ orchestrate.LifecycleService = orchestrateLifecycle{} +) diff --git a/server/internal/scheduler/wiring_lifecycle_test.go b/server/internal/scheduler/wiring_lifecycle_test.go new file mode 100644 index 0000000..284978d --- /dev/null +++ b/server/internal/scheduler/wiring_lifecycle_test.go @@ -0,0 +1,22 @@ +package scheduler + +import "testing" + +func TestEventForStatus(t *testing.T) { + cases := map[string]string{ + "up": "marked_up", + "draining": "draining", + "blocked_suspect": "blocked_suspect", + "blocked_confirmed": "blocked_confirmed", + "destroyed": "destroyed", + "probing": "provisioned", + "provisioning": "provisioned", + "down": "", // no event enum → skip writing an event row + "bogus": "", + } + for status, want := range cases { + if got := eventForStatus(status); got != want { + t.Errorf("eventForStatus(%q) = %q, want %q", status, got, want) + } + } +}