elastic

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Nov 6, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Overview

Package elastic is the Elasticsearch (8.x) and OpenSearch (2.x) log source. It sends only allowlisted read requests on the source's own indices through the egress client, with bodies it builds from typed values (a time range, terms on trusted keyword fields, sort and search_after, two aggregation kinds); text is matched locally, after masking. It asks only for the message field and the trusted fields, and drops anything else a response holds. Plans are fixed time slices whose cost is the bytes actually fetched, as for Loki.

Index

Constants

View Source 
const (
	// MaxSize is the most hits of one search page.
	MaxSize = 1000
	// MaxHistogramBuckets is the most date_histogram buckets of a count.
	MaxHistogramBuckets = 1000
	// MaxTermsSize is the most buckets of one terms aggregation.
	MaxTermsSize = 100
	// SearchTimeout is the server-side timeout of a search.
	SearchTimeout = "30s"
)

Caps of what locklog asks a backend.

View Source 
const (
	KindElastic    = "elastic"
	KindOpenSearch = "opensearch"
)

Source kinds (config kind = "elastic" or "opensearch"): the backend Check expects.

View Source 
const (
	FlavourElasticsearch = "elasticsearch"
	FlavourOpenSearch    = "opensearch"
)

Flavours of a backend, from GET /.

View Source 
const (
	// InFlight is the requests in flight per source.
	InFlight = 4
	// MaxSlices bounds the time slices of one plan.
	MaxSlices = 512
	// DefaultTimeField and DefaultMessageField are the ECS names.
	DefaultTimeField    = "@timestamp"
	DefaultMessageField = "message"
)

Defaults.

View Source 
const AroundWindow = 15 * time.Minute

AroundWindow bounds how far Around looks on each side.

View Source 
const MaxCountBy = 2

MaxCountBy bounds the fields of a server-side count (nested terms).

View Source 
const MaxDescribeValues = 64

MaxDescribeValues is the most values Describe lists of a trusted field.

View Source 
const MinSlice = time.Second

MinSlice is the narrowest slice: one holding more than a call's budget is refused.

Variables

View Source 
var ErrScope = errors.New("elasticsearch source: the scope (time field, selector) is not verified")

ErrScope marks a Check failure that concerns the source's scope: the time field or a selector field is not what Check requires (a date; a plain keyword field), or Check failed before it could type-check them. The selector and the time range are sent with every request whatever Check found, so such a failure refuses the source even off production (on a text field a selector term matches a token, and a time field that is not a date compares as text: either widens the scope).

View Source 
var MaxAroundBytes int64 = 64 << 20

MaxAroundBytes bounds the response bytes of one Around when its context carries no byte limit (source.WithByteLimit). A var for tests.

Functions

This section is empty.

Types

type Agg 

type Agg struct {
	DateHistogram *DateHistogram `json:"date_histogram,omitempty"`
	Terms         *TermsAgg      `json:"terms,omitempty"`
	Aggs          map[string]Agg `json:"aggs,omitempty"`
}

Agg is one aggregation: exactly one of DateHistogram and Terms, with sub-aggregations.

type BoolQuery 

type BoolQuery struct {
	Filter  []Clause `json:"filter,omitempty"`
	MustNot []Clause `json:"must_not,omitempty"`
}

BoolQuery is a conjunction of filters and negated filters.

type Clause 

type Clause struct {
	Range map[string]RangeSpec `json:"range,omitempty"`
	Term  map[string]string    `json:"term,omitempty"`
	Terms map[string][]string  `json:"terms,omitempty"`
}

Clause is one filter clause: exactly one of its members is set.

type Client 

type Client interface {
	Send(ctx context.Context, method, path string, q url.Values, body []byte) (egress.Response, error)
}

Client is the egress client of the source (egress.Client).

type CountBody 

type CountBody struct {
	Query Query `json:"query"`
}

CountBody is the body of a _count.

type DateHistogram 

type DateHistogram struct {
	Field         string `json:"field"`
	FixedInterval string `json:"fixed_interval"`
	Offset        string `json:"offset,omitempty"`
	MinDocCount   int    `json:"min_doc_count"`
}

DateHistogram is a date_histogram aggregation.

type FieldValues 

type FieldValues struct {
	Field  string   `json:"f"`
	Values []string `json:"v"`
}

FieldValues is one term of a filter: the field equals one of Values.

type Filter 

type Filter struct {
	In    []FieldValues `json:"in,omitempty"`
	NotIn []FieldValues `json:"not_in,omitempty"`
}

