elasticsearch

README

Elasticsearch output

It sends events into Elasticsearch. It uses _bulk API to send events in batches. If a network error occurs, the batch will infinitely try to be delivered to the random endpoint.

Config params

endpoints []string required

The list of elasticsearch endpoints in the following format: SCHEMA://HOST:PORT

use_gzip bool default=false

If set, the plugin will use gzip encoding.

gzip_compression_level string default=default options=default|no|best-speed|best-compression|huffman-only

Gzip compression level. Used if use_gzip=true.

username string

Username for HTTP Basic Authentication.

password string

Password for HTTP Basic Authentication.

api_key string

Base64-encoded token for authorization; if set, overrides username/password.

ca_cert string Path or content of a PEM-encoded CA file.

keep_alive KeepAliveConfig

Keep-alive config.

KeepAliveConfig params:

• max_idle_conn_duration - idle keep-alive connections are closed after this duration. By default idle connections are closed after 10s.
• max_conn_duration - keep-alive connections are closed after this duration. If set to 0 - connection duration is unlimited. By default connection duration is 5m.

index_format string default=file-d-%

It defines the pattern of elasticsearch index name. Use % character as a placeholder. Use index_values to define values for the replacement. E.g. if index_format="my-index-%-%" and index_values="service,time" and event is {"service"="my-service"} then index for that event will be my-index-my-service-2020-01-05. First % replaced with service field of the event and the second replaced with current time(see time_format option)

index_values []string default=[@time]

A comma-separated list of event fields which will be used for replacement index_format. There is a special field time which equals the current time. Use the time_format to define a time format. E.g. [service, time]

time_format string default=2006-01-02

The time format pattern to use as value for the time placeholder.

Check out func Parse doc for details.

connection_timeout cfg.Duration default=5s

It defines how much time to wait for the connection.

workers_count cfg.Expression default=gomaxprocs*4

It defines how many workers will be instantiated to send batches.

batch_size cfg.Expression default=capacity/4

A maximum quantity of events to pack into one batch.

batch_size_bytes cfg.Expression default=0

A minimum size of events in a batch to send. If both batch_size and batch_size_bytes are set, they will work together.

batch_flush_timeout cfg.Duration default=200ms

After this timeout batch will be sent even if batch isn't full.

batch_op_type string default=index options=index|create

Operation type to be used in batch requests. It can be index or create. Default is index.

Check out _bulk API doc for details.

retry int default=10

Retries of insertion. If File.d cannot insert for this number of attempts, File.d will fall with non-zero exit code or skip message (see fatal_on_failed_insert).

fatal_on_failed_insert bool default=false

After an insert error, fall with a non-zero exit code or not Experimental feature

retention cfg.Duration default=1s

Retention milliseconds for retry to DB.

retention_exponentially_multiplier int default=2

Multiplier for exponential increase of retention between retries

Generated using insane-doc

Documentation

Index

• Constants
• func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig)
• type Config
• type KeepAliveConfig
• type Plugin
  • func (p *Plugin) Out(event *pipeline.Event)
  • func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginParams)
  • func (p *Plugin) Stop()

Constants

const (
	NDJSONContentType = "application/x-ndjson"
)

Functions

func Factory

func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig)

Types

type Config

