aggregate

Documentation

Overview

Package aggregate provides per-IP sliding-window event aggregation. The Aggregator is safe for concurrent use. All methods honour context cancellation where a loop is involved.

Index

• Constants
• type Aggregator
  • func New(windows []time.Duration, maxSamples int) *Aggregator
  • func (a *Aggregator) Add(ev sdk.Event)
  • func (a *Aggregator) Aggregate(ip netip.Addr, window time.Duration, now time.Time) sdk.Aggregate
  • func (a *Aggregator) Flush(ctx context.Context, cutoff time.Time)
  • func (a *Aggregator) Len() int
  • func (a *Aggregator) Windows() []time.Duration
  • func (a *Aggregator) WithMaxIPs(maxIPs int) *Aggregator

Constants

const DefaultMaxSamples = 4096

DefaultMaxSamples is the default cap on events stored in sdk.Aggregate.Sample. The cap is intentionally large so that rule-engine field-level matching (e.g. counting 404s within the sample) remains accurate at typical thresholds. The AI layer must further reduce / redact the sample before sending it to a language model; never forward Sample directly to an LLM.

Types

type Aggregator

type Aggregator struct {
	// contains filtered or unexported fields
}

Aggregator maintains per-IP sliding-window event counts and capped samples. Events older than the longest configured window are evicted on every Add call. When maxIPs > 0 the bucket for the least-recently-seen IP is evicted once the count exceeds the cap, bounding memory growth regardless of attack breadth.

func New

func New(windows []time.Duration, maxSamples int) *Aggregator

New creates an Aggregator with the given sliding windows and sample cap. windows must be non-empty and contain only positive durations. maxSamples ≤ 0 falls back to DefaultMaxSamples.

func (*Aggregator) Add

func (a *Aggregator) Add(ev sdk.Event)

Add records ev in the per-IP bucket, then evicts events older than the longest configured window relative to ev.Time.

func (*Aggregator) Aggregate

func (a *Aggregator) Aggregate(ip netip.Addr, window time.Duration, now time.Time) sdk.Aggregate

Aggregate returns the event summary for ip over window as of now.

Kinds contains exact counts per event kind for all events in [now-window, now]. Sample holds up to maxSamples of those events in arrival order; it is used by the rule engine for field-level matching. The caller must cap and redact Sample before forwarding it to an AI provider.

func (*Aggregator) Flush

func (a *Aggregator) Flush(ctx context.Context, cutoff time.Time)

Flush evicts stale entries and removes IP buckets with no remaining events. cutoff should typically be time.Now().Add(-maxWindow). Call periodically to bound memory growth (e.g. once per maxWindow).

func (*Aggregator) Len

func (a *Aggregator) Len() int

Len returns the number of distinct IPs currently tracked.

func (*Aggregator) Windows

func (a *Aggregator) Windows() []time.Duration

Windows returns the configured sliding windows. Useful for callers that need to know which window durations to request.

func (*Aggregator) WithMaxIPs

func (a *Aggregator) WithMaxIPs(maxIPs int) *Aggregator

WithMaxIPs sets the maximum number of per-IP buckets retained in memory. When the limit is exceeded, the bucket least-recently-seen is evicted. A value ≤ 0 disables the cap (the default). WithMaxIPs returns the receiver for chaining with New().