Filter is the server-side selection of a plan, without its time range: the profile's selector and the trusted terms. Its canonical JSON is the chunk's stream, so that a reread selects as the scan did.

func DecodeFilter 

func DecodeFilter(s string) (Filter, error)

DecodeFilter parses a stream back into its filter; only a canonical encoding is accepted.

func (Filter) Encode 

func (f Filter) Encode() (string, error)

Encode returns the canonical JSON of f.

type IndexPrivileges 

type IndexPrivileges struct {
	Names      []string `json:"names"`
	Privileges []string `json:"privileges"`
}

IndexPrivileges are the privileges asked on the source's indices.

type Info 

type Info struct {
	Flavour string
	Version string
	// Write lists the write or manage privileges the credentials hold
	// (Elasticsearch), or the roles not known to be read-only
	// (OpenSearch). Empty on a read-only key.
	Write []string
	// Verified is set when the privileges could be read.
	Verified bool
}

Info is what Check learnt of the backend.

type Options 

type Options struct {
	Name string
	// Kind is KindElastic (default) or KindOpenSearch: the flavour Check
	// requires of the backend.
	Kind   string
	Client Client
	Parser *parse.Parser
	// Indices are the indices, aliases, data streams or trailing-'*'
	// patterns read (validated: egress.ValidIndex).
	Indices []string
	// Selector is the profile's scope: field = value terms ANDed into
	// every query (optional: the indices already scope the source).
	Selector map[string]string
	// TrustedFields are keyword fields shown (after the detectors),
	// selectable and countable; every other field is dropped.
	TrustedFields []string
	// TimeField is the record time (default @timestamp); MessageField the
	// text parsed as the line (default message).
	TimeField    string
	MessageField string
	// TieBreaker is a keyword field unique per document (an event id):
	// pages then resume exactly after the last hit. Without it, a page
	// resumes at the last hit's timestamp and skips the hits it returned;
	// more than Size documents at one timestamp is then an error.
	TieBreaker string
	// ReadOnlyRoles are OpenSearch roles the human vouches grant no
	// write or manage permission (besides the built-in readall roles).
	ReadOnlyRoles []string
	// InFlight overrides InFlight; Size the hits per page (default and
	// cap MaxSize).
	InFlight int
	Size     int
	Now      func() time.Time
}

Options configure a source.

type PrivilegesBody 

type PrivilegesBody struct {
	Cluster []string          `json:"cluster"`
	Index   []IndexPrivileges `json:"index"`
}

PrivilegesBody is the body of Elasticsearch's _has_privileges.

type Query 

type Query struct {
	Bool BoolQuery `json:"bool"`
}

Query is the query of a search or a count.

type RangeSpec 

type RangeSpec struct {
	GTE    string `json:"gte"`
	LT     string `json:"lt"`
	Format string `json:"format"`
}

RangeSpec is a range on the time field, [GTE, LT).

type SearchBody 

type SearchBody struct {
	Size           int                   `json:"size"`
	Timeout        string                `json:"timeout"`
	TrackTotalHits bool                  `json:"track_total_hits"`
	Source         bool                  `json:"_source"`
	Fields         []string              `json:"fields,omitempty"`
	Query          Query                 `json:"query"`
	Sort           []map[string]SortSpec `json:"sort,omitempty"`
	SearchAfter    []any                 `json:"search_after,omitempty"`
	Aggs           map[string]Agg        `json:"aggs,omitempty"`
}

SearchBody is the body of a _search.

type SortSpec 

type SortSpec struct {
	Order        string `json:"order"`
	NumericType  string `json:"numeric_type,omitempty"`
	UnmappedType string `json:"unmapped_type,omitempty"`
}

SortSpec is the sort of one field.

type Source 

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

Source is an Elasticsearch / OpenSearch source. It is safe for concurrent use.

func New 

func New(o Options) (*Source, error)

New validates o.

func (*Source) Around 

func (s *Source) Around(ctx context.Context, ref source.RawRef, before, after int) ([]source.RawRecord, error)

Around returns up to before documents before ref, ref's document and up to after documents after it, from the same filter. The window is fetched under the byte limit of ctx (else MaxAroundBytes): past it, Around fails with source.ErrByteLimit rather than buffer the window whole.

func (*Source) Check 

func (s *Source) Check(ctx context.Context, production bool) (info Info, retErr error)

