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.
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 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.
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 ¶
This section is empty.
Types ¶
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 workloads' current addresses and the address groups' CIDRs.
func (*Index) HasWorkload ¶
HasWorkload reports whether the index knows a workload by id.
func (*Index) Resolve ¶
Resolve returns the peer behind addr. A 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 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
}
Result reports one ingested window.
type Service ¶
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.
Source Files