saver

Documentation

Overview

Package saver contains all logic for writing records to files.

1. Sets up a channel that accepts slices of *netlink.ArchivalRecord
2. Maintains a map of Connections, one for each connection.
3. Uses several marshallers goroutines to serialize data and and write to zstd files.
4. Rotates Connection output files every 10 minutes for long lasting connections.
5. uses a cache to detect meaningful state changes, and avoid excessive writes.

Index

• Variables
• type CacheLogger
• type Connection
  • func (conn *Connection) Rotate(Host string, Pod string, FileAgeLimit time.Duration) error
• type MarshalChan
• type Saver
  • func NewSaver(host string, pod string, numMarshaller int, srv eventsocket.Server, ...) *Saver
  • func (svr *Saver) Close()
  • func (svr *Saver) LogCacheStats(localCount, errCount int)
  • func (svr *Saver) MessageSaverLoop(readerChannel <-chan netlink.MessageBlock)
• type Task
• type TcpStats

Variables

var (
	ErrNoMarshallers = errors.New("Saver has zero Marshallers")
)

Errors generated by saver functions.

Types

type CacheLogger

type CacheLogger interface {
	LogCacheStats(localCount, errCount int)
}

CacheLogger is any object with a LogCacheStats method.

type Connection

type Connection struct {
	Inode      uint32 // TODO - also use the UID???
	ID         inetdiag.SockID
	UID        uint32
	Slice      string    // 4 hex, indicating which machine segment this is on.
	StartTime  time.Time // Time the connection was initiated.
	Sequence   int       // Typically zero, but increments for long running connections.
	Expiration time.Time // Time we will swap files and increment Sequence.
	Writer     io.WriteCloser
}

Connection objects handle all output associated with a single connection.

func (*Connection) Rotate

func (conn *Connection) Rotate(Host string, Pod string, FileAgeLimit time.Duration) error

Rotate opens the next writer for a connection. Note that long running connections will have data in multiple directories, because, for all segments after the first one, we choose the directory based on the time Rotate() was called, and not on the StartTime of the connection. Long-running connections with data on multiple days will therefore likely have data in multiple date directories. (This behavior is new as of April 2020. Prior to then, all files were placed in the directory corresponding to the StartTime.)

type MarshalChan

type MarshalChan chan<- Task

MarshalChan is a channel of marshalling tasks.

type Saver

type Saver struct {
	Host          string // mlabN
	Pod           string // 3 alpha + 2 decimal
	FileAgeLimit  time.Duration
	MarshalChans  []MarshalChan
	Done          *sync.WaitGroup // All marshallers will call Done on this.
	Connections   map[uint64]*Connection
	ClosingStats  map[uint64]TcpStats // BytesReceived and BytesSent for connections that are closing.
	ClosingTotals TcpStats
	// contains filtered or unexported fields
}

Saver provides functionality for saving tcpinfo diffs to connection files. It handles arbitrary connections, and only writes to file when the significant fields change. (TODO - what does "significant fields" mean). TODO - just export an interface, instead of the implementation.

func NewSaver

func NewSaver(host string, pod string, numMarshaller int, srv eventsocket.Server, anon anonymize.IPAnonymizer, ex *netlink.ExcludeConfig) *Saver

NewSaver creates a new Saver for the given host and pod. numMarshaller controls how many marshalling goroutines are used to distribute the marshalling workload.

func (*Saver) Close

func (svr *Saver) Close()

Close shuts down all the marshallers, and waits for all files to be closed.

func (*Saver) LogCacheStats

func (svr *Saver) LogCacheStats(localCount, errCount int)

LogCacheStats prints out some basic cache stats. TODO(https://github.com/m-lab/tcp-info/issues/32) - should also export all of these as Prometheus metrics.

func (*Saver) MessageSaverLoop

func (svr *Saver) MessageSaverLoop(readerChannel <-chan netlink.MessageBlock)

MessageSaverLoop runs a loop to receive batches of ArchivalRecords. Local connections

type Task

type Task struct {
	// nil message means close the writer.
	Message *netlink.ArchivalRecord
	Writer  io.WriteCloser
}

Task represents a single marshalling task, specifying the message and the writer.

type TcpStats

type TcpStats struct {
	Sent     uint64 // BytesSent
	Received uint64 // BytesReceived
}

TcpStats is used to save the connection stats as connection is closing.