buffer

Documentation

Overview

Package buffer holds the agent's report ring: a RAM buffer bounded by both a byte cap (15 MB) and a time horizon (60 min), with drop-oldest on overflow. It is at-least-once: a report stays until the server acknowledges the batch that carried it with a 202 (remove-on-202), so a failed send loses nothing. Replay is newest-first so live data reaches the charts immediately and a backlog heals backwards. On a clean exit the ring is spooled to the state directory and reloaded on start, so a restart costs no data point and steady-state disk IO is zero.

Index

• Constants
• type Emitter
• type Ring
  • func New(clock platform.Clock, emitter Emitter, maxBytes int, horizon time.Duration) *Ring
  • func (r *Ring) Add(report *wire.Report)
  • func (r *Ring) Bytes() int
  • func (r *Ring) DroppedTotal() uint64
  • func (r *Ring) FillPct() float64
  • func (r *Ring) Len() int
  • func (r *Ring) Remove(seqs []uint64)
  • func (r *Ring) Restore(reports []*wire.Report)
  • func (r *Ring) Snapshot() []*wire.Report
  • func (r *Ring) TakeBatch(maxReports, maxBytes int) (reports []*wire.Report, seqs []uint64)
• type Spool
  • func NewSpool(fs platform.FS, path string, maxBytes int) *Spool
  • func (s *Spool) Load() (reports []*wire.Report, events []*wire.Event, err error)
  • func (s *Spool) Save(reports []*wire.Report, events []*wire.Event) error

Constants

const (
	DefaultMaxBytes = 15 * 1024 * 1024
	DefaultHorizon  = 60 * time.Minute
)

Defaults for the ring bounds.

Types

type Emitter

type Emitter interface {
	Emit(t wire.EventType, meta map[string]string)
}

Emitter raises a buffer_overflow event when the ring drops reports. It matches internal/event.Emitter without importing it.

type Ring

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

Ring is the bounded report buffer. It is safe for concurrent use: the collect loop Adds while the sender TakeBatch/Removes.

func New

func New(clock platform.Clock, emitter Emitter, maxBytes int, horizon time.Duration) *Ring

New builds a ring. maxBytes <= 0 uses DefaultMaxBytes; horizon <= 0 uses DefaultHorizon. emitter may be nil (overflow is then counted but not evented).

func (*Ring) Add

func (r *Ring) Add(report *wire.Report)

Add appends a report as the newest entry, then evicts the oldest entries that fall outside the time horizon or push the ring over the byte cap. Evictions are counted into the cumulative dropped total and, if any occurred, raise one buffer_overflow event with the number dropped.

func (*Ring) Bytes

func (r *Ring) Bytes() int

Bytes returns the current buffered byte total.

func (*Ring) DroppedTotal

func (r *Ring) DroppedTotal() uint64

DroppedTotal returns the cumulative number of reports dropped since start. It feeds the sys.agent.dropped_reports metric; the UI computes deltas.

func (*Ring) FillPct

func (r *Ring) FillPct() float64

FillPct returns the ring fill as a percentage of the byte cap, for sys.agent.buffer_fill_pct.

func (*Ring) Len

func (r *Ring) Len() int

Len returns the number of buffered reports.

func (*Ring) Remove

func (r *Ring) Remove(seqs []uint64)

Remove drops the entries with the given seqs (after a 202 or a terminal 4xx discard). Unknown seqs are ignored.

func (*Ring) Restore

func (r *Ring) Restore(reports []*wire.Report)

Restore appends spooled reports (oldest-first) on start, enforcing the byte cap and horizon silently: a restore is not an overflow, so it raises no buffer_overflow event.

func (*Ring) Snapshot

func (r *Ring) Snapshot() []*wire.Report

Snapshot returns all buffered reports oldest-first, for the exit spool.

func (*Ring) TakeBatch

func (r *Ring) TakeBatch(maxReports, maxBytes int) (reports []*wire.Report, seqs []uint64)

TakeBatch returns up to maxReports of the NEWEST reports whose combined proto size does not exceed maxBytes, newest-first, without removing them. The returned seqs identify the batch for Remove after a 202. A batch always contains at least the single newest report even if it alone exceeds maxBytes.

type Spool

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

Spool persists the ring and pending events to a single file on the state directory at shutdown and reloads them on start. It reuses the MetricBatch wire message as the on-disk container (reports plus events), so the spool format is exactly what would have gone on the wire. The file is consumed on load (removed) so a crash after load does not replay stale data twice.

func NewSpool

func NewSpool(fs platform.FS, path string, maxBytes int) *Spool

NewSpool builds a spool at path. maxBytes <= 0 uses DefaultMaxBytes.

func (*Spool) Load

func (s *Spool) Load() (reports []*wire.Report, events []*wire.Event, err error)

Load reads and removes the spool file, returning the persisted reports and events. A missing file returns no data and no error.

func (*Spool) Save

func (s *Spool) Save(reports []*wire.Report, events []*wire.Event) error

Save marshals reports and events into one MetricBatch and writes it atomically at 0600. If the serialized batch would exceed the byte cap, the oldest reports are dropped until it fits (events are kept: they are the stronger class and tiny). An empty batch removes any stale spool file instead of writing an empty one.