signalsources

package
v1.4.15 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 24, 2026 License: MIT Imports: 23 Imported by: 0

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

func ClampCursor added in v1.4.7 

func ClampCursor(candidate, since, now time.Time) time.Time

ClampCursor bounds a tailing source's next cursor to the closed interval [since, now]:

  • it never rewinds below `since` (the lower bound the worker asked for, so an empty or all-older tick reports "still here" rather than moving back);
  • it never advances beyond `now` (the wall clock at pull time), so an untrusted future-dated document cannot strand the cursor ahead of real time and blank every following query.

`now` should be the same clock reading the source used to upper-bound its scan window, so the returned cursor and the query stay consistent within the tick. When `since` is itself already in the future (a cursor persisted before this convention existed), the result collapses to `now`, letting the next tick resume real tailing instead of querying an empty future window forever.

func ErrRequiresEnterprise added in v1.4.4 

func ErrRequiresEnterprise(typeName string) error

ErrRequiresEnterprise builds the standard error returned when a known enterprise source type is configured but no module has registered it.

func Register added in v1.4.4 

func Register(typeName string, factory Factory)

Register makes a source type constructible by the agent factory. It is intended to be called from an init() in a module that provides additional source types (e.g. Versus Enterprise). Registering the same type name twice, or with a nil factory, panics — both indicate a programming error at wiring time, not a runtime condition.

func RegisterKind added in v1.4.4 

func RegisterKind(sourceType string, kind Kind)

RegisterKind records the KIND a source type belongs to. It is intended to be called from an init() (OSS for the built-in log types; the enterprise module for prometheus/traces). Registering with an empty type or empty kind, or registering the same type twice, panics — each indicates a wiring bug, not a runtime condition, exactly like Register.

func Registered added in v1.4.4 

func Registered() []string

Registered returns the sorted list of registered source type names. Useful for diagnostics and admin surfaces.

func RequiresEnterprise added in v1.4.4 

func RequiresEnterprise(typeName string) bool

