Documentation
¶
Overview ¶
Package signalsources contains concrete SignalSource implementations.
Each source implements pkg/core.SignalSource and must:
- Be cursor-aware: `since` defines the lower bound, the returned cursor is the upper bound that should be passed back next tick.
- Be polite: respect AgentSourceConfig.Elasticsearch.PageSize (or its equivalent) and never load arbitrarily many docs into memory.
- Be best-effort: a single failed tick must not crash the worker — return the error and let the worker decide whether to retry.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ElasticsearchSource ¶
type ElasticsearchSource struct {
// contains filtered or unexported fields
}
ElasticsearchSource pulls log documents from one or more Elasticsearch addresses using the `_search` API with a `range` filter on the configured time field. It uses sort-by-time + `search_after` for stable pagination.
This intentionally avoids the official ES client to keep the dependency surface small. The set of features used (basic auth, API-key auth, `_search`, `range`, `query_string`, `sort`, `search_after`) is stable across ES 7.x and 8.x.
func NewElasticsearchSource ¶
func NewElasticsearchSource(name string, cfg config.AgentElasticsearchSourceConfig) (*ElasticsearchSource, error)
NewElasticsearchSource validates config and returns a ready source.
func (*ElasticsearchSource) Name ¶
func (s *ElasticsearchSource) Name() string
func (*ElasticsearchSource) Pull ¶
func (s *ElasticsearchSource) Pull(ctx context.Context, since time.Time) ([]core.Signal, time.Time, error)
Pull issues a `_search` query with `range[time_field] > since` and walks pages with `search_after` until the page is short or we've collected enough docs. The returned cursor is the maximum timestamp seen so duplicate reads are skipped on the next tick.
type FileSource ¶
type FileSource struct {
// contains filtered or unexported fields
}
FileSource tails a log file on disk. It is intended primarily for local testing and small-scale deployments — production users should prefer the Elasticsearch source.
Behavior:
- Position is tracked as a byte offset stored in a sidecar cursor file so it survives process restarts.
- If the file shrinks between ticks (truncate / rotate / re-create) the source reads from the start.
- The Pull `since` argument is ignored: the byte offset is the source of truth. The returned cursor timestamp is just `time.Now()` so the worker has something to log.
- Lines longer than MaxLineBytes are truncated (with a marker).
- Empty / whitespace-only lines are skipped.
func NewFileSource ¶
func NewFileSource(name string, cfg config.AgentFileSourceConfig) (*FileSource, error)
NewFileSource validates configuration and locates the cursor sidecar file. It does NOT open the log file; that happens lazily inside Pull so a missing file at startup doesn't crash the worker (the file may appear later).
func (*FileSource) Name ¶
func (s *FileSource) Name() string
Source Files