opensearch

README

Opensearch Output Plugin

The opensearch output plugin writes to Opensearch via HTTP using Bulk API.

Configuration

[[outputs]]
  [outputs.opensearch]
    # list of Opensearch nodes to use
    # only one will be used on each write
    urls = [ "http://localhost:9200" ]

    # discover nodes periodically
    # with this option it is not necessary to list all nodes in config
    # zero for disable
    discover_interval = "0s"

    # username and password for HTTP Basic Authentication
    username = ""
    password = ""

    # bulk operation, "create" or "index"
    # data streams supports only "create" operation
    operation = "create"

    # if false, all event will be used as a document
    data_only = true

    # label with ingest pipeline name
    # plugin creates one indexer per unique label value
    pipeline_label = ""

    # label which value will be used to route operations to a specific shard
    routing_label = ""

    # time after which inactive indexers will be closed
    # if configured value a zero, idle producers will never be closed
    # if configured value less than 1m but not zero, it will be set to 1m
    idle_timeout = "1h"

    # interval between events buffer flushes if buffer length less than it's capacity
    batch_interval = "5s"

    # events buffer size
    batch_buffer = 100    

    # gzip compression usage flag
    enable_compression = true

    # timeout for HTTP requests
    request_timeout = "10s"

    # maximum number of attempts to execute bulk request
    # before events will be marked as failed
    # 
    # only requests that ended with:
    # - network timeout error 
    # - HTTP 4xx, 5xx codes
    # will be retried
    retry_attempts = 0 # zero for endless attempts

    # interval between retries to execute bulk request
    retry_after = "5s"

    ## TLS configuration
    # if true, TLS client will be used
    tls_enable = false
    # trusted root certificates for server
    tls_ca_file = "/etc/neptunus/ca.pem"
    # used for TLS client certificate authentication
    tls_key_file = "/etc/neptunus/key.pem"
    tls_cert_file = "/etc/neptunus/cert.pem"
    # minimum TLS version, not limited by default
    tls_min_version = "TLS12"
    # send the specified TLS server name via SNI
    tls_server_name = "exmple.svc.local"
    # use TLS but skip chain & host verification
    tls_insecure_skip_verify = false

Documentation

Index

• type Opensearch
  • func (o *Opensearch) Close() error
  • func (o *Opensearch) Init() error
  • func (o *Opensearch) Run()
• type TransportLogger
  • func (l *TransportLogger) LogRoundTrip(_ *http.Request, _ *http.Response, e error, _ time.Time, d time.Duration) error
  • func (l *TransportLogger) RequestBodyEnabled() bool
  • func (l *TransportLogger) ResponseBodyEnabled() bool

Types

type Opensearch

type Opensearch struct {
	*core.BaseOutput  `mapstructure:"-"`
	URLs              []string      `mapstructure:"urls"`
	Username          string        `mapstructure:"username"`
	Password          string        `mapstructure:"password"`
	EnableCompression bool          `mapstructure:"enable_compression"`
	DiscoverInterval  time.Duration `mapstructure:"discover_interval"`
	RequestTimeout    time.Duration `mapstructure:"request_timeout"`
	IdleTimeout       time.Duration `mapstructure:"idle_timeout"`
	PipelineLabel     string        `mapstructure:"pipeline_label"`
	RoutingLabel      string        `mapstructure:"routing_label"`
	DataOnly          bool          `mapstructure:"data_only"`
	Operation         string        `mapstructure:"operation"`

	*tls.TLSClientConfig          `mapstructure:",squash"`
	*batcher.Batcher[*core.Event] `mapstructure:",squash"`
	*retryer.Retryer              `mapstructure:",squash"`
	// contains filtered or unexported fields
}

func (*Opensearch) Close

func (o *Opensearch) Close() error

func (*Opensearch) Init

func (o *Opensearch) Init() error

func (*Opensearch) Run

func (o *Opensearch) Run()

type TransportLogger

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

func (*TransportLogger) LogRoundTrip

func (l *TransportLogger) LogRoundTrip(_ *http.Request, _ *http.Response, e error, _ time.Time, d time.Duration) error

func (*TransportLogger) RequestBodyEnabled

func (l *TransportLogger) RequestBodyEnabled() bool

func (*TransportLogger) ResponseBodyEnabled

func (l *TransportLogger) ResponseBodyEnabled() bool