merge: 自动更换编排 + 养机灰度 [tsk_mxEFSvKHX64G]
This commit is contained in:
wangjia
@@ -0,0 +1,200 @@
|
||||
// Package orchestrate implements the replacement orchestration state machine
|
||||
// (task 15E). It is driven every 30 s by the OrchestrateLoop (task 15H).
|
||||
//
|
||||
// Key components:
|
||||
// - Replacer.Tick — drains the 15D replace queue and advances in-flight records.
|
||||
// - Grayscale.Advance — gradually ramps new/recovered node weight 10→25→50→75→100.
|
||||
package orchestrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/scheduler/probe"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Phase constants
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Phase is the orchestration phase of a replacement record.
|
||||
type Phase string
|
||||
|
||||
const (
|
||||
PhasePending Phase = "pending" // waiting for breaker / quota approval
|
||||
PhaseCreating Phase = "creating" // CreateNode called; waiting for new node
|
||||
PhaseProbing Phase = "probing" // waiting for consecutive probe passes
|
||||
PhaseActivating Phase = "activating" // promoting new node; starting grayscale
|
||||
PhaseDrainingOld Phase = "draining_old" // destroying old (already-down) node
|
||||
PhaseDone Phase = "done" // finished successfully
|
||||
PhaseFailed Phase = "failed" // terminal failure; human review needed
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Tunable constants
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
const (
|
||||
// MaxAttempts is the maximum number of create+probe tries before marking failed.
|
||||
MaxAttempts = 3
|
||||
|
||||
// ProbeCyclesRequired is the number of consecutive Tick cycles with a passing
|
||||
// probe snapshot required before the new node is promoted to "up".
|
||||
ProbeCyclesRequired = 2
|
||||
|
||||
// ProbeTimeout is the maximum time allowed in the probing phase per attempt.
|
||||
ProbeTimeout = 15 * time.Minute
|
||||
|
||||
// GrayscaleInterval is the time between successive weight ramp steps.
|
||||
GrayscaleInterval = 6 * time.Hour
|
||||
)
|
||||
|
||||
// GrayscaleWeights is the weight ladder for the warmup ramp.
|
||||
// New / recovered nodes start at 10 and advance every GrayscaleInterval.
|
||||
var GrayscaleWeights = []int{10, 25, 50, 75, 100}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Data types
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// NodeSpec describes the replacement node to be provisioned.
|
||||
type NodeSpec struct {
|
||||
Tier string
|
||||
Region string
|
||||
Role string
|
||||
ProviderID string // preferred provider for this attempt
|
||||
RealitySNI string // SNI rotation
|
||||
RealityPBK string
|
||||
HY2Port int
|
||||
NameZH string
|
||||
NameEn string
|
||||
Tags []string
|
||||
}
|
||||
|
||||
// NodeInfo holds the properties of an existing node needed to build a
|
||||
// replacement spec or to look up tier / region for the breaker.
|
||||
type NodeInfo struct {
|
||||
ID string
|
||||
Tier string
|
||||
Region string
|
||||
Role string
|
||||
ProviderID string
|
||||
RealitySNI string
|
||||
RealityPBK string
|
||||
HY2Port int
|
||||
NameZH string
|
||||
NameEn string
|
||||
Tags []string
|
||||
}
|
||||
|
||||
// ProviderInfo describes a cloud provider available for provisioning.
|
||||
type ProviderInfo struct {
|
||||
ID string // provider identifier
|
||||
Regions []string // supported regions (empty means all regions)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Service interfaces
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// ProvisionService is the #14 provisioning interface.
|
||||
// The real implementation is provided by task #14; a mock is used in tests.
|
||||
type ProvisionService interface {
|
||||
// CreateNode provisions a new node. idempotencyKey makes the call crash-safe:
|
||||
// replaying the same key returns the already-created node without booting again.
|
||||
CreateNode(ctx context.Context, spec NodeSpec, idempotencyKey string) (nodeID string, err error)
|
||||
|
||||
// DestroyNode tears down a node and releases its IP.
|
||||
DestroyNode(ctx context.Context, nodeID string) error
|
||||
|
||||
// RotateIP swaps the elastic IP on an existing node without re-creating it.
|
||||
RotateIP(ctx context.Context, nodeID string) (newNodeID string, err error)
|
||||
|
||||
// ListProviders returns providers available for the given tier and region.
|
||||
// The returned slice is ordered; callers use rotation for provider selection.
|
||||
ListProviders(ctx context.Context, tier, region string) ([]ProviderInfo, error)
|
||||
}
|
||||
|
||||
// LifecycleService provides node lifecycle management (real impl: task #5).
|
||||
type LifecycleService interface {
|
||||
// GetNode returns the current properties of the node, or nil if not found.
|
||||
GetNode(ctx context.Context, nodeID string) (*NodeInfo, error)
|
||||
|
||||
// TransitionStatus performs an optimistic-lock status transition.
|
||||
// Returns (1, nil) on success, (0, nil) on lock conflict.
|
||||
TransitionStatus(ctx context.Context, nodeID string, from, to string, detail map[string]any) (int, error)
|
||||
|
||||
// SetWeight updates the routing weight for the node.
|
||||
SetWeight(ctx context.Context, nodeID string, weight int) error
|
||||
|
||||
// BumpVersion increments the global directory version so clients re-fetch.
|
||||
BumpVersion(ctx context.Context) error
|
||||
|
||||
// WriteAuditLog records an audit trail entry.
|
||||
WriteAuditLog(ctx context.Context, actor, action, target, meta string) error
|
||||
}
|
||||
|
||||
// ProbeSnapshotter reads the most-recent probe snapshots for a node (15A store).
|
||||
type ProbeSnapshotter interface {
|
||||
SnapshotsByNode(ctx context.Context, nodeID string) (map[string]probe.ProbeSnapshot, error)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Breaker (15F stub)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Breaker is the 15F circuit-breaker interface.
|
||||
type Breaker interface {
|
||||
// Allow returns true if a new replacement is permitted to proceed.
|
||||
Allow(tier, region string) bool
|
||||
|
||||
// Record increments the failure counter after a replaced node has been destroyed.
|
||||
Record(tier, region string)
|
||||
}
|
||||
|
||||
// StubBreaker always permits replacements.
|
||||
// TODO(15F): replace with the real Breaker once task 15F is implemented.
|
||||
type StubBreaker struct{}
|
||||
|
||||
// Allow implements Breaker; always returns true.
|
||||
func (StubBreaker) Allow(_, _ string) bool { return true }
|
||||
|
||||
// Record implements Breaker; no-op until 15F is implemented.
|
||||
func (StubBreaker) Record(_, _ string) {}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Notifier (15G stub)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Notifier is the 15G alerting interface.
|
||||
type Notifier interface {
|
||||
NotifyFault(ctx context.Context, nodeID, reason string) error
|
||||
}
|
||||
|
||||
// LogNotifier logs faults via slog. Used when no real notifier is wired.
|
||||
type LogNotifier struct{}
|
||||
|
||||
// NotifyFault implements Notifier.
|
||||
func (LogNotifier) NotifyFault(_ context.Context, nodeID, reason string) error {
|
||||
slog.Warn("orchestrate: replacement failed — manual review required",
|
||||
"node_id", nodeID,
|
||||
"reason", reason,
|
||||
)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Clock (for testability)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Clock abstracts wall-clock time so tests can fast-forward without sleeping.
|
||||
type Clock interface {
|
||||
Now() time.Time
|
||||
}
|
||||
|
||||
// RealClock is the production clock.
|
||||
type RealClock struct{}
|
||||
|
||||
// Now implements Clock.
|
||||
func (RealClock) Now() time.Time { return time.Now().UTC() }
|
||||
@@ -0,0 +1,183 @@
|
||||
package orchestrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Redis key namespace for grayscale records
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
const (
|
||||
// grayKeyPrefix is the namespace for per-node grayscale warm-up records.
|
||||
// Full key: sched:gray:{nodeID}
|
||||
grayKeyPrefix = "sched:gray:"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// GrayscaleRecord
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// GrayscaleRecord is the Redis-persisted state of a node in weight warm-up.
|
||||
// Key: sched:gray:{nodeID} Value: JSON-encoded GrayscaleRecord.
|
||||
type GrayscaleRecord struct {
|
||||
NodeID string `json:"nodeId"`
|
||||
CurrentWeight int `json:"currentWeight"`
|
||||
StartedAt time.Time `json:"startedAt"`
|
||||
LastAdvancedAt time.Time `json:"lastAdvancedAt"`
|
||||
}
|
||||
|
||||
// startGrayscale creates a new GrayscaleRecord for nodeID at weight 10 (the
|
||||
// initial weight). It is called from stepActivating immediately after the node
|
||||
// is transitioned to "up".
|
||||
func startGrayscale(ctx context.Context, rdb *redis.Client, clock Clock, nodeID string) error {
|
||||
now := clock.Now()
|
||||
rec := GrayscaleRecord{
|
||||
NodeID: nodeID,
|
||||
CurrentWeight: GrayscaleWeights[0], // 10
|
||||
StartedAt: now,
|
||||
LastAdvancedAt: now,
|
||||
}
|
||||
data, err := json.Marshal(rec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return rdb.Set(ctx, grayKeyPrefix+nodeID, data, 0).Err()
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Grayscale
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Grayscale advances the weight ramp for all nodes currently in warm-up.
|
||||
// It is called by the CapacityLoop (task 15H); the implementation lives here
|
||||
// (task 15E) while wiring is done in 15H.
|
||||
//
|
||||
// Weight ladder: 10 → 25 → 50 → 75 → 100, one step every 6 h.
|
||||
// When weight 100 is reached, the grayscale record is deleted.
|
||||
type Grayscale struct {
|
||||
rdb *redis.Client
|
||||
lc LifecycleService
|
||||
clock Clock
|
||||
}
|
||||
|
||||
// NewGrayscale creates a Grayscale with the given dependencies.
|
||||
// If clock is nil, RealClock is used.
|
||||
func NewGrayscale(rdb *redis.Client, lc LifecycleService, clock Clock) *Grayscale {
|
||||
if clock == nil {
|
||||
clock = RealClock{}
|
||||
}
|
||||
return &Grayscale{rdb: rdb, lc: lc, clock: clock}
|
||||
}
|
||||
|
||||
// Advance scans all sched:gray:* keys and advances any node whose
|
||||
// LastAdvancedAt is at least GrayscaleInterval (6 h) in the past.
|
||||
func (g *Grayscale) Advance(ctx context.Context) error {
|
||||
keys, err := g.scanGrayKeys(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("grayscale: scan keys: %w", err)
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
if advErr := g.advanceNode(ctx, key); advErr != nil {
|
||||
slog.Error("grayscale: advance node", "key", key, "error", advErr)
|
||||
// continue; one error must not halt other nodes
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// advanceNode loads the record at key, advances the weight if the interval has
|
||||
// elapsed, calls SetWeight, writes an audit log, and deletes the record when
|
||||
// the final weight (100) is reached.
|
||||
func (g *Grayscale) advanceNode(ctx context.Context, key string) error {
|
||||
val, err := g.rdb.Get(ctx, key).Result()
|
||||
if err == redis.Nil {
|
||||
return nil // already deleted
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("grayscale: get %s: %w", key, err)
|
||||
}
|
||||
var rec GrayscaleRecord
|
||||
if err := json.Unmarshal([]byte(val), &rec); err != nil {
|
||||
return fmt.Errorf("grayscale: unmarshal %s: %w", key, err)
|
||||
}
|
||||
|
||||
now := g.clock.Now()
|
||||
if now.Sub(rec.LastAdvancedAt) < GrayscaleInterval {
|
||||
return nil // interval not elapsed yet
|
||||
}
|
||||
|
||||
// Find the next weight in the ladder.
|
||||
nextWeight, ok := nextGrayscaleWeight(rec.CurrentWeight)
|
||||
if !ok {
|
||||
// Already at maximum; remove the record.
|
||||
_ = g.rdb.Del(ctx, key).Err()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Apply the new weight.
|
||||
if err := g.lc.SetWeight(ctx, rec.NodeID, nextWeight); err != nil {
|
||||
return fmt.Errorf("grayscale: set weight %d for %s: %w", nextWeight, rec.NodeID, err)
|
||||
}
|
||||
|
||||
// Audit trail.
|
||||
meta := fmt.Sprintf(`{"from":%d,"to":%d,"node":%q}`, rec.CurrentWeight, nextWeight, rec.NodeID)
|
||||
_ = g.lc.WriteAuditLog(ctx, "grayscale", "weight_advanced", "node:"+rec.NodeID, meta)
|
||||
|
||||
slog.Info("grayscale: weight advanced",
|
||||
"node", rec.NodeID,
|
||||
"from", rec.CurrentWeight,
|
||||
"to", nextWeight,
|
||||
)
|
||||
|
||||
if nextWeight >= GrayscaleWeights[len(GrayscaleWeights)-1] {
|
||||
// Final weight reached: remove the grayscale record.
|
||||
_ = g.rdb.Del(ctx, key).Err()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Persist updated record.
|
||||
rec.CurrentWeight = nextWeight
|
||||
rec.LastAdvancedAt = now
|
||||
data, err := json.Marshal(rec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return g.rdb.Set(ctx, key, data, 0).Err()
|
||||
}
|
||||
|
||||
// nextGrayscaleWeight returns the next weight after current in GrayscaleWeights.
|
||||
// Returns (0, false) if current is already at or beyond the final weight.
|
||||
func nextGrayscaleWeight(current int) (int, bool) {
|
||||
for i, w := range GrayscaleWeights {
|
||||
if w == current && i+1 < len(GrayscaleWeights) {
|
||||
return GrayscaleWeights[i+1], true
|
||||
}
|
||||
}
|
||||
return 0, false
|
||||
}
|
||||
|
||||
// scanGrayKeys returns all Redis keys matching sched:gray:*.
|
||||
func (g *Grayscale) scanGrayKeys(ctx context.Context) ([]string, error) {
|
||||
var keys []string
|
||||
iter := g.rdb.Scan(ctx, 0, grayKeyPrefix+"*", 0).Iterator()
|
||||
for iter.Next(ctx) {
|
||||
k := iter.Val()
|
||||
// Skip any keys that don't have the expected format (e.g. the index key).
|
||||
if strings.HasPrefix(k, grayKeyPrefix) {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
}
|
||||
if err := iter.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return keys, nil
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -0,0 +1,821 @@
|
||||
package orchestrate_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
|
||||
"github.com/wangjia/pangolin/server/internal/scheduler/probe"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// In-process Redis
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func newTestRedis(t *testing.T) (*redis.Client, *miniredis.Miniredis) {
|
||||
t.Helper()
|
||||
mr := miniredis.RunT(t)
|
||||
rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
||||
return rdb, mr
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Mock ProvisionService
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type createCall struct {
|
||||
Spec orchestrate.NodeSpec
|
||||
IdempotencyKey string
|
||||
ReturnedID string
|
||||
}
|
||||
|
||||
type mockProvision struct {
|
||||
mu sync.Mutex
|
||||
seq int
|
||||
|
||||
providers []orchestrate.ProviderInfo
|
||||
createCalls []createCall
|
||||
destroyCalls []string
|
||||
createErr error
|
||||
|
||||
// idem maps idempotency key → nodeID (simulates provider idempotency).
|
||||
idem map[string]string
|
||||
}
|
||||
|
||||
func newMockProvision(providers ...orchestrate.ProviderInfo) *mockProvision {
|
||||
return &mockProvision{
|
||||
providers: providers,
|
||||
idem: map[string]string{},
|
||||
}
|
||||
}
|
||||
|
||||
func (m *mockProvision) CreateNode(_ context.Context, spec orchestrate.NodeSpec, idemKey string) (string, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.createErr != nil {
|
||||
return "", m.createErr
|
||||
}
|
||||
// Idempotency replay.
|
||||
if existing, ok := m.idem[idemKey]; ok {
|
||||
m.createCalls = append(m.createCalls, createCall{spec, idemKey, existing})
|
||||
return existing, nil
|
||||
}
|
||||
m.seq++
|
||||
nodeID := fmt.Sprintf("new-node-%d", m.seq)
|
||||
m.idem[idemKey] = nodeID
|
||||
m.createCalls = append(m.createCalls, createCall{spec, idemKey, nodeID})
|
||||
return nodeID, nil
|
||||
}
|
||||
|
||||
func (m *mockProvision) DestroyNode(_ context.Context, nodeID string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.destroyCalls = append(m.destroyCalls, nodeID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockProvision) RotateIP(_ context.Context, _ string) (string, error) { return "", nil }
|
||||
|
||||
func (m *mockProvision) ListProviders(_ context.Context, _, _ string) ([]orchestrate.ProviderInfo, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return append([]orchestrate.ProviderInfo{}, m.providers...), nil
|
||||
}
|
||||
|
||||
func (m *mockProvision) createCount() int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return len(m.createCalls)
|
||||
}
|
||||
|
||||
func (m *mockProvision) destroyCount() int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return len(m.destroyCalls)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Mock LifecycleService
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type transitionEvent struct {
|
||||
NodeID, From, To string
|
||||
}
|
||||
|
||||
type mockLC struct {
|
||||
mu sync.Mutex
|
||||
nodes map[string]*orchestrate.NodeInfo
|
||||
weights map[string]int
|
||||
version int64
|
||||
auditLogs []string
|
||||
transitions []transitionEvent
|
||||
}
|
||||
|
||||
func newMockLC(nodes ...*orchestrate.NodeInfo) *mockLC {
|
||||
m := &mockLC{
|
||||
nodes: make(map[string]*orchestrate.NodeInfo),
|
||||
weights: make(map[string]int),
|
||||
}
|
||||
for _, n := range nodes {
|
||||
cp := *n
|
||||
m.nodes[n.ID] = &cp
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func (m *mockLC) addNode(n *orchestrate.NodeInfo) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
cp := *n
|
||||
m.nodes[n.ID] = &cp
|
||||
}
|
||||
|
||||
func (m *mockLC) GetNode(_ context.Context, nodeID string) (*orchestrate.NodeInfo, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
n, ok := m.nodes[nodeID]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
cp := *n
|
||||
return &cp, nil
|
||||
}
|
||||
|
||||
func (m *mockLC) TransitionStatus(_ context.Context, nodeID, from, to string, _ map[string]any) (int, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.transitions = append(m.transitions, transitionEvent{nodeID, from, to})
|
||||
return 1, nil // always succeed in tests
|
||||
}
|
||||
|
||||
func (m *mockLC) SetWeight(_ context.Context, nodeID string, weight int) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.weights[nodeID] = weight
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockLC) BumpVersion(_ context.Context) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.version++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockLC) WriteAuditLog(_ context.Context, actor, action, target, _ string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.auditLogs = append(m.auditLogs, actor+"|"+action+"|"+target)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockLC) weightOf(nodeID string) int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.weights[nodeID]
|
||||
}
|
||||
|
||||
func (m *mockLC) versionOf() int64 {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.version
|
||||
}
|
||||
|
||||
func (m *mockLC) hasAudit(entry string) bool {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, l := range m.auditLogs {
|
||||
if l == entry {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Mock ProbeSnapshotter
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type mockSnaps struct {
|
||||
mu sync.Mutex
|
||||
data map[string]map[string]probe.ProbeSnapshot
|
||||
}
|
||||
|
||||
func (s *mockSnaps) SnapshotsByNode(_ context.Context, nodeID string) (map[string]probe.ProbeSnapshot, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
d, ok := s.data[nodeID]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
out := make(map[string]probe.ProbeSnapshot, len(d))
|
||||
for k, v := range d {
|
||||
out[k] = v
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *mockSnaps) setPass(nodeID string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.data == nil {
|
||||
s.data = make(map[string]map[string]probe.ProbeSnapshot)
|
||||
}
|
||||
s.data[nodeID] = passingSnapshots()
|
||||
}
|
||||
|
||||
// passingSnapshots returns a snapshot set where 2/3 domestic ISPs pass + overseas OK.
|
||||
func passingSnapshots() map[string]probe.ProbeSnapshot {
|
||||
return map[string]probe.ProbeSnapshot{
|
||||
"CN:Telecom": {
|
||||
Vantage: probe.VantagePoint{Country: "CN", ISP: "ChinaTelecom"},
|
||||
Report: probe.NodeReport{
|
||||
L1: probe.L1Result{OK: true},
|
||||
L3: &probe.L3Result{OK: true},
|
||||
},
|
||||
},
|
||||
"CN:Unicom": {
|
||||
Vantage: probe.VantagePoint{Country: "CN", ISP: "ChinaUnicom"},
|
||||
Report: probe.NodeReport{
|
||||
L1: probe.L1Result{OK: true},
|
||||
L3: &probe.L3Result{OK: true},
|
||||
},
|
||||
},
|
||||
"CN:Mobile": {
|
||||
Vantage: probe.VantagePoint{Country: "CN", ISP: "ChinaMobile"},
|
||||
Report: probe.NodeReport{
|
||||
L1: probe.L1Result{OK: false},
|
||||
L3: &probe.L3Result{OK: false},
|
||||
},
|
||||
},
|
||||
"SG:AWS": {
|
||||
Vantage: probe.VantagePoint{Country: "SG", ISP: "AWS"},
|
||||
Report: probe.NodeReport{L1: probe.L1Result{OK: true}},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Mock Notifier
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type mockNotifier struct {
|
||||
mu sync.Mutex
|
||||
calls []string
|
||||
}
|
||||
|
||||
func (n *mockNotifier) NotifyFault(_ context.Context, nodeID, _ string) error {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
n.calls = append(n.calls, nodeID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (n *mockNotifier) count() int {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
return len(n.calls)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Mock Clock
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type mockClock struct {
|
||||
mu sync.Mutex
|
||||
now time.Time
|
||||
}
|
||||
|
||||
func newMockClock(t time.Time) *mockClock { return &mockClock{now: t} }
|
||||
|
||||
func (c *mockClock) Now() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.now
|
||||
}
|
||||
|
||||
func (c *mockClock) advance(d time.Duration) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.now = c.now.Add(d)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test harness
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type harness struct {
|
||||
rdb *redis.Client
|
||||
prov *mockProvision
|
||||
lc *mockLC
|
||||
snaps *mockSnaps
|
||||
notifier *mockNotifier
|
||||
clock *mockClock
|
||||
replacer *orchestrate.Replacer
|
||||
}
|
||||
|
||||
func newHarness(t *testing.T, nodes ...*orchestrate.NodeInfo) *harness {
|
||||
t.Helper()
|
||||
rdb, _ := newTestRedis(t)
|
||||
prov := newMockProvision(
|
||||
orchestrate.ProviderInfo{ID: "provider-A"},
|
||||
orchestrate.ProviderInfo{ID: "provider-B"},
|
||||
orchestrate.ProviderInfo{ID: "provider-C"},
|
||||
)
|
||||
lc := newMockLC(nodes...)
|
||||
snaps := &mockSnaps{}
|
||||
notifier := &mockNotifier{}
|
||||
clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
|
||||
|
||||
r := orchestrate.NewReplacer(orchestrate.Config{
|
||||
RDB: rdb,
|
||||
Prov: prov,
|
||||
LC: lc,
|
||||
Snaps: snaps,
|
||||
Notifier: notifier,
|
||||
Clock: clock,
|
||||
})
|
||||
return &harness{rdb: rdb, prov: prov, lc: lc, snaps: snaps, notifier: notifier, clock: clock, replacer: r}
|
||||
}
|
||||
|
||||
func (h *harness) pushQueue(t *testing.T, oldNodeID, replacementUUID string) {
|
||||
t.Helper()
|
||||
data, _ := json.Marshal(map[string]string{
|
||||
"nodeId": oldNodeID,
|
||||
"replacementUuid": replacementUUID,
|
||||
})
|
||||
if err := h.rdb.LPush(context.Background(), "detect:replace:queue", data).Err(); err != nil {
|
||||
t.Fatalf("pushQueue: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *harness) tick(t *testing.T) {
|
||||
t.Helper()
|
||||
if err := h.replacer.Tick(context.Background()); err != nil {
|
||||
t.Fatalf("Tick() error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// State-machine phase explanation (for reference):
|
||||
//
|
||||
// Tick 1: drainQueue creates pending record → advanceAll: stepPending → creating
|
||||
// Tick 2: stepCreating → CreateNode called → probing
|
||||
// Tick 3: stepProbing (probeStreak=1, need 2)
|
||||
// Tick 4: stepProbing (probeStreak=2) → activating
|
||||
// Tick 5: stepActivating → SetWeight(10)+TransitionStatus+BumpVersion+grayscale → draining_old
|
||||
// Tick 6: stepDrainingOld → DestroyNode(oldNode) → done
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test: happy-path replacement (6 ticks)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestHappyPathReplacement(t *testing.T) {
|
||||
const (
|
||||
oldNode = "old-node-001"
|
||||
replacementUUID = "uuid-happy"
|
||||
)
|
||||
h := newHarness(t,
|
||||
&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg", Role: "entry"},
|
||||
)
|
||||
h.pushQueue(t, oldNode, replacementUUID)
|
||||
|
||||
// Tick 1: pending → creating (CreateNode not yet called).
|
||||
h.tick(t)
|
||||
if h.prov.createCount() != 0 {
|
||||
t.Errorf("tick 1: CreateNode should not be called yet; got %d calls", h.prov.createCount())
|
||||
}
|
||||
|
||||
// Tick 2: creating → CreateNode → probing.
|
||||
h.tick(t)
|
||||
if h.prov.createCount() != 1 {
|
||||
t.Fatalf("tick 2: CreateNode calls = %d; want 1", h.prov.createCount())
|
||||
}
|
||||
call0 := h.prov.createCalls[0]
|
||||
// The idempotency key for attempt 0 must equal the replacement UUID.
|
||||
if call0.IdempotencyKey != replacementUUID {
|
||||
t.Errorf("CreateNode idempotency key = %q; want %q", call0.IdempotencyKey, replacementUUID)
|
||||
}
|
||||
|
||||
newNodeID := call0.ReturnedID
|
||||
// Simulate agent self-registration: add the new node to the lifecycle mock.
|
||||
h.lc.addNode(&orchestrate.NodeInfo{ID: newNodeID, Tier: "free", Region: "hkg"})
|
||||
// Set probe snapshots to PASS for the new node.
|
||||
h.snaps.setPass(newNodeID)
|
||||
|
||||
h.tick(t) // Tick 3: probing streak=1 (< 2 required).
|
||||
h.tick(t) // Tick 4: probing streak=2 → phase saved as activating.
|
||||
h.tick(t) // Tick 5: activating → SetWeight(10) + TransitionStatus(probing→up) + BumpVersion + grayscale → draining_old.
|
||||
|
||||
// After stepActivating:
|
||||
if w := h.lc.weightOf(newNodeID); w != 10 {
|
||||
t.Errorf("new node weight after activating = %d; want 10", w)
|
||||
}
|
||||
if h.lc.versionOf() == 0 {
|
||||
t.Error("BumpVersion was not called after activating")
|
||||
}
|
||||
|
||||
h.tick(t) // Tick 6: draining_old → DestroyNode(oldNode) → done.
|
||||
|
||||
// Old node must be destroyed.
|
||||
oldDestroyed := false
|
||||
for _, id := range h.prov.destroyCalls {
|
||||
if id == oldNode {
|
||||
oldDestroyed = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !oldDestroyed {
|
||||
t.Errorf("old node %s not destroyed; destroyCalls = %v", oldNode, h.prov.destroyCalls)
|
||||
}
|
||||
|
||||
// Grayscale record must exist for the new node.
|
||||
if _, err := h.rdb.Get(context.Background(), "sched:gray:"+newNodeID).Result(); err == redis.Nil {
|
||||
t.Errorf("grayscale record for %s not found", newNodeID)
|
||||
}
|
||||
|
||||
// Audit log must contain replacement_done.
|
||||
if !h.lc.hasAudit("orchestrate|replacement_done|node:" + oldNode) {
|
||||
t.Errorf("replacement_done audit log not found; got: %v", h.lc.auditLogs)
|
||||
}
|
||||
|
||||
// New node must NOT have been destroyed.
|
||||
for _, id := range h.prov.destroyCalls {
|
||||
if id == newNodeID {
|
||||
t.Errorf("new node %s was incorrectly destroyed", newNodeID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test: probe failure exhausts MaxAttempts → failed + alert (7 ticks + 3 clock advances)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestProbeFailMaxAttempts(t *testing.T) {
|
||||
const (
|
||||
oldNode = "old-node-fail"
|
||||
replacementUUID = "uuid-fail"
|
||||
)
|
||||
h := newHarness(t,
|
||||
&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg", Role: "entry"},
|
||||
)
|
||||
h.pushQueue(t, oldNode, replacementUUID)
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// ── Attempt 0 ────────────────────────────────────────────────────────────
|
||||
|
||||
h.tick(t) // Tick 1: pending → creating.
|
||||
h.tick(t) // Tick 2: creating → CreateNode(idem=uuid) → probing.
|
||||
|
||||
if h.prov.createCount() != 1 {
|
||||
t.Fatalf("attempt 0: CreateNode calls = %d; want 1", h.prov.createCount())
|
||||
}
|
||||
if h.prov.createCalls[0].IdempotencyKey != replacementUUID {
|
||||
t.Errorf("attempt 0 idem key = %q; want %q",
|
||||
h.prov.createCalls[0].IdempotencyKey, replacementUUID)
|
||||
}
|
||||
|
||||
newNode0 := h.prov.createCalls[0].ReturnedID
|
||||
h.snaps.setPass(newNode0) // even if snapshots pass, timeout overrides
|
||||
|
||||
h.clock.advance(orchestrate.ProbeTimeout + time.Second) // cause timeout
|
||||
|
||||
h.tick(t) // Tick 3: probing → timeout → DestroyNode(newNode0) → creating (attempts=1).
|
||||
|
||||
if h.prov.destroyCount() != 1 {
|
||||
t.Fatalf("after attempt 0 timeout: DestroyNode calls = %d; want 1", h.prov.destroyCount())
|
||||
}
|
||||
if h.prov.destroyCalls[0] != newNode0 {
|
||||
t.Errorf("attempt 0: expected %s destroyed; got %s", newNode0, h.prov.destroyCalls[0])
|
||||
}
|
||||
// Old node must NOT have been destroyed yet.
|
||||
for _, id := range h.prov.destroyCalls {
|
||||
if id == oldNode {
|
||||
t.Errorf("old node destroyed during attempt 0 (want: only after success)")
|
||||
}
|
||||
}
|
||||
|
||||
// ── Attempt 1 ────────────────────────────────────────────────────────────
|
||||
|
||||
h.tick(t) // Tick 4: creating → CreateNode(idem=uuid:retry:1) → probing.
|
||||
|
||||
if h.prov.createCount() != 2 {
|
||||
t.Fatalf("attempt 1: CreateNode calls = %d; want 2", h.prov.createCount())
|
||||
}
|
||||
call1 := h.prov.createCalls[1]
|
||||
if call1.IdempotencyKey != replacementUUID+":retry:1" {
|
||||
t.Errorf("attempt 1 idem key = %q; want %q:retry:1",
|
||||
call1.IdempotencyKey, replacementUUID)
|
||||
}
|
||||
|
||||
newNode1 := call1.ReturnedID
|
||||
h.snaps.setPass(newNode1)
|
||||
h.clock.advance(orchestrate.ProbeTimeout + time.Second)
|
||||
|
||||
h.tick(t) // Tick 5: probing → timeout → DestroyNode(newNode1) → creating (attempts=2).
|
||||
|
||||
if h.prov.destroyCount() != 2 {
|
||||
t.Fatalf("after attempt 1 timeout: DestroyNode calls = %d; want 2", h.prov.destroyCount())
|
||||
}
|
||||
|
||||
// ── Attempt 2 (final) ────────────────────────────────────────────────────
|
||||
|
||||
h.tick(t) // Tick 6: creating → CreateNode(idem=uuid:retry:2) → probing.
|
||||
|
||||
if h.prov.createCount() != 3 {
|
||||
t.Fatalf("attempt 2: CreateNode calls = %d; want 3", h.prov.createCount())
|
||||
}
|
||||
call2 := h.prov.createCalls[2]
|
||||
if call2.IdempotencyKey != replacementUUID+":retry:2" {
|
||||
t.Errorf("attempt 2 idem key = %q; want %q:retry:2",
|
||||
call2.IdempotencyKey, replacementUUID)
|
||||
}
|
||||
|
||||
newNode2 := call2.ReturnedID
|
||||
h.snaps.setPass(newNode2)
|
||||
h.clock.advance(orchestrate.ProbeTimeout + time.Second)
|
||||
|
||||
h.tick(t) // Tick 7: probing → timeout → DestroyNode(newNode2) → FAILED.
|
||||
|
||||
if h.prov.destroyCount() != 3 {
|
||||
t.Fatalf("after attempt 2 timeout: DestroyNode calls = %d; want 3", h.prov.destroyCount())
|
||||
}
|
||||
|
||||
// Old node must never have been destroyed.
|
||||
for _, id := range h.prov.destroyCalls {
|
||||
if id == oldNode {
|
||||
t.Errorf("old node %s was incorrectly destroyed during failed replacement", oldNode)
|
||||
}
|
||||
}
|
||||
|
||||
// NotifyFault must be called exactly once.
|
||||
if n := h.notifier.count(); n != 1 {
|
||||
t.Errorf("NotifyFault calls = %d; want 1", n)
|
||||
}
|
||||
|
||||
// Record must be in failed phase.
|
||||
raw, err := h.rdb.Get(ctx, "sched:replace:"+replacementUUID).Result()
|
||||
if err != nil {
|
||||
t.Fatalf("read final record: %v", err)
|
||||
}
|
||||
var p struct {
|
||||
Phase orchestrate.Phase `json:"phase"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(raw), &p); err != nil {
|
||||
t.Fatalf("unmarshal phase: %v", err)
|
||||
}
|
||||
if p.Phase != orchestrate.PhaseFailed {
|
||||
t.Errorf("final phase = %q; want %q", p.Phase, orchestrate.PhaseFailed)
|
||||
}
|
||||
|
||||
// Each attempt must have used a different provider (rotation check).
|
||||
p0 := h.prov.createCalls[0].Spec.ProviderID
|
||||
p1 := h.prov.createCalls[1].Spec.ProviderID
|
||||
p2 := h.prov.createCalls[2].Spec.ProviderID
|
||||
if p0 == p1 || p1 == p2 || p0 == p2 {
|
||||
t.Errorf("providers should differ across retries; got [%s, %s, %s]", p0, p1, p2)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test: crash recovery — CreateNode NOT called again when creating is complete
|
||||
//
|
||||
// Simulates a crash where CreateNode succeeded and newNode was set in memory
|
||||
// but the record was NOT yet saved with phase=probing. On restart the record
|
||||
// shows phase=creating with newNode set; we expect the Replacer to resume
|
||||
// from probing without calling CreateNode again.
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestCrashRecovery(t *testing.T) {
|
||||
const (
|
||||
oldNode = "old-node-crash"
|
||||
replacementUUID = "uuid-crash"
|
||||
simulatedNewNode = "new-node-crash"
|
||||
)
|
||||
h := newHarness(t,
|
||||
&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg", Role: "entry"},
|
||||
)
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Inject a pre-crash record: phase=creating, newNode already set.
|
||||
// This models a crash AFTER CreateNode returned but BEFORE the record was
|
||||
// saved with phase=probing.
|
||||
type crashRecord struct {
|
||||
Phase string `json:"phase"`
|
||||
OldNode string `json:"oldNode"`
|
||||
NewNode string `json:"newNode"`
|
||||
CurrentProviderID string `json:"currentProviderId"`
|
||||
Attempts int `json:"attempts"`
|
||||
ProviderTried []string `json:"providerTried"`
|
||||
ProbeStreak int `json:"probeStreak"`
|
||||
PhaseStartedAt time.Time `json:"phaseStartedAt"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
}
|
||||
recJSON, _ := json.Marshal(crashRecord{
|
||||
Phase: "creating",
|
||||
OldNode: oldNode,
|
||||
NewNode: simulatedNewNode,
|
||||
ProviderTried: []string{},
|
||||
PhaseStartedAt: h.clock.Now(),
|
||||
UpdatedAt: h.clock.Now(),
|
||||
})
|
||||
if err := h.rdb.Set(ctx, "sched:replace:"+replacementUUID, recJSON, 0).Err(); err != nil {
|
||||
t.Fatalf("inject crash record: %v", err)
|
||||
}
|
||||
if err := h.rdb.SAdd(ctx, "sched:replace:index", replacementUUID).Err(); err != nil {
|
||||
t.Fatalf("inject index: %v", err)
|
||||
}
|
||||
|
||||
// Add the simulated new node to the lifecycle mock.
|
||||
h.lc.addNode(&orchestrate.NodeInfo{ID: simulatedNewNode, Tier: "free", Region: "hkg"})
|
||||
h.snaps.setPass(simulatedNewNode)
|
||||
|
||||
// Tick 1 ("restart"): phase=creating, newNode already set → skip CreateNode → probing.
|
||||
h.tick(t)
|
||||
if h.prov.createCount() != 0 {
|
||||
t.Fatalf("tick 1 (recovery): CreateNode called %d times; want 0", h.prov.createCount())
|
||||
}
|
||||
|
||||
h.tick(t) // Tick 2: probing streak=1.
|
||||
h.tick(t) // Tick 3: probing streak=2 → activating.
|
||||
h.tick(t) // Tick 4: activating → SetWeight(10) + TransitionStatus + BumpVersion + grayscale → draining_old.
|
||||
h.tick(t) // Tick 5: draining_old → DestroyNode(oldNode) → done.
|
||||
|
||||
// CreateNode must never have been called.
|
||||
if h.prov.createCount() != 0 {
|
||||
t.Errorf("CreateNode called %d times during crash recovery; want 0", h.prov.createCount())
|
||||
}
|
||||
|
||||
// New node must be at weight 10.
|
||||
if w := h.lc.weightOf(simulatedNewNode); w != 10 {
|
||||
t.Errorf("new node weight = %d; want 10", w)
|
||||
}
|
||||
|
||||
// Old node must be destroyed.
|
||||
oldDestroyed := false
|
||||
for _, id := range h.prov.destroyCalls {
|
||||
if id == oldNode {
|
||||
oldDestroyed = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !oldDestroyed {
|
||||
t.Errorf("old node %s not destroyed in crash-recovery path; destroyCalls = %v",
|
||||
oldNode, h.prov.destroyCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test: idempotency — duplicate queue entries produce a single replacement
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestQueueReplayIdempotent(t *testing.T) {
|
||||
const (
|
||||
oldNode = "old-node-idem"
|
||||
replacementUUID = "uuid-idem"
|
||||
)
|
||||
h := newHarness(t,
|
||||
&orchestrate.NodeInfo{ID: oldNode, Tier: "free", Region: "hkg"},
|
||||
)
|
||||
|
||||
// Push the same entry twice (duplicate delivery).
|
||||
h.pushQueue(t, oldNode, replacementUUID)
|
||||
h.pushQueue(t, oldNode, replacementUUID)
|
||||
|
||||
// Tick 1: both entries drained; SetNX ensures a single record; stepPending → creating.
|
||||
h.tick(t)
|
||||
if h.prov.createCount() != 0 {
|
||||
t.Errorf("tick 1: CreateNode should not be called yet; got %d", h.prov.createCount())
|
||||
}
|
||||
|
||||
// Tick 2: creating → CreateNode called exactly once.
|
||||
h.tick(t)
|
||||
if h.prov.createCount() != 1 {
|
||||
t.Errorf("CreateNode calls = %d after two queue entries; want 1", h.prov.createCount())
|
||||
}
|
||||
|
||||
// Only one UUID must be in the in-flight index.
|
||||
members, err := h.rdb.SMembers(context.Background(), "sched:replace:index").Result()
|
||||
if err != nil {
|
||||
t.Fatalf("smembers: %v", err)
|
||||
}
|
||||
if len(members) != 1 {
|
||||
t.Errorf("index member count = %d; want 1", len(members))
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test: grayscale weight ramp 10 → 25 → 50 → 75 → 100 (each step = 6 h)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestGrayscaleAdvance(t *testing.T) {
|
||||
const nodeID = "gray-node-001"
|
||||
|
||||
rdb, _ := newTestRedis(t)
|
||||
lc := newMockLC(&orchestrate.NodeInfo{ID: nodeID})
|
||||
clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
|
||||
|
||||
// Seed a grayscale record at weight 10.
|
||||
grayKey := "sched:gray:" + nodeID
|
||||
type grayRec struct {
|
||||
NodeID string `json:"nodeId"`
|
||||
CurrentWeight int `json:"currentWeight"`
|
||||
StartedAt time.Time `json:"startedAt"`
|
||||
LastAdvancedAt time.Time `json:"lastAdvancedAt"`
|
||||
}
|
||||
data, _ := json.Marshal(grayRec{
|
||||
NodeID: nodeID,
|
||||
CurrentWeight: 10,
|
||||
StartedAt: clock.Now(),
|
||||
LastAdvancedAt: clock.Now(),
|
||||
})
|
||||
if err := rdb.Set(context.Background(), grayKey, data, 0).Err(); err != nil {
|
||||
t.Fatalf("seed gray record: %v", err)
|
||||
}
|
||||
|
||||
g := orchestrate.NewGrayscale(rdb, lc, clock)
|
||||
ctx := context.Background()
|
||||
|
||||
// Advance the clock by GrayscaleInterval (6 h) and call Advance() four times.
|
||||
// Expected progression: 10 → 25 → 50 → 75 → 100 (record deleted).
|
||||
wantWeights := []int{25, 50, 75, 100}
|
||||
|
||||
for step, want := range wantWeights {
|
||||
clock.advance(orchestrate.GrayscaleInterval)
|
||||
|
||||
if err := g.Advance(ctx); err != nil {
|
||||
t.Fatalf("step %d Advance(): %v", step, err)
|
||||
}
|
||||
|
||||
if got := lc.weightOf(nodeID); got != want {
|
||||
t.Errorf("step %d: weight = %d; want %d", step, got, want)
|
||||
}
|
||||
|
||||
// Audit log must record the weight advance.
|
||||
auditKey := "grayscale|weight_advanced|node:" + nodeID
|
||||
if !lc.hasAudit(auditKey) {
|
||||
t.Errorf("step %d: audit log %q not found; logs = %v", step, auditKey, lc.auditLogs)
|
||||
}
|
||||
}
|
||||
|
||||
// After weight 100, the grayscale record must be deleted.
|
||||
if _, err := rdb.Get(ctx, grayKey).Result(); err != redis.Nil {
|
||||
t.Error("grayscale record should be deleted after weight 100")
|
||||
}
|
||||
|
||||
// Final weight must be 100.
|
||||
if lc.weightOf(nodeID) != 100 {
|
||||
t.Errorf("final weight = %d; want 100", lc.weightOf(nodeID))
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test: grayscale does not advance before the interval elapses
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestGrayscaleNoAdvanceBeforeInterval(t *testing.T) {
|
||||
const nodeID = "gray-node-wait"
|
||||
|
||||
rdb, _ := newTestRedis(t)
|
||||
lc := newMockLC(&orchestrate.NodeInfo{ID: nodeID})
|
||||
clock := newMockClock(time.Unix(1_700_000_000, 0).UTC())
|
||||
|
||||
grayKey := "sched:gray:" + nodeID
|
||||
type grayRec struct {
|
||||
NodeID string `json:"nodeId"`
|
||||
CurrentWeight int `json:"currentWeight"`
|
||||
StartedAt time.Time `json:"startedAt"`
|
||||
LastAdvancedAt time.Time `json:"lastAdvancedAt"`
|
||||
}
|
||||
data, _ := json.Marshal(grayRec{
|
||||
NodeID: nodeID,
|
||||
CurrentWeight: 10,
|
||||
StartedAt: clock.Now(),
|
||||
LastAdvancedAt: clock.Now(),
|
||||
})
|
||||
rdb.Set(context.Background(), grayKey, data, 0)
|
||||
|
||||
g := orchestrate.NewGrayscale(rdb, lc, clock)
|
||||
|
||||
// Advance only 5 h (< 6 h interval).
|
||||
clock.advance(5 * time.Hour)
|
||||
|
||||
if err := g.Advance(context.Background()); err != nil {
|
||||
t.Fatalf("Advance(): %v", err)
|
||||
}
|
||||
|
||||
// Weight must NOT have been updated.
|
||||
if w := lc.weightOf(nodeID); w != 0 {
|
||||
t.Errorf("weight was updated too early: got %d; want 0", w)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user