Documentation
¶
Overview ¶
Package opensearch exports Atlas's durable event history to an OpenSearch (or API-compatible Elasticsearch) index. It is a WAL-tailing sink (ADR-0114): an Exporter tails the committed event log off the processor goroutine and bulk-indexes each record, so the engine's hot path is untouched (invariant I1) and a record is exported only once it is durable (invariant I2). Delivery is at-least-once and idempotent — each record's document id is its globally-unique log position, so a retry after a crash or a transient failure overwrites rather than duplicates.
The sink is opt-in and server-configured (endpoint, credentials, index), never authored in a model — mirroring the clio worker's shape (ADR-0036). An empty endpoint disables it.
Index ¶
Constants ¶
const DefaultIndex = "atlas-events"
DefaultIndex is the OpenSearch index events are written to when the config does not name one.
Variables ¶
var ErrSearchRefused = fmt.Errorf("opensearch: search refused")
ErrSearchRefused is returned when the cluster answered and declined: it is reachable and this server may not have what it asked for. Callers separate it from a transport failure because the two send an operator to different places — one to the credentials, the other to the network.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client interface {
// Bulk indexes items into index using the _bulk API, in one request. It must
// return an error if the request fails or the cluster reports any per-item
// failure, so an at-least-once caller retries the whole batch (safe because
// document ids are stable).
Bulk(ctx context.Context, index string, items []Item) error
}
Client indexes documents into OpenSearch. It is an interface so the Exporter is testable without a live cluster and so a different transport can be swapped in.
type Config ¶
Config is the server-side configuration of the OpenSearch sink. URL is the cluster base URL (e.g. "https://opensearch:9200"); an empty URL disables the exporter. Username/Password are optional HTTP basic-auth credentials. Index is the target index (defaults to DefaultIndex).
type DurableFunc ¶
DurableFunc reports the highest log position that is durably committed — the state store's LastAppliedPosition. The exporter never indexes a record beyond it, which is how a WAL-tail honors durable-before-visible (invariant I2, ADR-0114): that watermark is written in the same committed batch as state, only after the WAL fsync, so any position at or below it is on disk.
type Exporter ¶
type Exporter struct {
// contains filtered or unexported fields
}
Exporter tails the durable event log and bulk-indexes new records into OpenSearch. It runs off the processor goroutine and holds no engine state; it only reads committed WAL files and the durable-position watermark. It is not safe for concurrent use: drive it from a single goroutine (one Tick at a time).
func New ¶
func New(walDir string, durable DurableFunc, client Client, index string, pos *PositionStore) (*Exporter, error)
New constructs an Exporter tailing the WAL under walDir, bounded by durable, indexing into index (defaulting to DefaultIndex) through client, and persisting its high-water mark via pos. It loads the persisted mark so a restart resumes where the last run left off.
func (*Exporter) HighWaterMark ¶
HighWaterMark returns the highest record position exported so far — the value persisted across restarts. Safe to call from another goroutine (the retention sweep gates deletes on it, ADR-0115).
func (*Exporter) Tick ¶
Tick runs one export pass: read newly-durable event records after the high-water mark (up to the durable watermark and the batch cap), bulk-index them, then advance and persist the mark. It returns the number of documents indexed. On any error the mark is left unchanged, so the same records are retried on the next Tick (idempotent by document id).
type HTTPClient ¶
type HTTPClient struct {
// contains filtered or unexported fields
}
HTTPClient talks to a real OpenSearch cluster over its HTTP _bulk API.
func NewHTTPClient ¶
func NewHTTPClient(cfg Config) *HTTPClient
NewHTTPClient builds an OpenSearch client for a configured sink. It uses a bounded-timeout HTTP client so a stalled cluster cannot wedge the export loop.
func (*HTTPClient) Bulk ¶
Bulk sends items as one newline-delimited _bulk request. Each item contributes two lines: an index action naming the target index and the stable document id, then the document body. A non-2xx status, or a 2xx whose body reports item-level errors, is returned as an error so the caller retries the batch.
func (*HTTPClient) Search ¶ added in v0.5.0
Search implements Searcher against a real cluster.
The query body is passed through verbatim: what comes back is bounded by the shape the caller asked for — an aggregation, or a page of hits with a `size` and an explicit `_source` — rather than by the size of the index. A caller that asks for unbounded hits gets the response bound above as an error, not a large answer.
type Item ¶
Item is one document to index: a stable, idempotent id (the record's log position) and its JSON body.
type PositionStore ¶
type PositionStore struct {
// contains filtered or unexported fields
}
PositionStore persists the exporter's high-water mark — the highest record [Position] indexed into OpenSearch — so a restart resumes without re-exporting (ADR-0114). It is a single small file written atomically (temp + rename), matching the durable-sidecar discipline (ADR-0019): a crash mid-write leaves either the old value or the new one, never a torn number.
func NewPositionStore ¶
func NewPositionStore(dir string) *PositionStore
NewPositionStore returns a store backed by dir/opensearch.pos. The directory is created on the first Save.
func (*PositionStore) Load ¶
func (p *PositionStore) Load() (uint64, error)
Load returns the persisted high-water mark, or 0 if none has been written yet (a fresh exporter starts from genesis). A present-but-corrupt file is an error rather than a silent reset, so a misconfiguration is not mistaken for "start over and re-export everything".
func (*PositionStore) Save ¶
func (p *PositionStore) Save(pos uint64) error
Save writes pos to a temp file and renames it over the target, so a concurrent or crashing write never leaves a partial value. It deliberately does not fsync: the mark is only a resume hint, and losing the very latest write to a crash just re-exports a few already-indexed records, which OpenSearch de-duplicates by document id (ADR-0114) — so the extra durability is not worth the cost.
type Searcher ¶ added in v0.5.0
type Searcher interface {
// Search posts query (an OpenSearch query DSL body) to index/_search and returns
// the raw response body. The context's deadline is the caller's bound on how long
// somebody else's cluster may hold up this request.
Search(ctx context.Context, index string, query []byte) ([]byte, error)
}
Searcher runs one search against an index. It is an interface so a caller is testable without a live cluster, exactly as Client is.
Source Files