diff --git a/server/internal/scheduler/orchestrate/breaker.go b/server/internal/scheduler/orchestrate/breaker.go new file mode 100644 index 0000000..1aba7b4 --- /dev/null +++ b/server/internal/scheduler/orchestrate/breaker.go @@ -0,0 +1,307 @@ +package orchestrate + +import ( + "context" + "fmt" + "log/slog" + "math" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/wangjia/pangolin/server/internal/idgen" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Redis key constants +// ───────────────────────────────────────────────────────────────────────────── + +const ( + // breakerWindowKey is the sliding-window ZSET key prefix. + // Full key: sched:breaker:{tier}:{region} + // Schema: member = replacement UUID, score = Unix timestamp (seconds). + breakerWindowKey = "sched:breaker:" + + // breakerTripKey is the trip-flag key prefix. + // Full key: sched:breaker:trip:{tier}:{region} + // TTL = BreakerWindowMin. Used for: alert dedup + admin visibility. + // The key auto-expires when the sliding window clears (enabling auto-recovery). + breakerTripKey = "sched:breaker:trip:" +) + +// ───────────────────────────────────────────────────────────────────────────── +// RedisBreaker +// ───────────────────────────────────────────────────────────────────────────── + +// RedisBreaker is the production 15F circuit-breaker. It satisfies the +// Breaker interface (Allow + Record) and additionally exposes Reset for the +// admin handler (task #8). +// +// Mechanism: +// +// ZSET sched:breaker:{tier}:{region} +// member = replacement UUID (unique per completed replacement) +// score = Unix timestamp of completion +// +// Window: BreakerWindowMin (default 60 min). Each Allow call +// runs ZREMRANGEBYSCORE first to expire out-of-window entries. +// +// Threshold N: +// N = ceil(poolTarget × BreakerFractionPct / 100), min BreakerMinN (3). +// +// Trip: +// When window-count ≥ N: Allow returns false. The first detection in a +// window epoch sets a trip-flag key with TTL = BreakerWindowMin and emits +// a critical alert (idempotent via SetNX). +// +// Auto-recovery: +// After BreakerWindowMin minutes with no new Record calls, all ZSET entries +// expire out of the window and the trip flag TTL also expires → Allow true. +// +// Manual reset: +// Admin calls Reset, which deletes the trip flag and the ZSET immediately. +type RedisBreaker struct { + rdb *redis.Client + pools PoolReader + notifier Notifier + lc LifecycleService // for audit log; may be nil + cfgMgr *ConfigManager + clock Clock +} + +// RedisBreakerConfig holds all dependencies for NewRedisBreaker. +type RedisBreakerConfig struct { + RDB *redis.Client + Pools PoolReader + Notifier Notifier + LC LifecycleService // for audit log; may be nil + CfgMgr *ConfigManager + Clock Clock +} + +// NewRedisBreaker constructs a RedisBreaker. If Clock is nil, RealClock is used. +func NewRedisBreaker(cfg RedisBreakerConfig) *RedisBreaker { + if cfg.Clock == nil { + cfg.Clock = RealClock{} + } + return &RedisBreaker{ + rdb: cfg.RDB, + pools: cfg.Pools, + notifier: cfg.Notifier, + lc: cfg.LC, + cfgMgr: cfg.CfgMgr, + clock: cfg.Clock, + } +} + +// Allow implements Breaker. Returns false when the pool's sliding-window +// replacement count has reached the threshold (breaker is tripped), blocking +// further replacements from starting. +// +// On Redis errors the breaker fails open (returns true) to avoid a single +// point of failure halting all replacements. +func (b *RedisBreaker) Allow(tier, region string) bool { + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + + cfg := b.cfgMgr.Current() + window := time.Duration(cfg.Capacity.BreakerWindowMin) * time.Minute + threshold := b.computeThreshold(ctx, tier, region, cfg) + + zKey := breakerWindowKey + tier + ":" + region + tripKey := breakerTripKey + tier + ":" + region + cutoff := b.clock.Now().Add(-window) + + // Expire entries outside the sliding window. + cutoffStr := fmt.Sprintf("%d", cutoff.Unix()) + if err := b.rdb.ZRemRangeByScore(ctx, zKey, "-inf", cutoffStr).Err(); err != nil { + slog.Error("breaker: ZREMRANGEBYSCORE failed; failing open", + "tier", tier, "region", region, "error", err) + return true // fail open + } + + // Count remaining (in-window) entries. + count, err := b.rdb.ZCard(ctx, zKey).Result() + if err != nil { + slog.Error("breaker: ZCARD failed; failing open", + "tier", tier, "region", region, "error", err) + return true // fail open + } + + if count < int64(threshold) { + return true // below threshold — allow replacement + } + + // Count ≥ threshold: trip. Emit alert exactly once per window epoch via SetNX. + set, setErr := b.rdb.SetNX(ctx, tripKey, "1", window).Result() + if setErr != nil { + slog.Error("breaker: trip flag set failed", + "tier", tier, "region", region, "error", setErr) + } + if set { + b.emitTripAlert(ctx, tier, region, int(count), threshold, cfg) + } + + slog.Warn("breaker: replacement blocked — circuit open", + "tier", tier, "region", region, + "window_count", count, "threshold", threshold, + ) + return false +} + +// Record implements Breaker. Records a completed replacement for the given +// pool in the sliding-window ZSET. Called by the orchestrator after the old +// node is successfully destroyed. +// +// Record also checks whether the count has just reached the threshold and, +// if so, trips the breaker and emits a critical alert. +func (b *RedisBreaker) Record(tier, region string) { + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + + cfg := b.cfgMgr.Current() + window := time.Duration(cfg.Capacity.BreakerWindowMin) * time.Minute + threshold := b.computeThreshold(ctx, tier, region, cfg) + + zKey := breakerWindowKey + tier + ":" + region + tripKey := breakerTripKey + tier + ":" + region + now := b.clock.Now() + + // Add a unique entry for this replacement. + member := idgen.NewString() + if err := b.rdb.ZAdd(ctx, zKey, redis.Z{ + Score: float64(now.Unix()), + Member: member, + }).Err(); err != nil { + slog.Error("breaker: ZADD failed", + "tier", tier, "region", region, "error", err) + return + } + + // Check if we have just reached the threshold; trip once per epoch. + count, err := b.rdb.ZCard(ctx, zKey).Result() + if err != nil { + slog.Error("breaker: ZCARD after record failed", + "tier", tier, "region", region, "error", err) + return + } + + if count >= int64(threshold) { + set, setErr := b.rdb.SetNX(ctx, tripKey, "1", window).Result() + if setErr != nil { + slog.Error("breaker: trip flag on record failed", + "tier", tier, "region", region, "error", setErr) + return + } + if set { + // First trip in this window epoch: emit critical alert. + b.emitTripAlert(ctx, tier, region, int(count), threshold, cfg) + } + } +} + +// Reset clears a tripped breaker for the given pool by deleting both the trip +// flag and the sliding-window ZSET. This lets new replacements proceed +// immediately without waiting for the window to auto-expire. +// +// An audit log entry is written with actor and action metadata. +// Reset is not part of the Breaker interface; it is called only by the admin +// handler (task #8). +func (b *RedisBreaker) Reset(ctx context.Context, tier, region, actor string) error { + zKey := breakerWindowKey + tier + ":" + region + tripKey := breakerTripKey + tier + ":" + region + + pipe := b.rdb.Pipeline() + pipe.Del(ctx, tripKey) + pipe.Del(ctx, zKey) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("breaker: reset %s/%s: %w", tier, region, err) + } + + slog.Info("breaker: manually reset by operator", + "tier", tier, "region", region, "actor", actor) + + // Audit trail. + meta := fmt.Sprintf( + `{"tier":%q,"region":%q,"actor":%q,"action":"breaker_reset"}`, + tier, region, actor, + ) + if b.lc != nil { + if err := b.lc.WriteAuditLog( + ctx, actor, "breaker_reset", "pool:"+tier+":"+region, meta, + ); err != nil { + slog.Error("breaker: write audit log failed", "error", err) + } + } + return nil +} + +// IsTripped reports whether the trip flag is currently set for the pool. +// Useful for admin status queries. +func (b *RedisBreaker) IsTripped(ctx context.Context, tier, region string) (bool, error) { + tripKey := breakerTripKey + tier + ":" + region + exists, err := b.rdb.Exists(ctx, tripKey).Result() + if err != nil { + return false, fmt.Errorf("breaker: IsTripped %s/%s: %w", tier, region, err) + } + return exists > 0, nil +} + +// WindowCount returns the number of completed replacements currently recorded +// in the sliding window for the pool. Useful for admin status queries. +func (b *RedisBreaker) WindowCount(ctx context.Context, tier, region string) (int64, error) { + cfg := b.cfgMgr.Current() + window := time.Duration(cfg.Capacity.BreakerWindowMin) * time.Minute + cutoff := b.clock.Now().Add(-window) + cutoffStr := fmt.Sprintf("%d", cutoff.Unix()) + + zKey := breakerWindowKey + tier + ":" + region + if err := b.rdb.ZRemRangeByScore(ctx, zKey, "-inf", cutoffStr).Err(); err != nil { + return 0, fmt.Errorf("breaker: WindowCount cleanup %s/%s: %w", tier, region, err) + } + count, err := b.rdb.ZCard(ctx, zKey).Result() + if err != nil { + return 0, fmt.Errorf("breaker: WindowCount ZCARD %s/%s: %w", tier, region, err) + } + return count, nil +} + +// ───────────────────────────────────────────────────────────────────────────── +// Helpers +// ───────────────────────────────────────────────────────────────────────────── + +// computeThreshold returns the trip threshold for a pool: +// +// N = ceil(poolTarget × BreakerFractionPct / 100), minimum BreakerMinN. +// +// Falls back to BreakerMinN when the pool cannot be found or has zero target. +func (b *RedisBreaker) computeThreshold(ctx context.Context, tier, region string, cfg *SchedConfig) int { + _, target, err := b.pools.PoolCapacity(ctx, tier, region) + if err != nil || target <= 0 { + return cfg.Capacity.BreakerMinN + } + n := int(math.Ceil(float64(target) * float64(cfg.Capacity.BreakerFractionPct) / 100.0)) + if n < cfg.Capacity.BreakerMinN { + n = cfg.Capacity.BreakerMinN + } + return n +} + +// emitTripAlert sends a critical alert to the Notifier for a tripped breaker. +func (b *RedisBreaker) emitTripAlert(ctx context.Context, tier, region string, count, threshold int, cfg *SchedConfig) { + poolID := "pool:" + tier + ":" + region + reason := fmt.Sprintf( + "circuit breaker tripped: %d replacements in %d-min window (threshold=%d) — manual Reset required", + count, cfg.Capacity.BreakerWindowMin, threshold, + ) + slog.Error("breaker: circuit breaker tripped", + "tier", tier, "region", region, + "window_count", count, "threshold", threshold, + ) + if b.notifier != nil { + if err := b.notifier.NotifyFault(ctx, poolID, reason); err != nil { + slog.Error("breaker: emit trip alert failed", + "pool", poolID, "error", err) + } + } +} diff --git a/server/internal/scheduler/orchestrate/breaker_test.go b/server/internal/scheduler/orchestrate/breaker_test.go new file mode 100644 index 0000000..26e6ac8 --- /dev/null +++ b/server/internal/scheduler/orchestrate/breaker_test.go @@ -0,0 +1,295 @@ +package orchestrate_test + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/wangjia/pangolin/server/internal/scheduler/orchestrate" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Mock PoolReader +// ───────────────────────────────────────────────────────────────────────────── + +// mockPoolReader is a simple in-memory PoolReader for tests. +type mockPoolReader struct { + mu sync.Mutex + pools map[string]orchestrate.PoolStat // key = tier+":"+region +} + +func newMockPoolReader() *mockPoolReader { + return &mockPoolReader{pools: make(map[string]orchestrate.PoolStat)} +} + +func (r *mockPoolReader) setPool(tier, region string, up, target int) { + r.mu.Lock() + defer r.mu.Unlock() + r.pools[tier+":"+region] = orchestrate.PoolStat{ + Tier: tier, Region: region, UpCount: up, Target: target, + } +} + +func (r *mockPoolReader) AllPools(_ context.Context) ([]orchestrate.PoolStat, error) { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]orchestrate.PoolStat, 0, len(r.pools)) + for _, s := range r.pools { + out = append(out, s) + } + return out, nil +} + +func (r *mockPoolReader) PoolCapacity(_ context.Context, tier, region string) (int, int, error) { + r.mu.Lock() + defer r.mu.Unlock() + s, ok := r.pools[tier+":"+region] + if !ok { + return 0, 0, nil + } + return s.UpCount, s.Target, nil +} + +// ───────────────────────────────────────────────────────────────────────────── +// Helper: build a RedisBreaker for tests +// ───────────────────────────────────────────────────────────────────────────── + +func newTestBreaker(t *testing.T, pools *mockPoolReader, notifier *mockNotifier, clock *mockClock) (*orchestrate.RedisBreaker, *orchestrate.ConfigManager) { + t.Helper() + rdb, _ := newTestRedis(t) + cfgMgr := orchestrate.NewConfigManager("", nil) + _ = cfgMgr.Load() + + return orchestrate.NewRedisBreaker(orchestrate.RedisBreakerConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + CfgMgr: cfgMgr, + Clock: clock, + }), cfgMgr +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: N-th Record trips the breaker; (N+1)-th Allow returns false +// ───────────────────────────────────────────────────────────────────────────── + +func TestBreakerTripsAtThreshold(t *testing.T) { + const ( + tier = "free" + region = "hkg" + ) + + pools := newMockPoolReader() + // Pool of 10 nodes → threshold = ceil(10 × 30%) = 3. + pools.setPool(tier, region, 10, 10) + + notifier := &mockNotifier{} + clock := newMockClock(time.Unix(1_700_000_000, 0).UTC()) + + b, _ := newTestBreaker(t, pools, notifier, clock) + + // Before any records: Allow should return true. + if !b.Allow(tier, region) { + t.Fatal("Allow should return true before any records") + } + + // Record N-1 replacements (threshold-1 = 2); Allow should still be true. + b.Record(tier, region) // count = 1 + b.Record(tier, region) // count = 2 + if !b.Allow(tier, region) { + t.Fatalf("Allow should return true with count < threshold (count=2, threshold=3)") + } + + // Record the N-th replacement; now count = 3 = threshold. + b.Record(tier, region) // count = 3 + + // (N+1)-th Allow: breaker is tripped → false. + if b.Allow(tier, region) { + t.Error("Allow should return false when count >= threshold (tripped)") + } + + // Critical alert must have been emitted exactly once. + if n := notifier.count(); n != 1 { + t.Errorf("NotifyFault calls = %d; want 1 (on trip)", n) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Window slide-out auto-recovers the breaker +// ───────────────────────────────────────────────────────────────────────────── + +func TestBreakerAutoRecoveryAfterWindowExpiry(t *testing.T) { + const ( + tier = "premium" + region = "sin" + ) + + pools := newMockPoolReader() + // Pool of 10 → threshold = 3. + pools.setPool(tier, region, 10, 10) + + notifier := &mockNotifier{} + clock := newMockClock(time.Unix(1_700_000_000, 0).UTC()) + b, _ := newTestBreaker(t, pools, notifier, clock) + + // Record N entries to trip. + b.Record(tier, region) + b.Record(tier, region) + b.Record(tier, region) + if b.Allow(tier, region) { + t.Fatal("breaker should be tripped after 3 records (threshold=3)") + } + + // Advance clock past the 1-h window; all ZSET entries are now stale. + clock.advance(61 * time.Minute) + + // Allow should return true: ZREMRANGEBYSCORE removes stale entries, + // count drops to 0 < threshold, and the trip flag TTL has expired. + if !b.Allow(tier, region) { + t.Error("Allow should return true after the window has expired (auto-recovery)") + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Reset manually clears the breaker with audit +// ───────────────────────────────────────────────────────────────────────────── + +func TestBreakerResetRestoresAllow(t *testing.T) { + const ( + tier = "free" + region = "tyo" + ) + + pools := newMockPoolReader() + pools.setPool(tier, region, 10, 10) // threshold = 3 + + notifier := &mockNotifier{} + lc := newMockLC() + clock := newMockClock(time.Unix(1_700_000_000, 0).UTC()) + rdb, _ := newTestRedis(t) + cfgMgr := orchestrate.NewConfigManager("", nil) + _ = cfgMgr.Load() + + b := orchestrate.NewRedisBreaker(orchestrate.RedisBreakerConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + LC: lc, + CfgMgr: cfgMgr, + Clock: clock, + }) + + // Trip the breaker. + b.Record(tier, region) + b.Record(tier, region) + b.Record(tier, region) + if b.Allow(tier, region) { + t.Fatal("breaker should be tripped") + } + + // Reset. + ctx := context.Background() + if err := b.Reset(ctx, tier, region, "admin-alice"); err != nil { + t.Fatalf("Reset error: %v", err) + } + + // Allow should now return true. + if !b.Allow(tier, region) { + t.Error("Allow should return true after Reset") + } + + // Audit log must mention the reset. + if !lc.hasAudit("admin-alice|breaker_reset|pool:" + tier + ":" + region) { + t.Errorf("audit log missing breaker_reset entry; got: %v", lc.auditLogs) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: BreakerMinN = 3 enforced on tiny pool (target < 10) +// ───────────────────────────────────────────────────────────────────────────── + +func TestBreakerMinNEnforcedOnSmallPool(t *testing.T) { + const ( + tier = "free" + region = "fra" + ) + + pools := newMockPoolReader() + // Pool of 2 nodes; 30% of 2 = 0.6 → ceil = 1, but min = 3. + pools.setPool(tier, region, 2, 2) + + notifier := &mockNotifier{} + clock := newMockClock(time.Unix(1_700_000_000, 0).UTC()) + b, _ := newTestBreaker(t, pools, notifier, clock) + + // 2 records should NOT trip (threshold = 3, not 1). + b.Record(tier, region) + b.Record(tier, region) + if !b.Allow(tier, region) { + t.Error("Allow should return true: count=2, threshold=min(3)") + } + + // 3rd record trips. + b.Record(tier, region) + if b.Allow(tier, region) { + t.Error("Allow should return false: count=3 >= minN(3)") + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Trip alert is emitted only once per window epoch (dedup) +// ───────────────────────────────────────────────────────────────────────────── + +func TestBreakerAlertSentOnlyOnce(t *testing.T) { + const ( + tier = "free" + region = "ams" + ) + + pools := newMockPoolReader() + pools.setPool(tier, region, 10, 10) // threshold = 3 + + notifier := &mockNotifier{} + clock := newMockClock(time.Unix(1_700_000_000, 0).UTC()) + b, _ := newTestBreaker(t, pools, notifier, clock) + + // Trip. + b.Record(tier, region) + b.Record(tier, region) + b.Record(tier, region) + + // Call Allow multiple times while tripped. + for i := 0; i < 5; i++ { + b.Allow(tier, region) + } + + // NotifyFault must have been called exactly once (from Record reaching N). + if n := notifier.count(); n != 1 { + t.Errorf("NotifyFault calls = %d; want exactly 1 (trip alert dedup)", n) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: pool not found → falls back to BreakerMinN (3) +// ───────────────────────────────────────────────────────────────────────────── + +func TestBreakerFallsBackToMinNForUnknownPool(t *testing.T) { + pools := newMockPoolReader() + // No pool registered for "unknown"/"xyz". + + notifier := &mockNotifier{} + clock := newMockClock(time.Unix(1_700_000_000, 0).UTC()) + b, _ := newTestBreaker(t, pools, notifier, clock) + + b.Record("unknown", "xyz") + b.Record("unknown", "xyz") + if !b.Allow("unknown", "xyz") { + t.Error("Allow should return true: count=2, threshold=minN(3) for unknown pool") + } + + b.Record("unknown", "xyz") + if b.Allow("unknown", "xyz") { + t.Error("Allow should return false: count=3 >= minN(3)") + } +} diff --git a/server/internal/scheduler/orchestrate/capacity.go b/server/internal/scheduler/orchestrate/capacity.go new file mode 100644 index 0000000..dbea5b0 --- /dev/null +++ b/server/internal/scheduler/orchestrate/capacity.go @@ -0,0 +1,242 @@ +package orchestrate + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/redis/go-redis/v9" +) + +// ───────────────────────────────────────────────────────────────────────────── +// PoolReader — pool capacity interface +// ───────────────────────────────────────────────────────────────────────────── + +// PoolStat holds capacity metrics for a single tier+region pool. +type PoolStat struct { + Tier string + Region string + UpCount int // number of nodes currently in "up" status + Target int // configured target node count for this pool +} + +// PoolReader provides pool capacity information for the capacity monitor and +// circuit breaker. The real implementation (wired by task 15H) queries the +// lifecycle service. +type PoolReader interface { + // AllPools returns all known pools with their current UpCount and Target. + AllPools(ctx context.Context) ([]PoolStat, error) + + // PoolCapacity returns (upCount, target) for the named pool. + // Returns (0, 0, nil) when the pool is not found. + PoolCapacity(ctx context.Context, tier, region string) (int, int, error) +} + +// ───────────────────────────────────────────────────────────────────────────── +// Redis key constants +// ───────────────────────────────────────────────────────────────────────────── + +const ( + // capAlertPoolPrefix is the dedup key prefix for pool-watermark alerts. + // Full key: sched:cap:pool:{tier}:{region} TTL = AlertSuppressMin + capAlertPoolPrefix = "sched:cap:pool:" + + // capAlertProbePrefix is the dedup key prefix for probe-lost-contact alerts. + // Full key: sched:cap:probe:{probeId} TTL = AlertSuppressMin + capAlertProbePrefix = "sched:cap:probe:" + + // probeHBKeyPrefix is the heartbeat key prefix written by the probe package. + // Full key: probe:hb:{probeId} TTL = 15 min (from probe.heartbeatTTL) + probeHBKeyPrefix = "probe:hb:" +) + +// ───────────────────────────────────────────────────────────────────────────── +// CapacityMonitor +// ───────────────────────────────────────────────────────────────────────────── + +// CapacityMonitorConfig holds all dependencies for CapacityMonitor. +type CapacityMonitorConfig struct { + RDB *redis.Client + Pools PoolReader + Notifier Notifier + Gray *Grayscale // grayscale weight-ramp driver + LC LifecycleService // for WriteAuditLog; may be nil + CfgMgr *ConfigManager + Clock Clock +} + +// CapacityMonitor checks pool watermarks and probe heartbeats, and drives +// the grayscale weight ramp. It is called by the CapacityLoop (task 15H) +// approximately every 1–2 minutes. +// +// Each RunOnce call: +// 1. Checks pool fill-rates; emits a throttled alert per pool below threshold. +// 2. Checks probe heartbeats for every configured probe ID; emits a throttled +// 「探针失联」alert for each probe whose heartbeat key is absent. +// 3. Calls Grayscale.Advance to advance any in-progress weight ramps. +type CapacityMonitor struct { + rdb *redis.Client + pools PoolReader + notifier Notifier + gray *Grayscale + lc LifecycleService + cfgMgr *ConfigManager + clock Clock +} + +// NewCapacityMonitor constructs a CapacityMonitor from the given config. +// If Clock is nil, RealClock is used. +func NewCapacityMonitor(cfg CapacityMonitorConfig) *CapacityMonitor { + if cfg.Clock == nil { + cfg.Clock = RealClock{} + } + return &CapacityMonitor{ + rdb: cfg.RDB, + pools: cfg.Pools, + notifier: cfg.Notifier, + gray: cfg.Gray, + lc: cfg.LC, + cfgMgr: cfg.CfgMgr, + clock: cfg.Clock, + } +} + +// RunOnce executes one full capacity-monitoring cycle. Errors from individual +// steps are logged but do not abort subsequent steps. +func (m *CapacityMonitor) RunOnce(ctx context.Context) error { + cfg := m.cfgMgr.Current() + + // Step 1: pool watermark checks. + if err := m.checkWatermarks(ctx, cfg); err != nil { + slog.Error("capacity: watermark check error", "error", err) + } + + // Step 2: probe heartbeat checks. + if err := m.checkProbeHeartbeats(ctx, cfg); err != nil { + slog.Error("capacity: probe heartbeat check error", "error", err) + } + + // Step 3: advance grayscale weight ramps (Grayscale.Advance is 15E's + // warm-up entry point; CapacityLoop drives it per the spec). + if m.gray != nil { + if err := m.gray.Advance(ctx); err != nil { + slog.Error("capacity: grayscale advance error", "error", err) + } + } + + return nil +} + +// checkWatermarks iterates all known pools. For each pool whose fill-rate +// (upCount/target) is below WatermarkThreshold, an alert is emitted — but +// at most once per AlertSuppressMin to suppress repeated notifications. +func (m *CapacityMonitor) checkWatermarks(ctx context.Context, cfg *SchedConfig) error { + stats, err := m.pools.AllPools(ctx) + if err != nil { + return fmt.Errorf("capacity: list pools: %w", err) + } + + threshold := cfg.Capacity.WatermarkThreshold + suppressTTL := time.Duration(cfg.Capacity.AlertSuppressMin) * time.Minute + + for _, stat := range stats { + if stat.Target <= 0 { + continue // skip misconfigured pools with no target + } + fillRate := float64(stat.UpCount) / float64(stat.Target) + if fillRate >= threshold { + continue // pool is healthy + } + + // Throttle: at most one alert per pool per AlertSuppressMin. + dedupKey := capAlertPoolPrefix + stat.Tier + ":" + stat.Region + set, setErr := m.rdb.SetNX(ctx, dedupKey, "1", suppressTTL).Result() + if setErr != nil { + slog.Error("capacity: watermark dedup key error", + "key", dedupKey, "error", setErr) + continue + } + if !set { + continue // alert already sent within suppression window + } + + poolID := "pool:" + stat.Tier + ":" + stat.Region + reason := fmt.Sprintf( + "capacity watermark below %.0f%%: %d/%d up (fill_rate=%.1f%%)", + threshold*100, stat.UpCount, stat.Target, fillRate*100, + ) + + slog.Warn("capacity: pool below watermark", + "tier", stat.Tier, + "region", stat.Region, + "up", stat.UpCount, + "target", stat.Target, + "fill_rate_pct", fmt.Sprintf("%.1f", fillRate*100), + ) + + if m.notifier != nil { + if notifyErr := m.notifier.NotifyFault(ctx, poolID, reason); notifyErr != nil { + slog.Error("capacity: notify watermark alert", + "pool", poolID, "error", notifyErr) + } + } + } + return nil +} + +// checkProbeHeartbeats verifies that every configured probe agent has a live +// heartbeat key in Redis. A missing key means the probe has not reported +// within the 15-minute heartbeat TTL (probe package constant heartbeatTTL). +// +// Semantics: absence of the key is "no recent data", NOT "node blocked". It +// must NOT feed into any determination logic — only ops alerting. +func (m *CapacityMonitor) checkProbeHeartbeats(ctx context.Context, cfg *SchedConfig) error { + probeIDs := cfg.Capacity.ExpectedProbeIDs + if len(probeIDs) == 0 { + return nil // no probes configured; nothing to check + } + + suppressTTL := time.Duration(cfg.Capacity.AlertSuppressMin) * time.Minute + + for _, pid := range probeIDs { + hbKey := probeHBKeyPrefix + pid + exists, err := m.rdb.Exists(ctx, hbKey).Result() + if err != nil { + slog.Error("capacity: probe heartbeat check error", + "probe_id", pid, "error", err) + continue + } + if exists > 0 { + continue // heartbeat key present — probe is alive + } + + // Heartbeat key absent: throttled 「探针失联」alert. + dedupKey := capAlertProbePrefix + pid + set, setErr := m.rdb.SetNX(ctx, dedupKey, "1", suppressTTL).Result() + if setErr != nil { + slog.Error("capacity: probe alert dedup key error", + "probe_id", pid, "error", setErr) + continue + } + if !set { + continue // alert already sent within suppression window + } + + probeNodeID := "probe:" + pid + reason := "探针失联: no heartbeat received in last 15 min (no data, not a block signal)" + + slog.Warn("capacity: probe lost contact", + "probe_id", pid, + "hb_key", hbKey, + ) + + if m.notifier != nil { + if notifyErr := m.notifier.NotifyFault(ctx, probeNodeID, reason); notifyErr != nil { + slog.Error("capacity: notify probe lost contact", + "probe_id", pid, "error", notifyErr) + } + } + } + return nil +} diff --git a/server/internal/scheduler/orchestrate/capacity_test.go b/server/internal/scheduler/orchestrate/capacity_test.go new file mode 100644 index 0000000..1f7b140 --- /dev/null +++ b/server/internal/scheduler/orchestrate/capacity_test.go @@ -0,0 +1,258 @@ +package orchestrate_test + +import ( + "context" + "os" + "testing" + "time" + + "github.com/wangjia/pangolin/server/internal/scheduler/orchestrate" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Utility: write a temporary YAML config file +// ───────────────────────────────────────────────────────────────────────────── + +func writeTempYAML(t *testing.T, content string) (string, error) { + t.Helper() + f, err := os.CreateTemp("", "sched_config_*.yaml") + if err != nil { + return "", err + } + t.Cleanup(func() { os.Remove(f.Name()) }) + if _, err := f.WriteString(content); err != nil { + f.Close() + return "", err + } + f.Close() + return f.Name(), nil +} + +// defaultCfgMgr returns a ConfigManager loaded with production defaults. +func defaultCfgMgr(t *testing.T) *orchestrate.ConfigManager { + t.Helper() + m := orchestrate.NewConfigManager("", nil) + if err := m.Load(); err != nil { + t.Fatalf("ConfigManager.Load: %v", err) + } + return m +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Pool below 70% watermark → alert emitted +// ───────────────────────────────────────────────────────────────────────────── + +func TestCapacityWatermarkAlert(t *testing.T) { + rdb, _ := newTestRedis(t) + pools := newMockPoolReader() + // 6 out of 10 up = 60% < 70% → alert expected. + pools.setPool("free", "hkg", 6, 10) + + notifier := &mockNotifier{} + cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + CfgMgr: defaultCfgMgr(t), + }) + + if err := cm.RunOnce(context.Background()); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + if notifier.count() != 1 { + t.Errorf("NotifyFault calls = %d; want 1 (watermark alert)", notifier.count()) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Pool at exactly 70% → no alert +// ───────────────────────────────────────────────────────────────────────────── + +func TestCapacityWatermarkNoAlertAtThreshold(t *testing.T) { + rdb, _ := newTestRedis(t) + pools := newMockPoolReader() + // 7/10 = 70% == threshold → healthy, no alert. + pools.setPool("free", "hkg", 7, 10) + + notifier := &mockNotifier{} + cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + CfgMgr: defaultCfgMgr(t), + }) + + if err := cm.RunOnce(context.Background()); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + if notifier.count() != 0 { + t.Errorf("NotifyFault calls = %d; want 0 (fill rate = 70%% >= threshold)", notifier.count()) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Repeated RunOnce within suppress window → only 1 alert +// ───────────────────────────────────────────────────────────────────────────── + +func TestCapacityWatermarkAlertSuppression(t *testing.T) { + rdb, mr := newTestRedis(t) + pools := newMockPoolReader() + pools.setPool("free", "sin", 5, 10) // 50% < 70% + + notifier := &mockNotifier{} + cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + CfgMgr: defaultCfgMgr(t), + }) + + ctx := context.Background() + + // Three RunOnce calls within the 10-min suppress window. + for i := 0; i < 3; i++ { + if err := cm.RunOnce(ctx); err != nil { + t.Fatalf("RunOnce %d: %v", i, err) + } + } + if notifier.count() != 1 { + t.Errorf("NotifyFault calls = %d; want 1 (suppressed repeats)", notifier.count()) + } + + // Advance miniredis TTL past the 10-min suppress window. + mr.FastForward(11 * time.Minute) + + // After suppress window expires → another alert. + if err := cm.RunOnce(ctx); err != nil { + t.Fatalf("RunOnce after expire: %v", err) + } + if notifier.count() != 2 { + t.Errorf("NotifyFault calls = %d; want 2 after suppress window expires", notifier.count()) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Missing probe heartbeat → alert; present probe → no alert +// ───────────────────────────────────────────────────────────────────────────── + +func TestCapacityProbeHeartbeatMissingAlert(t *testing.T) { + rdb, _ := newTestRedis(t) + pools := newMockPoolReader() + notifier := &mockNotifier{} + + ctx := context.Background() + // probe-b has a live heartbeat; probe-a does not. + if err := rdb.Set(ctx, "probe:hb:probe-b", "1", 15*time.Minute).Err(); err != nil { + t.Fatalf("seed probe-b heartbeat: %v", err) + } + + yamlContent := ` +capacity: + expected_probe_ids: + - probe-a + - probe-b + alert_suppress_min: 10 +` + tmpFile, err := writeTempYAML(t, yamlContent) + if err != nil { + t.Fatalf("write temp YAML: %v", err) + } + cfgMgr := orchestrate.NewConfigManager(tmpFile, nil) + if err := cfgMgr.Load(); err != nil { + t.Fatalf("Load YAML config: %v", err) + } + + cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + CfgMgr: cfgMgr, + }) + + if err := cm.RunOnce(ctx); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + // Only probe-a should trigger an alert. + if notifier.count() != 1 { + t.Errorf("NotifyFault calls = %d; want 1 (only probe-a missing)", notifier.count()) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Both probes alive → no alert +// ───────────────────────────────────────────────────────────────────────────── + +func TestCapacityProbeHeartbeatBothAlive(t *testing.T) { + rdb, _ := newTestRedis(t) + pools := newMockPoolReader() + notifier := &mockNotifier{} + + ctx := context.Background() + rdb.Set(ctx, "probe:hb:probe-a", "1", 15*time.Minute) + rdb.Set(ctx, "probe:hb:probe-b", "1", 15*time.Minute) + + yamlContent := ` +capacity: + expected_probe_ids: + - probe-a + - probe-b +` + tmpFile, err := writeTempYAML(t, yamlContent) + if err != nil { + t.Fatalf("write temp YAML: %v", err) + } + cfgMgr := orchestrate.NewConfigManager(tmpFile, nil) + if err := cfgMgr.Load(); err != nil { + t.Fatalf("Load YAML: %v", err) + } + + cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + CfgMgr: cfgMgr, + }) + + if err := cm.RunOnce(ctx); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + if notifier.count() != 0 { + t.Errorf("NotifyFault calls = %d; want 0 (both probes alive)", notifier.count()) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Multiple pools — only below-threshold pools alert +// ───────────────────────────────────────────────────────────────────────────── + +func TestCapacityMultiplePoolsSelectiveAlert(t *testing.T) { + rdb, _ := newTestRedis(t) + pools := newMockPoolReader() + // hkg: 60% → alert + pools.setPool("free", "hkg", 6, 10) + // sin: 90% → no alert + pools.setPool("free", "sin", 9, 10) + // tyo: 50% → alert + pools.setPool("free", "tyo", 5, 10) + + notifier := &mockNotifier{} + cm := orchestrate.NewCapacityMonitor(orchestrate.CapacityMonitorConfig{ + RDB: rdb, + Pools: pools, + Notifier: notifier, + CfgMgr: defaultCfgMgr(t), + }) + + if err := cm.RunOnce(context.Background()); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + // Expect 2 alerts: hkg and tyo; sin is healthy. + if n := notifier.count(); n != 2 { + t.Errorf("NotifyFault calls = %d; want 2 (hkg + tyo)", n) + } +} diff --git a/server/internal/scheduler/orchestrate/config.go b/server/internal/scheduler/orchestrate/config.go new file mode 100644 index 0000000..62487b8 --- /dev/null +++ b/server/internal/scheduler/orchestrate/config.go @@ -0,0 +1,402 @@ +package orchestrate + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "os" + "os/signal" + "strconv" + "strings" + "sync/atomic" + "syscall" + + "gopkg.in/yaml.v3" +) + +// ───────────────────────────────────────────────────────────────────────────── +// SchedConfig — master configuration tree +// ───────────────────────────────────────────────────────────────────────────── + +// SchedConfig is the single source of truth for all scheduler thresholds. +// It covers: 15D detection, 15F capacity/breaker, and 15E probe/grayscale. +// +// Values are layered in order: +// 1. Defaults (DefaultSchedConfig). +// 2. Optional YAML file (path set via SCHED_CONFIG_FILE env var, or passed +// to NewConfigManager). +// 3. Environment variable overrides (applied on top of YAML). +// +// The live config is accessed via ConfigManager.Current() — a lock-free +// atomic.Pointer read. Hot reload is triggered by SIGHUP; each reload +// computes a JSON diff and calls the provided AuditFn. +type SchedConfig struct { + // Detect holds 15D detection-engine thresholds. + Detect DetectSection `yaml:"detect"` + // Capacity holds 15F capacity-monitor and breaker thresholds. + Capacity CapacitySection `yaml:"capacity"` + // Probe holds 15E probe-window and grayscale-ramp settings. + Probe ProbeSection `yaml:"probe"` +} + +// DetectSection mirrors detect.DetectConfig; task 15H bridges the two at +// wiring time. +type DetectSection struct { + // DomesticFailNumerator / DomesticFailDenominator define the ISP-failure + // fraction that triggers the suspect rule (default: 2/3). + DomesticFailNumerator int `yaml:"domestic_fail_numerator"` + DomesticFailDenominator int `yaml:"domestic_fail_denominator"` + + // SuspectStreakMin is the number of consecutive failing cycles before a + // node enters blocked_suspect (default: 2). + SuspectStreakMin int `yaml:"suspect_streak_min"` + + // TrafficDropThreshold is the minimum percentage drop in online connections + // over 15 min that activates the traffic-warning rule (default: 80.0). + TrafficDropThreshold float64 `yaml:"traffic_drop_threshold"` + + // TrafficBaselineMin is the minimum current online-connection count for + // the traffic-warning relaxation to apply (default: 20). + TrafficBaselineMin int `yaml:"traffic_baseline_min"` + + // ConfirmedStreakMin is the number of consecutive cycles in blocked_suspect + // before promotion to blocked_confirmed (default: 6). + ConfirmedStreakMin int `yaml:"confirmed_streak_min"` + + // RecoverStreakMin is the number of consecutive passing cycles while in + // blocked_suspect required to recover to up (default: 2). + RecoverStreakMin int `yaml:"recover_streak_min"` + + // SuspectWeight is the routing weight applied when a node first enters + // blocked_suspect (default: 10). + SuspectWeight int `yaml:"suspect_weight"` +} + +// CapacitySection holds 15F capacity-monitor and circuit-breaker thresholds. +type CapacitySection struct { + // WatermarkThreshold is the minimum pool fill-rate (upCount/target) before + // an alert is emitted (default: 0.70 = 70%). + WatermarkThreshold float64 `yaml:"watermark_threshold"` + + // AlertSuppressMin is the minimum gap in minutes between repeated alerts + // for the same pool or probe (default: 10 min). + AlertSuppressMin int `yaml:"alert_suppress_min"` + + // BreakerWindowMin is the sliding-window duration in minutes for the + // circuit breaker (default: 60 min = 1 h). + BreakerWindowMin int `yaml:"breaker_window_min"` + + // BreakerFractionPct is the percentage of pool target capacity that, when + // replaced within one window, trips the breaker (default: 30). + BreakerFractionPct int `yaml:"breaker_fraction_pct"` + + // BreakerMinN is the absolute lower bound for the breaker trip threshold + // (default: 3). Overrides fraction when fraction yields a smaller number. + BreakerMinN int `yaml:"breaker_min_n"` + + // ManualAlertThresh is the consecutive probe-failure count for a single + // node that triggers a 转人工 (escalate-to-human) alert. Should match + // 15E's MaxAttempts (default: 3). + ManualAlertThresh int `yaml:"manual_alert_thresh"` + + // ExpectedProbeIDs is the list of probe-agent IDs that must maintain live + // heartbeat keys in Redis. A missing key triggers a 「探针失联」alert. + ExpectedProbeIDs []string `yaml:"expected_probe_ids"` +} + +// ProbeSection holds 15E probe-window and grayscale-ramp settings. +type ProbeSection struct { + // GrayscaleIntervalHours is the time in hours between successive grayscale + // weight-ramp steps (default: 6 h). + GrayscaleIntervalHours int `yaml:"grayscale_interval_hours"` + + // ProbeTimeoutMin is the maximum time in minutes allowed for a node to + // pass probing per attempt (default: 15 min). + ProbeTimeoutMin int `yaml:"probe_timeout_min"` + + // MaxAttempts is the maximum number of create+probe attempts per + // replacement record before marking it failed (default: 3). + MaxAttempts int `yaml:"max_attempts"` + + // ProbeCyclesRequired is the number of consecutive passing probe ticks + // needed before a new node is promoted to up (default: 2). + ProbeCyclesRequired int `yaml:"probe_cycles_required"` +} + +// DefaultSchedConfig returns a SchedConfig pre-filled with production defaults. +func DefaultSchedConfig() SchedConfig { + return SchedConfig{ + Detect: DetectSection{ + DomesticFailNumerator: 2, + DomesticFailDenominator: 3, + SuspectStreakMin: 2, + TrafficDropThreshold: 80.0, + TrafficBaselineMin: 20, + ConfirmedStreakMin: 6, + RecoverStreakMin: 2, + SuspectWeight: 10, + }, + Capacity: CapacitySection{ + WatermarkThreshold: 0.70, + AlertSuppressMin: 10, + BreakerWindowMin: 60, + BreakerFractionPct: 30, + BreakerMinN: 3, + ManualAlertThresh: 3, + }, + Probe: ProbeSection{ + GrayscaleIntervalHours: 6, + ProbeTimeoutMin: 15, + MaxAttempts: 3, + ProbeCyclesRequired: 2, + }, + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// ConfigManager +// ───────────────────────────────────────────────────────────────────────────── + +// AuditFn is called after each successful config reload with the old and new +// configurations serialised as JSON strings. The caller is responsible for +// deciding how to persist the diff (e.g. via LifecycleService.WriteAuditLog). +// May be nil to skip auditing. +type AuditFn func(ctx context.Context, oldJSON, newJSON string) + +// ConfigManager holds the live SchedConfig and supports SIGHUP-triggered +// hot reload. Config reads are lock-free (atomic.Pointer). +// +// Usage: +// +// mgr := NewConfigManager("/etc/pangolin/sched.yaml", auditFn) +// if err := mgr.Load(); err != nil { log.Fatal(err) } +// mgr.WatchSIGHUP(ctx) +// cfg := mgr.Current() +type ConfigManager struct { + path string // optional YAML file path (may be empty) + cfg atomic.Pointer[SchedConfig] + auditFn AuditFn +} + +// NewConfigManager creates a ConfigManager. +// +// path: optional path to the YAML config file. If SCHED_CONFIG_FILE is +// set in the environment it overrides this argument. +// auditFn: called on every reload when the config changes; may be nil. +// +// The manager is seeded with DefaultSchedConfig so Current() is never nil. +func NewConfigManager(path string, fn AuditFn) *ConfigManager { + if p := os.Getenv("SCHED_CONFIG_FILE"); p != "" { + path = p + } + m := &ConfigManager{path: path, auditFn: fn} + dflt := DefaultSchedConfig() + m.cfg.Store(&dflt) + return m +} + +// Current returns the live config. Never nil. +func (m *ConfigManager) Current() *SchedConfig { + return m.cfg.Load() +} + +// Load reads the config (YAML file if configured, then env overrides) and +// atomically replaces the live config. Safe to call multiple times. +func (m *ConfigManager) Load() error { + next, err := loadSchedConfig(m.path) + if err != nil { + return err + } + m.swap(context.Background(), next) + return nil +} + +// WatchSIGHUP starts a goroutine that calls Reload whenever SIGHUP is +// received. The goroutine stops when ctx is cancelled. +func (m *ConfigManager) WatchSIGHUP(ctx context.Context) { + ch := make(chan os.Signal, 1) + signal.Notify(ch, syscall.SIGHUP) + go func() { + defer signal.Stop(ch) + for { + select { + case <-ctx.Done(): + return + case <-ch: + if err := m.Reload(ctx); err != nil { + slog.Error("sched config: SIGHUP reload failed", "error", err) + } else { + slog.Info("sched config: reloaded via SIGHUP") + } + } + } + }() +} + +// Reload re-reads the config and hot-swaps the live value. Idempotent. +func (m *ConfigManager) Reload(ctx context.Context) error { + next, err := loadSchedConfig(m.path) + if err != nil { + return err + } + m.swap(ctx, next) + return nil +} + +// swap atomically replaces the live config and records a diff in the audit +// log when the config actually changed. +func (m *ConfigManager) swap(ctx context.Context, next *SchedConfig) { + old := m.cfg.Swap(next) + if m.auditFn == nil { + return + } + oldJSON, _ := json.Marshal(old) + newJSON, _ := json.Marshal(next) + if string(oldJSON) == string(newJSON) { + return // no change; skip audit + } + m.auditFn(ctx, string(oldJSON), string(newJSON)) +} + +// ───────────────────────────────────────────────────────────────────────────── +// Loading logic +// ───────────────────────────────────────────────────────────────────────────── + +// loadSchedConfig builds a SchedConfig using the layering strategy: +// defaults → YAML file → env-var overrides. +func loadSchedConfig(path string) (*SchedConfig, error) { + cfg := DefaultSchedConfig() + + // Layer 2: optional YAML file. + if path != "" { + data, err := os.ReadFile(path) + if err != nil && !os.IsNotExist(err) { + return nil, fmt.Errorf("sched config: read %s: %w", path, err) + } + if err == nil { + if err := yaml.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("sched config: parse %s: %w", path, err) + } + } + } + + // Layer 3: environment variable overrides. + applySchedEnvOverrides(&cfg) + + return &cfg, nil +} + +// applySchedEnvOverrides overlays environment-variable values on top of cfg. +// Only non-empty env vars are applied; empty vars preserve the current value. +func applySchedEnvOverrides(cfg *SchedConfig) { + // ── 15D detect ──────────────────────────────────────────────────────────── + if v := posIntEnv("DETECT_DOMESTIC_FAIL_NUM"); v > 0 { + cfg.Detect.DomesticFailNumerator = v + } + if v := posIntEnv("DETECT_DOMESTIC_FAIL_DEN"); v > 0 { + cfg.Detect.DomesticFailDenominator = v + } + if v := posIntEnv("DETECT_SUSPECT_STREAK_MIN"); v > 0 { + cfg.Detect.SuspectStreakMin = v + } + if v := posFloatEnv("DETECT_TRAFFIC_DROP_THRESHOLD"); v > 0 { + cfg.Detect.TrafficDropThreshold = v + } + if v := posIntEnv("DETECT_TRAFFIC_BASELINE_MIN"); v > 0 { + cfg.Detect.TrafficBaselineMin = v + } + if v := posIntEnv("DETECT_CONFIRMED_STREAK_MIN"); v > 0 { + cfg.Detect.ConfirmedStreakMin = v + } + if v := posIntEnv("DETECT_RECOVER_STREAK_MIN"); v > 0 { + cfg.Detect.RecoverStreakMin = v + } + if v := posIntEnv("DETECT_SUSPECT_WEIGHT"); v > 0 { + cfg.Detect.SuspectWeight = v + } + + // ── 15F capacity / breaker ──────────────────────────────────────────────── + if v := posFloatEnv("CAP_WATERMARK_THRESHOLD"); v > 0 { + cfg.Capacity.WatermarkThreshold = v + } + if v := posIntEnv("CAP_ALERT_SUPPRESS_MIN"); v > 0 { + cfg.Capacity.AlertSuppressMin = v + } + if v := posIntEnv("CAP_BREAKER_WINDOW_MIN"); v > 0 { + cfg.Capacity.BreakerWindowMin = v + } + if v := posIntEnv("CAP_BREAKER_FRACTION_PCT"); v > 0 { + cfg.Capacity.BreakerFractionPct = v + } + if v := posIntEnv("CAP_BREAKER_MIN_N"); v > 0 { + cfg.Capacity.BreakerMinN = v + } + if v := posIntEnv("CAP_MANUAL_ALERT_THRESH"); v > 0 { + cfg.Capacity.ManualAlertThresh = v + } + if v := os.Getenv("CAP_EXPECTED_PROBE_IDS"); v != "" { + cfg.Capacity.ExpectedProbeIDs = splitCSVEnv(v) + } + + // ── 15E probe / grayscale ───────────────────────────────────────────────── + if v := posIntEnv("PROBE_GRAYSCALE_INTERVAL_H"); v > 0 { + cfg.Probe.GrayscaleIntervalHours = v + } + if v := posIntEnv("PROBE_TIMEOUT_MIN"); v > 0 { + cfg.Probe.ProbeTimeoutMin = v + } + if v := posIntEnv("PROBE_MAX_ATTEMPTS"); v > 0 { + cfg.Probe.MaxAttempts = v + } + if v := posIntEnv("PROBE_CYCLES_REQUIRED"); v > 0 { + cfg.Probe.ProbeCyclesRequired = v + } +} + +// posIntEnv reads key as a positive integer. Returns 0 if the env var is +// unset, empty, zero, or non-parseable (logs a warning for parse errors). +func posIntEnv(key string) int { + v := os.Getenv(key) + if v == "" { + return 0 + } + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + if err != nil { + slog.Warn("sched config: invalid integer env var", "key", key, "value", v) + } + return 0 + } + return n +} + +// posFloatEnv reads key as a positive float64. Returns 0 on error. +func posFloatEnv(key string) float64 { + v := os.Getenv(key) + if v == "" { + return 0 + } + f, err := strconv.ParseFloat(v, 64) + if err != nil || f <= 0 { + if err != nil { + slog.Warn("sched config: invalid float env var", "key", key, "value", v) + } + return 0 + } + return f +} + +// splitCSVEnv splits a comma-separated env-var value, trimming whitespace and +// dropping empty fields. +func splitCSVEnv(s string) []string { + parts := strings.Split(s, ",") + out := parts[:0] + for _, p := range parts { + p = strings.TrimSpace(p) + if p != "" { + out = append(out, p) + } + } + return out +} diff --git a/server/internal/scheduler/orchestrate/config_test.go b/server/internal/scheduler/orchestrate/config_test.go new file mode 100644 index 0000000..921c3a8 --- /dev/null +++ b/server/internal/scheduler/orchestrate/config_test.go @@ -0,0 +1,289 @@ +package orchestrate_test + +import ( + "context" + "os" + "testing" + + "github.com/wangjia/pangolin/server/internal/scheduler/orchestrate" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Default values +// ───────────────────────────────────────────────────────────────────────────── + +func TestDefaultSchedConfig(t *testing.T) { + cfg := orchestrate.DefaultSchedConfig() + + // Spot-check a representative value from each section. + if cfg.Detect.DomesticFailNumerator != 2 { + t.Errorf("Detect.DomesticFailNumerator = %d; want 2", cfg.Detect.DomesticFailNumerator) + } + if cfg.Capacity.WatermarkThreshold != 0.70 { + t.Errorf("Capacity.WatermarkThreshold = %f; want 0.70", cfg.Capacity.WatermarkThreshold) + } + if cfg.Capacity.BreakerMinN != 3 { + t.Errorf("Capacity.BreakerMinN = %d; want 3", cfg.Capacity.BreakerMinN) + } + if cfg.Probe.GrayscaleIntervalHours != 6 { + t.Errorf("Probe.GrayscaleIntervalHours = %d; want 6", cfg.Probe.GrayscaleIntervalHours) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: YAML file overrides defaults +// ───────────────────────────────────────────────────────────────────────────── + +func TestConfigManagerYAMLOverrides(t *testing.T) { + yaml := ` +detect: + suspect_streak_min: 5 + confirmed_streak_min: 12 +capacity: + watermark_threshold: 0.60 + breaker_min_n: 5 + expected_probe_ids: + - pb-1 + - pb-2 +probe: + grayscale_interval_hours: 8 + max_attempts: 4 +` + tmpFile, err := writeTempYAML(t, yaml) + if err != nil { + t.Fatalf("writeTempYAML: %v", err) + } + + m := orchestrate.NewConfigManager(tmpFile, nil) + if err := m.Load(); err != nil { + t.Fatalf("Load: %v", err) + } + + cfg := m.Current() + + if cfg.Detect.SuspectStreakMin != 5 { + t.Errorf("Detect.SuspectStreakMin = %d; want 5", cfg.Detect.SuspectStreakMin) + } + if cfg.Detect.ConfirmedStreakMin != 12 { + t.Errorf("Detect.ConfirmedStreakMin = %d; want 12", cfg.Detect.ConfirmedStreakMin) + } + if cfg.Capacity.WatermarkThreshold != 0.60 { + t.Errorf("Capacity.WatermarkThreshold = %f; want 0.60", cfg.Capacity.WatermarkThreshold) + } + if cfg.Capacity.BreakerMinN != 5 { + t.Errorf("Capacity.BreakerMinN = %d; want 5", cfg.Capacity.BreakerMinN) + } + if len(cfg.Capacity.ExpectedProbeIDs) != 2 { + t.Errorf("ExpectedProbeIDs count = %d; want 2", len(cfg.Capacity.ExpectedProbeIDs)) + } else { + if cfg.Capacity.ExpectedProbeIDs[0] != "pb-1" || cfg.Capacity.ExpectedProbeIDs[1] != "pb-2" { + t.Errorf("ExpectedProbeIDs = %v; want [pb-1 pb-2]", cfg.Capacity.ExpectedProbeIDs) + } + } + if cfg.Probe.GrayscaleIntervalHours != 8 { + t.Errorf("Probe.GrayscaleIntervalHours = %d; want 8", cfg.Probe.GrayscaleIntervalHours) + } + if cfg.Probe.MaxAttempts != 4 { + t.Errorf("Probe.MaxAttempts = %d; want 4", cfg.Probe.MaxAttempts) + } + + // Unset fields must retain defaults. + if cfg.Detect.DomesticFailNumerator != 2 { + t.Errorf("Detect.DomesticFailNumerator should be default 2; got %d", cfg.Detect.DomesticFailNumerator) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Environment variable overrides YAML +// ───────────────────────────────────────────────────────────────────────────── + +func TestConfigManagerEnvOverridesYAML(t *testing.T) { + yaml := ` +detect: + suspect_streak_min: 5 +capacity: + breaker_min_n: 7 +` + tmpFile, err := writeTempYAML(t, yaml) + if err != nil { + t.Fatalf("writeTempYAML: %v", err) + } + + // Set env vars AFTER writing YAML to ensure they override. + t.Setenv("DETECT_SUSPECT_STREAK_MIN", "9") + t.Setenv("CAP_BREAKER_MIN_N", "15") + + m := orchestrate.NewConfigManager(tmpFile, nil) + if err := m.Load(); err != nil { + t.Fatalf("Load: %v", err) + } + + cfg := m.Current() + + // Env var should override YAML value. + if cfg.Detect.SuspectStreakMin != 9 { + t.Errorf("Detect.SuspectStreakMin = %d; want 9 (env override)", cfg.Detect.SuspectStreakMin) + } + if cfg.Capacity.BreakerMinN != 15 { + t.Errorf("Capacity.BreakerMinN = %d; want 15 (env override)", cfg.Capacity.BreakerMinN) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: SCHED_CONFIG_FILE env var overrides path argument +// ───────────────────────────────────────────────────────────────────────────── + +func TestConfigManagerSCHED_CONFIG_FILE(t *testing.T) { + yaml := ` +capacity: + breaker_window_min: 120 +` + tmpFile, err := writeTempYAML(t, yaml) + if err != nil { + t.Fatalf("writeTempYAML: %v", err) + } + + t.Setenv("SCHED_CONFIG_FILE", tmpFile) + + // Pass a wrong path as arg; env var should win. + m := orchestrate.NewConfigManager("/nonexistent/path.yaml", nil) + if err := m.Load(); err != nil { + t.Fatalf("Load: %v", err) + } + + if cfg := m.Current(); cfg.Capacity.BreakerWindowMin != 120 { + t.Errorf("BreakerWindowMin = %d; want 120 (read from SCHED_CONFIG_FILE)", cfg.Capacity.BreakerWindowMin) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Reload changes values and triggers audit callback with diff +// ───────────────────────────────────────────────────────────────────────────── + +func TestConfigManagerReloadWithAudit(t *testing.T) { + yaml1 := ` +capacity: + watermark_threshold: 0.70 + breaker_min_n: 3 +` + tmpFile, err := writeTempYAML(t, yaml1) + if err != nil { + t.Fatalf("writeTempYAML: %v", err) + } + + var auditCalls []struct{ old, new string } + auditFn := func(ctx context.Context, old, new string) { + auditCalls = append(auditCalls, struct{ old, new string }{old, new}) + } + + m := orchestrate.NewConfigManager(tmpFile, orchestrate.AuditFn(auditFn)) + if err := m.Load(); err != nil { + t.Fatalf("initial Load: %v", err) + } + + initialThreshold := m.Current().Capacity.WatermarkThreshold + + // Rewrite the YAML with a changed value. + yaml2 := ` +capacity: + watermark_threshold: 0.55 + breaker_min_n: 5 +` + if err := os.WriteFile(tmpFile, []byte(yaml2), 0o644); err != nil { + t.Fatalf("rewrite YAML: %v", err) + } + + // Reload should pick up the new values. + if err := m.Reload(context.Background()); err != nil { + t.Fatalf("Reload: %v", err) + } + + newThreshold := m.Current().Capacity.WatermarkThreshold + + if initialThreshold == newThreshold { + t.Errorf("WatermarkThreshold not updated after reload: both = %f", newThreshold) + } + if newThreshold != 0.55 { + t.Errorf("WatermarkThreshold after reload = %f; want 0.55", newThreshold) + } + if m.Current().Capacity.BreakerMinN != 5 { + t.Errorf("BreakerMinN after reload = %d; want 5", m.Current().Capacity.BreakerMinN) + } + + // Audit callback must have been invoked once (old != new). + if len(auditCalls) != 1 { + t.Errorf("audit callback count = %d; want 1", len(auditCalls)) + } else { + if auditCalls[0].old == auditCalls[0].new { + t.Error("audit callback old == new; diff expected") + } + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: Reload with no change → audit NOT called (diff is empty) +// ───────────────────────────────────────────────────────────────────────────── + +func TestConfigManagerReloadNoChangeSkipsAudit(t *testing.T) { + yaml := ` +capacity: + watermark_threshold: 0.75 +` + tmpFile, err := writeTempYAML(t, yaml) + if err != nil { + t.Fatalf("writeTempYAML: %v", err) + } + + auditCount := 0 + auditFn := func(_ context.Context, _, _ string) { auditCount++ } + + m := orchestrate.NewConfigManager(tmpFile, orchestrate.AuditFn(auditFn)) + if err := m.Load(); err != nil { + t.Fatalf("initial Load: %v", err) + } + + // Reload without changing the YAML file. + if err := m.Reload(context.Background()); err != nil { + t.Fatalf("Reload: %v", err) + } + + // No diff → audit should NOT be called on the second load. + // (First Load always swaps from defaults to YAML; that may be different.) + // What we really check: the reload didn't trigger another audit if nothing changed. + // The auditFn may have been called 1 time (from Load) but NOT from Reload. + // + // Actually, Load also calls swap. If the defaults differ from the YAML, + // the initial Load increments auditCount. Then Reload sees no diff → no extra call. + countAfterReload := auditCount + + // Reload again — still unchanged → same count. + if err := m.Reload(context.Background()); err != nil { + t.Fatalf("second Reload: %v", err) + } + if auditCount != countAfterReload { + t.Errorf("audit called %d times after second reload; expected 0 extra calls (no diff)", auditCount-countAfterReload) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: comma-separated env var for ExpectedProbeIDs +// ───────────────────────────────────────────────────────────────────────────── + +func TestConfigManagerProbeIDsFromEnv(t *testing.T) { + t.Setenv("CAP_EXPECTED_PROBE_IDS", "alpha, beta , gamma") + + m := orchestrate.NewConfigManager("", nil) + if err := m.Load(); err != nil { + t.Fatalf("Load: %v", err) + } + ids := m.Current().Capacity.ExpectedProbeIDs + if len(ids) != 3 { + t.Fatalf("ExpectedProbeIDs = %v; want 3 entries", ids) + } + want := []string{"alpha", "beta", "gamma"} + for i, w := range want { + if ids[i] != w { + t.Errorf("ids[%d] = %q; want %q", i, ids[i], w) + } + } +}