type Config struct {
	// > @3@4@5@6
	// >
	// > The list of elasticsearch endpoints in the following format: `SCHEMA://HOST:PORT`
	Endpoints []string `json:"endpoints"  required:"true"` // *

	// > @3@4@5@6
	// >
	// > If set, the plugin will use gzip encoding.
	UseGzip bool `json:"use_gzip" default:"false"` // *

	// > @3@4@5@6
	// >
	// > Gzip compression level. Used if `use_gzip=true`.
	GzipCompressionLevel string `json:"gzip_compression_level" default:"default" options:"default|no|best-speed|best-compression|huffman-only"` // *

	// > @3@4@5@6
	// >
	// > Username for HTTP Basic Authentication.
	Username string `json:"username"` // *

	// > @3@4@5@6
	// >
	// > Password for HTTP Basic Authentication.
	Password string `json:"password"` // *

	// > @3@4@5@6
	// >
	// > Base64-encoded token for authorization; if set, overrides username/password.
	APIKey string `json:"api_key"` // *

	// > @3@4@5@6
	// > Path or content of a PEM-encoded CA file.
	CACert string `json:"ca_cert"` // *

	// > @3@4@5@6
	// >
	// > Keep-alive config.
	// >
	// > `KeepAliveConfig` params:
	// > * `max_idle_conn_duration` - idle keep-alive connections are closed after this duration.
	// > By default idle connections are closed after `10s`.
	// > * `max_conn_duration` - keep-alive connections are closed after this duration.
	// > If set to `0` - connection duration is unlimited.
	// > By default connection duration is `5m`.
	KeepAlive KeepAliveConfig `json:"keep_alive" child:"true"` // *

	// > @3@4@5@6
	// >
	// > It defines the pattern of elasticsearch index name. Use `%` character as a placeholder. Use `index_values` to define values for the replacement.
	// > E.g. if `index_format="my-index-%-%"` and `index_values="service,@@time"` and event is `{"service"="my-service"}`
	// > then index for that event will be `my-index-my-service-2020-01-05`. First `%` replaced with `service` field of the event and the second
	// > replaced with current time(see `time_format` option)
	IndexFormat string `json:"index_format" default:"file-d-%"` // *

	// > @3@4@5@6
	// >
	// > A comma-separated list of event fields which will be used for replacement `index_format`.
	// > There is a special field `@@time` which equals the current time. Use the `time_format` to define a time format.
	// > E.g. `[service, @@time]`
	IndexValues []string `json:"index_values" default:"[@time]" slice:"true"` // *

	// > @3@4@5@6
	// >
	// > The time format pattern to use as value for the `@@time` placeholder.
	// > > Check out [func Parse doc](https://golang.org/pkg/time/#Parse) for details.
	TimeFormat string `json:"time_format" default:"2006-01-02"` // *

	// > @3@4@5@6
	// >
	// > It defines how much time to wait for the connection.
	ConnectionTimeout  cfg.Duration `json:"connection_timeout" default:"5s" parse:"duration"` // *
	ConnectionTimeout_ time.Duration

	// > @3@4@5@6
	// >
	// > It defines how many workers will be instantiated to send batches.
	WorkersCount  cfg.Expression `json:"workers_count" default:"gomaxprocs*4" parse:"expression"` // *
	WorkersCount_ int

	// > @3@4@5@6
	// >
	// > A maximum quantity of events to pack into one batch.
	BatchSize  cfg.Expression `json:"batch_size" default:"capacity/4"  parse:"expression"` // *
	BatchSize_ int

	// > @3@4@5@6
	// >
	// > A minimum size of events in a batch to send.
	// > If both batch_size and batch_size_bytes are set, they will work together.
	BatchSizeBytes  cfg.Expression `json:"batch_size_bytes" default:"0" parse:"expression"` // *
	BatchSizeBytes_ int

	// > @3@4@5@6
	// >
	// > After this timeout batch will be sent even if batch isn't full.
	BatchFlushTimeout  cfg.Duration `json:"batch_flush_timeout" default:"200ms" parse:"duration"` // *
	BatchFlushTimeout_ time.Duration

	// > @3@4@5@6
	// >
	// > Operation type to be used in batch requests. It can be `index` or `create`. Default is `index`.
	// > > Check out [_bulk API doc](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html) for details.
	BatchOpType string `json:"batch_op_type" default:"index" options:"index|create"` // *

	// > @3@4@5@6
	// >
	// > Retries of insertion. If File.d cannot insert for this number of attempts,
	// > File.d will fall with non-zero exit code or skip message (see fatal_on_failed_insert).
	Retry int `json:"retry" default:"10"` // *

	// > @3@4@5@6
	// >
	// > After an insert error, fall with a non-zero exit code or not
	// > **Experimental feature**
	FatalOnFailedInsert bool `json:"fatal_on_failed_insert" default:"false"` // *

	// > @3@4@5@6
	// >
	// > Retention milliseconds for retry to DB.
	Retention  cfg.Duration `json:"retention" default:"1s" parse:"duration"` // *
	Retention_ time.Duration

	// > @3@4@5@6
	// >
	// > Multiplier for exponential increase of retention between retries
	RetentionExponentMultiplier int `json:"retention_exponentially_multiplier" default:"2"` // *
}

! config-params ^ config-params

type KeepAliveConfig

type KeepAliveConfig struct {
	// Idle keep-alive connections are closed after this duration.
	MaxIdleConnDuration  cfg.Duration `json:"max_idle_conn_duration" parse:"duration" default:"10s"`
	MaxIdleConnDuration_ time.Duration

	// Keep-alive connections are closed after this duration.
	MaxConnDuration  cfg.Duration `json:"max_conn_duration" parse:"duration" default:"5m"`
	MaxConnDuration_ time.Duration
}

type Plugin

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

func (*Plugin) Out

func (p *Plugin) Out(event *pipeline.Event)

func (*Plugin) Start

func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginParams)

func (*Plugin) Stop

func (p *Plugin) Stop()