Documentation
¶
Overview ¶
Package ingest receives pre-aggregated flow windows from agents, resolves each record's peer at ingest (address → workload identity via the registry), and writes the windows and their evidence gaps through the FlowStore interface (ADR-0009, ADR-0019). It does not deduplicate: v1 observes inbound connections only (ADR-0010), so each connection is reported once, by the workload it reached, and stored as reported. Each report's clock reading, the agent's clock at send against the instant of receipt, is recorded on the workload, and a skewed one is logged; a skewed window is stored as reported all the same (ADR-0015 as amended).
Package ingest receives pre-aggregated flow windows from agents, validates them, resolves each record's source address to the peer behind it at that moment (a workload with its labels, an address group, or nothing), and writes the result through the FlowStore interface (ADR-0009, ADR-0019).
Index ¶
Constants ¶
const ( // MeasuredBySentAt is a reading from the agent's clock at send. MeasuredBySentAt = "sent_at" // MeasuredByWindowEnd is a reading from the window's end, for an // agent that predates sent_at. MeasuredByWindowEnd = "window_end" )
Measures name how a clock reading was taken, for the log.
const ClockSkewWindows = 2
ClockSkewWindows is the clock-skew tolerance in flow aggregation windows: a report whose agent clock differs from the control plane's by more than this many windows is skewed. Two windows is generous for transport (a report is stamped immediately before it is sent and its stream is bounded at thirty seconds) and tight enough for real drift, which runs to minutes or hours: a VM's clock after its host slept, or a host whose time sync has failed. A skewed agent stamps every window it reports with the wrong clock, so its flows land at the wrong times. It is a constant, not configuration: the window length is the knob.
const MaxGapsPerRequest = 4096
MaxGapsPerRequest bounds the evidence gaps one request may carry; past it the rest are rejected and counted. An agent holds at most a thousand or so and ships them in one request.
const SilenceWindows = 2
SilenceWindows is how many flow windows may pass between consecutive reports of a workload before the silence between them is an evidence gap (ADR-0019 as amended). The reporter ships one window per window length, so two allow one late report — the same allowance after which the read model stops calling the workload's telemetry live (readmodel.TelemetryLiveWindows; a test keeps the two equal). Below it, an agent restarted quickly enough to miss no whole window records no gap.
Variables ¶
var ErrInvalidWindow = errors.New("ingest: window bounds are missing or inverted")
ErrInvalidWindow is returned for a report whose window bounds are missing or inverted. Individual records that fail validation are skipped, not fatal: telemetry is loss-tolerant and one malformed record must not cost the rest of the window.
Functions ¶
func SilenceAtRetirement ¶ added in v0.5.0
func SilenceAtRetirement(prev *time.Time, proc *registry.AgentProcess, at time.Time, window time.Duration) (flowstore.Gap, bool)
SilenceAtRetirement is the agent-silence gap a retirement at instant at closes, if one is open (ADR-0023; issue #126): the gap deriveSilence would have written had a report arrived at that instant, by the same rules. A silence since the last receipt prev longer than SilenceWindows flow windows, which no heartbeat shows one agent process running through, is one gap from prev to at. A workload that never reported, or whose agent was demonstrably running and quiet, has none. No report can follow a retirement, so there is no arriving window to refine the bound by and the gap is never derived twice.
Types ¶
type ClockReading ¶ added in v0.2.0
type ClockReading struct {
// Offset is the agent's clock minus the control plane's: negative
// when the agent is behind.
Offset time.Duration
// MeasuredBy is MeasuredBySentAt or MeasuredByWindowEnd.
MeasuredBy string
// Skewed says the offset is beyond the tolerance.
Skewed bool
// ReceivedAt is when the control plane received the report.
ReceivedAt time.Time
}
ClockReading is what one flow report says about its agent's clock.
func ReadClock ¶ added in v0.2.0
func ReadClock(req *innerwallv1.ReportFlowsRequest, receivedAt time.Time, window time.Duration) ClockReading
ReadClock compares one report with the instant it was received, under a tolerance of ClockSkewWindows flow windows.
A report that carries sent_at is read from it: the agent's clock at send against the control plane's at receipt, skewed in either direction. A report without it, from an agent that predates the field, is read from its window's end, and is skewed only when that end is ahead of receipt by more than the tolerance. A window can arrive any time after it ends, because the agent holds closed windows in its buffer while the control plane is unreachable and delivers them late by design (ADR-0019), so a late window says nothing about the clock; a window cannot legitimately arrive before it ends, so an early one does.
type ClockRecorder ¶ added in v0.2.0
type ClockRecorder interface {
RecordClockReading(ctx context.Context, id identity.WorkloadID, offset time.Duration, skewed bool, receivedAt time.Time) error
}
ClockRecorder records what a flow report said about its agent's clock on the workload: a skewed reading's offset and instant, or that the latest reading was within tolerance. The store implements it.
type Directory ¶
type Directory interface {
ListWorkloads(ctx context.Context) ([]registry.Workload, error)
ListAddressGroups(ctx context.Context) ([]policy.AddressGroup, error)
}
Directory is the registry view resolution needs: every workload with its current addresses and labels, and every address group. The store implements both.
type Index ¶
type Index struct {
// contains filtered or unexported fields
}
Index resolves source addresses to peers as the registry stands at one instant. It is built once per reported window and discarded, so every record in a window is resolved against the same view and no replica keeps state between windows (ADR-0017).
func BuildIndex ¶
func BuildIndex(workloads []registry.Workload, groups []policy.AddressGroup) *Index
BuildIndex indexes the active workloads' current addresses and the address groups' CIDRs. A retired workload's last addresses are no longer its own and may already belong to whatever replaced it, so nothing resolves to it (ADR-0023); such an address falls through to the groups.
func (*Index) HasWorkload ¶
HasWorkload reports whether the index knows a workload by id.
func (*Index) Resolve ¶
Resolve returns the peer behind addr. An active workload's current address wins over any address group containing it; the most specific group wins over a wider one; an address matching neither is recorded as itself. The workload's labels are copied so the snapshot is independent of later registry changes.
type ReportRecorder ¶ added in v0.2.0
type ReportRecorder interface {
RecordFlowReport(ctx context.Context, id identity.WorkloadID, receivedAt time.Time) (prev *time.Time, proc *registry.AgentProcess, advanced bool, err error)
}
ReportRecorder records the instant the control plane received a flow report from a workload. A report proves the agent is running whatever the state of its sync stream, which travels on a connection of its own. It returns the receipt this one replaced (nil for the first ever), the agent process the workload's latest heartbeat proved running (nil when none is recorded), and whether it advanced at all: a stale receipt, handled out of order, replaces nothing. The implementation must hand the replaced instant to exactly one of two concurrent reports, so a silence is derived once, and read the process under the same lock; the store implements it with one locking statement.
type Result ¶
type Result struct {
// Accepted is how many records were stored.
Accepted int
// Rejected is how many records failed validation and were skipped.
Rejected int
// Gaps is how many evidence gaps were accepted for storage (a gap
// already stored is accepted and stored once), and GapsRejected how
// many failed validation and were skipped.
Gaps int
GapsRejected int
// Derived is how many gaps this report made the control plane derive
// and store itself: one when it ended a silence longer than
// SilenceWindows flow windows, zero otherwise (ADR-0019 as amended).
Derived int
}
Result reports one ingested window.
type Service ¶
type Service struct {
Directory Directory
Flows flowstore.FlowStore
// Clock records each report's clock reading on its workload; nil
// records nothing, and a skewed report is still logged.
Clock ClockRecorder
// Reports records the receipt of every report on its workload; nil
// records nothing.
Reports ReportRecorder
// FlowWindow is the flow aggregation window the control plane hands
// agents in SyncConfig; the clock-skew tolerance is ClockSkewWindows
// of it. Zero skips the clock check.
FlowWindow time.Duration
Log *slog.Logger
// Now is the clock receipt is read from; time.Now if nil.
Now func() time.Time
}
Service ingests reported windows.
func (*Service) Ingest ¶
func (s *Service) Ingest(ctx context.Context, id identity.WorkloadID, req *innerwallv1.ReportFlowsRequest) (Result, error)
Ingest validates and stores one window reported by workload id. The identity comes from the caller, which took it from the connection credential; nothing in the request names the reporter. Resolution reads the registry once per window, so every record sees the same view. The report's clock reading is recorded on the workload and a skewed one is logged, but skew rejects nothing: the window is stored as reported.
The clock is read before the window's bounds are validated. The send instant is independent of the bounds, and a clock stepped while a window was open is what inverts them, so a report refused for its bounds still records the skew that most likely produced it.
Source Files