DebTik/internal/sdwan/health.go

package sdwan

import (
	"context"
	"log"
	"math"
	"os/exec"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"
)

var (
	rttRegex  = regexp.MustCompile(`time=([0-9.]+)\s*ms`)
	lossRegex = regexp.MustCompile(`([0-9.]+)%\s*packet\s*loss`)
)

type probeSample struct {
	latency float64
	success bool
	t       time.Time
}

// HealthMonitor periodically probes SD-WAN member interfaces and updates SLA telemetry.
type HealthMonitor struct {
	mu            sync.RWMutex
	samples       map[string][]probeSample // memberID -> rolling probe samples
	cancel        context.CancelFunc
	done          chan struct{}
	onStateChange func()
	applyResults  func([]*SDWANGroup) bool
	probe         func(string, string) (float64, bool)
}

func NewHealthMonitor(onStateChange func()) *HealthMonitor {
	return &HealthMonitor{
		samples:       make(map[string][]probeSample),
		onStateChange: onStateChange,
	}
}

// Start begins periodic background health checking for active SD-WAN members.
func (h *HealthMonitor) Start(getGroups func() []*SDWANGroup) {
	ctx, cancel := context.WithCancel(context.Background())
	h.cancel = cancel
	h.done = make(chan struct{})

	go func() {
		defer close(h.done)
		ticker := time.NewTicker(time.Second)
		defer ticker.Stop()
		lastProbe := make(map[string]time.Time)

		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				groups := dueProbeGroups(getGroups(), lastProbe, time.Now())
				if len(groups) == 0 {
					continue
				}
				changed := h.runProbeIteration(groups)
				if ctx.Err() != nil {
					return
				}
				if h.applyResults != nil {
					changed = h.applyResults(groups)
				}
				if changed && h.onStateChange != nil {
					h.onStateChange()
				}
			}
		}
	}()
}

// dueProbeGroups respects the configured interval for each group. New groups
// are probed on the next tick; removed or disabled groups lose their schedule.
func dueProbeGroups(groups []*SDWANGroup, last map[string]time.Time, now time.Time) []*SDWANGroup {
	active := make(map[string]bool)
	var due []*SDWANGroup
	for _, group := range groups {
		if group == nil || group.Disabled {
			continue
		}
		active[group.ID] = true
		interval := group.ProbeInterval
		if interval <= 0 {
			interval = 3
		}
		if previous, ok := last[group.ID]; !ok || now.Sub(previous) >= time.Duration(interval)*time.Second {
			due = append(due, group)
			last[group.ID] = now
		}
	}
	for id := range last {
		if !active[id] {
			delete(last, id)
		}
	}
	return due
}

// Stop terminates the probing goroutine.
func (h *HealthMonitor) Stop() {
	if h.cancel != nil {
		h.cancel()
		if h.done != nil {
			<-h.done
		}
	}
}

func (h *HealthMonitor) runProbeIteration(groups []*SDWANGroup) bool {
	var wg sync.WaitGroup
	slots := make(chan struct{}, 8)
	changes := make(chan bool, 1)
	for _, group := range groups {
		if group == nil || group.Disabled {
			continue
		}
		for _, member := range group.Members {
			if member == nil {
				continue
			}
			if member.Disabled || member.Interface == "" {
				member.Status = StatusDisabled
				continue
			}
			wg.Add(1)
			slots <- struct{}{}
			go func(m *SDWANMember) {
				defer wg.Done()
				defer func() { <-slots }()
				target := m.TargetProbe
				if target == "" {
					target = "1.1.1.1"
				}
				probe := h.executeProbe
				if h.probe != nil {
					probe = h.probe
				}
				latency, success := probe(m.Interface, target)
				h.mu.Lock()
				sampleKey := m.ID + "\x00" + m.Interface + "\x00" + target
				history := append(h.samples[sampleKey], probeSample{latency: latency, success: success, t: time.Now()})
				if len(history) > 10 {
					history = history[len(history)-10:]
				}
				h.samples[sampleKey] = history
				history = append([]probeSample(nil), history...)
				h.mu.Unlock()
				previous := m.Status
				m.LastChecked = time.Now()
				if success {
					m.ConsecutiveSuccesses++
					m.ConsecutiveFails = 0
				} else {
					m.ConsecutiveFails++
					m.ConsecutiveSuccesses = 0
				}
				h.calculateMetrics(m, history)
				if m.ConsecutiveFails >= 5 {
					m.Status = StatusDown
				} else if previous == StatusDown && m.ConsecutiveSuccesses < 2 {
					m.Status = StatusDown
				} else {
					maxLat := float64(m.MaxLatency)
					if maxLat <= 0 {
						maxLat = 250
					}
					maxLoss := float64(m.MaxLoss)
					if maxLoss <= 0 {
						maxLoss = 20
					}
					if m.PacketLoss > maxLoss || m.LiveLatency > maxLat {
						m.Status = StatusDegraded
					} else {
						m.Status = StatusHealthy
					}
				}
				if (previous == StatusDown) != (m.Status == StatusDown) {
					select {
					case changes <- true:
					default:
					}
					log.Printf("[SD-WAN] Health alert: %s status changed from %s to %s (latency: %.1fms, loss: %.1f%%)", m.Interface, previous, m.Status, m.LiveLatency, m.PacketLoss)
				}
			}(member)
		}
	}
	wg.Wait()
	select {
	case <-changes:
		return true
	default:
		return false
	}
}

func (h *HealthMonitor) executeProbe(iface, target string) (float64, bool) {
	// Sanitize probe target: fallback if empty, ending in .0, or known rate-limited anycast target
	cleanTarget := strings.TrimSpace(target)
	if cleanTarget == "" || cleanTarget == "188.114.97.0" || strings.HasSuffix(cleanTarget, ".0") {
		cleanTarget = "1.1.1.1"
	}

	// ping -c 2 -W 2 -i 0.2 -I <iface> <cleanTarget>
	cmd := exec.Command("ping", "-c", "2", "-W", "2", "-i", "0.2", "-I", iface, cleanTarget)
	out, err := cmd.CombinedOutput()
	if err != nil {
		return 0, false
	}

	str := string(out)
	if strings.Contains(str, "100% packet loss") || strings.Contains(str, " 0 received") || strings.Contains(str, "0 packets received") {
		return 0, false
	}

	matches := rttRegex.FindStringSubmatch(str)
	if len(matches) > 1 {
		if val, err := strconv.ParseFloat(matches[1], 64); err == nil {
			return val, true
		}
	}

	return 1.0, true
}

func (h *HealthMonitor) calculateMetrics(m *SDWANMember, history []probeSample) {
	if len(history) == 0 {
		return
	}

	total := len(history)
	failed := 0
	var sumLat float64
	var countSuccess int
	var latencies []float64

	for _, s := range history {
		if !s.success {
			failed++
		} else {
			sumLat += s.latency
			countSuccess++
			latencies = append(latencies, s.latency)
		}
	}

	m.PacketLoss = (float64(failed) / float64(total)) * 100.0

	if countSuccess > 0 {
		avg := sumLat / float64(countSuccess)
		m.LiveLatency = math.Round(avg*10) / 10

		// Jitter calculation (average difference between consecutive probes)
		if len(latencies) >= 2 {
			var diffSum float64
			for i := 1; i < len(latencies); i++ {
				diffSum += math.Abs(latencies[i] - latencies[i-1])
			}
			m.Jitter = math.Round((diffSum/float64(len(latencies)-1))*10) / 10
		}
	} else {
		m.LiveLatency = 0
		m.Jitter = 0
	}
}