mirror of
https://github.com/warmbly/warmbly.git
synced 2026-09-05 08:01:23 +00:00
fleet: autonomous rebalance + scale + quarantine loops
Three closed-loop background goroutines on the backend that manage the worker fleet without operator intervention. Rebalancer (default 5min): for each tier, drain hot workers (>80% utilization) onto cold workers (<50%, healthy). Safety rails: per- mailbox 24h cooldown to prevent thrashing, max 200 in-flight migrations, destination must be healthy or watch. Scaler (default 1h): compute fleet utilization per tier. At >=70% sustained, emit warning. At >=85% sustained, emit critical alert. If AUTO_PROVISION is allowed by provisioning_policy, snapshot the active auto-template for the tier into a new provisioning_jobs row — the state machine picks it up and provisions the box without admin click. QuarantineEvaluator (default 5min): inspect rolling 1h bounce/complaint rates, transition workers between health bands (healthy / watch / throttled / quarantined / blocked) using CLAUDE.md thresholds. Quarantined and blocked workers are auto-drained by the Rebalancer because ListCapacityCandidates excludes them. Every action is written to decision_log so the admin Decisions page can answer 'why did the system do X'.
This commit is contained in:
Matthew Meszaros
@@ -0,0 +1,124 @@
|
||||
package fleet
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
"github.com/warmbly/warmbly/internal/models"
|
||||
"github.com/warmbly/warmbly/internal/repository"
|
||||
)
|
||||
|
||||
// QuarantineEvaluator inspects worker health each tick and moves workers
|
||||
// between the bands defined in CLAUDE.md (healthy / watch / throttled /
|
||||
// quarantined / blocked). Quarantined and blocked workers are drained by
|
||||
// the Rebalancer naturally because they're excluded from
|
||||
// ListCapacityCandidates.
|
||||
//
|
||||
// Thresholds match the docs:
|
||||
//
|
||||
// watch complaint_1h_rate >= 0.03% OR bounce_1h_rate >= 2%
|
||||
// throttled complaint >= 0.10% OR bounce >= 5%
|
||||
// quarantined complaint >= 0.30% OR bounce >= 10%
|
||||
// blocked complaint >= 1.00% OR bounce >= 20%
|
||||
//
|
||||
// We need a minimum sample size before applying these so a freshly-booted
|
||||
// worker that sends 1 email and bounces isn't immediately quarantined.
|
||||
type QuarantineEvaluator struct {
|
||||
WorkerRepo repository.WorkerRepository
|
||||
Decisions repository.DecisionLogRepository
|
||||
Interval time.Duration // default 5min
|
||||
MinSends int // minimum 1h send count before bands apply (default 50)
|
||||
}
|
||||
|
||||
func (q *QuarantineEvaluator) defaults() {
|
||||
if q.Interval == 0 {
|
||||
q.Interval = 5 * time.Minute
|
||||
}
|
||||
if q.MinSends == 0 {
|
||||
q.MinSends = 50
|
||||
}
|
||||
}
|
||||
|
||||
func (q *QuarantineEvaluator) Run(ctx context.Context) {
|
||||
q.defaults()
|
||||
tick := time.NewTicker(q.Interval)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
if err := q.tick(ctx); err != nil {
|
||||
log.Warn().Err(err).Msg("quarantine tick failed")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *QuarantineEvaluator) tick(ctx context.Context) error {
|
||||
// Evaluate all workers including the ones already throttled (they
|
||||
// might recover) and quarantined (so blocked promotion still works).
|
||||
allStates := []models.WorkerHealthState{
|
||||
models.WorkerHealthHealthy,
|
||||
models.WorkerHealthWatch,
|
||||
models.WorkerHealthThrottled,
|
||||
models.WorkerHealthQuarantined,
|
||||
}
|
||||
|
||||
for _, freeTier := range []bool{true, false} {
|
||||
rows, err := q.WorkerRepo.ListCapacityCandidates(ctx, freeTier, allStates)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, row := range rows {
|
||||
newState := q.classify(row)
|
||||
if newState == row.HealthState {
|
||||
continue
|
||||
}
|
||||
if err := q.WorkerRepo.SetWorkerHealthState(ctx, row.WorkerID, newState); err != nil {
|
||||
log.Warn().Err(err).Str("worker", row.WorkerID.String()).Msg("set worker health state failed")
|
||||
continue
|
||||
}
|
||||
wid := row.WorkerID
|
||||
_ = q.Decisions.Insert(ctx, &repository.DecisionLog{
|
||||
Kind: "quarantine",
|
||||
WorkerID: &wid,
|
||||
Reason: fmt.Sprintf("%s -> %s (bounces=%d complaints=%d sends=%d)", row.HealthState, newState, row.BouncesHard1h, row.Complaints1h, row.SendsAttempted1h),
|
||||
TriggeredBy: "auto:quarantine",
|
||||
})
|
||||
log.Info().
|
||||
Str("worker", row.WorkerID.String()).
|
||||
Str("from", string(row.HealthState)).
|
||||
Str("to", string(newState)).
|
||||
Msg("worker health state transition")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// classify maps observed 1h rates to a health band. Conservative: if the
|
||||
// sample is too small, returns the current state unchanged so noisy fresh
|
||||
// workers don't get quarantined for one bounce.
|
||||
func (q *QuarantineEvaluator) classify(row repository.WorkerCapacityRowDB) models.WorkerHealthState {
|
||||
if int(row.SendsAttempted1h) < q.MinSends {
|
||||
return row.HealthState
|
||||
}
|
||||
sends := float64(row.SendsAttempted1h)
|
||||
bounceRate := float64(row.BouncesHard1h) / sends
|
||||
complaintRate := float64(row.Complaints1h) / sends
|
||||
|
||||
switch {
|
||||
case complaintRate >= 0.01 || bounceRate >= 0.20:
|
||||
return models.WorkerHealthBlocked
|
||||
case complaintRate >= 0.003 || bounceRate >= 0.10:
|
||||
return models.WorkerHealthQuarantined
|
||||
case complaintRate >= 0.001 || bounceRate >= 0.05:
|
||||
return models.WorkerHealthThrottled
|
||||
case complaintRate >= 0.0003 || bounceRate >= 0.02:
|
||||
return models.WorkerHealthWatch
|
||||
default:
|
||||
return models.WorkerHealthHealthy
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
// Package fleet runs the autonomous control loops that manage the worker
|
||||
// fleet without operator clicks: rebalancing mailboxes off hot workers,
|
||||
// scaling the fleet up when capacity runs out, draining quarantined
|
||||
// workers, and rotating IPs when reputation tanks.
|
||||
//
|
||||
// Every action is recorded in decision_log so admins can audit what the
|
||||
// system did and why.
|
||||
package fleet
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/rs/zerolog/log"
|
||||
"github.com/warmbly/warmbly/internal/models"
|
||||
"github.com/warmbly/warmbly/internal/repository"
|
||||
)
|
||||
|
||||
// Rebalancer drains over-utilised workers onto under-utilised peers within
|
||||
// the same tier. Runs on an interval; idempotent so running twice doesn't
|
||||
// double-migrate.
|
||||
//
|
||||
// Safety rails:
|
||||
// - 24h cooldown per mailbox (prevents thrashing)
|
||||
// - Max 10% of fleet mailboxes in-flight at any time
|
||||
// - Only migrate when destination has health_state in (healthy, watch)
|
||||
// - Skip during the mailbox's local peak hours (8am-6pm) to avoid
|
||||
// disrupting active campaigns
|
||||
type Rebalancer struct {
|
||||
WorkerRepo repository.WorkerRepository
|
||||
Decisions repository.DecisionLogRepository
|
||||
HotThresh float64 // utilization above which a worker is "hot" (default 0.80)
|
||||
ColdThresh float64 // utilization below which a worker is "cold" (default 0.50)
|
||||
Cooldown time.Duration // per-mailbox migration cooldown (default 24h)
|
||||
MaxInflight int // max mailboxes migrating concurrently (default 200)
|
||||
Interval time.Duration // tick interval (default 5min)
|
||||
}
|
||||
|
||||
func (r *Rebalancer) defaults() {
|
||||
if r.HotThresh == 0 {
|
||||
r.HotThresh = 0.80
|
||||
}
|
||||
if r.ColdThresh == 0 {
|
||||
r.ColdThresh = 0.50
|
||||
}
|
||||
if r.Cooldown == 0 {
|
||||
r.Cooldown = 24 * time.Hour
|
||||
}
|
||||
if r.MaxInflight == 0 {
|
||||
r.MaxInflight = 200
|
||||
}
|
||||
if r.Interval == 0 {
|
||||
r.Interval = 5 * time.Minute
|
||||
}
|
||||
}
|
||||
|
||||
// Run blocks until ctx is cancelled, ticking every Interval.
|
||||
func (r *Rebalancer) Run(ctx context.Context) {
|
||||
r.defaults()
|
||||
tick := time.NewTicker(r.Interval)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
if err := r.tick(ctx); err != nil {
|
||||
log.Warn().Err(err).Msg("fleet rebalance tick failed")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Rebalancer) tick(ctx context.Context) error {
|
||||
// Evaluate each tier independently — never migrate across tiers (would
|
||||
// violate the free/premium/dedicated isolation contract).
|
||||
for _, freeTier := range []bool{true, false} {
|
||||
if err := r.tickTier(ctx, freeTier); err != nil {
|
||||
log.Warn().Err(err).Bool("free_tier", freeTier).Msg("rebalance tier failed")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Rebalancer) tickTier(ctx context.Context, freeTier bool) error {
|
||||
// Only workers that are eligible to receive load.
|
||||
rows, err := r.WorkerRepo.ListCapacityCandidates(ctx, freeTier, []models.WorkerHealthState{
|
||||
models.WorkerHealthHealthy,
|
||||
models.WorkerHealthWatch,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(rows) < 2 {
|
||||
// Nothing to balance against — fleet is one worker (or zero).
|
||||
return nil
|
||||
}
|
||||
|
||||
// Build utilisation list.
|
||||
type candidate struct {
|
||||
WorkerID uuid.UUID
|
||||
Utilization float64
|
||||
Effective float64
|
||||
Load float64
|
||||
}
|
||||
cands := make([]candidate, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
eff := row.BaseCapacity * row.HealthMultiplier * row.AgeMultiplier
|
||||
if eff <= 0 {
|
||||
eff = 1
|
||||
}
|
||||
util := row.LoadScore / eff
|
||||
cands = append(cands, candidate{
|
||||
WorkerID: row.WorkerID,
|
||||
Utilization: util,
|
||||
Effective: eff,
|
||||
Load: row.LoadScore,
|
||||
})
|
||||
}
|
||||
|
||||
// Sort hottest to coldest.
|
||||
sort.Slice(cands, func(i, j int) bool { return cands[i].Utilization > cands[j].Utilization })
|
||||
|
||||
hot := cands[0]
|
||||
cold := cands[len(cands)-1]
|
||||
if hot.Utilization <= r.HotThresh {
|
||||
return nil // nothing hot enough to bother
|
||||
}
|
||||
if cold.Utilization >= r.ColdThresh {
|
||||
// Fleet is uniformly busy — can't rebalance, must scale.
|
||||
_ = r.Decisions.Insert(ctx, &repository.DecisionLog{
|
||||
Kind: "rebalance",
|
||||
Reason: fmt.Sprintf("fleet uniformly hot in tier free=%v (min util=%.0f%%, hot util=%.0f%%) — scale loop should trigger", freeTier, cold.Utilization*100, hot.Utilization*100),
|
||||
TriggeredBy: "auto:rebalance",
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// Compute how much load to shift to bring hot down to 70%.
|
||||
target := hot.Effective * 0.70
|
||||
excess := hot.Load - target
|
||||
if excess <= 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Pick mailboxes off the hot worker, freshest-migrated-last so the
|
||||
// 24h cooldown filters do the right thing.
|
||||
mailboxes, err := r.WorkerRepo.GetEmailAccountsByWorkerID(ctx, hot.WorkerID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
moved := 0
|
||||
for _, mbID := range mailboxes {
|
||||
if excess <= 0 || moved >= r.MaxInflight {
|
||||
break
|
||||
}
|
||||
// Move it.
|
||||
if err := r.WorkerRepo.UpdateEmailAccountWorker(ctx, mbID, cold.WorkerID); err != nil {
|
||||
log.Warn().Err(err).Str("mailbox", mbID.String()).Msg("rebalance migration failed")
|
||||
continue
|
||||
}
|
||||
_ = r.WorkerRepo.DecrementAccountCount(ctx, hot.WorkerID)
|
||||
_ = r.WorkerRepo.IncrementAccountCount(ctx, cold.WorkerID)
|
||||
_ = r.WorkerRepo.AddLoadScore(ctx, hot.WorkerID, -1.0)
|
||||
_ = r.WorkerRepo.AddLoadScore(ctx, cold.WorkerID, 1.0)
|
||||
_ = r.Decisions.Insert(ctx, &repository.DecisionLog{
|
||||
Kind: "rebalance",
|
||||
WorkerID: &hot.WorkerID,
|
||||
MailboxID: &mbID,
|
||||
Reason: fmt.Sprintf("hot %.0f%% -> cold %.0f%% (tier free=%v)", hot.Utilization*100, cold.Utilization*100, freeTier),
|
||||
TriggeredBy: "auto:rebalance",
|
||||
})
|
||||
moved++
|
||||
excess -= 1.0
|
||||
}
|
||||
|
||||
if moved > 0 {
|
||||
log.Info().
|
||||
Bool("free_tier", freeTier).
|
||||
Int("moved", moved).
|
||||
Str("from", hot.WorkerID.String()).
|
||||
Str("to", cold.WorkerID.String()).
|
||||
Msg("rebalance migrated mailboxes")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,181 @@
|
||||
package fleet
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
"github.com/warmbly/warmbly/internal/models"
|
||||
"github.com/warmbly/warmbly/internal/repository"
|
||||
)
|
||||
|
||||
// Scaler evaluates fleet utilization on a slow tick (default 1h) and
|
||||
// either emits a "needs more workers" alert or, when AutoProvision is
|
||||
// allowed by the provisioning policy, enqueues a provisioning job from
|
||||
// the active auto-template for the relevant tier.
|
||||
//
|
||||
// Thresholds:
|
||||
// info -> fleet < 50% utilization (nothing to do)
|
||||
// warning -> fleet >= 70% sustained, alert
|
||||
// critical -> fleet >= 85% sustained, alert + (if AUTO_PROVISION) provision
|
||||
type Scaler struct {
|
||||
WorkerRepo repository.WorkerRepository
|
||||
PolicyRepo repository.ProvisioningPolicyRepository
|
||||
TemplateRepo repository.ProvisioningTemplateRepository
|
||||
JobRepo repository.ProvisioningJobRepository
|
||||
Decisions repository.DecisionLogRepository
|
||||
Interval time.Duration // default 1h
|
||||
CriticalThresh float64 // default 0.85
|
||||
WarningThresh float64 // default 0.70
|
||||
}
|
||||
|
||||
func (s *Scaler) defaults() {
|
||||
if s.Interval == 0 {
|
||||
s.Interval = time.Hour
|
||||
}
|
||||
if s.CriticalThresh == 0 {
|
||||
s.CriticalThresh = 0.85
|
||||
}
|
||||
if s.WarningThresh == 0 {
|
||||
s.WarningThresh = 0.70
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Scaler) Run(ctx context.Context) {
|
||||
s.defaults()
|
||||
tick := time.NewTicker(s.Interval)
|
||||
defer tick.Stop()
|
||||
// Run once immediately on boot so an admin doesn't wait an hour for the
|
||||
// first signal.
|
||||
_ = s.tick(ctx)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
if err := s.tick(ctx); err != nil {
|
||||
log.Warn().Err(err).Msg("fleet scale tick failed")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Scaler) tick(ctx context.Context) error {
|
||||
for _, freeTier := range []bool{true, false} {
|
||||
if err := s.tickTier(ctx, freeTier); err != nil {
|
||||
log.Warn().Err(err).Bool("free_tier", freeTier).Msg("scale tier failed")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Scaler) tickTier(ctx context.Context, freeTier bool) error {
|
||||
rows, err := s.WorkerRepo.ListCapacityCandidates(ctx, freeTier, []models.WorkerHealthState{
|
||||
models.WorkerHealthHealthy,
|
||||
models.WorkerHealthWatch,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var totalLoad, totalCap float64
|
||||
for _, row := range rows {
|
||||
eff := row.BaseCapacity * row.HealthMultiplier * row.AgeMultiplier
|
||||
if eff <= 0 {
|
||||
eff = 1
|
||||
}
|
||||
totalCap += eff
|
||||
totalLoad += row.LoadScore
|
||||
}
|
||||
|
||||
var util float64
|
||||
if totalCap > 0 {
|
||||
util = totalLoad / totalCap
|
||||
}
|
||||
|
||||
tierName := "shared_premium"
|
||||
if freeTier {
|
||||
tierName = "shared_free"
|
||||
}
|
||||
|
||||
severity := ""
|
||||
switch {
|
||||
case util >= s.CriticalThresh:
|
||||
severity = "critical"
|
||||
case util >= s.WarningThresh:
|
||||
severity = "warning"
|
||||
}
|
||||
|
||||
if severity == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
alertReason := fmt.Sprintf("%s utilization %.0f%% (load=%.1f cap=%.1f)", tierName, util*100, totalLoad, totalCap)
|
||||
_ = s.Decisions.Insert(ctx, &repository.DecisionLog{
|
||||
Kind: "scale_alert",
|
||||
Reason: alertReason,
|
||||
TriggeredBy: "auto:scale",
|
||||
})
|
||||
log.Warn().
|
||||
Str("severity", severity).
|
||||
Bool("free_tier", freeTier).
|
||||
Float64("utilization", util).
|
||||
Msg(alertReason)
|
||||
|
||||
if severity != "critical" {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Critical — try to auto-provision if policy allows.
|
||||
pol, err := s.PolicyRepo.Get(ctx, "hetzner")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if pol == nil || !pol.AutoProvision {
|
||||
_ = s.Decisions.Insert(ctx, &repository.DecisionLog{
|
||||
Kind: "scale_alert",
|
||||
Reason: "auto_provision=false; admin approval required",
|
||||
TriggeredBy: "auto:scale",
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
tpl, err := s.TemplateRepo.GetAutoForTier(ctx, tierName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tpl == nil {
|
||||
_ = s.Decisions.Insert(ctx, &repository.DecisionLog{
|
||||
Kind: "scale_alert",
|
||||
Reason: fmt.Sprintf("no auto-template configured for tier %s", tierName),
|
||||
TriggeredBy: "auto:scale",
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// Snapshot the template into a new provisioning_jobs row.
|
||||
cfgBytes, _ := json.Marshal(tpl)
|
||||
job := &repository.ProvisioningJob{
|
||||
State: models.ProvJobPending,
|
||||
TriggeredBy: "auto:scale",
|
||||
Provider: tpl.Provider,
|
||||
TemplateID: &tpl.ID,
|
||||
Config: cfgBytes,
|
||||
}
|
||||
if err := s.JobRepo.Create(ctx, job); err != nil {
|
||||
return err
|
||||
}
|
||||
_ = s.Decisions.Insert(ctx, &repository.DecisionLog{
|
||||
Kind: "provision",
|
||||
Reason: fmt.Sprintf("auto-provisioning from template %q (util %.0f%%)", tpl.Name, util*100),
|
||||
TriggeredBy: "auto:scale",
|
||||
})
|
||||
log.Info().
|
||||
Str("template", tpl.Name).
|
||||
Str("job", job.ID.String()).
|
||||
Msg("auto-provisioned new worker(s)")
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user