feat(scheduler): 自动更换编排 + 养机灰度 [tsk_mxEFSvKHX64G]
实现 15E 任务：Replacer 状态机（pending→creating→probing→activating→draining_old→done/failed）
+ Grayscale 权重梯度（10→25→50→75→100，每步 6h）。
关键设计：
- Redis JSON 记录（sched:replace:{uuid}）+ SetNX 崩溃安全，防止重复 CreateNode
- 幂等 key：attempt 0 = uuid，retry N = uuid:retry:N
- probePass：国内 ≥2/3 ISP 通过 + 海外 L1 全通，连续 2 Tick 才晋级
- 探活超时 15min → failProbeAttempt → 轮换 provider；3 次失败 → failed + NotifyFault
- StubBreaker（15F 未就绪）；NewReplacer(Config{}) 依赖注入
- 6 个集成测试全部通过（happy path / max-retry / crash recovery / queue idempotency / grayscale）
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
wangjia
@@ -0,0 +1,625 @@
|
||||
package orchestrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/scheduler/probe"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Redis key constants
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
const (
|
||||
// replaceQueueKey is the Redis list populated by 15D when a node is confirmed
|
||||
// blocked. We read from the right (RPOP) for FIFO ordering; 15D writes with LPUSH.
|
||||
replaceQueueKey = "detect:replace:queue"
|
||||
|
||||
// replaceKeyPrefix is the namespace for per-replacement orchestration records.
|
||||
// Full key: sched:replace:{uuid}
|
||||
replaceKeyPrefix = "sched:replace:"
|
||||
|
||||
// replaceIndexKey is a Redis set holding all in-flight replacement UUIDs.
|
||||
replaceIndexKey = "sched:replace:index"
|
||||
|
||||
// replaceTTL is how long a terminal (done / failed) record is kept for auditing.
|
||||
replaceTTL = 7 * 24 * time.Hour
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Persistent record types
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// ReplaceRecord is the crash-recoverable orchestration state stored in Redis.
|
||||
// Key: sched:replace:{uuid} Value: JSON-encoded ReplaceRecord.
|
||||
//
|
||||
// All phase transitions read the record first and write it last; a crashed
|
||||
// process resumes from the persisted state without repeating already-executed
|
||||
// actions (CreateNode is guarded by an idempotency key; DestroyNode is
|
||||
// idempotent by design).
|
||||
type ReplaceRecord struct {
|
||||
Phase Phase `json:"phase"`
|
||||
OldNode string `json:"oldNode"` // node ID from the 15D replace queue
|
||||
NewNode string `json:"newNode,omitempty"` // new node ID, set after CreateNode
|
||||
CurrentProviderID string `json:"currentProviderId,omitempty"` // provider for the current attempt
|
||||
Attempts int `json:"attempts"` // how many create+probe attempts so far
|
||||
ProviderTried []string `json:"providerTried"` // provider IDs from failed attempts
|
||||
ProbeStreak int `json:"probeStreak"` // consecutive passing probe Ticks
|
||||
PhaseStartedAt time.Time `json:"phaseStartedAt"` // when the current phase began
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
}
|
||||
|
||||
// queueEntry is the JSON shape pushed by 15D (detect/engine.go pushReplaceQueue).
|
||||
type queueEntry struct {
|
||||
NodeID string `json:"nodeId"`
|
||||
ReplacementUUID string `json:"replacementUuid"`
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Replacer
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Replacer drives the replacement orchestration state machine.
|
||||
// It is called every 30 s by the OrchestrateLoop (task 15H).
|
||||
type Replacer struct {
|
||||
rdb *redis.Client
|
||||
prov ProvisionService
|
||||
lc LifecycleService
|
||||
snaps ProbeSnapshotter
|
||||
breaker Breaker
|
||||
notifier Notifier
|
||||
clock Clock
|
||||
}
|
||||
|
||||
// Config holds all dependencies for NewReplacer.
|
||||
type Config struct {
|
||||
RDB *redis.Client
|
||||
Prov ProvisionService
|
||||
LC LifecycleService
|
||||
Snaps ProbeSnapshotter
|
||||
Breaker Breaker
|
||||
Notifier Notifier
|
||||
Clock Clock
|
||||
}
|
||||
|
||||
// NewReplacer constructs a Replacer. Nil optional deps are replaced with stubs.
|
||||
func NewReplacer(cfg Config) *Replacer {
|
||||
if cfg.Breaker == nil {
|
||||
cfg.Breaker = StubBreaker{}
|
||||
}
|
||||
if cfg.Notifier == nil {
|
||||
cfg.Notifier = LogNotifier{}
|
||||
}
|
||||
if cfg.Clock == nil {
|
||||
cfg.Clock = RealClock{}
|
||||
}
|
||||
return &Replacer{
|
||||
rdb: cfg.RDB,
|
||||
prov: cfg.Prov,
|
||||
lc: cfg.LC,
|
||||
snaps: cfg.Snaps,
|
||||
breaker: cfg.Breaker,
|
||||
notifier: cfg.Notifier,
|
||||
clock: cfg.Clock,
|
||||
}
|
||||
}
|
||||
|
||||
// Tick drains the 15D replace queue and advances every in-flight record one
|
||||
// step. It is designed to be called every 30 s and to be idempotent across
|
||||
// crashes: all state is persisted in Redis before any external action, so a
|
||||
// restart resumes from the saved phase.
|
||||
func (r *Replacer) Tick(ctx context.Context) error {
|
||||
if err := r.drainQueue(ctx); err != nil {
|
||||
// Non-fatal: log and continue so in-flight records still advance.
|
||||
slog.Error("orchestrate: drain replace queue", "error", err)
|
||||
}
|
||||
return r.advanceAll(ctx)
|
||||
}
|
||||
|
||||
// drainQueue pops entries from detect:replace:queue and creates pending records.
|
||||
func (r *Replacer) drainQueue(ctx context.Context) error {
|
||||
const maxDrain = 100 // guard against burst
|
||||
for i := 0; i < maxDrain; i++ {
|
||||
raw, err := r.rdb.RPop(ctx, replaceQueueKey).Result()
|
||||
if err == redis.Nil {
|
||||
break // queue empty
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: rpop: %w", err)
|
||||
}
|
||||
var entry queueEntry
|
||||
if err := json.Unmarshal([]byte(raw), &entry); err != nil {
|
||||
slog.Error("orchestrate: bad queue entry", "raw", raw, "error", err)
|
||||
continue
|
||||
}
|
||||
if err := r.ensureRecord(ctx, entry.ReplacementUUID, entry.NodeID); err != nil {
|
||||
slog.Error("orchestrate: ensure record",
|
||||
"uuid", entry.ReplacementUUID,
|
||||
"node", entry.NodeID,
|
||||
"error", err,
|
||||
)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ensureRecord creates a pending orchestration record for uuid/oldNode if one
|
||||
// does not already exist, then adds uuid to the in-flight index. It is safe
|
||||
// to call multiple times (SetNX + SAdd are both idempotent).
|
||||
func (r *Replacer) ensureRecord(ctx context.Context, uuid, oldNode string) error {
|
||||
key := replaceKeyPrefix + uuid
|
||||
rec := ReplaceRecord{
|
||||
Phase: PhasePending,
|
||||
OldNode: oldNode,
|
||||
ProviderTried: []string{},
|
||||
PhaseStartedAt: r.clock.Now(),
|
||||
UpdatedAt: r.clock.Now(),
|
||||
}
|
||||
data, err := json.Marshal(rec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// SetNX: only stores if the key does not exist, preserving any in-progress record.
|
||||
r.rdb.SetNX(ctx, key, data, 0) //nolint:errcheck // best-effort; SAdd follows
|
||||
return r.rdb.SAdd(ctx, replaceIndexKey, uuid).Err()
|
||||
}
|
||||
|
||||
// advanceAll loads every UUID from the in-flight index and advances each record.
|
||||
func (r *Replacer) advanceAll(ctx context.Context) error {
|
||||
uuids, err := r.rdb.SMembers(ctx, replaceIndexKey).Result()
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: smembers index: %w", err)
|
||||
}
|
||||
for _, uuid := range uuids {
|
||||
if advErr := r.advanceRecord(ctx, uuid); advErr != nil {
|
||||
slog.Error("orchestrate: advance record", "uuid", uuid, "error", advErr)
|
||||
// continue; one failure must not halt other replacements
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// advanceRecord loads a single record and advances it by one phase step.
|
||||
func (r *Replacer) advanceRecord(ctx context.Context, uuid string) error {
|
||||
rec, err := r.loadRecord(ctx, uuid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if rec == nil {
|
||||
// Key expired or was deleted; clean up the index entry.
|
||||
_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
|
||||
return nil
|
||||
}
|
||||
|
||||
switch rec.Phase {
|
||||
case PhasePending:
|
||||
return r.stepPending(ctx, uuid, rec)
|
||||
case PhaseCreating:
|
||||
return r.stepCreating(ctx, uuid, rec)
|
||||
case PhaseProbing:
|
||||
return r.stepProbing(ctx, uuid, rec)
|
||||
case PhaseActivating:
|
||||
return r.stepActivating(ctx, uuid, rec)
|
||||
case PhaseDrainingOld:
|
||||
return r.stepDrainingOld(ctx, uuid, rec)
|
||||
case PhaseDone, PhaseFailed:
|
||||
// Terminal: the record exists only for audit; remove from active index.
|
||||
_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
|
||||
return nil
|
||||
default:
|
||||
return fmt.Errorf("orchestrate: unknown phase %q for %s", rec.Phase, uuid)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Phase step functions
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// stepPending checks the circuit breaker and watermark/quota, then advances
|
||||
// to creating. If the breaker blocks, the record stays pending (retried next Tick).
|
||||
func (r *Replacer) stepPending(ctx context.Context, uuid string, rec *ReplaceRecord) error {
|
||||
nodeInfo, err := r.lc.GetNode(ctx, rec.OldNode)
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: get node %s: %w", rec.OldNode, err)
|
||||
}
|
||||
if nodeInfo == nil {
|
||||
return fmt.Errorf("orchestrate: old node %s not found", rec.OldNode)
|
||||
}
|
||||
|
||||
// Circuit-breaker check (15F).
|
||||
if !r.breaker.Allow(nodeInfo.Tier, nodeInfo.Region) {
|
||||
slog.Info("orchestrate: breaker blocked replacement",
|
||||
"uuid", uuid, "tier", nodeInfo.Tier, "region", nodeInfo.Region)
|
||||
return nil // stay pending; retry next Tick
|
||||
}
|
||||
|
||||
// Watermark / quota check — stub (always passes).
|
||||
// TODO(15F): implement real capacity-quota guard here.
|
||||
|
||||
rec.Phase = PhaseCreating
|
||||
rec.PhaseStartedAt = r.clock.Now()
|
||||
rec.UpdatedAt = r.clock.Now()
|
||||
return r.saveRecord(ctx, uuid, rec)
|
||||
}
|
||||
|
||||
// stepCreating calls CreateNode (idempotent via idempotency key) and advances
|
||||
// to probing. If NewNode is already set (crash recovery after CreateNode
|
||||
// succeeded but before the record was saved with phase=probing), CreateNode is
|
||||
// NOT called again.
|
||||
func (r *Replacer) stepCreating(ctx context.Context, uuid string, rec *ReplaceRecord) error {
|
||||
if rec.NewNode == "" {
|
||||
nodeInfo, err := r.lc.GetNode(ctx, rec.OldNode)
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: get node %s: %w", rec.OldNode, err)
|
||||
}
|
||||
if nodeInfo == nil {
|
||||
return fmt.Errorf("orchestrate: old node %s not found", rec.OldNode)
|
||||
}
|
||||
|
||||
providerID, err := r.pickProvider(ctx, nodeInfo.Tier, nodeInfo.Region, rec.ProviderTried)
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: pick provider: %w", err)
|
||||
}
|
||||
|
||||
spec := NodeSpec{
|
||||
Tier: nodeInfo.Tier,
|
||||
Region: nodeInfo.Region,
|
||||
Role: nodeInfo.Role,
|
||||
ProviderID: providerID,
|
||||
RealitySNI: nodeInfo.RealitySNI,
|
||||
RealityPBK: nodeInfo.RealityPBK,
|
||||
HY2Port: nodeInfo.HY2Port,
|
||||
NameZH: nodeInfo.NameZH,
|
||||
NameEn: nodeInfo.NameEn,
|
||||
Tags: nodeInfo.Tags,
|
||||
}
|
||||
idemKey := idempotencyKey(uuid, rec.Attempts)
|
||||
|
||||
newNodeID, err := r.prov.CreateNode(ctx, spec, idemKey)
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: create node (attempt %d): %w", rec.Attempts, err)
|
||||
}
|
||||
rec.NewNode = newNodeID
|
||||
rec.CurrentProviderID = providerID
|
||||
}
|
||||
|
||||
// Advance to probing; reset streak and set phase start time.
|
||||
rec.Phase = PhaseProbing
|
||||
rec.ProbeStreak = 0
|
||||
rec.PhaseStartedAt = r.clock.Now()
|
||||
rec.UpdatedAt = r.clock.Now()
|
||||
return r.saveRecord(ctx, uuid, rec)
|
||||
}
|
||||
|
||||
// stepProbing checks probe snapshots for the new node.
|
||||
// On 2 consecutive passing Ticks it advances to activating.
|
||||
// On timeout (15 min) or repeated failure it retries with a different provider,
|
||||
// or marks the record failed after MaxAttempts.
|
||||
func (r *Replacer) stepProbing(ctx context.Context, uuid string, rec *ReplaceRecord) error {
|
||||
now := r.clock.Now()
|
||||
|
||||
// Timeout guard.
|
||||
if now.Sub(rec.PhaseStartedAt) > ProbeTimeout {
|
||||
return r.failProbeAttempt(ctx, uuid, rec, "probe timeout (15 min)")
|
||||
}
|
||||
|
||||
// Read probe snapshots for the new node (from the 15A probe store).
|
||||
snapshots, err := r.snaps.SnapshotsByNode(ctx, rec.NewNode)
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: snapshots for %s: %w", rec.NewNode, err)
|
||||
}
|
||||
|
||||
if probePass(snapshots) {
|
||||
rec.ProbeStreak++
|
||||
} else {
|
||||
rec.ProbeStreak = 0
|
||||
}
|
||||
|
||||
if rec.ProbeStreak >= ProbeCyclesRequired {
|
||||
// Probe window passed: advance to activating.
|
||||
rec.Phase = PhaseActivating
|
||||
rec.PhaseStartedAt = now
|
||||
rec.UpdatedAt = now
|
||||
return r.saveRecord(ctx, uuid, rec)
|
||||
}
|
||||
|
||||
// Not yet passed: update streak and wait for next Tick.
|
||||
rec.UpdatedAt = now
|
||||
return r.saveRecord(ctx, uuid, rec)
|
||||
}
|
||||
|
||||
// failProbeAttempt destroys the failed new node and either schedules a retry
|
||||
// (with a different provider) or marks the replacement as permanently failed.
|
||||
func (r *Replacer) failProbeAttempt(ctx context.Context, uuid string, rec *ReplaceRecord, reason string) error {
|
||||
slog.Warn("orchestrate: probing failed",
|
||||
"uuid", uuid,
|
||||
"new_node", rec.NewNode,
|
||||
"attempt", rec.Attempts,
|
||||
"reason", reason,
|
||||
)
|
||||
|
||||
// Destroy the bad new node (bad IP must not enter the pool).
|
||||
if rec.NewNode != "" {
|
||||
if err := r.prov.DestroyNode(ctx, rec.NewNode); err != nil {
|
||||
slog.Error("orchestrate: destroy failed new node",
|
||||
"uuid", uuid, "node", rec.NewNode, "error", err)
|
||||
// Continue: still mark the attempt as failed.
|
||||
}
|
||||
// Track which provider was tried so the next attempt avoids it.
|
||||
if rec.CurrentProviderID != "" {
|
||||
rec.ProviderTried = appendUnique(rec.ProviderTried, rec.CurrentProviderID)
|
||||
}
|
||||
rec.NewNode = ""
|
||||
rec.CurrentProviderID = ""
|
||||
}
|
||||
rec.Attempts++
|
||||
|
||||
if rec.Attempts >= MaxAttempts {
|
||||
// All attempts exhausted: mark failed and alert.
|
||||
rec.Phase = PhaseFailed
|
||||
rec.UpdatedAt = r.clock.Now()
|
||||
if err := r.saveRecord(ctx, uuid, rec); err != nil {
|
||||
return err
|
||||
}
|
||||
_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
|
||||
_ = r.rdb.Expire(ctx, replaceKeyPrefix+uuid, replaceTTL).Err()
|
||||
|
||||
alertReason := fmt.Sprintf("probing failed after %d attempts: %s", rec.Attempts, reason)
|
||||
if notifyErr := r.notifier.NotifyFault(ctx, rec.OldNode, alertReason); notifyErr != nil {
|
||||
slog.Error("orchestrate: notify fault", "uuid", uuid, "error", notifyErr)
|
||||
}
|
||||
slog.Error("orchestrate: replacement permanently failed — manual review required",
|
||||
"uuid", uuid, "old_node", rec.OldNode, "attempts", rec.Attempts)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Schedule retry: go back to creating (different provider picked next call).
|
||||
rec.Phase = PhaseCreating
|
||||
rec.ProbeStreak = 0
|
||||
rec.PhaseStartedAt = r.clock.Now()
|
||||
rec.UpdatedAt = r.clock.Now()
|
||||
return r.saveRecord(ctx, uuid, rec)
|
||||
}
|
||||
|
||||
// stepActivating sets initial weight, promotes the new node from probing→up,
|
||||
// bumps the directory version, and registers grayscale warm-up.
|
||||
func (r *Replacer) stepActivating(ctx context.Context, uuid string, rec *ReplaceRecord) error {
|
||||
// Set initial weight for grayscale ramp.
|
||||
if err := r.lc.SetWeight(ctx, rec.NewNode, GrayscaleWeights[0]); err != nil {
|
||||
return fmt.Errorf("orchestrate: set initial weight: %w", err)
|
||||
}
|
||||
|
||||
// Optimistic-lock transition: probing → up.
|
||||
detail := map[string]any{
|
||||
"from": "probing",
|
||||
"to": "up",
|
||||
"replacement_uuid": uuid,
|
||||
"old_node": rec.OldNode,
|
||||
}
|
||||
affected, err := r.lc.TransitionStatus(ctx, rec.NewNode, "probing", "up", detail)
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: transition probing→up: %w", err)
|
||||
}
|
||||
if affected == 0 {
|
||||
// Lock conflict: another writer changed the state. Retry next Tick.
|
||||
slog.Info("orchestrate: probing→up conflict, will retry",
|
||||
"uuid", uuid, "new_node", rec.NewNode)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Bump directory version so clients re-fetch the updated node list.
|
||||
if err := r.lc.BumpVersion(ctx); err != nil {
|
||||
slog.Error("orchestrate: bump version", "uuid", uuid, "error", err)
|
||||
}
|
||||
|
||||
// Register the new node for grayscale warm-up.
|
||||
if err := startGrayscale(ctx, r.rdb, r.clock, rec.NewNode); err != nil {
|
||||
slog.Error("orchestrate: start grayscale", "uuid", uuid, "node", rec.NewNode, "error", err)
|
||||
}
|
||||
|
||||
rec.Phase = PhaseDrainingOld
|
||||
rec.PhaseStartedAt = r.clock.Now()
|
||||
rec.UpdatedAt = r.clock.Now()
|
||||
return r.saveRecord(ctx, uuid, rec)
|
||||
}
|
||||
|
||||
// stepDrainingOld destroys the old node, records the event in the breaker, and
|
||||
// finalises the replacement.
|
||||
func (r *Replacer) stepDrainingOld(ctx context.Context, uuid string, rec *ReplaceRecord) error {
|
||||
// Look up old node for breaker recording (best-effort; node may be gone).
|
||||
nodeInfo, _ := r.lc.GetNode(ctx, rec.OldNode)
|
||||
|
||||
// Destroy the old (already-down) node and release its IP.
|
||||
if err := r.prov.DestroyNode(ctx, rec.OldNode); err != nil {
|
||||
return fmt.Errorf("orchestrate: destroy old node %s: %w", rec.OldNode, err)
|
||||
}
|
||||
|
||||
// Inform the 15F circuit breaker of this successful replacement.
|
||||
if nodeInfo != nil {
|
||||
r.breaker.Record(nodeInfo.Tier, nodeInfo.Region)
|
||||
}
|
||||
|
||||
rec.Phase = PhaseDone
|
||||
rec.UpdatedAt = r.clock.Now()
|
||||
if err := r.saveRecord(ctx, uuid, rec); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
slog.Info("orchestrate: replacement complete",
|
||||
"uuid", uuid, "old_node", rec.OldNode, "new_node", rec.NewNode)
|
||||
|
||||
// Write audit log.
|
||||
_ = r.lc.WriteAuditLog(ctx, "orchestrate", "replacement_done",
|
||||
"node:"+rec.OldNode,
|
||||
fmt.Sprintf(`{"uuid":%q,"new_node":%q}`, uuid, rec.NewNode),
|
||||
)
|
||||
|
||||
// Remove from active index; set TTL for 7-day audit retention.
|
||||
_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
|
||||
_ = r.rdb.Expire(ctx, replaceKeyPrefix+uuid, replaceTTL).Err()
|
||||
return nil
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Probe-pass check
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// probePass returns true when the snapshot set indicates the new node is healthy:
|
||||
// - Domestic (CN): ≥ 2/3 ISPs are passing.
|
||||
// - Overseas: at least one vantage is present and all overseas vantages pass L1.
|
||||
func probePass(snapshots map[string]probe.ProbeSnapshot) bool {
|
||||
if len(snapshots) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
domesticISPs := make(map[string]bool) // ISP name → failed?
|
||||
overseasTotal := 0
|
||||
overseasOK := 0
|
||||
|
||||
for _, snap := range snapshots {
|
||||
v := snap.Vantage
|
||||
rpt := snap.Report
|
||||
switch {
|
||||
case v.Country == "CN":
|
||||
// Normalise "3rd-" prefix (third-party vantages share ISP grouping).
|
||||
isp := strings.TrimPrefix(v.ISP, "3rd-")
|
||||
failed := isSnapshotFailed(rpt)
|
||||
if prev, seen := domesticISPs[isp]; seen {
|
||||
domesticISPs[isp] = prev || failed
|
||||
} else {
|
||||
domesticISPs[isp] = failed
|
||||
}
|
||||
case v.Country != "":
|
||||
overseasTotal++
|
||||
if rpt.L1.OK {
|
||||
overseasOK++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(domesticISPs) == 0 || overseasTotal == 0 {
|
||||
return false // insufficient probe data
|
||||
}
|
||||
|
||||
// Domestic: ≥ 2/3 ISPs must be passing (not failed).
|
||||
passISPs := 0
|
||||
for _, failed := range domesticISPs {
|
||||
if !failed {
|
||||
passISPs++
|
||||
}
|
||||
}
|
||||
total := len(domesticISPs)
|
||||
domesticOK := passISPs*3 >= total*2
|
||||
|
||||
// Overseas: every vantage must pass L1.
|
||||
overseasPass := overseasOK == overseasTotal
|
||||
|
||||
return domesticOK && overseasPass
|
||||
}
|
||||
|
||||
// isSnapshotFailed mirrors detect/signals.go isReportFailed: check the highest
|
||||
// available layer (L3 > L2 > L1).
|
||||
func isSnapshotFailed(rpt probe.NodeReport) bool {
|
||||
if rpt.L3 != nil {
|
||||
return !rpt.L3.OK
|
||||
}
|
||||
if !rpt.L1.OK {
|
||||
return true
|
||||
}
|
||||
if rpt.L2 != nil && !rpt.L2.OK {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Provider selection
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// pickProvider selects a provider for a new attempt. It prefers providers NOT
|
||||
// in previouslyTried so each retry rotates to a different vendor.
|
||||
func (r *Replacer) pickProvider(ctx context.Context, tier, region string, previouslyTried []string) (string, error) {
|
||||
providers, err := r.prov.ListProviders(ctx, tier, region)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("orchestrate: list providers: %w", err)
|
||||
}
|
||||
if len(providers) == 0 {
|
||||
return "", fmt.Errorf("orchestrate: no providers for tier=%s region=%s", tier, region)
|
||||
}
|
||||
|
||||
tried := make(map[string]bool, len(previouslyTried))
|
||||
for _, id := range previouslyTried {
|
||||
tried[id] = true
|
||||
}
|
||||
|
||||
for _, p := range providers {
|
||||
if !tried[p.ID] {
|
||||
return p.ID, nil
|
||||
}
|
||||
}
|
||||
// All providers have been tried; fall back to the first one.
|
||||
return providers[0].ID, nil
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Redis helpers
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// loadRecord reads and unmarshals a ReplaceRecord. Returns (nil, nil) if absent.
|
||||
func (r *Replacer) loadRecord(ctx context.Context, uuid string) (*ReplaceRecord, error) {
|
||||
val, err := r.rdb.Get(ctx, replaceKeyPrefix+uuid).Result()
|
||||
if err == redis.Nil {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("orchestrate: load %s: %w", uuid, err)
|
||||
}
|
||||
var rec ReplaceRecord
|
||||
if err := json.Unmarshal([]byte(val), &rec); err != nil {
|
||||
return nil, fmt.Errorf("orchestrate: unmarshal %s: %w", uuid, err)
|
||||
}
|
||||
return &rec, nil
|
||||
}
|
||||
|
||||
// saveRecord marshals and persists rec under sched:replace:{uuid}.
|
||||
func (r *Replacer) saveRecord(ctx context.Context, uuid string, rec *ReplaceRecord) error {
|
||||
data, err := json.Marshal(rec)
|
||||
if err != nil {
|
||||
return fmt.Errorf("orchestrate: marshal %s: %w", uuid, err)
|
||||
}
|
||||
if err := r.rdb.Set(ctx, replaceKeyPrefix+uuid, data, 0).Err(); err != nil {
|
||||
return fmt.Errorf("orchestrate: save %s: %w", uuid, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Misc helpers
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// idempotencyKey derives the CreateNode idempotency key for the given attempt.
|
||||
// Attempt 0 uses the replacement UUID directly; retries append a suffix to
|
||||
// ensure each attempt is independently idempotent.
|
||||
func idempotencyKey(replacementUUID string, attempt int) string {
|
||||
if attempt == 0 {
|
||||
return replacementUUID
|
||||
}
|
||||
return fmt.Sprintf("%s:retry:%d", replacementUUID, attempt)
|
||||
}
|
||||
|
||||
// appendUnique appends s to slice only if s is not already present.
|
||||
func appendUnique(slice []string, s string) []string {
|
||||
for _, v := range slice {
|
||||
if v == s {
|
||||
return slice
|
||||
}
|
||||
}
|
||||
return append(slice, s)
|
||||
}
|
||||
Reference in New Issue
Block a user