pulse-lets-go/cmd/emulator/main.go

// Эмулятор метрик для pulse-lets-go.
// Публикует метрики от эмулируемых нод в NATS, имитируя разные сценарии нагрузки.
//
//	Сценарии:
//	  normal     — все ноды здоровы, случайная загрузка (проверка weighted score + best_node)
//	  overload   — последняя нода перегружена (active_calls >= max_calls → lethal)
//	  cpu-low    — последняя нода с низким idle_cpu (< 5 → lethal)
//	  fail-high  — последняя нода с высоким call_failure_rate (> 15 → lethal)
//	  node-down  — последняя нода status="down" (→ lethal)
//	  stale      — первая нода перестаёт отправлять метрики через 10 сек (→ lethal)
//	  chaos      — случайное чередование lethal-состояний у случайных нод
package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"math/rand"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	natsgo "github.com/nats-io/nats.go"
)

// --- Модель метрики ---

type nodeMetric struct {
	NodeID          string  `json:"node_id"`
	TS              int64   `json:"ts"`
	Status          string  `json:"status"`
	ActiveCalls     int     `json:"active_calls"`
	MaxCalls        int     `json:"max_calls"`
	IdleCPU         float64 `json:"idle_cpu"`
	LoadAvg         float64 `json:"load_avg"`
	CallFailureRate float64 `json:"call_failure_rate"`
	SIPGateway      string  `json:"sip_gateway,omitempty"`
}

// --- Спецификация значения (фиксированное или случайный диапазон) ---

type valueSpec struct {
	fixed  float64
	min    float64
	max    float64
	isRand bool
}

func fixed(v float64) valueSpec { return valueSpec{fixed: v} }
func rrange(min, max float64) valueSpec {
	return valueSpec{min: min, max: max, isRand: true}
}

func (v valueSpec) get() float64 {
	if v.isRand {
		return v.min + rand.Float64()*(v.max-v.min)
	}
	return v.fixed
}

// --- Конфигурация ноды в эмуляторе ---

type nodeSpec struct {
	nodeID     string
	status     valueSpec // строка: ok / down
	calls      valueSpec
	maxCalls   float64 // фиксированное
	idleCPU    valueSpec
	loadAvg    valueSpec
	failRate   valueSpec
	sipGateway string // SIP-адрес для авто-создания транка
}

func (ns *nodeSpec) generate() nodeMetric {
	return nodeMetric{
		NodeID:          ns.nodeID,
		TS:              time.Now().Unix(),
		Status:          statusFromValue(ns.status.get()),
		ActiveCalls:     int(ns.calls.get()),
		MaxCalls:        int(ns.maxCalls),
		IdleCPU:         ns.idleCPU.get(),
		LoadAvg:         ns.loadAvg.get(),
		CallFailureRate: ns.failRate.get(),
		SIPGateway:      ns.sipGateway,
	}
}

func statusFromValue(v float64) string {
	if v < 0.5 {
		return "down"
	}
	return "ok"
}

// --- Сценарии ---

func scenarioNormal() []nodeSpec {
	return []nodeSpec{
		{
			nodeID: "pbx-01", status: fixed(1),
			calls: rrange(10, 100), maxCalls: 250,
			idleCPU: rrange(40, 95), loadAvg: rrange(0.1, 1.0), failRate: rrange(0, 2),
			sipGateway: "sip:pbx-01.lan:5060",
		},
		{
			nodeID: "pbx-02", status: fixed(1),
			calls: rrange(50, 200), maxCalls: 250,
			idleCPU: rrange(15, 45), loadAvg: rrange(0.5, 2.0), failRate: rrange(1, 8),
			sipGateway: "sip:pbx-02.lan:5060",
		},
		{
			nodeID: "pbx-03", status: fixed(1),
			calls: rrange(5, 60), maxCalls: 250,
			idleCPU: rrange(60, 90), loadAvg: rrange(0.05, 0.5), failRate: rrange(0, 1),
			sipGateway: "sip:pbx-03.lan:5060",
		},
	}
}

func scenarioOverload() []nodeSpec {
	specs := scenarioNormal()
	// Последняя нода перегружена: active_calls = max_calls → lethal
	specs[len(specs)-1].calls = fixed(250)
	specs[len(specs)-1].idleCPU = fixed(10)
	specs[len(specs)-1].loadAvg = fixed(5.0)
	return specs
}

func scenarioCPULow() []nodeSpec {
	specs := scenarioNormal()
	specs[len(specs)-1].idleCPU = fixed(3) // < 5 → lethal
	return specs
}

func scenarioFailHigh() []nodeSpec {
	specs := scenarioNormal()
	specs[len(specs)-1].failRate = fixed(20) // > 15 → lethal
	return specs
}

func scenarioNodeDown() []nodeSpec {
	specs := scenarioNormal()
	specs[len(specs)-1].status = fixed(0) // status=down → lethal
	return specs
}

func scenarioStale() []nodeSpec {
	// Identical to normal — stale управляется логикой в main (пропуск отправки)
	return scenarioNormal()
}

// --- Main ---

