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 CloudWatchLogsSource ¶ added in v1.4.0
type CloudWatchLogsSource struct {
// contains filtered or unexported fields
}
CloudWatchLogsSource pulls log events from one AWS CloudWatch Logs log group via FilterLogEvents. Authentication is the standard AWS SDK chain (env vars / shared credentials / IAM role on the host).
FilterLogEvents is the right primitive here because:
- it is real-time (no async query lifecycle like Insights),
- it accepts a `startTime` filter so cursoring is trivial, and
- it returns events sorted by ingestion time across streams.
The cursor is the maximum event timestamp seen on the previous tick (in milliseconds, the unit CloudWatch uses internally).
func NewCloudWatchLogsSource ¶ added in v1.4.0
func NewCloudWatchLogsSource(name string, cfg config.AgentCloudWatchLogsSourceConfig) (*CloudWatchLogsSource, error)
NewCloudWatchLogsSource validates config and returns a ready source. It loads the default AWS SDK config for the configured region.
func (*CloudWatchLogsSource) Name ¶ added in v1.4.0
func (s *CloudWatchLogsSource) Name() string
func (*CloudWatchLogsSource) Pull ¶ added in v1.4.0
func (s *CloudWatchLogsSource) Pull(ctx context.Context, since time.Time) ([]core.Signal, time.Time, error)
Pull issues a FilterLogEvents request with `startTime = since + 1ms` (CloudWatch's startTime is inclusive). It walks NextToken pagination until the page is short, the requested page size has been collected, or we've made `maxPages` calls (safety cap).
Events are appended in API order. The cursor is the maximum event timestamp seen, never lower than `since`.
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
type LokiSource ¶ added in v1.4.0
type LokiSource struct {
// contains filtered or unexported fields
}
LokiSource pulls log entries from Grafana Loki using the `query_range` HTTP endpoint with `direction=forward`. Loki returns entries grouped by stream (label set); this source flattens them into a single timestamp-sorted batch and tracks the maximum timestamp seen as the cursor for the next tick.
Loki entry timestamps are nanoseconds since epoch encoded as a string.
func NewLokiSource ¶ added in v1.4.0
func NewLokiSource(name string, cfg config.AgentLokiSourceConfig) (*LokiSource, error)
NewLokiSource validates config and returns a ready source.
func (*LokiSource) Name ¶ added in v1.4.0
func (s *LokiSource) Name() string
func (*LokiSource) Pull ¶ added in v1.4.0
Pull issues a `query_range` request with `start = since + 1ns` (Loki's `start` is inclusive) and `end = now`. Results are returned forward (oldest first) so the cursor at the end is the max timestamp seen.
We do not paginate: Loki caps the result set at PageSize. If the cap is hit we still advance the cursor to the last entry's timestamp so the next tick continues from there. This is the standard pattern for streaming pulls.
Source Files