Files
pangolin/server/internal/scheduler/wiring.go
T
wangjia ebc9c1e702 feat(scheduler): 装配 scheduler — 三 loop + Redis 选主 + 优雅退出 [tsk_rtYVDLkmc5mo]
## 内容

### server/internal/scheduler/deps.go
新增 scheduler 级接口文件,定义所有外部依赖接口:
- DetectEngine / OrchestrateReplacer / OrchestrateGrayscale(15D/15E)
- CapacityService + StubCapacityService(15F 存根)
- ProbeStateReader(15A)、ThirdPartyProber / TargetProvider(15C)
- Notifier + LogNotifier(15G)、StaticTargetProvider

### server/internal/scheduler/scheduler.go
Scheduler 主体实现:
- Config 结构体:注入所有依赖 + 可配置的选主时间参数(方便测试)
- Run(ctx) = 安装 SIGTERM/SIGINT 信号 + Start(ctx)
- Start(ctx) = 启动三 loop goroutine + 可选 15C 采集 goroutine + 30s 优雅退出
- runLoop: 非主实例每 5s 抢租约(SET NX PX 30000)
- runLeaderLoop: 主实例按 interval tick,defer 释放租约
- keepLease: 每 10s Lua 校验后续租(PEXPIRE),失败即退出
- releaseLease: Lua 校验后 DEL,加速接管
- capacityTick: 15F check → 15E grayscale.Advance → 15F decay → 探针失联告警
- runThirdPartyProber: 无主从约束全副本跑(理由注释在代码中)

### server/internal/scheduler/wiring.go
BuildStubConfig(): 用存根 LC/Prov 构建可运行的 Config,
供 server main 和集成测试使用;待 #5/#14 就绪后替换真实实现。

### server/internal/scheduler/scheduler_test.go (go test -race 全通过)
- TestLeaderElection: 两实例竞争,任意时刻只有一个在 tick
- TestFollowerTakeover: 主实例退出后备实例在 ≤LeaseTTL 内接管
- TestGracefulShutdown: ctx 取消后三个 leader key 被主动 DEL
- TestCapacityTickProbeDisconnect: 失联探针触发 Notifier
- TestE2EMockScenario: 端到端 mock 剧本
  up→suspect(降权10)→confirmed→down + 替换队列 →
  CreateNode→probing→activating(weight10+BumpVersion+grayscale)→
  draining_old→done,全链 node_events/audit_log 对账

### server/cmd/server/main.go
- 新增 SCHED_ENABLED=true 开关,灰度启动 scheduler
- 复用 REDIS_ADDR/REDIS_PASSWORD,probe.Store 共享给 scheduler
- 通过 signal.NotifyContext 传递 SIGTERM,确保优雅退出

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-15 22:45:56 +08:00

114 lines
4.7 KiB
Go

package scheduler
// wiring.go — construction helpers for starting the scheduler in the server
// monolith before tasks #5 (LifecycleService) and #14 (ProvisionService) are
// fully implemented.
//
// BuildStubConfig returns a ready-to-use Config whose lifecycle and provision
// dependencies are replaced with no-op stubs. The scheduler's three loops
// still run and hold leader leases, but detect/orchestrate ticks are no-ops.
// Replace the stubs with real implementations once #5/#14 are ready.
import (
"context"
"github.com/redis/go-redis/v9"
"github.com/wangjia/pangolin/server/internal/scheduler/detect"
"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
"github.com/wangjia/pangolin/server/internal/scheduler/probe"
)
// 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).
// - Integration tests that need a scheduler goroutine to be running.
//
// To swap in real implementations once they exist:
//
// cfg := scheduler.BuildStubConfig(rdb, probeStore)
// cfg.Engine = detect.NewEngine(realLC, ...)
// cfg.Replacer = orchestrate.NewReplacer(orchestrate.Config{LC: realLC, Prov: realProv, ...})
// cfg.Grayscale = orchestrate.NewGrayscale(rdb, realLC, nil)
// sched := scheduler.New(cfg)
func BuildStubConfig(rdb *redis.Client, probeStore *probe.Store) Config {
// Stub lifecycle for 15D — no nodes registered, so Tick is a safe no-op.
stubDetectLC := detect.NewMockLifecycle(nil)
// Stub lifecycle and provision for 15E.
stubOrchLC := &stubOrchLC{}
stubProv := &stubProvision{}
streaks := detect.NewStreakStore(rdb)
engine := detect.NewEngine(
probeStore, // 15A probe snapshot reader
stubDetectLC, // 15D lifecycle (stub until #5)
streaks,
rdb,
nil, // notifier — LogNotifier used by default
nil, // config — DefaultConfig used
)
replacer := orchestrate.NewReplacer(orchestrate.Config{
RDB: rdb,
Prov: stubProv, // provision stub until #14
LC: stubOrchLC, // lifecycle stub until #5
Snaps: probeStore,
Breaker: nil, // StubBreaker used by default
Notifier: nil, // LogNotifier used by default
Clock: nil, // RealClock used by default
})
grayscale := orchestrate.NewGrayscale(rdb, stubOrchLC, nil)
return Config{
RDB: rdb,
Engine: engine,
Replacer: replacer,
Grayscale: grayscale,
// Capacity, Notifier, ProbeStore, Prober, Targets: nil → stubs/disabled
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Stub implementations for orchestrate.LifecycleService and ProvisionService
//
// These stubs satisfy the interface contracts while doing nothing harmful.
// Replace with the real services (#5 / #14) when those tasks are complete.
// ─────────────────────────────────────────────────────────────────────────────
// stubOrchLC implements orchestrate.LifecycleService as a no-op.
// GetNode always returns nil (node not found); all mutating calls succeed
// silently. This prevents any orchestration work from proceeding until a real
// lifecycle service is wired in.
type stubOrchLC struct{}
func (stubOrchLC) GetNode(_ context.Context, _ string) (*orchestrate.NodeInfo, error) {
return nil, nil // "node not found" → orchestrate skips this replacement
}
func (stubOrchLC) TransitionStatus(_ context.Context, _, _, _ string, _ map[string]any) (int, error) {
return 1, 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.
// CreateNode always returns an error so pending replacements stay pending
// rather than silently creating phantom resources.
type stubProvision struct{}
func (stubProvision) CreateNode(_ context.Context, _ orchestrate.NodeSpec, _ string) (string, error) {
return "", context.DeadlineExceeded // signal "not ready" without alarming
}
func (stubProvision) DestroyNode(_ context.Context, _ string) error { return nil }
func (stubProvision) RotateIP(_ context.Context, _ string) (string, error) { return "", nil }
func (stubProvision) ListProviders(_ context.Context, _, _ string) ([]orchestrate.ProviderInfo, error) {
return nil, nil
}