func main() {
	scenario := flag.String("scenario", "normal",
		"Сценарий: normal | overload | cpu-low | fail-high | node-down | stale | chaos")
	natsURL := flag.String("nats", "nats://localhost:4222", "NATS URL")
	interval := flag.Duration("interval", 5*time.Second, "Интервал отправки метрик")
	flag.Parse()

	nc, err := natsgo.Connect(*natsURL, natsgo.MaxReconnects(-1))
	if err != nil {
		log.Fatalf("ошибка NATS: %v", err)
	}
	defer nc.Close()

	var specs []nodeSpec
	switch *scenario {
	case "normal":
		specs = scenarioNormal()
	case "overload":
		specs = scenarioOverload()
	case "cpu-low":
		specs = scenarioCPULow()
	case "fail-high":
		specs = scenarioFailHigh()
	case "node-down":
		specs = scenarioNodeDown()
	case "stale":
		specs = scenarioStale()
	case "chaos":
		specs = scenarioChaos()
	default:
		log.Fatalf("неизвестный сценарий: %s", *scenario)
	}

	log.Printf("[emulator] сценарий=%s, нод=%d, NATS=%s, интервал=%s",
		*scenario, len(specs), *natsURL, *interval)

	if *scenario == "chaos" {
		runChaos(nc, specs, *interval)
		return
	}

	if *scenario == "stale" {
		runStale(nc, specs, *interval)
		return
	}

	runNormal(nc, specs, *interval)
}

// runNormal — обычный цикл: все ноды шлют метрики каждые interval.
func runNormal(nc *natsgo.Conn, specs []nodeSpec, interval time.Duration) {
	log.Printf("[emulator] запущен, Ctrl+C для остановки")
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)

	// Первая отправка сразу
	publishAll(nc, specs)

	for {
		select {
		case <-ticker.C:
			publishAll(nc, specs)
		case <-quit:
			log.Println("[emulator] остановлен")
			return
		}
	}
}

// runStale — первая нода шлёт метрики только первые 10 секунд, потом замолкает.
func runStale(nc *natsgo.Conn, specs []nodeSpec, interval time.Duration) {
	log.Printf("[emulator] сценарий stale: нода %s замолкнет через 10 сек", specs[0].nodeID)

	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)

	startTime := time.Now()
	staleNodeID := specs[0].nodeID

	publishAll(nc, specs)

	for {
		select {
		case <-ticker.C:
			elapsed := time.Since(startTime)
			for _, spec := range specs {
				if spec.nodeID == staleNodeID && elapsed > 10*time.Second {
					continue // пропускаем stale-ноду
				}
				metric := spec.generate()
				publish(nc, metric)
			}
		case <-quit:
			log.Println("[emulator] остановлен")
			return
		}
	}
}

// runChaos — случайно чередует lethal-состояния у случайных нод.
func runChaos(nc *natsgo.Conn, specs []nodeSpec, interval time.Duration) {
	log.Printf("[emulator] сценарий chaos: случайные lethal-состояния")

	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)

	// Сохраняем оригинальные спецификации для восстановления
	original := make([]nodeSpec, len(specs))
	copy(original, specs)

	var mu sync.Mutex

	publishAll(nc, specs)

	for {
		select {
		case <-ticker.C:
			mu.Lock()
			// Каждые 15-30 секунд случайно портим случайную ноду
			if rand.Intn(4) == 0 { // ~25% шанс на каждом тике
				idx := rand.Intn(len(specs))
				kind := rand.Intn(5)
				switch kind {
				case 0:
					specs[idx].status = fixed(0) // down
				case 1:
					specs[idx].calls = fixed(250) // overload
				case 2:
					specs[idx].idleCPU = fixed(2) // idle_cpu < 5
				case 3:
					specs[idx].failRate = fixed(25) // failure rate > 15
				case 4:
					// Восстановить оригинал
					specs[idx] = original[idx]
				}
			}
			publishAll(nc, specs)
			mu.Unlock()
		case <-quit:
			log.Println("[emulator] остановлен")
			return
		}
	}
}

func scenarioChaos() []nodeSpec {
	// Начинаем с normal
	return scenarioNormal()
}

// --- Helpers ---

func publishAll(nc *natsgo.Conn, specs []nodeSpec) {
	for _, spec := range specs {
		metric := spec.generate()
		publish(nc, metric)
	}
}

func publish(nc *natsgo.Conn, m nodeMetric) {
	subject := fmt.Sprintf("pulse.metrics.%s", m.NodeID)
	data, err := json.Marshal(m)
	if err != nil {
		log.Printf("[emulator] ошибка сериализации %s: %v", m.NodeID, err)
		return
	}
	if err := nc.Publish(subject, data); err != nil {
		log.Printf("[emulator] ошибка публикации %s: %v", m.NodeID, err)
		return
	}
	log.Printf("[emulator] → %s: calls=%d/%d idle=%.0f%% load=%.2f fail=%.1f%% status=%s gw=%s",
		m.NodeID, m.ActiveCalls, m.MaxCalls, m.IdleCPU, m.LoadAvg, m.CallFailureRate, m.Status, m.SIPGateway)
}