observation

Documentation

Overview

Package observation provides an in-process, multi-subscriber feed for traffic observations.

Index

• func BindMetrics(reg metrics.Registry)
• type Feature
• type Feed
  • func Global() *Feed
  • func NewFeed() *Feed
  • func (f *Feed) Close()
  • func (f *Feed) Dropped() int64
  • func (f *Feed) Publish(obs Observation)
  • func (f *Feed) Subscribe(name string, fn func(Observation)) int
  • func (f *Feed) Unsubscribe(id int)
• type FlowKey
• type Kind
• type Observation

Functions

func BindMetrics

func BindMetrics(reg metrics.Registry)

BindMetrics registers observation feed counters with the given registry.

Types

type Feature

type Feature uint8

const (
	FeatureRxBytes Feature = iota + 1
	FeatureRxPackets
	FeatureFlowBytes
	FeatureFlowPackets
	FeatureNewFlowCount
)

type Feed

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

Feed is an in-process, typed, multi-subscriber observation bus. Publishers call Publish (non-blocking); subscribers receive on their own goroutine via a buffered channel.

func Global

func Global() *Feed

Global returns the process-wide observation feed.

func NewFeed

func NewFeed() *Feed

NewFeed creates a ready-to-use Feed.

func (*Feed) Close

func (f *Feed) Close()

Close unsubscribes all subscribers and waits for their goroutines to finish.

func (*Feed) Dropped

func (f *Feed) Dropped() int64

Dropped returns the total number of observations dropped across all subscribers due to full buffers.

func (*Feed) Publish

func (f *Feed) Publish(obs Observation)

Publish fans out an observation to all subscribers. Non-blocking: if a subscriber's buffer is full the observation is dropped and the drop counter increments. The publisher's goroutine is never stalled.

func (*Feed) Subscribe

func (f *Feed) Subscribe(name string, fn func(Observation)) int

Subscribe registers a handler that will be called for every published observation on its own goroutine. Returns an ID for Unsubscribe.

func (*Feed) Unsubscribe

func (f *Feed) Unsubscribe(id int)

Unsubscribe removes the subscriber with the given ID and waits for its goroutine to exit. The data channel is left open so a concurrent Publish holding a stale snapshot does not panic on a closed channel.

type FlowKey

type FlowKey struct {
	Src     netip.Addr
	Dst     netip.Addr
	SrcPort uint16
	DstPort uint16
	Proto   uint8
}

type Kind

type Kind uint8

const (
	KindInterface Kind = iota + 1
	KindSourceIP
	KindDestIP
	KindFlow
)

type Observation

type Observation struct {
	Kind    Kind
	Iface   string
	Flow    FlowKey
	Feature Feature
	Value   float64
	At      time.Time
}