RequiresEnterprise reports whether typeName is a known source type that has moved to Versus Enterprise. The factory uses it to turn an unregistered enterprise type into an actionable error.

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) and `endTime = now`. 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` and never past `now` (ClampCursor). The `endTime = now` bound plus the clamp are the same invariant the Elasticsearch source enforces: a future-dated event — an untrusted producer timestamp — must not advance the cursor past the wall clock, or every following `startTime = cursor + 1ms` query would return nothing until that future time actually arrives (the tailing stall). Future-dated events are intentionally not tailed; minor clock skew is recovered once the wall clock passes it.

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.

Tailing is lossless and exactly-once for the near-real-time case. Each tick queries an INCLUSIVE lower bound (`gte`) offset a bounded reorderWindow below the poll cursor, so documents indexed at — or slightly behind — the boundary timestamp (same-millisecond bursts, refresh lag, minor clock skew / late ingestion) are still seen instead of being stranded forever behind a strict `gt`. To avoid folding those re-scanned documents into the model twice, the source tracks the `_id`s it has already emitted whose timestamp falls inside the current reorder window and skips them on the next tick. That dedup set is pruned to one window each tick, so it stays bounded to (window × ingest rate) rather than growing with all-time history.

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 an INCLUSIVE `range[time_field] >= lower` (where lower = since - reorderWindow) and walks pages with `search_after` until the page is short or we've collected enough docs. Documents already delivered on a previous tick — tracked by `_id` within the reorder window — are skipped so each is learned exactly once. The returned cursor is the maximum timestamp seen (never below `since`), so it advances tick-over-tick as new data arrives and stays put when the source is idle.

The scan is also upper-bounded at `now` (`range[time_field] <= now`) and the returned cursor is clamped to `now` (ClampCursor). Without this a single future-dated document — an untrusted producer timestamp — would advance the cursor past the wall clock, after which every following `>= cursor` query matches nothing real until that future time arrives (the "learns the first batch then stops until Clear-all" stall, reproduced live with docs dated 2048). Bounding at `now` keeps the tail on real data; future-dated docs are intentionally not tailed. Minor clock skew (a producer a few seconds ahead) is not lost: once the wall clock passes such a document it falls inside the next tick's inclusive `[cursor - reorderWindow, now]` re-scan.

func (*ElasticsearchSource) Rewind added in v1.4.7 

func (s *ElasticsearchSource) Rewind(_ context.Context) error

Rewind clears the boundary dedup set so a catalog clear (which rewinds the worker poll cursor to the lookback window) makes this source re-emit — and therefore relearn — its whole window from scratch, exactly like a fresh process start. Without it the pre-clear `_id`s would suppress the very docs the operator asked to relearn.

It implements core.SourceRewinder. The poll cursor is the source's primary position, but the dedup set is a second, source-owned piece of state the cursor reset cannot reach; Rewind reconciles it. Safe to call concurrently with Pull (both take mu) and leaves the source in the state a freshly constructed instance would have.

type Factory added in v1.4.4 

type Factory func(name string, options map[string]any) (core.SignalSource, error)

Factory builds a core.SignalSource for one configured source instance from its instance name and the generic per-source options block. The options map is the decoded YAML under the source's `options:` key (see config.AgentSourceConfig.Options); a Factory is responsible for decoding it into whatever concrete config struct it owns (e.g. via mapstructure).

func Lookup added in v1.4.4 

func Lookup(typeName string) (Factory, bool)

Lookup returns the factory registered for typeName, if any.

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

func (*FileSource) Pull 

func (s *FileSource) Pull(_ context.Context, _ time.Time) ([]core.Signal, time.Time, error)

Pull reads new content from the file since the last recorded byte offset. Errors from a single tick are returned (worker logs and continues); the offset is only advanced for content that was successfully read.

func (*FileSource) Rewind added in v1.4.7 

func (s *FileSource) Rewind(_ context.Context) error

Rewind resets the read position to what a brand-new FileSource would use when it finds no persisted sidecar cursor: offset 0 when from_beginning is set (so the whole file is re-read), else the current EOF (so history the operator chose to skip stays skipped). It also removes the sidecar so a restart mid-rewind starts from the same place.

This implements core.SourceRewinder. The file source's byte offset is its own cursor of truth and it ignores the worker's `since` cursor, so a catalog clear that only rewinds the worker cursor would leave this source pinned at EOF and unable to re-emit already-consumed lines. Rewind reconciles the two so a clear makes the SAME running worker re-read the file in place — the in-memory equivalent of recreating the container.

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 Kind added in v1.4.4 

type Kind string

Kind is the family a signal-source type belongs to. The string values match the seam Kind() the typed brains report ("logs"/"metrics"/"traces"), so a registered kind can be compared against a brain's Kind() in a drift test. 

const (
	KindLogs    Kind = "logs"
	KindMetrics Kind = "metrics"
	KindTraces  Kind = "traces"
)

func KindOf added in v1.4.4 

func KindOf(sourceType string) Kind

KindOf returns the registered KIND for a source type, or KindLogs when the type is unknown/unregistered. The log default keeps any unrecognised type behaving exactly as it did before the taxonomy existed.

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 

func (s *LokiSource) Pull(ctx context.Context, since time.Time) ([]core.Signal, time.Time, error)

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 MetricMeta added in v1.4.4 

type MetricMeta struct {
	Type string // "counter" | "gauge" | "histogram" | "summary" | "untyped"
	Help string
	Unit string
}

MetricMeta is the type/help/unit metadata Prometheus records for a metric name (from /api/v1/metadata).

type MetricSample added in v1.4.4 

type MetricSample struct {
	Timestamp time.Time
	Value     float64
}

MetricSample is one (timestamp, value) point of a metric series.

type MetricSeries added in v1.4.4 

type MetricSeries struct {
	Metric  map[string]string
	Samples []MetricSample
}

MetricSeries is one labelled series returned by a PromQL query.

type PrometheusAuth added in v1.4.4 

type PrometheusAuth struct {
	BearerToken string
	Username    string
	Password    string
}

PrometheusAuth carries the optional credentials for a Prometheus endpoint. BearerToken takes priority over Username/Password.

type PrometheusQuerier added in v1.4.4 

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

PrometheusQuerier issues instant and range PromQL queries against a Prometheus HTTP endpoint. Construct it once and reuse it.

func NewPrometheusQuerier added in v1.4.4 

func NewPrometheusQuerier(address string, auth PrometheusAuth, insecureSkipVerify bool) (*PrometheusQuerier, error)

NewPrometheusQuerier validates the address and returns a ready querier.

func (*PrometheusQuerier) LabelValues added in v1.4.4 

func (q *PrometheusQuerier) LabelValues(ctx context.Context, label string, start, end time.Time, matchers ...string) ([]string, error)

LabelValues returns the observed values of a label (GET /api/v1/label/<name>/values), optionally constrained by series selectors (e.g. `up`, `http_requests_total{job="api"}`).

start/end bound the read to a time window (see Metadata for why this matters on Prometheus-compatible backends). Pass the zero time for either to omit it.

func (*PrometheusQuerier) Metadata added in v1.4.4 

func (q *PrometheusQuerier) Metadata(ctx context.Context, start, end time.Time) (map[string]MetricMeta, error)

Metadata returns the metric → metadata map advertised by the target (GET /api/v1/metadata). A metric may carry several metadata entries; Prometheus guarantees they are consistent, so we keep the first.

start/end bound the read to a time window: vanilla Prometheus ignores them, but Prometheus-compatible backends (Mimir, Cortex, Thanos, VictoriaMetrics, Grafana Cloud) return empty for the metadata/label/series endpoints unless an explicit window is supplied. Pass the zero time for either to omit it.

func (*PrometheusQuerier) QueryInstant added in v1.4.4 

func (q *PrometheusQuerier) QueryInstant(ctx context.Context, query string, t time.Time) ([]MetricSeries, error)

QueryInstant runs a PromQL instant query at time t and returns the vector result as a slice of single-sample series.

func (*PrometheusQuerier) QueryRange added in v1.4.4 

func (q *PrometheusQuerier) QueryRange(ctx context.Context, query string, start, end time.Time, step time.Duration) ([]MetricSeries, error)

QueryRange runs a PromQL `query_range` over [start, end] at the given step resolution and returns the matrix result as a slice of series.

func (*PrometheusQuerier) Series added in v1.4.4 

func (q *PrometheusQuerier) Series(ctx context.Context, start, end time.Time, matchers ...string) ([]map[string]string, error)

Series returns the label sets of series matching the given selectors (GET /api/v1/series). At least one selector is required by Prometheus.

start/end bound the read to a time window (see Metadata for why this matters on Prometheus-compatible backends). Pass the zero time for either to omit it.

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

func (*SplunkSource) Pull added in v1.4.3 

func (s *SplunkSource) Pull(ctx context.Context, since time.Time) ([]core.Signal, time.Time, error)

Pull issues an `export` request over the (since, now) window. Results are returned as one JSON object per line (output_mode=json); each line has shape `{"preview":false,"offset":N,"result":{...}}`.

type TempoQuerier added in v1.4.4 

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

TempoQuerier issues TraceQL searches against a Tempo HTTP endpoint.

func NewTempoQuerier added in v1.4.4 

func NewTempoQuerier(address string, auth PrometheusAuth, insecureSkipVerify bool) (*TempoQuerier, error)

NewTempoQuerier validates the address and returns a ready querier.

func (*TempoQuerier) Search added in v1.4.4 

func (q *TempoQuerier) Search(ctx context.Context, query string, start, end time.Time, limit int) ([]TraceSummary, error)

Search runs a TraceQL `query` over [start, end] and returns up to `limit` trace summaries (newest first).

func (*TempoQuerier) TagValues added in v1.4.4 

func (q *TempoQuerier) TagValues(ctx context.Context, tag string) ([]string, error)

TagValues returns the observed values of one tag (GET /api/search/tag/<tag>/values). The values are de-duplicated and sorted.

func (*TempoQuerier) Tags added in v1.4.4 

func (q *TempoQuerier) Tags(ctx context.Context) ([]string, error)

Tags returns the searchable tag names advertised by Tempo. It prefers the v2 scoped endpoint (/api/v2/search/tags) and falls back to the v1 flat list (/api/search/tags) for older Tempo. The result is de-duplicated and sorted.

type TraceSummary added in v1.4.4 

type TraceSummary struct {
	TraceID    string
	Service    string
	Operation  string
	DurationMs float64
	Start      time.Time
	Error      bool
}

TraceSummary is one trace returned by a search, flattened to the fields the analyze agent reasons over.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.