observability

Documentation

Overview

Package observability implements persistent serial JSONL, pcapng output, and stable directional link-counter representations.

Index

• func Capture(ctx context.Context, writer *PcapNGWriter, sources []PacketSource, ...) error
• func CreateCaptureFile(workingDir, topology, link, output string, now time.Time) (*os.File, string, error)
• func InsertVLANTag(frame []byte, tag VLANTag) []byte
• func U64(v uint64) *uint64
• type CaptureOptions
• type CounterCollector
  • func (c CounterCollector) Collect(ctx context.Context, topology string, links []LinkRef) (CounterReport, error)
• type CounterReport
  • func Delta(before, after CounterReport) (CounterReport, error)
• type CounterSource
• type CounterValue
• type DeviceCounters
• type Direction
• type EndpointCounters
• type EndpointRef
• type ImpairmentCounters
• type InterfaceDescription
• type LinkCounters
• type LinkRef
• type ListenFunc
• type Packet
• type PacketSource
• type PcapNGWriter
  • func NewPcapNGWriter(w io.Writer, snaplen uint32, interfaces []InterfaceDescription) (*PcapNGWriter, error)
  • func (p *PcapNGWriter) Close() error
  • func (p *PcapNGWriter) WritePacket(packet Packet) error
• type SerialCollector
  • func (c *SerialCollector) Serve(ctx context.Context) error
• type SerialRecord
  • func MergeSerial(perNode map[string][]SerialRecord) []SerialRecord
  • func ReadSerial(path, node string, tail int) ([]SerialRecord, error)
• type TrafficCounters
• type VLANTag

Functions

func Capture

func Capture(ctx context.Context, writer *PcapNGWriter, sources []PacketSource, options CaptureOptions) error

Capture multiplexes one or two directional sources into one pcapng writer. Cancellation (including SIGINT propagated by the CLI) closes a valid file and is considered a successful stop.

func CreateCaptureFile

func CreateCaptureFile(workingDir, topology, link, output string, now time.Time) (*os.File, string, error)

CreateCaptureFile resolves a requested output below workingDir and opens it exclusively. Empty output receives a collision-resistant default name.

func InsertVLANTag

func InsertVLANTag(frame []byte, tag VLANTag) []byte

InsertVLANTag reconstructs a tag stripped by host receive offload. It does not add a duplicate tag if the frame already carries the supplied TPID.

func U64

func U64(v uint64) *uint64

Types

type CaptureOptions

type CaptureOptions struct {
	Duration       time.Duration
	Count          uint64
	OwnershipCheck func(context.Context) error
	CheckInterval  time.Duration
}

type CounterCollector

type CounterCollector struct {
	Source CounterSource
	Now    func() time.Time
}

func (CounterCollector) Collect

func (c CounterCollector) Collect(ctx context.Context, topology string, links []LinkRef) (CounterReport, error)

type CounterReport

type CounterReport struct {
	Topology    string         `json:"topology"`
	CollectedAt time.Time      `json:"collected_at"`
	Links       []LinkCounters `json:"links"`
}

func Delta

func Delta(before, after CounterReport) (CounterReport, error)

type CounterSource

type CounterSource interface {
	ReadEndpoint(context.Context, EndpointRef) (EndpointCounters, error)
}

type CounterValue

type CounterValue = *uint64

type DeviceCounters

type DeviceCounters struct {
	RXErrors CounterValue `json:"rx_errors"`
	RXDrops  CounterValue `json:"rx_drops"`
	TXErrors CounterValue `json:"tx_errors"`
	TXDrops  CounterValue `json:"tx_drops"`
	MTUDrops CounterValue `json:"mtu_drops"`
}

type Direction

type Direction string

const (
	Ingress Direction = "ingress"
	Egress  Direction = "egress"
)

type EndpointCounters

type EndpointCounters struct {
	Node       string             `json:"node"`
	Interface  string             `json:"interface"`
	Ingress    TrafficCounters    `json:"ingress"`
	Egress     TrafficCounters    `json:"egress"`
	Impairment ImpairmentCounters `json:"impairment"`
	Device     DeviceCounters     `json:"device"`
}

type EndpointRef

type EndpointRef struct{ Node, Interface, IngressDevice, EgressDevice, ImpairmentDevice string }

type ImpairmentCounters

type ImpairmentCounters struct {
	OfferedPackets            CounterValue `json:"offered_packets"`
	OfferedBytes              CounterValue `json:"offered_bytes"`
	DeliveredPackets          CounterValue `json:"delivered_packets"`
	DeliveredBytes            CounterValue `json:"delivered_bytes"`
	ConfiguredLossDrops       CounterValue `json:"configured_loss_drops"`
	NetemQueueDrops           CounterValue `json:"netem_queue_drops"`
	RateLimiterDrops          CounterValue `json:"rate_limiter_drops"`
	RateLimiterOverlimits     CounterValue `json:"rate_limiter_overlimits"`
	RateLimiterRequeues       CounterValue `json:"rate_limiter_requeues"`
	RateLimiterBacklogPackets CounterValue `json:"rate_limiter_backlog_packets"`
	RateLimiterBacklogBytes   CounterValue `json:"rate_limiter_backlog_bytes"`
}

type InterfaceDescription

type InterfaceDescription struct {
	Name, Comment string
	Direction     Direction
}

type LinkCounters

type LinkCounters struct {
	Name                string             `json:"name"`
	GenerationID        string             `json:"generation_id"`
	GenerationStartedAt time.Time          `json:"generation_started_at"`
	Endpoints           []EndpointCounters `json:"endpoints"`
}

type LinkRef

type LinkRef struct {
	Name, GenerationID  string
	GenerationStartedAt time.Time
	Endpoints           []EndpointRef
}

type ListenFunc

type ListenFunc func(network, address string) (net.Listener, error)

type Packet

type Packet struct {
	Interface      int
	Direction      Direction
	Timestamp      time.Time
	Data           []byte
	OriginalLength uint32
	VLAN           *VLANTag
}

type PacketSource

type PacketSource interface {
	ReadPacket(context.Context) (Packet, error)
	Close() error
}

type PcapNGWriter

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

func NewPcapNGWriter

func NewPcapNGWriter(w io.Writer, snaplen uint32, interfaces []InterfaceDescription) (*PcapNGWriter, error)

func (*PcapNGWriter) Close

func (p *PcapNGWriter) Close() error

func (*PcapNGWriter) WritePacket

func (p *PcapNGWriter) WritePacket(packet Packet) error

type SerialCollector

type SerialCollector struct {
	Node, SocketPath, LogPath string
	Listen                    ListenFunc
	Now                       func() time.Time
	MaxLineBytes              int
}

func (*SerialCollector) Serve

func (c *SerialCollector) Serve(ctx context.Context) error

type SerialRecord

type SerialRecord struct {
	Timestamp time.Time `json:"timestamp"`
	Line      string    `json:"line"`
	Node      string    `json:"-"`
}

func MergeSerial

func MergeSerial(perNode map[string][]SerialRecord) []SerialRecord

func ReadSerial

func ReadSerial(path, node string, tail int) ([]SerialRecord, error)

type TrafficCounters

type TrafficCounters struct {
	Packets CounterValue `json:"packets"`
	Bytes   CounterValue `json:"bytes"`
}

type VLANTag

type VLANTag struct{ TPID, TCI uint16 }