mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 21:20:00 +00:00
alerts: answer history reads from the shared fold without full copies (#3093)
## Why A heap profile of the long-running dogfood install (77.6 MB live after a forced GC) shows the alert history projection holding 36 MB (47%). It keeps one merged alert per occurrence for the event log's 90-day retention: 1,895 occurrences there, with resolved snapshots averaging about 20 KB. Reads made it much worse (#1803): - every history read deep-copied the whole projection before applying its limit, so the attention summary's 500-row read copied all 1,895 occurrences; - every read with a `since` (the 30-day alert-quality window, date-filtered history) built a fresh projection and re-decoded every snapshot in its window. That churn is why the same install holds about 158 MB of Go heap for 78 MB of live data. ## What changed - One full-history fold, built and extended as before. Each occurrence also keeps a small mark per folded event (time, durable ID, and the `LastSeen` it contributed) and the index of its latest acknowledged snapshot. - A `since` read is answered from that fold. The window starts at each occurrence's first event that the log's own `occurred_at >= since` text comparison admits. `LastSeen` and acknowledgement come from that point on, and occurrences are ordered by their first in-window event, exactly as a walk of only the window would produce. - Overlay, sort and limit run on lightweight rows under the projection lock. Only returned rows are copied, once, by `canonicalizeAlertHistoryForOutput`. - The alerts contract notes the windowed-read rule and the pinning tests. ## Measured (pulse-dev, log shaped like the dogfood install's: about 1,900 occurrences, 40k events) | Read | main | this PR | |---|---|---| | 30-day window | 68.9 ms, 19.8 MB allocated | 5.2 ms, 2.8 MB | | Latest 500 | 12.7 ms, 8.5 MB | 3.6 ms, 2.0 MB | | Full history | 20.8 ms, 12.9 MB | 13.4 ms, 6.9 MB | The 36 MB the fold itself holds is unchanged. The new marks add about 56 bytes per lifecycle event, about 2 MB at 40k events. ## Checks - `TestWindowedHistoryMatchesAWalkOfOnlyTheWindow` compares every window and limit against an independent walk of only the window (the original algorithm). It covers 25 random histories with acknowledgements on both sides of the boundary, out-of-order `LastSeen`, uneven fractional seconds where RFC3339Nano text order and time order disagree, same-instant events, a history clear, and live overlays. - `TestWindowedHistoryReusesTheFullHistoryFold`, `TestLimitedHistoryReadCopiesOnlyReturnedRows` and `TestHistoryReadResultsDoNotAliasTheSharedFold` pin reuse, copying and isolation. - Negative controls: ignoring the window, building a fresh fold per `since` read, and copying every row before the limit each fail their test. - Removing the per-row clone did not fail the aliasing test, because the output copy already isolates results. That double copy was dropped, and the test guards isolation end to end. - Vet and the whole `internal/alerts/...` tree pass on pulse-dev. ## Not established How much held heap this saves on the dogfood install itself. That needs this build running there. Related to #1803
This commit is contained in:
rcourtman
@@ -1028,6 +1028,19 @@ Live active-alert overlays and returned objects are independent clones. Explicit
|
||||
Since queries preserve their existing event-window semantics without evicting
|
||||
the normal full-history fold. Repeated attention requests must not replay all
|
||||
retained snapshots independently or accumulate behind the history connection pool.
|
||||
Since queries are answered from that same fold (#1803). Each occurrence keeps one
|
||||
mark per folded event (time, durable ID and the LastSeen it contributed) and the
|
||||
index of its latest acknowledged snapshot. A window starts at the occurrence's
|
||||
first event that the log's own `occurred_at >= since` text comparison admits,
|
||||
takes LastSeen and acknowledgement from that point on, and orders occurrences by
|
||||
their first in-window event as a walk of only the window would. A read selects
|
||||
and limits rows before copying, so only returned rows are cloned.
|
||||
`TestWindowedHistoryMatchesAWalkOfOnlyTheWindow` holds this equal to an
|
||||
independent walk of only the window, and
|
||||
`TestLimitedHistoryReadCopiesOnlyReturnedRows`,
|
||||
`TestWindowedHistoryReusesTheFullHistoryFold` and
|
||||
`TestHistoryReadResultsDoNotAliasTheSharedFold` pin the copy, reuse and
|
||||
isolation rules.
|
||||
|
||||
|
||||
PBS node-status availability is distinct from connectivity. While a connected
|
||||
|
||||
@@ -17,19 +17,102 @@ import (
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog"
|
||||
)
|
||||
|
||||
// historyEventMark is what one folded lifecycle event contributed to its
|
||||
// occurrence. The marks let a Since window be answered from the full fold
|
||||
// instead of decoding every snapshot in the window again (#1803).
|
||||
type historyEventMark struct {
|
||||
at time.Time
|
||||
id int64
|
||||
lastSeen time.Time // the snapshot's LastSeen, or the resolution time if later
|
||||
}
|
||||
|
||||
// historyOccurrence accumulates one alert occurrence from its lifecycle
|
||||
// events, oldest to newest.
|
||||
type historyOccurrence struct {
|
||||
alert Alert
|
||||
firstEvent time.Time
|
||||
lastEvent time.Time
|
||||
resolved bool
|
||||
alert Alert
|
||||
events []historyEventMark
|
||||
// lastAcked indexes the latest event whose snapshot was acknowledged, or
|
||||
// is -1. A window starting after it never saw the acknowledgement.
|
||||
lastAcked int
|
||||
// The latest snapshot's own acknowledgement fields, before merging carried
|
||||
// an earlier acknowledgement forward onto them.
|
||||
rawAcknowledged bool
|
||||
rawAckTime *time.Time
|
||||
rawAckUser string
|
||||
}
|
||||
|
||||
func historyOccurrenceKey(alertID string, snapshot *Alert) string {
|
||||
return alertID + "\x00" + snapshot.StartTime.UTC().Format(time.RFC3339Nano)
|
||||
}
|
||||
|
||||
// windowStart returns the index of the first event the log's Since filter
|
||||
// (occurred_at >= since, compared as RFC3339Nano text) admits, or
|
||||
// len(events) when none is. Events are folded in the log's (occurred_at,
|
||||
// id) text order, so the comparison is monotonic.
|
||||
func (occ *historyOccurrence) windowStart(sinceKey string) int {
|
||||
return sort.Search(len(occ.events), func(i int) bool {
|
||||
return occ.events[i].at.UTC().Format(time.RFC3339Nano) >= sinceKey
|
||||
})
|
||||
}
|
||||
|
||||
// windowAck is the acknowledgement that folding events[start:] alone would
|
||||
// produce: an acknowledgement is carried forward, so only one inside the
|
||||
// window survives into it.
|
||||
func (occ *historyOccurrence) windowAck(start int) (bool, *time.Time, string) {
|
||||
if occ.lastAcked >= start {
|
||||
return occ.alert.Acknowledged, occ.alert.AckTime, occ.alert.AckUser
|
||||
}
|
||||
return occ.rawAcknowledged, occ.rawAckTime, occ.rawAckUser
|
||||
}
|
||||
|
||||
// windowLastSeen is the LastSeen that folding events[start:] alone would
|
||||
// produce: the latest contribution inside the window.
|
||||
func (occ *historyOccurrence) windowLastSeen(start int) time.Time {
|
||||
if start == 0 {
|
||||
return occ.alert.LastSeen
|
||||
}
|
||||
var latest time.Time
|
||||
for _, mark := range occ.events[start:] {
|
||||
if mark.lastSeen.After(latest) {
|
||||
latest = mark.lastSeen
|
||||
}
|
||||
}
|
||||
return latest
|
||||
}
|
||||
|
||||
// historyRow is one candidate history entry. Rows reference the shared fold
|
||||
// and are copied only when selected for output.
|
||||
type historyRow struct {
|
||||
key string
|
||||
occ *historyOccurrence // nil for an active alert the log never saw fire
|
||||
start int
|
||||
firstEvent time.Time
|
||||
overlay *Alert // the live active alert shown instead of the folded state
|
||||
}
|
||||
|
||||
func (row *historyRow) ack() (bool, *time.Time, string) {
|
||||
if row.overlay != nil {
|
||||
return row.overlay.Acknowledged, row.overlay.AckTime, row.overlay.AckUser
|
||||
}
|
||||
return row.occ.windowAck(row.start)
|
||||
}
|
||||
|
||||
// materialize returns the row's entry. It shares the fold's maps and pointers;
|
||||
// projectedHistory deep-copies every returned entry through
|
||||
// canonicalizeAlertHistoryForOutput before releasing the projection lock, so
|
||||
// that is the only copy a read makes.
|
||||
func (row *historyRow) materialize() Alert {
|
||||
if row.overlay != nil {
|
||||
return *row.overlay
|
||||
}
|
||||
alert := row.occ.alert
|
||||
if row.start > 0 {
|
||||
alert.LastSeen = row.occ.windowLastSeen(row.start)
|
||||
alert.Acknowledged, alert.AckTime, alert.AckUser = row.occ.windowAck(row.start)
|
||||
}
|
||||
return alert
|
||||
}
|
||||
|
||||
// AlertHistoryFromEvents projects the alert history list from the event
|
||||
// log. It reports ok=false when no event log is enabled, letting callers
|
||||
// fall back to the JSON-backed history manager.
|
||||
@@ -38,73 +121,19 @@ func (m *Manager) AlertHistoryFromEvents(since time.Time, limit int) ([]Alert, b
|
||||
if store == nil || !m.eventHistoryAuthoritative.Load() {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
occurrences, order, err := m.historyOccurrences(store, since)
|
||||
results, err := m.projectedHistory(store, since, limit)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// Overlay the live active alerts: an occurrence still firing shows its
|
||||
// current state (fresh LastSeen and value), exactly as the JSON
|
||||
// history's continuously updated entries do. Active alerts the log
|
||||
// never saw fire (log enabled mid-flight) are appended.
|
||||
m.mu.RLock()
|
||||
active := make([]*Alert, 0, len(m.activeAlerts))
|
||||
for storageKey, alert := range m.activeAlerts {
|
||||
if alert == nil {
|
||||
continue
|
||||
}
|
||||
exported := cloneAlertForOutput(alert)
|
||||
if exported == nil {
|
||||
continue
|
||||
}
|
||||
if exported.ID == "" {
|
||||
exported.ID = effectiveAlertID(alert, storageKey)
|
||||
}
|
||||
active = append(active, exported)
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
for _, alert := range active {
|
||||
key := historyOccurrenceKey(alert.ID, alert)
|
||||
if occ, exists := occurrences[key]; exists {
|
||||
ack := occ.alert.Acknowledged
|
||||
ackTime, ackUser := occ.alert.AckTime, occ.alert.AckUser
|
||||
occ.alert = *alert
|
||||
if ack && !occ.alert.Acknowledged {
|
||||
occ.alert.Acknowledged = true
|
||||
occ.alert.AckTime = ackTime
|
||||
occ.alert.AckUser = ackUser
|
||||
}
|
||||
occ.resolved = false
|
||||
} else if since.IsZero() || alert.StartTime.After(since) {
|
||||
occurrences[key] = &historyOccurrence{
|
||||
alert: *alert,
|
||||
firstEvent: alert.StartTime,
|
||||
lastEvent: alert.LastSeen,
|
||||
}
|
||||
order = append(order, key)
|
||||
}
|
||||
}
|
||||
|
||||
sort.SliceStable(order, func(i, j int) bool {
|
||||
return occurrences[order[i]].firstEvent.Before(occurrences[order[j]].firstEvent)
|
||||
})
|
||||
|
||||
results := make([]Alert, 0, len(order))
|
||||
for i := len(order) - 1; i >= 0; i-- {
|
||||
if limit > 0 && len(results) >= limit {
|
||||
break
|
||||
}
|
||||
results = append(results, occurrences[order[i]].alert)
|
||||
}
|
||||
return m.applyCurrentNodeDisplayNames(canonicalizeAlertHistoryForOutput(results)), true
|
||||
return m.applyCurrentNodeDisplayNames(results), true
|
||||
}
|
||||
|
||||
// historyProjection is a disposable fold of the durable log, never a second
|
||||
// source of truth. Its cursor is bounded by a committed event ID. Old imports,
|
||||
// retention and store replacement rebuild it using the original chronological
|
||||
// fold. A moving Since window keeps its original event-filter semantics and
|
||||
// does not evict the full-history fold used by attention and normal polling.
|
||||
// fold. A Since window is answered from this full-history fold with the same
|
||||
// result as walking only the window's events (windowStart, windowAck and
|
||||
// windowLastSeen), so windowed reads neither rebuild nor evict it.
|
||||
type historyProjection struct {
|
||||
store *eventlog.Store
|
||||
boundary eventlog.ReplayBoundary
|
||||
@@ -119,7 +148,7 @@ func newHistoryProjection(store *eventlog.Store) *historyProjection {
|
||||
return &historyProjection{store: store, occurrences: make(map[string]*historyOccurrence)}
|
||||
}
|
||||
|
||||
func (p *historyProjection) fold(since time.Time, throughID int64) error {
|
||||
func (p *historyProjection) fold(throughID int64) error {
|
||||
if throughID == 0 || throughID == p.boundary.LastID {
|
||||
return nil
|
||||
}
|
||||
@@ -130,7 +159,7 @@ func (p *historyProjection) fold(since time.Time, throughID int64) error {
|
||||
eventlog.TypeSnoozed, eventlog.TypeUnsnoozed, eventlog.TypeEscalated,
|
||||
eventlog.TypeHistoryImported, eventlog.TypeHistoryCleared,
|
||||
},
|
||||
Since: since, AfterID: p.boundary.LastID, ThroughID: throughID,
|
||||
AfterID: p.boundary.LastID, ThroughID: throughID,
|
||||
}, func(event eventlog.Event) error {
|
||||
// Match the log's (occurred_at, id) ordering. Equal timestamps are
|
||||
// safe because all appended IDs exceed the preceding boundary.
|
||||
@@ -156,72 +185,173 @@ func (p *historyProjection) fold(since time.Time, throughID int64) error {
|
||||
key := historyOccurrenceKey(event.AlertID, &snapshot)
|
||||
occ, exists := p.occurrences[key]
|
||||
if !exists {
|
||||
occ = &historyOccurrence{alert: snapshot, firstEvent: event.OccurredAt}
|
||||
occ = &historyOccurrence{alert: snapshot, lastAcked: -1}
|
||||
p.occurrences[key] = occ
|
||||
p.order = append(p.order, key)
|
||||
} else {
|
||||
occ.alert = mergeHistoryAlertSnapshots(occ.alert, snapshot)
|
||||
}
|
||||
occ.lastEvent = event.OccurredAt
|
||||
mark := historyEventMark{at: event.OccurredAt, id: event.ID, lastSeen: snapshot.LastSeen}
|
||||
if event.Type == eventlog.TypeResolved {
|
||||
occ.resolved = true
|
||||
// The JSON history's resolve path stamps the entry's LastSeen
|
||||
// with the resolution time so the row reflects the true
|
||||
// duration; mirror that.
|
||||
if event.OccurredAt.After(occ.alert.LastSeen) {
|
||||
occ.alert.LastSeen = event.OccurredAt
|
||||
}
|
||||
} else {
|
||||
occ.resolved = false
|
||||
if event.OccurredAt.After(mark.lastSeen) {
|
||||
mark.lastSeen = event.OccurredAt
|
||||
}
|
||||
}
|
||||
occ.events = append(occ.events, mark)
|
||||
if snapshot.Acknowledged {
|
||||
occ.lastAcked = len(occ.events) - 1
|
||||
}
|
||||
occ.rawAcknowledged, occ.rawAckTime, occ.rawAckUser = snapshot.Acknowledged, snapshot.AckTime, snapshot.AckUser
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func (m *Manager) historyOccurrences(store *eventlog.Store, since time.Time) (map[string]*historyOccurrence, []string, error) {
|
||||
m.historyProjectionMu.Lock()
|
||||
defer m.historyProjectionMu.Unlock()
|
||||
// currentHistoryProjection folds the log through its current boundary and
|
||||
// returns the shared projection. The caller must hold historyProjectionMu.
|
||||
func (m *Manager) currentHistoryProjection(store *eventlog.Store) (*historyProjection, error) {
|
||||
for {
|
||||
boundary, err := store.ReplayBoundary()
|
||||
if err != nil {
|
||||
m.historyProjection = nil
|
||||
return nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
p := m.historyProjection
|
||||
if !since.IsZero() || p == nil || p.store != store ||
|
||||
if p == nil || p.store != store ||
|
||||
p.boundary.RetentionRevision != boundary.RetentionRevision || p.boundary.LastID > boundary.LastID {
|
||||
p = newHistoryProjection(store)
|
||||
}
|
||||
err = p.fold(since, boundary.LastID)
|
||||
err = p.fold(boundary.LastID)
|
||||
if errors.Is(err, errHistoryEventBeforeCursor) {
|
||||
p = newHistoryProjection(store)
|
||||
err = p.fold(since, boundary.LastID)
|
||||
err = p.fold(boundary.LastID)
|
||||
}
|
||||
if err != nil {
|
||||
m.historyProjection = nil
|
||||
return nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
after, err := store.ReplayBoundary()
|
||||
if err != nil {
|
||||
m.historyProjection = nil
|
||||
return nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
if after.RetentionRevision != boundary.RetentionRevision {
|
||||
m.historyProjection = nil
|
||||
continue
|
||||
}
|
||||
p.boundary = boundary
|
||||
if since.IsZero() {
|
||||
m.historyProjection = p
|
||||
}
|
||||
// The live-state overlay and callers may mutate their result. Never
|
||||
// let those changes contaminate the durable fold or another reader.
|
||||
occurrences := make(map[string]*historyOccurrence, len(p.occurrences))
|
||||
for key, occurrence := range p.occurrences {
|
||||
copy := *occurrence
|
||||
copy.alert = *occurrence.alert.Clone()
|
||||
occurrences[key] = ©
|
||||
}
|
||||
return occurrences, append([]string(nil), p.order...), nil
|
||||
m.historyProjection = p
|
||||
return p, nil
|
||||
}
|
||||
}
|
||||
|
||||
// projectedHistory selects the history list, newest first, under the
|
||||
// projection lock, and copies only the entries it returns. Callers may mutate
|
||||
// the result without touching the shared fold or another reader's result.
|
||||
func (m *Manager) projectedHistory(store *eventlog.Store, since time.Time, limit int) ([]Alert, error) {
|
||||
m.historyProjectionMu.Lock()
|
||||
defer m.historyProjectionMu.Unlock()
|
||||
p, err := m.currentHistoryProjection(store)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
sinceKey := ""
|
||||
if !since.IsZero() {
|
||||
sinceKey = since.UTC().Format(time.RFC3339Nano)
|
||||
}
|
||||
rows := make([]historyRow, 0, len(p.order))
|
||||
for _, key := range p.order {
|
||||
occ := p.occurrences[key]
|
||||
start := 0
|
||||
if sinceKey != "" {
|
||||
if start = occ.windowStart(sinceKey); start == len(occ.events) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
rows = append(rows, historyRow{key: key, occ: occ, start: start, firstEvent: occ.events[start].at})
|
||||
}
|
||||
if sinceKey != "" {
|
||||
// Walking only the window appends each occurrence at its first event
|
||||
// inside the window, in the log's (occurred_at, id) order.
|
||||
keys := make([]string, len(rows))
|
||||
for i := range rows {
|
||||
keys[i] = rows[i].firstEvent.UTC().Format(time.RFC3339Nano)
|
||||
}
|
||||
order := make([]int, len(rows))
|
||||
for i := range order {
|
||||
order[i] = i
|
||||
}
|
||||
sort.SliceStable(order, func(a, b int) bool {
|
||||
ra, rb := order[a], order[b]
|
||||
if keys[ra] != keys[rb] {
|
||||
return keys[ra] < keys[rb]
|
||||
}
|
||||
return rows[ra].occ.events[rows[ra].start].id < rows[rb].occ.events[rows[rb].start].id
|
||||
})
|
||||
sorted := make([]historyRow, len(rows))
|
||||
for i, row := range order {
|
||||
sorted[i] = rows[row]
|
||||
}
|
||||
rows = sorted
|
||||
}
|
||||
|
||||
// Overlay the live active alerts: an occurrence still firing shows its
|
||||
// current state (fresh LastSeen and value), exactly as the JSON
|
||||
// history's continuously updated entries do. Active alerts the log
|
||||
// never saw fire (log enabled mid-flight) are appended.
|
||||
m.mu.RLock()
|
||||
active := make([]*Alert, 0, len(m.activeAlerts))
|
||||
for storageKey, alert := range m.activeAlerts {
|
||||
if alert == nil {
|
||||
continue
|
||||
}
|
||||
exported := cloneAlertForOutput(alert)
|
||||
if exported == nil {
|
||||
continue
|
||||
}
|
||||
if exported.ID == "" {
|
||||
exported.ID = effectiveAlertID(alert, storageKey)
|
||||
}
|
||||
active = append(active, exported)
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
index := make(map[string]int, len(rows))
|
||||
for i := range rows {
|
||||
index[rows[i].key] = i
|
||||
}
|
||||
for _, alert := range active {
|
||||
key := historyOccurrenceKey(alert.ID, alert)
|
||||
if i, exists := index[key]; exists {
|
||||
acknowledged, ackTime, ackUser := rows[i].ack()
|
||||
overlay := *alert
|
||||
if acknowledged && !overlay.Acknowledged {
|
||||
overlay.Acknowledged, overlay.AckTime, overlay.AckUser = true, ackTime, ackUser
|
||||
}
|
||||
rows[i].overlay = &overlay
|
||||
} else if since.IsZero() || alert.StartTime.After(since) {
|
||||
overlay := *alert
|
||||
index[key] = len(rows)
|
||||
rows = append(rows, historyRow{key: key, firstEvent: alert.StartTime, overlay: &overlay})
|
||||
}
|
||||
}
|
||||
|
||||
sort.SliceStable(rows, func(i, j int) bool {
|
||||
return rows[i].firstEvent.Before(rows[j].firstEvent)
|
||||
})
|
||||
|
||||
size := len(rows)
|
||||
if limit > 0 && limit < size {
|
||||
size = limit
|
||||
}
|
||||
results := make([]Alert, 0, size)
|
||||
for i := len(rows) - 1; i >= 0 && len(results) < size; i-- {
|
||||
results = append(results, rows[i].materialize())
|
||||
}
|
||||
return canonicalizeAlertHistoryForOutput(results), nil
|
||||
}
|
||||
|
||||
@@ -10,7 +10,10 @@ package alerts
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"reflect"
|
||||
"runtime"
|
||||
"sort"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -291,3 +294,263 @@ func TestHistoryProjectionWalksEveryLifecycleEventBeyondQueryCap(t *testing.T) {
|
||||
t.Fatalf("limited projection order = %q ... %q", limited[0].ID, limited[24].ID)
|
||||
}
|
||||
}
|
||||
|
||||
// Windowed and limited history reads are answered from the one shared
|
||||
// full-history fold (#1803). Before, every read deep-copied the whole fold and
|
||||
// every Since read re-decoded each snapshot in its window. These tests hold
|
||||
// the shared answer equal to an independent walk of only the window's events,
|
||||
// which is the original algorithm.
|
||||
|
||||
// referenceWindowedHistory is the original read path: a fresh fold of only
|
||||
// the events the log's Since filter admits, the live overlay, then the sort.
|
||||
func referenceWindowedHistory(t *testing.T, m *Manager, since time.Time, limit int) []Alert {
|
||||
t.Helper()
|
||||
type occurrence struct {
|
||||
alert Alert
|
||||
firstEvent time.Time
|
||||
}
|
||||
occurrences := map[string]*occurrence{}
|
||||
var order []string
|
||||
err := m.eventLogStore().WalkOldest(eventlog.Filter{
|
||||
Types: []string{
|
||||
eventlog.TypeFired, eventlog.TypeRefired, eventlog.TypeResolved,
|
||||
eventlog.TypeAcknowledged, eventlog.TypeUnacknowledged,
|
||||
eventlog.TypeSnoozed, eventlog.TypeUnsnoozed, eventlog.TypeEscalated,
|
||||
eventlog.TypeHistoryImported, eventlog.TypeHistoryCleared,
|
||||
},
|
||||
Since: since,
|
||||
}, func(event eventlog.Event) error {
|
||||
if event.Type == eventlog.TypeHistoryCleared {
|
||||
occurrences = map[string]*occurrence{}
|
||||
order = order[:0]
|
||||
return nil
|
||||
}
|
||||
var snapshot Alert
|
||||
if len(event.Snapshot) == 0 || json.Unmarshal(event.Snapshot, &snapshot) != nil {
|
||||
return nil
|
||||
}
|
||||
key := historyOccurrenceKey(event.AlertID, &snapshot)
|
||||
occ, exists := occurrences[key]
|
||||
if !exists {
|
||||
occ = &occurrence{alert: snapshot, firstEvent: event.OccurredAt}
|
||||
occurrences[key] = occ
|
||||
order = append(order, key)
|
||||
} else {
|
||||
occ.alert = mergeHistoryAlertSnapshots(occ.alert, snapshot)
|
||||
}
|
||||
if event.Type == eventlog.TypeResolved && event.OccurredAt.After(occ.alert.LastSeen) {
|
||||
occ.alert.LastSeen = event.OccurredAt
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("reference walk: %v", err)
|
||||
}
|
||||
m.mu.RLock()
|
||||
var active []*Alert
|
||||
for storageKey, alert := range m.activeAlerts {
|
||||
exported := cloneAlertForOutput(alert)
|
||||
if exported.ID == "" {
|
||||
exported.ID = effectiveAlertID(alert, storageKey)
|
||||
}
|
||||
active = append(active, exported)
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
for _, alert := range active {
|
||||
key := historyOccurrenceKey(alert.ID, alert)
|
||||
if occ, exists := occurrences[key]; exists {
|
||||
ack, ackTime, ackUser := occ.alert.Acknowledged, occ.alert.AckTime, occ.alert.AckUser
|
||||
occ.alert = *alert
|
||||
if ack && !occ.alert.Acknowledged {
|
||||
occ.alert.Acknowledged, occ.alert.AckTime, occ.alert.AckUser = true, ackTime, ackUser
|
||||
}
|
||||
} else if since.IsZero() || alert.StartTime.After(since) {
|
||||
occurrences[key] = &occurrence{alert: *alert, firstEvent: alert.StartTime}
|
||||
order = append(order, key)
|
||||
}
|
||||
}
|
||||
sort.SliceStable(order, func(i, j int) bool {
|
||||
return occurrences[order[i]].firstEvent.Before(occurrences[order[j]].firstEvent)
|
||||
})
|
||||
results := []Alert{}
|
||||
for i := len(order) - 1; i >= 0; i-- {
|
||||
if limit > 0 && len(results) >= limit {
|
||||
break
|
||||
}
|
||||
results = append(results, occurrences[order[i]].alert)
|
||||
}
|
||||
return m.applyCurrentNodeDisplayNames(canonicalizeAlertHistoryForOutput(results))
|
||||
}
|
||||
|
||||
// writeRandomHistory appends a lifecycle history with acknowledgements on
|
||||
// both sides of window boundaries, out-of-order LastSeen values, uneven
|
||||
// fractional seconds (so RFC3339Nano text order and time order can differ),
|
||||
// same-instant events and a history clear. It returns every event time.
|
||||
func writeRandomHistory(t *testing.T, m *Manager, seed int64) []time.Time {
|
||||
t.Helper()
|
||||
rng := rand.New(rand.NewSource(seed))
|
||||
store := m.eventLogStore()
|
||||
base := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
|
||||
kinds := []string{
|
||||
eventlog.TypeFired, eventlog.TypeRefired, eventlog.TypeAcknowledged,
|
||||
eventlog.TypeUnacknowledged, eventlog.TypeSnoozed, eventlog.TypeEscalated, eventlog.TypeResolved,
|
||||
}
|
||||
fractions := []time.Duration{0, 500 * time.Millisecond, 50 * time.Millisecond, 123456789}
|
||||
var times []time.Time
|
||||
at := base
|
||||
for i := 0; i < 120; i++ {
|
||||
if rng.Intn(4) != 0 {
|
||||
at = at.Add(time.Duration(rng.Intn(3)) * time.Second)
|
||||
}
|
||||
occurred := at.Add(fractions[rng.Intn(len(fractions))])
|
||||
times = append(times, occurred)
|
||||
if i == 60 {
|
||||
if err := store.AppendDurable(eventlog.Event{Type: eventlog.TypeHistoryCleared, AlertID: "history", OccurredAt: occurred}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
id := fmt.Sprintf("alert-%d", rng.Intn(5))
|
||||
snapshot := Alert{
|
||||
ID: id,
|
||||
ResourceID: "vm-" + id,
|
||||
StartTime: base.Add(time.Duration(rng.Intn(3)) * time.Hour),
|
||||
LastSeen: occurred.Add(time.Duration(rng.Intn(240)-120) * time.Second),
|
||||
Value: float64(i),
|
||||
}
|
||||
kind := kinds[rng.Intn(len(kinds))]
|
||||
if kind == eventlog.TypeAcknowledged || (kind != eventlog.TypeUnacknowledged && rng.Intn(5) == 0) {
|
||||
ackTime := occurred
|
||||
snapshot.Acknowledged, snapshot.AckTime, snapshot.AckUser = true, &ackTime, fmt.Sprintf("user-%d", i)
|
||||
}
|
||||
payload, err := json.Marshal(snapshot)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.AppendDurable(eventlog.Event{Type: kind, AlertID: id, OccurredAt: occurred, Snapshot: payload}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
return times
|
||||
}
|
||||
|
||||
func TestWindowedHistoryMatchesAWalkOfOnlyTheWindow(t *testing.T) {
|
||||
for seed := int64(1); seed <= 25; seed++ {
|
||||
t.Run(fmt.Sprintf("seed-%d", seed), func(t *testing.T) {
|
||||
m := newHistoryParityManager(t)
|
||||
times := writeRandomHistory(t, m, seed)
|
||||
// Live overlay: one alert the log knows (with a fresh value) and
|
||||
// one it never saw fire.
|
||||
m.mu.Lock()
|
||||
m.activeAlerts["live-known"] = &Alert{ID: "alert-1", ResourceID: "vm-alert-1",
|
||||
StartTime: time.Date(2026, 9, 1, 1, 0, 0, 0, time.UTC), LastSeen: times[len(times)-1], Value: 999}
|
||||
m.activeAlerts["live-new"] = &Alert{ID: "alert-new", ResourceID: "vm-new",
|
||||
StartTime: times[len(times)/2], LastSeen: times[len(times)-1], Value: 7}
|
||||
m.mu.Unlock()
|
||||
|
||||
windows := []time.Time{{}, times[0].Add(-time.Hour), times[len(times)-1].Add(time.Hour)}
|
||||
for i := 0; i < len(times); i += 7 {
|
||||
windows = append(windows, times[i], times[i].Add(time.Nanosecond), times[i].Add(-time.Nanosecond))
|
||||
}
|
||||
for _, since := range windows {
|
||||
for _, limit := range []int{0, 1, 3} {
|
||||
got, ok := m.AlertHistoryFromEvents(since, limit)
|
||||
if !ok {
|
||||
t.Fatal("history unavailable")
|
||||
}
|
||||
want := referenceWindowedHistory(t, m, since, limit)
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("since=%s limit=%d: shared fold differs from a walk of only the window\n got: %+v\nwant: %+v",
|
||||
since.Format(time.RFC3339Nano), limit, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestWindowedHistoryReusesTheFullHistoryFold(t *testing.T) {
|
||||
m := newHistoryParityManager(t)
|
||||
times := writeRandomHistory(t, m, 7)
|
||||
if _, ok := m.AlertHistoryFromEvents(time.Time{}, 0); !ok {
|
||||
t.Fatal("history unavailable")
|
||||
}
|
||||
m.historyProjectionMu.Lock()
|
||||
full := m.historyProjection
|
||||
m.historyProjectionMu.Unlock()
|
||||
if full == nil {
|
||||
t.Fatal("a full-history read did not keep its fold")
|
||||
}
|
||||
for _, since := range []time.Time{times[10], times[90], times[len(times)-1]} {
|
||||
if _, ok := m.AlertHistoryFromEvents(since, 0); !ok {
|
||||
t.Fatal("history unavailable")
|
||||
}
|
||||
m.historyProjectionMu.Lock()
|
||||
current := m.historyProjection
|
||||
m.historyProjectionMu.Unlock()
|
||||
if current != full {
|
||||
t.Fatalf("a Since read at %s replaced the shared fold instead of reusing it", since)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHistoryReadResultsDoNotAliasTheSharedFold(t *testing.T) {
|
||||
m := newHistoryParityManager(t)
|
||||
times := writeRandomHistory(t, m, 11)
|
||||
m.mu.Lock()
|
||||
m.activeAlerts["live-known"] = &Alert{ID: "alert-2", ResourceID: "vm-alert-2",
|
||||
StartTime: time.Date(2026, 9, 1, 2, 0, 0, 0, time.UTC), LastSeen: times[len(times)-1]}
|
||||
m.mu.Unlock()
|
||||
for _, since := range []time.Time{{}, times[30], times[100]} {
|
||||
first, _ := m.AlertHistoryFromEvents(since, 0)
|
||||
before := referenceWindowedHistory(t, m, since, 0)
|
||||
for i := range first {
|
||||
first[i].Value = -1
|
||||
first[i].LastSeen = time.Time{}
|
||||
if first[i].AckTime != nil {
|
||||
*first[i].AckTime = time.Time{}
|
||||
}
|
||||
}
|
||||
if again, _ := m.AlertHistoryFromEvents(since, 0); !reflect.DeepEqual(again, before) {
|
||||
t.Fatalf("since=%s: mutating a result changed the next read", since)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A limited read copies only the rows it returns, not the whole fold.
|
||||
func TestLimitedHistoryReadCopiesOnlyReturnedRows(t *testing.T) {
|
||||
m := newHistoryParityManager(t)
|
||||
store := m.eventLogStore()
|
||||
at := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
|
||||
metadata := map[string]interface{}{}
|
||||
for i := 0; i < 40; i++ {
|
||||
metadata[fmt.Sprintf("key-%d", i)] = fmt.Sprintf("value-%d-with-some-detail", i)
|
||||
}
|
||||
for i := 0; i < 2000; i++ {
|
||||
occurred := at.Add(time.Duration(i) * time.Minute)
|
||||
snapshot, err := json.Marshal(Alert{ID: fmt.Sprintf("alert-%d", i), ResourceID: "vm", StartTime: occurred,
|
||||
LastSeen: occurred, Message: "disk usage above threshold on a long resource name", Metadata: metadata})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.AppendDurable(eventlog.Event{Type: eventlog.TypeFired, AlertID: fmt.Sprintf("alert-%d", i),
|
||||
OccurredAt: occurred, Snapshot: snapshot}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
allocated := func(limit int) uint64 {
|
||||
var before, after runtime.MemStats
|
||||
runtime.GC()
|
||||
runtime.ReadMemStats(&before)
|
||||
if got, ok := m.AlertHistoryFromEvents(time.Time{}, limit); !ok || (limit > 0 && len(got) != limit) {
|
||||
t.Fatalf("history read with limit %d failed", limit)
|
||||
}
|
||||
runtime.ReadMemStats(&after)
|
||||
return after.TotalAlloc - before.TotalAlloc
|
||||
}
|
||||
allocated(0) // build the shared fold
|
||||
everything, ten := allocated(0), allocated(10)
|
||||
if ten*5 > everything {
|
||||
t.Fatalf("a 10-row read allocated %d bytes against %d for all 2000 rows; it must not copy the whole fold", ten, everything)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user