collect

type Aggregator struct {
	// Gaps receives the overflow gap when a window closes; nil records
	// nothing.
	Gaps *Gaps
	// contains filtered or unexported fields
}

type Buffer struct {
	// Gaps records every overflow as a buffer_overflow gap over the
	// dropped window's bounds, with the records dropped; nil records
	// nothing.
	Gaps *Gaps
	// contains filtered or unexported fields
}

type Collector struct {
	Source Source
	Buffer *Buffer
	// Gaps holds the evidence gaps waiting to be shipped: a window that
	// closes empty is still queued while any are, so they ride it.
	Gaps *Gaps
	Log  *slog.Logger
	// Now is the clock; time.Now if nil.
	Now func() time.Time
	// Monotonic reads a clock that never steps, as the time since any
	// fixed origin; the runtime's monotonic clock if nil. A window's
	// length is measured on it, so a wall clock stepped while a window was
	// open is seen when the window closes.
	Monotonic func() time.Duration
	// WindowKeys bounds the distinct keys one open window holds;
	// DefaultWindowKeys when zero.
	WindowKeys int
	// BackoffBase and BackoffCap bound the wait before a failed source is
	// restarted: one second and one minute when zero.
	BackoffBase time.Duration
	BackoffCap  time.Duration
	// contains filtered or unexported fields
}

type Dialer func(ctx context.Context) (innerwallv1.AgentServiceClient, io.Closer, error)

type Gap struct {
	Kind     GapKind
	Source   GapSource
	From, To time.Time
	Count    uint64
	HasCount bool
}

type GapKind string

type GapSource string

type Gapped interface {
	EvidenceGaps() (*Gaps, GapSource)
}

type Gaps struct {
	Log *slog.Logger
	// Capacity bounds the held gaps; DefaultGapCapacity when zero.
	Capacity int
	// contains filtered or unexported fields
}

type Key struct {
	Src      netip.Addr
	Dst      netip.Addr
	DstPort  uint16
	Protocol innerwallv1.Protocol
	Decision innerwallv1.PolicyDecision
	RuleID   string
}

type Observation struct {
	At          time.Time
	Src         netip.Addr
	Dst         netip.Addr
	DstPort     uint16
	Protocol    innerwallv1.Protocol
	Decision    innerwallv1.PolicyDecision
	RuleID      string
	Connections uint64
	Bytes       uint64
	ProcessName string
}

type Reporter struct {
	Buffer *Buffer
	// Gaps are taken and shipped with each window, and put back when its
	// stream fails; nil ships none.
	Gaps *Gaps
	Dial Dialer
	Log  *slog.Logger
	// BackoffBase and BackoffCap bound the wait after a failed report:
	// one second and five minutes when zero.
	BackoffBase time.Duration
	BackoffCap  time.Duration
	// SendTimeout bounds one window's stream; 30 seconds when zero.
	SendTimeout time.Duration
	// Now is the clock; time.Now if nil.
	Now func() time.Time
	// contains filtered or unexported fields
}

type Source interface {
	Run(ctx context.Context, emit func(Observation)) error
}

type Sources []Source

type Window struct {
	Start   time.Time
	End     time.Time
	Records []*innerwallv1.FlowRecord
}