signalsources

package
v1.4.24 Latest Latest
Warning

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

 Go to latest Published: Aug 27, 2026 License: MIT Imports: 25 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

View Source 
const (
	SigNozSignalLogs    = "logs"
	SigNozSignalTraces  = "traces"
	SigNozSignalMetrics = "metrics"

	SigNozRequestTypeRaw        = "raw"
	SigNozRequestTypeTimeSeries = "time_series"
	SigNozRequestTypeScalar     = "scalar"
)

Signal names accepted by `spec.signal` and request types accepted by `requestType` in a v5 query body.

View Source 
const SigNozQueryRangePath = "/api/v5/query_range"

SigNozQueryRangePath is the ONLY path this client may ever request. It is a literal, not a template: there is nothing to interpolate and therefore nothing to steer.

Variables

This section is empty.

Functions

func AttachTailDedupBackend added in v1.4.23 

func AttachTailDedupBackend(sources []core.SignalSource, backend TailDedupBackend) int

AttachTailDedupBackend hands one backend to every source in the slice that keeps a boundary dedup set, and reports how many took it. Sources that do not tail — and every source when the backend is nil — are left untouched.

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 FormatTailWindows added in v1.4.23 

func FormatTailWindows(windows []TailWindow) string

FormatTailWindows renders the widenings as one comma-separated line.

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).

Each tick re-reads an inclusive span below that cursor and suppresses the events already delivered inside it with a persisted event-id set (TailDedup) that survives a restart — the same convention the Elasticsearch and SigNoz tails follow. `reorder_window` sets how much of that span is lateness tolerance; the agent adds one catalog persist interval on top so a killed process's unflushed events are re-read rather than skipped. A source with no `reorder_window` therefore still re-reads the persist interval, which is what stops the shipped default from losing events on an abrupt restart.

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) Commit added in v1.4.23 

func (s *CloudWatchLogsSource) Commit(ctx context.Context) error

