beszel-ipv6/agent/network_monitor_history.go

package agent

import (
	"math"
	"sync"
	"time"

	"github.com/henrygd/beszel/internal/entities/monitor"
)

// Monitors run at user-defined intervals (e.g., every 10s).
// To keep memory usage low and constant, data is stored in two layers:
// 1. Raw samples: The most recent individual results (kept for monitorRawRetention).
// 2. Minute buckets: A ring buffer of 61 buckets, each representing one
//    wall-clock minute. Samples collected within the same minute are aggregated
//    (sum, min, max, count) into a single bucket.
//
// Short-term requests (<= 61s) use raw samples.
// Long-term requests (up to 1h) use the minute buckets to avoid storing thousands
// of individual data points.

const (
	// monitorRawRetention is the duration to keep individual samples
	monitorRawRetention = 61 * time.Second
	// monitorMinuteBucketLen is the number of 1-minute buckets to keep (1 hour + 1 for partials)
	monitorMinuteBucketLen int32 = 61
)

// monitorHistory owns retention and aggregation, independently of probe execution.
type monitorHistory struct {
	mu          sync.Mutex
	sampleCount int64
	samples     []monitorSample
	buckets     [monitorMinuteBucketLen]monitorBucket
}

func newMonitorHistory() *monitorHistory {
	// Start small for typical intervals; append grows the buffer for faster probes.
	return &monitorHistory{samples: make([]monitorSample, 0, 4)}
}

func (h *monitorHistory) clone() *monitorHistory {
	h.mu.Lock()
	defer h.mu.Unlock()
	cloned := newMonitorHistory()
	cloned.samples = append(cloned.samples, h.samples...)
	cloned.buckets = h.buckets
	cloned.sampleCount = h.sampleCount
	return cloned
}

func (h *monitorHistory) result(duration time.Duration, now time.Time) (monitor.Result, bool) {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.resultLocked(duration, now)
}

func (h *monitorHistory) record(sample monitorSample) monitor.Result {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.addSampleLocked(sample)
	result, _ := h.resultLocked(time.Minute, sample.timestamp)
	return result
}

// monitorSample stores one monitor attempt and its collection time.
type monitorSample struct {
	responseUs int64 // -1 means loss
	timestamp  time.Time
}

// monitorBucket stores one minute of aggregated monitor data.
type monitorBucket struct {
	minute int32
	filled bool
	stats  monitorAggregate
}

// monitorAggregate accumulates successful response stats and total sample counts.
type monitorAggregate struct {
	sumUs        int64
	minUs        int64
	maxUs        int64
	totalCount   int64
	successCount int64
}

// newMonitorAggregate initializes an aggregate with an unset minimum value.
func newMonitorAggregate() monitorAggregate {
	return monitorAggregate{minUs: math.MaxInt64}
}

// addResponse folds a single monitor sample into the aggregate.
func (agg *monitorAggregate) addResponse(responseUs int64) {
	agg.totalCount++
	if responseUs < 0 {
		return
	}
	agg.successCount++
	agg.sumUs += responseUs
	if responseUs < agg.minUs {
		agg.minUs = responseUs
	}
	if responseUs > agg.maxUs {
		agg.maxUs = responseUs
	}
}

// addAggregate merges another aggregate into this one.
func (agg *monitorAggregate) addAggregate(other monitorAggregate) {
	if other.totalCount == 0 {
		return
	}
	agg.totalCount += other.totalCount
	agg.successCount += other.successCount
	agg.sumUs += other.sumUs
	if other.successCount == 0 {
		return
	}
	if agg.minUs == math.MaxInt64 || other.minUs < agg.minUs {
		agg.minUs = other.minUs
	}
	if other.maxUs > agg.maxUs {
		agg.maxUs = other.maxUs
	}
}

// hasData reports whether the aggregate contains any samples.
func (agg monitorAggregate) hasData() bool {
	return agg.totalCount > 0
}

// result converts the aggregate into the monitor result format.
func (agg monitorAggregate) result() monitor.Result {
	avg := agg.avgResponse()
	result := monitor.Result{
		AvgResponse:  avg,
		MinResponse:  agg.minUs,
		MaxResponse:  agg.maxUs,
		PacketLoss:   agg.lossPercentage(),
		TotalCount:   agg.totalCount,
		SuccessCount: agg.successCount,
		ResponseSum:  agg.sumUs,
	}
	if agg.successCount == 0 {
		result.MinResponse, result.MaxResponse = 0, 0
	}
	return result
}

