corpus

Documentation

Overview

Package corpus implements the streaming corpus ingest of requirement R2 (§5.1).

The reader emits generic field paths and leaves kind inference to the field registry of §5.1. No corpus's column layout appears in this package as code: the mapping from delimited columns to field paths is a Schema value, so onboarding a new source is a configuration change and nothing more, which is exactly what evaluation hypothesis E6 measures.

§5.3 distinguishes a field that is absent from one that is present but unusable. An empty column (two adjacent delimiters) is absent and never enters dom(e); a column equal to the schema's missing token (LANL's literal "?") enters dom(e) as an unusable value, so detectors abstain over it rather than treating it as never observed.

Ingest is deterministic (R4): the reader consults no wall clock and no source of randomness. Timestamps come from the row's own time column, multiplied into microseconds by the schema's unit.

Index

• type Reader
  • func NewReader(r io.Reader, schema Schema) *Reader
  • func (r *Reader) FilteredByEntity() int64
  • func (r *Reader) Malformed() int64
  • func (r *Reader) Next() (*event.Event, error)
  • func (r *Reader) Rows() int64
• type RowError
  • func (e *RowError) Error() string
• type Schema
  • func LoadSchema(path string) (Schema, error)

Types

type Reader

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

Reader streams events from a delimited text stream. It never loads the corpus into memory — LANL's auth file is 7.6 GB compressed and over a billion rows — only the current line is ever held.

encoding/csv is deliberately not used: LANL fields never contain quotes or embedded delimiters, and csv's quote handling would silently corrupt values containing a double quote.

func NewReader

func NewReader(r io.Reader, schema Schema) *Reader

NewReader returns a Reader decoding r according to schema.

func (*Reader) FilteredByEntity

func (r *Reader) FilteredByEntity() int64

FilteredByEntity returns the count of rows the entity filter excluded.

func (*Reader) Malformed

func (r *Reader) Malformed() int64

Malformed returns the count of rows that produced a *RowError so far.

func (*Reader) Next

func (r *Reader) Next() (*event.Event, error)

Next returns the next event, or io.EOF at the end of the stream.

Malformed rows (wrong column count, unparseable time, missing entity) are not silently dropped: they are returned as *RowError carrying the line number and the reason; the caller decides. Next can be called again after a *RowError, resuming at the next row.

func (*Reader) Rows

func (r *Reader) Rows() int64

Rows returns the count of rows read so far (including malformed and filtered).

type RowError

type RowError struct {
	// Line is the 1-based line number of the malformed row.
	Line int64
	// Reason describes why the row could not become an event.
	Reason string
}

RowError reports a single malformed row: wrong column count, an unparseable timestamp, or a missing entity. Malformed rows are surfaced, never silently dropped; the caller decides whether to log, count, or abort. Reader.Next may be called again after a RowError and resumes at the following row.

func (*RowError) Error

func (e *RowError) Error() string

Error implements the error interface.

type Schema

type Schema struct {
	Source     event.SourceID
	Delimiter  byte            // ',' for LANL
	TimeColumn int             // 0-based column carrying the integer timestamp
	TimeUnit   event.Timestamp // multiplier to microseconds; event.Second for LANL

	// TimeLayout, when non-empty, parses the time column as a formatted timestamp
	// using this Go reference layout instead of an integer tick count, and measures
	// event time from Epoch. Corpora differ in how they encode time — LANL counts
	// seconds from an arbitrary epoch, CERT writes "01/02/2010 02:24:51" — and that
	// is an encoding concern of the reader, not a property of a source's fields.
	// Supporting it here is a generic capability that serves any corpus using the
	// same encoding; E6's zero-code-change claim concerns admitting unseen FIELDS,
	// and a run whose onboarding required this capability records that fact.
	TimeLayout string

	// Epoch is the instant corresponding to event time zero when TimeLayout is set.
	// Event time is deliberately relative: the framework's decay and circular
	// timing depend only on differences and on time of day, and keeping the origin
	// explicit keeps replays reproducible.
	Epoch        time.Time
	EntityColumn int               // 0-based column whose value is the entity ε(e)
	Columns      []event.FieldPath // field path per column; "" = column not emitted as a field
	MissingToken string            // a value equal to this is present-but-unusable; "?" for LANL (§5.3)

	// EntityFilter, when non-nil, admits only rows whose entity value satisfies it,
	// before any event is constructed. This is an ingest-side application of the
	// run's entity-population restriction, which the run records in its coverage
	// statement; it exists because building and digesting events for rows the
	// population excludes costs the majority of a full-corpus pass. Filtered rows
	// are counted, not errors.
	EntityFilter func(string) bool
}

Schema maps delimited columns to field paths. It is configuration, not code (R2, E6): a new source is onboarded by writing a Schema literal, never by changing the reader.

The time column and the entity column are emitted as fields only if their Columns entry is non-empty. The entity column normally is also a field (LANL's auth.source_user, for example), because the co-occurrence graph of §8 wants it.

func LoadSchema

func LoadSchema(path string) (Schema, error)

LoadSchema reads a schema configuration file.