pipeline

Documentation

Overview

Package pipeline turns raw capture observations into the flow events the rest of the application consumes, so that nothing downstream needs to know whether an observation came from a socket table or a packet capture.

Index

• Constants
• type Bus
  • func NewBus(depth int) *Bus
  • func (b *Bus) Publish(m Message)
  • func (b *Bus) Subscribe() (<-chan Message, func())
  • func (b *Bus) Subscribers() int
• type DomainLabeller
• type Engine
  • func NewEngine(st *store.Store, bus *Bus, sources []capture.Source) *Engine
  • func (e *Engine) ActiveFlows() []types.Flow
  • func (e *Engine) Health() Health
  • func (e *Engine) Run(ctx context.Context) error
• type FlowEvent
• type Health
  • func (h Health) Healthy() bool
• type Message
• type Normalizer
  • func NewNormalizer() *Normalizer
  • func (n *Normalizer) Active() []types.Flow
  • func (n *Normalizer) Apply(ev types.RawEvent) []FlowEvent
  • func (n *Normalizer) Len() int
• type Phase

Constants

const DefaultGraceRounds = 2

DefaultGraceRounds is how many consecutive samples a flow must be absent from before it is called closed. Polling sources race with short-lived sockets, and UDP "connections" in particular flicker in and out of the table; requiring two misses stops one unlucky sample from closing and reopening a live flow.

const FlushInterval = 2 * time.Second

FlushInterval is how often accumulated flow changes are written to the store. Writing every observation individually would be pointless churn: a flow that is merely still open produces one row update either way.

Types

type Bus

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

Bus fans live updates out to every connected client.

It is deliberately lossy. A dashboard that cannot keep up with a busy network gets the newest events and drops the backlog, because a live view showing stale traffic is worse than one showing a gap, and because a slow websocket must never be able to stall ingest.

func NewBus

func NewBus(depth int) *Bus

NewBus returns a bus with the given per-subscriber queue depth.

func (*Bus) Publish

func (b *Bus) Publish(m Message)

Publish delivers a message to every subscriber, dropping the oldest queued message for any subscriber that has fallen behind.

func (*Bus) Subscribe

func (b *Bus) Subscribe() (<-chan Message, func())

Subscribe registers a listener and returns it with its cancel function.

func (*Bus) Subscribers

func (b *Bus) Subscribers() int

Subscribers reports the current listener count.

type DomainLabeller

type DomainLabeller interface {
	Category(domain string) (string, bool)
}

DomainLabeller categorizes a domain, e.g. as a tracker or telemetry endpoint. Satisfied by enrich.Labeller; an interface so the pipeline does not depend on the enrichment package.

type Engine

type Engine struct {
	Store   *store.Store
	Bus     *Bus
	Sources []capture.Source
	// Labeller is optional. Without one, lookups are recorded unlabelled.
	Labeller DomainLabeller

	// OnSighting receives device identity learned by a capture source, such as
	// the hostname and vendor in a DHCP request. Set by the caller so the
	// pipeline does not need to know about the discovery store.
	OnSighting func(types.Sighting)
	// contains filtered or unexported fields
}

Engine runs the capture sources and moves what they produce through the normalizer into the store and out to connected dashboards.

func NewEngine

func NewEngine(st *store.Store, bus *Bus, sources []capture.Source) *Engine

NewEngine builds the pipeline.

func (*Engine) ActiveFlows

func (e *Engine) ActiveFlows() []types.Flow

ActiveFlows returns the flows currently believed open.

func (*Engine) Health

func (e *Engine) Health() Health

Health reports whether observations are reaching storage.

func (*Engine) Run

func (e *Engine) Run(ctx context.Context) error

Run starts every available source and processes events until ctx is done.

type FlowEvent

type FlowEvent struct {
	Phase Phase
	Flow  types.Flow
}

FlowEvent is a change to one tracked flow.

type Health

type Health struct {
	// Writes and Failures count flush attempts since start.
	Writes   int64 `json:"writes"`
	Failures int64 `json:"failures"`
	// Consecutive is the current run of failures. Anything above zero means data
	// is being dropped right now.
	Consecutive int64 `json:"consecutive_failures"`
	// LastError is the most recent failure, in English. It names a programming
	// or environment fault rather than anything the user did, so it is shown
	// as-is rather than translated.
	LastError string     `json:"last_error,omitempty"`
	LastFail  *time.Time `json:"last_failure,omitempty"`
	LastWrite *time.Time `json:"last_write,omitempty"`
}

Health is a snapshot of whether observations are actually reaching storage.

func (Health) Healthy

func (h Health) Healthy() bool

Healthy reports whether observations are currently reaching storage.

type Message

type Message struct {
	Type string `json:"type"` // flow | dns | device | finding | status
	Data any    `json:"data"`
}

Message is one live update pushed to connected dashboards.

type Normalizer

type Normalizer struct {
	// GraceRounds is the close delay described above. Zero means the default.
	GraceRounds int
	// Now is the clock, overridable for tests.
	Now func() time.Time
	// contains filtered or unexported fields
}

Normalizer converts capture events into flow events. Polling sources deliver complete snapshots and the normalizer diffs consecutive ones; streaming sources deliver deltas, which are applied directly.

It is not safe for concurrent use: drive it from a single goroutine.

func NewNormalizer

func NewNormalizer() *Normalizer

NewNormalizer returns a ready normalizer.

func (*Normalizer) Active

func (n *Normalizer) Active() []types.Flow

Active returns every flow currently believed open.

func (*Normalizer) Apply

func (n *Normalizer) Apply(ev types.RawEvent) []FlowEvent

Apply feeds one raw event in and returns the flow events it produced.

func (*Normalizer) Len

func (n *Normalizer) Len() int

Len reports how many flows are being tracked.

type Phase

type Phase string

Phase is what happened to a flow.

const (
	PhaseOpen   Phase = "open"
	PhaseUpdate Phase = "update"
	PhaseClose  Phase = "close"
)