// avgResponse returns the rounded average of successful samples.
func (agg monitorAggregate) avgResponse() int64 {
	if agg.successCount == 0 {
		return 0
	}
	return agg.sumUs / agg.successCount

}

// lossPercentage returns the rounded failure rate for the aggregate.
func (agg monitorAggregate) lossPercentage() float64 {
	if agg.totalCount == 0 {
		return 0
	}
	return math.Round(float64(agg.totalCount-agg.successCount)/float64(agg.totalCount)*10000) / 100
}

// resultLocked returns the aggregated monitor result for the requested duration along with a bool indicating whether any data was available.
func (h *monitorHistory) resultLocked(duration time.Duration, now time.Time) (monitor.Result, bool) {
	agg := h.aggregateLocked(duration, now)
	if !agg.hasData() {
		// short realtime windows (e.g. the 1s window used for 1m/realtime charts) often fall
		// between monitor samples since monitors run at longer, user-defined intervals; fall back to
		// the most recent sample so realtime requests still report current status.
		agg = h.latestSampleAggregateLocked()
	}
	hourAgg := h.aggregateLocked(time.Hour, now)
	if !agg.hasData() {
		return monitor.Result{}, false
	}

	result := agg.result()
	if len(h.samples) > 0 {
		result.LastProbeAt = h.samples[len(h.samples)-1].timestamp.UnixMilli()
	}

	result.AvgResponse1h = hourAgg.avgResponse()
	result.MinResponse1h = hourAgg.minUs
	result.MaxResponse1h = hourAgg.maxUs
	result.PacketLoss1h = hourAgg.lossPercentage()
	result.SampleCount = h.sampleCount

	if hourAgg.successCount == 0 {
		result.MinResponse1h, result.MaxResponse1h = 0, 0
	}
	return result, true
}

// latestSampleAggregateLocked returns an aggregate containing only the most recent sample, if any.
func (h *monitorHistory) latestSampleAggregateLocked() monitorAggregate {
	agg := newMonitorAggregate()
	if len(h.samples) == 0 {
		return agg
	}
	agg.addResponse(h.samples[len(h.samples)-1].responseUs)
	return agg
}

// aggregateLocked collects monitor data for the requested time window.
func (h *monitorHistory) aggregateLocked(duration time.Duration, now time.Time) monitorAggregate {
	cutoff := now.Add(-duration)
	// Keep short windows exact; longer windows read from minute buckets to avoid raw-sample retention.
	if duration <= monitorRawRetention {
		return aggregateSamplesSince(h.samples, cutoff)
	}
	return aggregateBucketsSince(h.buckets[:], cutoff, now)
}

// aggregateSamplesSince aggregates raw samples newer than the cutoff.
func aggregateSamplesSince(samples []monitorSample, cutoff time.Time) monitorAggregate {
	agg := newMonitorAggregate()
	for _, sample := range samples {
		if sample.timestamp.Before(cutoff) {
			continue
		}
		agg.addResponse(sample.responseUs)
	}
	return agg
}

// aggregateBucketsSince aggregates minute buckets overlapping the requested window.
func aggregateBucketsSince(buckets []monitorBucket, cutoff, now time.Time) monitorAggregate {
	agg := newMonitorAggregate()
	startMinute := int32(cutoff.Unix() / 60)
	endMinute := int32(now.Unix() / 60)
	for _, bucket := range buckets {
		if !bucket.filled || bucket.minute < startMinute || bucket.minute > endMinute {
			continue
		}
		agg.addAggregate(bucket.stats)
	}
	return agg
}

// addSampleLocked stores a fresh sample in both raw and per-minute retention buffers.
func (h *monitorHistory) addSampleLocked(sample monitorSample) {
	h.sampleCount++
	cutoff := sample.timestamp.Add(-monitorRawRetention)
	start := 0
	for i := range h.samples {
		if !h.samples[i].timestamp.Before(cutoff) {
			start = i
			break
		}
		if i == len(h.samples)-1 {
			start = len(h.samples)
		}
	}
	if start > 0 {
		size := copy(h.samples, h.samples[start:])
		h.samples = h.samples[:size]
	}
	h.samples = append(h.samples, sample)

	minute := int32(sample.timestamp.Unix() / 60)
	// Each slot stores one wall-clock minute, so the ring stays fixed-size at ~1h per monitor.
	bucket := &h.buckets[minute%monitorMinuteBucketLen]
	if !bucket.filled || bucket.minute != minute {
		bucket.minute = minute
		bucket.filled = true
		bucket.stats = newMonitorAggregate()
	}
	bucket.stats.addResponse(sample.responseUs)
}