Commit makes the event ids this source has delivered durable. The worker calls it only after flushing the events they describe, so a process that dies can re-deliver events but can never suppress events it failed to store. It implements core.SourceCommitter.

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 over `[since - scanWindow, now]` with an inclusive start, and suppresses the events already delivered inside that span with the dedup set. When no window applies at all the start falls back to `since + 1ms` (CloudWatch's startTime is inclusive), which re-reads nothing. 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.

func (*CloudWatchLogsSource) Rewind added in v1.4.23 

func (s *CloudWatchLogsSource) Rewind(ctx context.Context) error

Rewind clears the boundary dedup set — in memory and in its backend — so a catalog clear makes this source re-emit, and therefore relearn, its whole window from scratch. It implements core.SourceRewinder.

func (*CloudWatchLogsSource) SetTailDedupBackend added in v1.4.23 

func (s *CloudWatchLogsSource) SetTailDedupBackend(b TailDedupBackend)

SetTailDedupBackend makes this source's boundary dedup set durable. It implements TailDedupBinder.

func (*CloudWatchLogsSource) SetTailReplaySpan added in v1.4.23 

func (s *CloudWatchLogsSource) SetTailReplaySpan(span time.Duration) (time.Duration, time.Duration)

SetTailReplaySpan widens what each tick re-reads so it also covers the events a killed process learned but never flushed. It implements TailReplaySpanSetter.

type DedupRow added in v1.4.23 

type DedupRow struct {
	ID string
	TS time.Time
}

DedupRow is one row a tailing source's query returned this tick: the id it dedupes on plus the timestamp that decides how long the id is retained.

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 span the next query re-reads and skips them on the next tick. That dedup set is the shared TailDedup: bounded by time and by size, and durable through its backend so a restart resumes on both halves of the position rather than replaying the window.

func NewElasticsearchSource 

func NewElasticsearchSource(name string, cfg config.AgentElasticsearchSourceConfig) (*ElasticsearchSource, error)

NewElasticsearchSource validates config and returns a ready source.

func (*ElasticsearchSource) Commit added in v1.4.23 

func (s *ElasticsearchSource) Commit(ctx context.Context) error

Commit makes the `_id`s this source has delivered durable. The worker calls it only after flushing the docs they describe, so a process that dies can re-deliver docs but can never suppress docs it failed to store. It implements core.SourceCommitter.

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(ctx context.Context) error

Rewind clears the boundary dedup set — in memory and in its backend — 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.

func (*ElasticsearchSource) SetTailDedupBackend added in v1.4.23 

func (s *ElasticsearchSource) SetTailDedupBackend(b TailDedupBackend)

SetTailDedupBackend makes this source's boundary dedup set durable. It implements TailDedupBinder.

func (*ElasticsearchSource) SetTailReplaySpan added in v1.4.23 

func (s *ElasticsearchSource) SetTailReplaySpan(span time.Duration) (time.Duration, time.Duration)

SetTailReplaySpan widens what each tick re-scans so it also covers the docs a killed process learned but never flushed. It implements TailReplaySpanSetter.

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 SigNozAggregation added in v1.4.23 

type SigNozAggregation struct {
	// MetricName is the series to read, e.g. "http_server_duration". Required.
	MetricName string
	// Temporality is the metric's OTLP temporality ("delta", "cumulative" or
	// "unspecified"). Empty lets SigNoz infer it from the metric's metadata.
	Temporality string
	// TimeAggregation reduces samples within one step ("avg", "sum", "min",
	// "max", "count", "rate", "increase", ...). Empty lets SigNoz default.
	TimeAggregation string
	// SpaceAggregation reduces series across the group-by dimensions ("avg",
	// "sum", "min", "max", "p90", "p95", "p99", ...). Empty lets SigNoz default.
	SpaceAggregation string
}

SigNozAggregation is one aggregation term of a builder query. For the `metrics` signal SigNoz carries the metric NAME here and nowhere else, so a metrics query without an aggregation has no subject at all — hence the constructor rejects it rather than letting SigNoz answer an opaque 400.

type SigNozBuilderQuery added in v1.4.23 

type SigNozBuilderQuery struct {
	// Name labels the query in the response ("A", "B", ...).
	Name string
	// Signal is one of SigNozSignalLogs / Traces / Metrics.
	Signal string
	// Filter is a v5 filter expression. Empty omits the filter entirely,
	// matching everything in the window.
	Filter string
	Order  []SigNozOrderBy
	Offset int
	Limit  int
	// Aggregations is required for the `metrics` signal and unused by the raw
	// logs/traces reads. Omitted from the body entirely when empty — SigNoz
	// validates the envelope and an empty array is not the same as absent.
	Aggregations []SigNozAggregation
	// StepInterval is the resolution of a time-series read. The wire format is
	// whole SECONDS; a value below one second, or zero, omits the key.
	StepInterval time.Duration
	// SelectFields extends the default column set of a `raw` read. Omitted from
	// the body entirely when empty, which leaves SigNoz's defaults in place.
	SelectFields []SigNozSelectField
}

SigNozBuilderQuery is one `builder_query` inside a composite query.

type SigNozOrderBy added in v1.4.23 

type SigNozOrderBy struct {
	Key       string // attribute name, e.g. "timestamp" or "id"
	Direction string // "asc" or "desc"
}

SigNozOrderBy is one ordering term of a builder query. SigNoz requires a tiebreak key alongside `timestamp` for a stable order — see SigNozSource.

type SigNozQuerier added in v1.4.23 

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

SigNozQuerier issues v5 query_range reads against one SigNoz instance.

func NewSigNozQuerier added in v1.4.23 

func NewSigNozQuerier(address, apiKey string, insecureSkipVerify bool) (*SigNozQuerier, error)

NewSigNozQuerier validates the endpoint and returns a ready client.

func (*SigNozQuerier) Endpoint added in v1.4.23 

func (q *SigNozQuerier) Endpoint() string

Endpoint is the single URL this client requests. Exposed so callers (and tests) can assert the path allowlist without reaching into the struct.

func (*SigNozQuerier) QueryRange added in v1.4.23 

func (q *SigNozQuerier) QueryRange(ctx context.Context, req SigNozQueryRangeRequest) ([]byte, error)

QueryRange issues one v5 query and returns the validated response body. It is the general entry point: `time_series` and `scalar` request types decode from the same body, so a metric or trace consumer can build on this without the client growing per-signal knowledge.

func (*SigNozQuerier) QueryRangeRaw added in v1.4.23 

func (q *SigNozQuerier) QueryRangeRaw(ctx context.Context, req SigNozQueryRangeRequest) ([]SigNozRawResult, error)

QueryRangeRaw issues a `raw` query and decodes the per-query row sets.

Decoding is deliberately forgiving about the envelope: SigNoz wraps the query response in `{"status":..,"data":..}` and the payload itself carries a `data.results` array, so both nestings are accepted. A body that matches neither yields an error, never a panic — a broken or hostile endpoint must not be able to wedge the worker.

type SigNozQueryRangeRequest added in v1.4.23 

type SigNozQueryRangeRequest struct {
	Start       time.Time
	End         time.Time
	RequestType string
	Queries     []SigNozBuilderQuery
}

SigNozQueryRangeRequest is one call to the v5 query endpoint. Start and End are wall-clock times; the wire format is epoch MILLISECONDS.

type SigNozRawResult added in v1.4.23 

type SigNozRawResult struct {
	QueryName string `json:"queryName"`
	// NextCursor is returned by SigNoz but not used for tailing: offset+limit
	// is the documented pagination for builder queries.
	NextCursor string          `json:"nextCursor"`
	Rows       []*SigNozRawRow `json:"rows"`
}

SigNozRawResult is the `raw` result for one named query.

type SigNozRawRow added in v1.4.23 

type SigNozRawRow struct {
	Timestamp time.Time              `json:"timestamp"`
	Data      map[string]interface{} `json:"data"`
}

SigNozRawRow is one record of a `raw` result. `timestamp` is omitted by SigNoz when zero, so consumers must be prepared to read the timestamp out of Data instead.

type SigNozSelectField added in v1.4.23 

type SigNozSelectField struct {
	// Name is the field to return, e.g. "has_error" or "service.name". Required.
	Name string
	// FieldContext disambiguates where the name lives ("resource", "attribute",
	// "span", "log", ...). Empty lets SigNoz resolve it.
	FieldContext string
	// FieldDataType is the field's type ("string", "bool", "int64", ...). Empty
	// lets SigNoz resolve it.
	FieldDataType string
}

SigNozSelectField names one column a `raw` read should return. SigNoz answers a raw query with a fixed default column set and returns anything else — a span's `has_error` flag, an OTLP attribute — only when it is selected here.

Selecting REPLACES that default set rather than extending it, so a caller must name every column it reads, not just the extra one. Naming a column the backend does not have fails the whole query with a 400.

type SigNozSource added in v1.4.23 

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

SigNozSource pulls log rows from SigNoz through the v5 query API (`requestType: raw`, `signal: logs`).

Cursor contract — the Elasticsearch model, deliberately NOT the Loki one. SigNoz has no server-side tail cursor: a builder query paginates with offset+limit, and request bounds are MILLISECOND precision. A bare-timestamp cursor of the "next tick starts strictly after the newest timestamp seen" kind therefore drops every row that shares the boundary millisecond with the last row of the previous tick. Instead:

  • each tick queries [cursor - reorderWindow, now] with an INCLUSIVE lower bound, ordered `timestamp asc, id asc` (SigNoz requires the `id` tiebreak for a stable order over timestamp);
  • `offset` is walked WITHIN a tick only, never carried across ticks — the window moves between ticks, so a carried offset would point at a different row set;
  • rows already delivered are tracked by `id` and skipped when the overlapping re-scan pulls them back, so each row is learned once.

The dedup set is the shared TailDedup: bounded to the span the next query can re-read, and durable through its backend so a restart resumes on both halves of the position — the timestamp from the worker's cursor store and the ids from the dedup backend — instead of replaying the window.

func NewSigNozSource added in v1.4.23 

func NewSigNozSource(name string, cfg config.AgentSignozSourceConfig) (*SigNozSource, error)

NewSigNozSource validates config and returns a ready source.

func (*SigNozSource) Commit added in v1.4.23 

func (s *SigNozSource) Commit(ctx context.Context) error

Commit makes the ids this source has delivered durable. The worker calls it only after flushing the rows they describe, so a process that dies can re-deliver rows but can never suppress rows it failed to store. It implements core.SourceCommitter.

func (*SigNozSource) Name added in v1.4.23 

func (s *SigNozSource) Name() string

func (*SigNozSource) Pull added in v1.4.23 

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

Pull walks one tick's worth of rows over [since - reorderWindow, now].

func (*SigNozSource) Rewind added in v1.4.23 

func (s *SigNozSource) Rewind(ctx context.Context) error

Rewind clears the boundary dedup set — in memory and in its backend — so a catalog clear makes this source re-emit, and therefore relearn, its whole window from scratch, exactly like a fresh process start. It implements core.SourceRewinder.

func (*SigNozSource) SetTailDedupBackend added in v1.4.23 

func (s *SigNozSource) SetTailDedupBackend(b TailDedupBackend)

SetTailDedupBackend makes this source's boundary dedup set durable. It implements TailDedupBinder.

func (*SigNozSource) SetTailReplaySpan added in v1.4.23 

func (s *SigNozSource) SetTailReplaySpan(span time.Duration) (time.Duration, time.Duration)

SetTailReplaySpan widens what each tick re-reads so it also covers the rows a killed process learned but never flushed. It implements TailReplaySpanSetter.

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 TailDedup added in v1.4.23 

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

TailDedup is the per-source boundary dedup set: the ids a tailing source has already delivered whose timestamps are still inside the span its next query will re-read.

Crash consistency. The set is staged in memory at the END of Pull and only becomes DURABLE when the worker calls Commit — which it does after the flush that puts the rows those ids describe into storage (core.SourceCommitter). That ordering is the whole point, and it has two halves:

  • Rows before ids. The catalog reaches storage on its own flush interval, so committing the ids inside Pull would record rows as delivered up to a whole interval before they were durable; a process that died in that gap would never re-read them and never have stored them. Ids trailing the rows makes the worst case a REPLAY, never a hole.
  • Ids before the cursor. Staging happens before the worker persists the advanced cursor, and the retention floor is one reorder window below the cursor the tick STARTED from, not the one it ends at — so the set still covers the query the next process issues whichever of the two cursors turns out to be the durable one.

The cost is a bounded duplicate when a process dies between the catalog flush and the commit that follows it. That is the acceptable failure direction: a replayed row is re-learned, a dropped row is gone.

func NewTailDedup added in v1.4.23 

func NewTailDedup(source string) *TailDedup

NewTailDedup returns an in-memory dedup set for one source. Call SetTailDedupBackend to make it durable.

func (*TailDedup) Clear added in v1.4.23 

func (d *TailDedup) Clear(ctx context.Context) error

Clear empties the set in memory AND in the backend. It is what a source's Rewind calls: a catalog clear-all rewinds the poll cursor so the operator's history is relearned, and a dedup set that survived that would suppress the very rows the relearn is supposed to re-read.

func (*TailDedup) Commit added in v1.4.23 

func (d *TailDedup) Commit(ctx context.Context) error

Commit pushes everything staged since the last commit to the backend, along with the retention floor and the size cap so the persisted set is bounded the same way the in-memory one is. The worker calls it after the flush that made the staged rows durable (core.SourceCommitter), so the persisted ids can only ever trail the rows, never lead them.

A failed save keeps the pending entries so the next commit retries them; the cost of an outage is a possible replay, never a lost row.

func (*TailDedup) Has added in v1.4.23 

func (d *TailDedup) Has(id string) bool

Has reports whether an id was already delivered and is still retained.

func (*TailDedup) Len added in v1.4.23 

func (d *TailDedup) Len() int

Len returns the number of retained ids.

func (*TailDedup) Load added in v1.4.23 

func (d *TailDedup) Load(ctx context.Context) error

Load hydrates the set from the backend once. It is called at the top of every Pull; after the first success it is a no-op, and a failure is retried on the next tick rather than being cached as "empty".

func (*TailDedup) SetBackend added in v1.4.23 

func (d *TailDedup) SetBackend(b TailDedupBackend)

SetBackend attaches (or detaches, with nil) the persistence backend. The next Load re-hydrates from it.

func (*TailDedup) Stage added in v1.4.23 

func (d *TailDedup) Stage(rows []DedupRow, floor time.Time)

Stage folds this tick's rows into the in-memory set, prunes everything stamped before floor and enforces the size cap. It touches NO backend: an id only becomes durable in Commit, once the rows it describes are durable too. Staging alone is enough to suppress the next tick's re-read inside this process, which is where all but the restart duplicates come from.

floor must be the query's inclusive lower bound (cursor-at-tick-start minus the reorder window), which is exactly the span the next query can re-read.

type TailDedupBackend added in v1.4.23 

type TailDedupBackend interface {
	// LoadDedup returns the persisted id → row-timestamp entries for a source.
	LoadDedup(ctx context.Context, source string) (map[string]time.Time, error)
	// SaveDedup adds entries, drops everything stamped before floor, and keeps
	// at most max ids (oldest dropped first).
	SaveDedup(ctx context.Context, source string, added map[string]time.Time, floor time.Time, max int) error
	// ClearDedup removes a source's whole persisted set.
	ClearDedup(ctx context.Context, source string) error
}

TailDedupBackend persists one source's boundary dedup set. It degrades the same way the worker's CursorStore does: no backend means in-memory only, so a development setup keeps working without Redis and behaves exactly as it did before the set was made durable.

`source` is the SignalSource name, so keys never bleed across sources. An implementation serving more than one organisation adds the org to its own key prefix; the OSS implementation below is single-tenant and does not.

func NewRedisTailDedupBackend added in v1.4.23 

func NewRedisTailDedupBackend(rdb redis.UniversalClient) TailDedupBackend

NewRedisTailDedupBackend returns a Redis-backed dedup backend, or nil when there is no Redis — in which case every source stays in-memory only.

func NewScopedTailDedupBackend added in v1.4.23 

func NewScopedTailDedupBackend(scope string, inner TailDedupBackend) TailDedupBackend

NewScopedTailDedupBackend wraps a backend so every source it stores is namespaced by `scope`. A single-tenant deployment needs nothing here and passes an empty scope, which returns the backend unchanged; a deployment serving one organisation per process passes its organisation id, so two organisations polling a SAME-NAMED source can never read, overwrite or evict each other's set — including when one clears its own.

The scope is an opaque string: this seam knows nothing about organisations, licensing or tenancy. It is the generic half of multi-tenant key isolation, and works over any TailDedupBackend, not just the Redis one.

type TailDedupBinder added in v1.4.23 

type TailDedupBinder interface {
	SetTailDedupBackend(TailDedupBackend)
}

TailDedupBinder is implemented by sources that keep a TailDedup, so the process wiring can hand every one of them the same backend without knowing which concrete types are in the slice.

type TailReplaySpanSetter added in v1.4.23 

type TailReplaySpanSetter interface {
	// SetTailReplaySpan adds span to the configured reorder window and returns
	// the configured window and the resulting effective one. It is idempotent:
	// calling it twice with the same span leaves the same effective window. A
	// non-positive span changes nothing and only reports.
	SetTailReplaySpan(span time.Duration) (configured, effective time.Duration)
}

TailReplaySpanSetter is implemented by tailing sources whose inclusive re-read span below the cursor doubles as the span a restarted process replays. It lets the agent wiring — the only layer that knows the catalog persist interval — widen every tail through one call, including tails registered from outside this package.

type TailWindow added in v1.4.23 

type TailWindow struct {
	Source     string
	Configured time.Duration
	Effective  time.Duration
}

TailWindow reports one source's configured reorder window against the effective span its queries actually re-read.

func ApplyTailReplaySpan added in v1.4.23 

func ApplyTailReplaySpan(sources []core.SignalSource, span time.Duration) []TailWindow

ApplyTailReplaySpan widens every tailing source in the slice so its re-read span covers one catalog persist interval on top of the configured reorder window, and reports what each source ended up with so the caller can say it once at boot. A non-positive span leaves every source untouched.

func TailWindows added in v1.4.23 

func TailWindows(sources []core.SignalSource) []TailWindow

TailWindows reports what each tailing source in the slice currently re-reads, changing nothing. It is how a process that wires its own worker checks that its tails were widened rather than assuming it.

func (TailWindow) String added in v1.4.23 

func (w TailWindow) String() string

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.