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 GraylogSource ¶ added in v1.4.3
type GraylogSource struct {
// contains filtered or unexported fields
}
GraylogSource pulls log messages from Graylog using the legacy `search/universal/absolute` REST endpoint. That endpoint is synchronous (no async query lifecycle), accepts a free-form Graylog query string, and returns messages sorted by timestamp — exactly the shape the agent worker wants.
Cursor contract: the source asks for `from = since` (Graylog `from` is INCLUSIVE) and filters the response client-side to messages with timestamp > since. The cursor returned is the maximum timestamp seen.
func NewGraylogSource ¶ added in v1.4.3
func NewGraylogSource(name string, cfg config.AgentGraylogSourceConfig) (*GraylogSource, error)
NewGraylogSource validates the config and constructs a ready source.
func (*GraylogSource) Name ¶ added in v1.4.3
func (s *GraylogSource) Name() string
func (*GraylogSource) Pull ¶ added in v1.4.3
func (s *GraylogSource) Pull(ctx context.Context, since time.Time) ([]core.Signal, time.Time, error)
Pull issues a `search/universal/absolute` request between (since, now) and returns every message strictly newer than `since`. The cursor is the max message timestamp seen this tick — when zero messages match, the cursor is unchanged so the next tick re-asks for the same window.
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.
type SplunkSource ¶ added in v1.4.3
type SplunkSource struct {
// contains filtered or unexported fields
}
SplunkSource pulls events from Splunk Enterprise / Splunk Cloud using the synchronous `search/v2/jobs/export` REST endpoint. Export is preferred over `oneshot` because it streams results sorted by `_time` without holding state on the indexer — exactly the cursor-friendly pattern the agent worker wants.
Cursor contract: `earliest_time` is sent as sub-second epoch (since.Unix() + fractional). `latest_time` is `now`. Splunk's `earliest_time` is INCLUSIVE — we drop any returned events whose `_time` is not strictly after `since` to honor the >`since` requirement. The returned cursor is the max `_time` observed.
func NewSplunkSource ¶ added in v1.4.3
func NewSplunkSource(name string, cfg config.AgentSplunkSourceConfig) (*SplunkSource, error)
NewSplunkSource validates the config and constructs a ready source.
func (*SplunkSource) Name ¶ added in v1.4.3
func (s *SplunkSource) Name() string
Source Files