Incident Investigation, Root Cause Analysis, and Postmortems Questions

package main

import (
	"fmt"
	"sync"
	"time"
)

// Event represents an incoming alert
type Event struct {
	Timestamp time.Time
	Signature string
	Severity  string
}

// Deduplicator suppresses duplicates within window W and emits summaries every X.
type Deduplicator struct {
	inCh      chan Event
	summaryCh chan map[string]int

	// configuration
	W        time.Duration
	X        time.Duration
	granular time.Duration // bucket size

	// internal
	buckets []map[string]struct{} // ring of seen signatures per bucket
	counts  map[string]int        // counts since last summary
	pos     int
	start   time.Time

	mu sync.Mutex // protects Close
	quit chan struct{}
}

func NewDeduplicator(W, X, granularity time.Duration) *Deduplicator {
	buckets := make([]map[string]struct{}, int(W/granularity))
	for i := range buckets {
		buckets[i] = make(map[string]struct{})
	}
	d := &Deduplicator{
		inCh:     make(chan Event, 1000),
		summaryCh: make(chan map[string]int),
		W:        W,
		X:        X,
		granular: granularity,
		buckets:  buckets,
		counts:   make(map[string]int),
		pos:      0,
		start:    time.Now(),
		quit:     make(chan struct{}),
	}
	go d.run()
	return d
}

func (d *Deduplicator) Ingest(e Event) {
	select {
	case d.inCh <- e:
	default:
		// drop or backpressure; here we drop to keep system stable
	}
}

func (d *Deduplicator) SummaryChannel() <-chan map[string]int { return d.summaryCh }

func (d *Deduplicator) rotateIfNeeded(now time.Time) {
	// compute how many buckets to advance
	elapsed := now.Sub(d.start)
	steps := int(elapsed / d.granular)
	if steps <= 0 {
		return
	}
	for i := 0; i < steps && i < len(d.buckets); i++ {
		// clear the next bucket and advance pos
		next := (d.pos + 1) % len(d.buckets)
		for k := range d.buckets[next] {
			delete(d.buckets[next], k)
		}
		d.pos = next
	}
	d.start = d.start.Add(time.Duration(steps) * d.granular)
}

func (d *Deduplicator) run() {
	summaryTicker := time.NewTicker(d.X)
	bucketTicker := time.NewTicker(d.granular)
	defer summaryTicker.Stop()
	defer bucketTicker.Stop()

	for {
		select {
		case <-d.quit:
			return
		case now := <-bucketTicker.C:
			// rotate one bucket each granularity tick
			d.rotateIfNeeded(now)
		case e := <-d.inCh:
			now := time.Now()
			d.rotateIfNeeded(now)
			// check dedupe: if signature seen in any bucket? we maintain presence across buckets via union check:
			seen := false
			// fast path: check against combined presence by scanning buckets (bounded by W/G)
			for i := 0; i < len(d.buckets); i++ {
				if _, ok := d.buckets[i][e.Signature]; ok {
					seen = true
					break
				}
			}
			if !seen {
				// record in current bucket
				d.buckets[d.pos][e.Signature] = struct{}{}
				// increment count for summary
				d.counts[e.Signature]++
			}
		case <-summaryTicker.C:
			// emit a copy of counts, then reset
			out := make(map[string]int, len(d.counts))
			for k, v := range d.counts {
				out[k] = v
			}
			select {
			case d.summaryCh <- out:
			default:
				// if nobody reading, drop or block; here we drop to avoid blocking
			}
			d.counts = make(map[string]int)
		}
	}
}

func (d *Deduplicator) Close() {
	d.mu.Lock()
	defer d.mu.Unlock()
	select {
	case <-d.quit:
		return
	default:
		close(d.quit)
	}
}

// Example usage
func main() {
	W := 5 * time.Minute
	X := 1 * time.Minute
	G := 10 * time.Second
	d := NewDeduplicator(W, X, G)
	defer d.Close()

	// reader of summaries
	go func() {
		for s := range d.SummaryChannel() {
			fmt.Println("Summary:", s)
		}
	}()

	now := time.Now()
	// simulate events
	d.Ingest(Event{Timestamp: now, Signature: "A", Severity: "critical"})
	d.Ingest(Event{Timestamp: now.Add(5 * time.Second), Signature: "A", Severity: "critical"}) // suppressed
	d.Ingest(Event{Timestamp: now.Add(70 * time.Second), Signature: "A", Severity: "critical"}) // depending on W and timing may be suppressed or allowed

	time.Sleep(2 * time.Minute)
}