Check reads the backend's flavour and version, the credentials' privileges and the trusted fields' types. A production profile refuses a key with any write or manage privilege, or one whose privileges cannot be read (security disabled, an unknown OpenSearch role). Trusted fields must be keyword fields; the time field a date.

Re-Check policy expected from the wiring: on a production profile, no Plan, Scan, count or Describe before Check has succeeded (the source is not served otherwise); Check may run again at any time (on reconnect, on a timer): while it runs, and after it fails, the source pushes no trusted term and serves no count, and streams planned under an earlier state that pushed a term now kept local are stale (source.ErrStale), down to the next record of a scan.

Concurrent Checks (#93), which the console's wiring never runs (one Check per connection, before the source is served):

  • a failure, even of an older or cancelled Check, also voids a newer Check still running: it fails safe, and the caller checks again after any failure;
  • Check shares the source's InFlight requests with scans: a slow scan can delay it up to the client's timeout.

func (*Source) Describe 

func (s *Source) Describe(ctx context.Context) (source.Schema, error)

Describe lists the fields of the source's indices (names: masked by the caller), marks the trusted ones, and lists the trusted fields' values of the last 24 hours when they are few.

func (*Source) IndexCounts 

func (s *Source) IndexCounts(ctx context.Context, q source.CountQuery) (source.CountResult, error)

IndexCounts answers a trusted-field count on the server: a _count for a selector-only question, else a search of size 0 with a date_histogram (at most MaxHistogramBuckets buckets) and nested terms aggregations (at most MaxTermsSize values each; the rest counts under index.Other). No document leaves the backend. Counts on untrusted fields, by more than MaxCountBy fields, on bounds finer than a millisecond, or with too many buckets are not served (source.ErrNotIndexable).

func (*Source) Kind 

func (s *Source) Kind() string

Kind returns the source kind (KindElastic or KindOpenSearch).

func (*Source) Name 

func (s *Source) Name() string

Name returns the source name.

func (*Source) Plan 

func (s *Source) Plan(ctx context.Context, r source.Range, sel source.Selector) (source.Byteplan, error)

Plan returns one chunk of epoch-aligned time slices covering r, for the filter built from the profile and the trusted terms. The slices' byte cost is unknown until fetched: the engine charges the bytes fetched.

func (*Source) Reread 

func (s *Source) Reread(ctx context.Context, ref source.RawRef) (source.RawRecord, error)

Reread fetches the document at ref again: ref.Offset is its timestamp and ref.Length its ordinal (from 1) among the documents of that nanosecond. The read is bounded by the byte limit of ctx, else MaxAroundBytes.

func (*Source) Scan 

func (s *Source) Scan(ctx context.Context, c source.Chunk) (source.Iterator, error)

Scan yields the records of the chunk's slices in time order. Slices are fetched lazily, at most InFlight of them ahead of the one Next is reading (each buffered whole, to be sorted), and every page's bytes count towards Processed as it arrives. The byte limit of ctx (source.WithByteLimit) bounds the bytes all of them fetch: past it, the fetches stop and Next fails with source.ErrByteLimit.

func (*Source) Split 

func (s *Source) Split(g source.Granule) []source.Granule

Split cuts slice g into the epoch-aligned slices of the next narrower width of subWidths (clipped to g), or returns nil when g is no wider than MinSlice. Documents of one nanosecond stay in one slice, so refs (a timestamp and an ordinal within it) are unchanged.

type TermsAgg 

type TermsAgg struct {
	Field   string  `json:"field"`
	Size    int     `json:"size"`
	Missing *string `json:"missing,omitempty"`
}

TermsAgg is a terms aggregation. Missing puts the documents without the field in the "" bucket (the absent value of a count).

Source Files

View all Source files

Directories

Path Synopsis
Package elastictest is a fake Elasticsearch 8 / OpenSearch 2 for the tests of the packages that open an Elasticsearch source (the session, the engine through it): an http.Handler over in-memory documents that answers the requests locklog sends (the root, the privilege checks, _resolve/index, _field_caps, _mapping, _settings, _count and _search with bool filters, sort, search_after and the two aggregation kinds), behind basic or API-key authentication unless Open.
 Package elastictest is a fake Elasticsearch 8 / OpenSearch 2 for the tests of the packages that open an Elasticsearch source (the session, the engine through it): an http.Handler over in-memory documents that answers the requests locklog sends (the root, the privilege checks, _resolve/index, _field_caps, _mapping, _settings, _count and _search with bool filters, sort, search_after and the two aggregation kinds), behind basic or API-key authentication unless Open.

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.