index

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

type Count struct {
	Vals []string `json:"v"`
	N    int64    `json:"n"`
}

type Granule struct {
	// N is the granule number: the granule covers stream offsets
	// [N*size, (N+1)*size), clipped to the stream.
	N     int64 `json:"g"`
	Start int64 `json:"s"`
	End   int64 `json:"e"`
	// Sealed is false for the tail granule of a stream that may grow;
	// it is indexed again once the stream has grown past it.
	Sealed  bool  `json:"sealed,omitempty"`
	Records int64 `json:"r"`
	// From and To bound the timed records (Unix ns); zero when none.
	From int64 `json:"from,omitempty"`
	To   int64 `json:"to,omitempty"`
	// Minutes are sorted by minute. Untimed counts records without a
	// time.
	Minutes []Minute `json:"min,omitempty"`
	Untimed []Count  `json:"untimed,omitempty"`
}

type Indexer interface {
	// IndexStep indexes about maxBytes and reports the bytes processed and
	// whether nothing was left to index. With upkeep false it does only
	// backfill (streams and granules never indexed), not the re-indexing
	// of grown tails.
	IndexStep(ctx context.Context, maxBytes int64, upkeep bool) (processed int64, done bool, err error)
}

type Minute struct {
	M      int64   `json:"m"`
	Counts []Count `json:"c"`
}

type RunOptions struct {
	// BytesPerHour is the index building budget (a rolling budget of its
	// own, apart from the agents' budgets).
	BytesPerHour int64
	// Busy reports that agent calls are running. While busy, only the
	// backfill runs (first start, a new file, at the full budget); the
	// upkeep of grown tails waits.
	Busy func() bool
	// Flush persists the index (Store.Flush); called after steps, at most
	// every FlushEvery.
	Flush      func() error
	FlushEvery time.Duration
	// Idle is the pause when every source is done (new data is picked up
	// on the next pass).
	Idle time.Duration
	// OnError is told of a step's error (the source is retried later).
	OnError func(Indexer, error)
	// Now and Sleep are the clock (tests).
	Now   func() time.Time
	Sleep func(context.Context, time.Duration) bool
}

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

type Stream struct {
	Version int `json:"version"`
	// Dev, Ino, Gen identify the stream; PrefixLen/PrefixHash are the
	// latest hashed prefix (a stream whose prefix no longer hashes the
	// same is another stream).
	Dev        uint64 `json:"dev"`
	Ino        uint64 `json:"ino"`
	Gen        string `json:"gen"`
	PrefixLen  int    `json:"prefix_len"`
	PrefixHash string `json:"prefix_hash"`
	// Compression of the file ("", "gzip", "zstd"). Size is the length
	// of the decompressed stream once a full pass measured it, else -1.
	Compression string `json:"compression,omitempty"`
	Size        int64  `json:"size"`
	// FileSize is a compressed file's own size when it was measured.
	FileSize int64 `json:"file_size,omitempty"`
	// Fields are the indexed fields, in tuple order; Dropped those
	// dropped for cardinality.
	Fields  []string `json:"fields"`
	Dropped []string `json:"dropped,omitempty"`
	// Tag is the source's tag the stream was indexed under (the docker
	// source's container name); another tag means another index.
	Tag string `json:"tag,omitempty"`
	// Granules, by granule number.
	Granules map[int64]*Granule `json:"-"`
	List     []*Granule         `json:"granules"`
	// contains filtered or unexported fields
}