Documentation
¶
Overview ¶
Package es provides Sypl's ElasticSearch support: the low-level client, and the ready-to-use `output.IOutput` factories. It lives in its own Go module (github.com/thalesfsp/sypl/es/v2) so the core sypl module carries no ElasticSearch dependency - import this module only if you log to ElasticSearch.
Client features: - Message's content by default is JSON-formatted. - Provides multiple ways to set the index name: static, or dynamic - which is evaluated at index time. - Two indexing strategies: `ElasticSearch` (one request per document), and `ElasticSearchBulk` (documents batched into _bulk requests via esutil's BulkIndexer - see `NewBulk`, and `NewBulkWithDynamicIndex`). The bulk client indexes asynchronously: failures are delivered through `BulkWithOnError`, and `Flush`/`Close` drain it.
Output factories (formerly in the `output` package): - `Output` (was `output.ElasticSearch`) - `OutputWithDynamicIndex` (was `output.ElasticSearchWithDynamicIndex`) - `OutputWithTagMap` (was `output.ElasticSearchWithTagMap`) - `BulkOutput` (was `output.ElasticSearchBulk`) - `BulkOutputWithDynamicIndex` (was `output.ElasticSearchBulkWithDynamicIndex`)
NOTE: It's the caller's responsibility to create the index, define its mapping, and settings.
Index ¶
- Variables
- func BulkOutput(indexName string, esConfig Config, maxLevel level.Level, bulkOpts []BulkOption, ...) output.IOutput
- func BulkOutputWithDynamicIndex(dynamicIndexFunc DynamicIndexFunc, esConfig Config, maxLevel level.Level, ...) output.IOutput
- func Output(indexName string, esConfig Config, maxLevel level.Level, ...) output.IOutput
- func OutputWithDynamicIndex(dynamicIndexFunc DynamicIndexFunc, esConfig Config, maxLevel level.Level, ...) output.IOutput
- func OutputWithTagMap(tagMap TagMap, esConfig Config, processors ...processor.IProcessor) []output.IOutput
- type BulkOption
- type Config
- type DynamicIndexFunc
- type ElasticSearch
- type ElasticSearchBulk
- type TagMap
- type TagMapItem
Constants ¶
This section is empty.
Variables ¶
var ErrBulkClosed = errors.New("elasticsearch bulk output is closed")
ErrBulkClosed is returned when writing to a closed bulk output.
Functions ¶
func BulkOutput ¶
func BulkOutput( indexName string, esConfig Config, maxLevel level.Level, bulkOpts []BulkOption, processors ...processor.IProcessor, ) output.IOutput
BulkOutput is a built-in `output` - named `ElasticSearchBulk`, that writes to ElasticSearch batching documents into _bulk requests - the high-throughput sibling of `Output`.
Capabilities: `Flush() error` (drains the indexer), and idempotent `Close() error`. Indexing is asynchronous: per-item failures are delivered through `BulkWithOnError`.
NOTE: Formerly `output.ElasticSearchBulk`. NOTE: By default, data is JSON-formatted. NOTE: It's the caller's responsibility to create the index, define its mapping, and settings.
func BulkOutputWithDynamicIndex ¶
func BulkOutputWithDynamicIndex( dynamicIndexFunc DynamicIndexFunc, esConfig Config, maxLevel level.Level, bulkOpts []BulkOption, processors ...processor.IProcessor, ) output.IOutput
BulkOutputWithDynamicIndex is a built-in `output` - named `ElasticSearchBulkWithDynamicIndex-{index}` that writes to ElasticSearch batching documents into _bulk requests. It allows to define a function that returns the index name to be used, evaluated at the index time.
Capabilities: `Flush() error` (drains the indexer), and idempotent `Close() error`. Indexing is asynchronous: per-item failures are delivered through `BulkWithOnError`.
NOTE: Formerly `output.ElasticSearchBulkWithDynamicIndex`. NOTE: By default, data is JSON-formatted. NOTE: It's the caller's responsibility to create the index, define its mapping, and settings.
func Output ¶
func Output( indexName string, esConfig Config, maxLevel level.Level, processors ...processor.IProcessor, ) output.IOutput
Output is a built-in `output` - named `ElasticSearch`, that writes to ElasticSearch.
NOTE: Formerly `output.ElasticSearch`. NOTE: By default, data is JSON-formatted. NOTE: It's the caller's responsibility to create the index, define its mapping, and settings.
func OutputWithDynamicIndex ¶
func OutputWithDynamicIndex( dynamicIndexFunc DynamicIndexFunc, esConfig Config, maxLevel level.Level, processors ...processor.IProcessor, ) output.IOutput
OutputWithDynamicIndex is a built-in `output` - named `ElasticSearchWithDynamicIndex` that writes to ElasticSearch. It allows to define a function that returns the index name to be used, evaluated at the index time.
NOTE: Formerly `output.ElasticSearchWithDynamicIndex`. NOTE: By default, data is JSON-formatted. NOTE: It's the caller's responsibility to create the index, define its mapping, and settings.
func OutputWithTagMap ¶
func OutputWithTagMap( tagMap TagMap, esConfig Config, processors ...processor.IProcessor, ) []output.IOutput
OutputWithTagMap is a built-in `output` - named `ElasticSearchWithTagMap-{tag}` that writes to ElasticSearch. It allows to define a map of tags and indexes. The index name is a function which defines the name of the index and is evaluated at the index time.
IT'S THE CALLER'S RESPONSIBILITY TO DEFINE A CATCH-ALL INDEX - IF DESIRED. TO ACHIEVE THIS, USE `*` AS THE TAG NAME.
NOTE: Formerly `output.ElasticSearchWithTagMap`. NOTE: By default, data is JSON-formatted. NOTE: It's the caller's responsibility to create the index, define its mapping, and settings.
Types ¶
type BulkOption ¶
type BulkOption func(*ElasticSearchBulk)
BulkOption configures the `ElasticSearchBulk` output.
func BulkWithCloseTimeout ¶
func BulkWithCloseTimeout(d time.Duration) BulkOption
BulkWithCloseTimeout bounds how long Flush, and Close wait for the indexer to drain. Defaults to 30s.
func BulkWithFlushBytes ¶
func BulkWithFlushBytes(n int) BulkOption
BulkWithFlushBytes sets the flush threshold in bytes. Defaults to esutil's 5MB.
func BulkWithFlushInterval ¶
func BulkWithFlushInterval(d time.Duration) BulkOption
BulkWithFlushInterval sets the periodic flush interval. Defaults to esutil's 30s.
func BulkWithNumWorkers ¶
func BulkWithNumWorkers(n int) BulkOption
BulkWithNumWorkers sets how many workers flush concurrently. Defaults to esutil's runtime.NumCPU().
func BulkWithOnError ¶
func BulkWithOnError(cb func(error)) BulkOption
BulkWithOnError sets the callback receiving per-item indexing failures, and indexer-level (e.g. flush) failures. The callback may be called concurrently.
type DynamicIndexFunc ¶
type DynamicIndexFunc func() string
DynamicIndexFunc is a function which defines the name of the index, and evaluated at the index time.
type ElasticSearch ¶
type ElasticSearch struct {
// Client is the ElasticSearch client.
Client *elasticsearch.Client
// Config is the ElasticSearch configuration.
Config Config
// DynamicIndex is a function which defines the name of the index, and
// evaluated at the index time.
DynamicIndex DynamicIndexFunc
}
ElasticSearch `Output` definition.
func New ¶
func New( indexName string, esConfig Config, ) *ElasticSearch
New returns a new `ElasticSearch` client.
func NewWithDynamicIndex ¶
func NewWithDynamicIndex( dynamicIndexFunc DynamicIndexFunc, esConfig Config, ) *ElasticSearch
NewWithDynamicIndex returns a new `ElasticSearch` client. It allows to define is a function which defines the name of the index, and evaluated at the index time.
type ElasticSearchBulk ¶
type ElasticSearchBulk struct {
// Client is the ElasticSearch client.
Client *elasticsearch.Client
// Config is the ElasticSearch configuration.
Config Config
// DynamicIndex is a function which defines the name of the index, and
// evaluated at the index time.
DynamicIndex DynamicIndexFunc
// contains filtered or unexported fields
}
ElasticSearchBulk `Output` definition: it batches documents through esutil's BulkIndexer instead of one request per document.
func NewBulk ¶
func NewBulk( indexName string, esConfig Config, opts ...BulkOption, ) *ElasticSearchBulk
NewBulk returns a new `ElasticSearchBulk` client indexing into `indexName`.
NOTE: Indexing is asynchronous, and batched - deliver failures through `BulkWithOnError`, and drain with `Flush`, or `Close`.
NOTE: _bulk items must be single-line JSON (NDJSON) - `Write` enforces it: multi-line documents are compacted, non-compactable ones rejected through the error callback.
func NewBulkWithDynamicIndex ¶
func NewBulkWithDynamicIndex( dynamicIndexFunc DynamicIndexFunc, esConfig Config, opts ...BulkOption, ) *ElasticSearchBulk
NewBulkWithDynamicIndex returns a new `ElasticSearchBulk` client. It allows to define a function which defines the name of the index, and evaluated at the index time.
NOTE: Indexing is asynchronous, and batched - deliver failures through `BulkWithOnError`, and drain with `Flush`, or `Close`.
func (*ElasticSearchBulk) Close ¶
func (es *ElasticSearchBulk) Close() error
Close drains the bulk indexer - waiting, bounded by the close timeout (default: 30s, see `BulkWithCloseTimeout`) - and shuts it down. It's idempotent: subsequent calls return the FIRST call's outcome without re-closing - parity with the async output. Writes after Close return `ErrBulkClosed` - never panic.
func (*ElasticSearchBulk) Flush ¶
func (es *ElasticSearchBulk) Flush() error
Flush drains the bulk indexer - waiting, bounded by the close timeout (default: 30s, see `BulkWithCloseTimeout`), until every enqueued document is sent - then swaps in a fresh indexer, so the output keeps working. After Close it's a no-op.
When REBUILDING the indexer fails, the error is returned - never log.Fatalf: this is a runtime path, and a logging library must not kill the host. Writes then degrade to the uninitialized-indexer guard until a later Flush rebuilds it.
func (*ElasticSearchBulk) Write ¶
func (es *ElasticSearchBulk) Write(data []byte) (int, error)
Write conforms to the `io.Writer` interface: it enqueues the document into the bulk indexer. Indexing is asynchronous - per-item failures are delivered to the `BulkWithOnError` callback, never panics. After Close, it returns `ErrBulkClosed`.
NDJSON safety: _bulk items must be single-line JSON. Multi-line documents are compacted; non-compactable ones are rejected - reported through the error callback, and returned as the write error - so one bad payload can't corrupt the whole batch.
type TagMap ¶
type TagMap = map[string]TagMapItem
TagMap is a map of tags to index names.
NOTE: Formerly `output.ElasticSearchTagMap`.
type TagMapItem ¶
type TagMapItem struct {
DynamicIndexFunc DynamicIndexFunc
Level level.Level
}
TagMapItem is an item of the ElasticSearch tag map.
NOTE: Formerly `output.ElasticSearchTagMapItem`.
func NewTagMapItem ¶
func NewTagMapItem( l level.Level, dynamicIndexFunc DynamicIndexFunc, ) TagMapItem
NewTagMapItem is a helper to create `TagMapItem`.
NOTE: Formerly `output.NewElasticSearchTagMapItem`.
Source Files