elasticsearch

Documentation

Index

• Variables
• type ElasticIndexClient
  • func NewElasticIndexClient(connectionFactory bulkServiceFactory, metrics *Metrics, ...) *ElasticIndexClient
  • func (c *ElasticIndexClient) Run(ctx context.Context)
  • func (c *ElasticIndexClient) Send(request *eventIndexRequest)
  • func (c *ElasticIndexClient) Stop()
• type Elasticsearch
  • func (i *Elasticsearch) ProcessAsync(event *firebolt.AsyncEvent)
  • func (i *Elasticsearch) Receive(msg fbcontext.Message) error
  • func (i *Elasticsearch) Setup(cfgMap map[string]string) error
  • func (i *Elasticsearch) Shutdown() error
• type IndexRequest
• type Metrics
  • func (m *Metrics) RegisterElasticIndexMetrics(min, max float64, count int)

Variables

var (
	// ErrMaxRetries is an error used to alert the caller of doBulkIndex
	// that the maximum number of retries was exceeded and the batch should
	// not try to be indexed again. Messages will be dropped in this case.
	ErrMaxRetries = errors.New("max partial bulk retries reached")
)

Types

type ElasticIndexClient

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

ElasticIndexClient is an implementation of the IndexClient interface.

func NewElasticIndexClient

func NewElasticIndexClient(
	connectionFactory bulkServiceFactory,
	metrics *Metrics,
	batchSize,
	maxRetries,
	timeoutSeconds,
	workerPool int,
	batchMaxWait time.Duration) *ElasticIndexClient

NewElasticIndexClient returns an IndexClient with a downstream elasticsearch connection.

func (*ElasticIndexClient) Run

func (c *ElasticIndexClient) Run(ctx context.Context)

Run runs the elasticsearch indexing client

func (*ElasticIndexClient) Send

func (c *ElasticIndexClient) Send(request *eventIndexRequest)

Send prepares a LogMessage to be sent to elasticsearch

func (*ElasticIndexClient) Stop

func (c *ElasticIndexClient) Stop()

Stop stops the elasticsearch indexing client

type Elasticsearch

type Elasticsearch struct {
	fbcontext.ContextAware
	// contains filtered or unexported fields
}

Elasticsearch is a Node that uses elastic_index_client to index documents in the configured ElasticSearch cluster.

func (*Elasticsearch) ProcessAsync

func (i *Elasticsearch) ProcessAsync(event *firebolt.AsyncEvent)

ProcessAsync enqueues the document index request for bulk indexing

func (*Elasticsearch) Receive

func (i *Elasticsearch) Receive(msg fbcontext.Message) error

Receive handles a message from another node or an external source

func (*Elasticsearch) Setup

func (i *Elasticsearch) Setup(cfgMap map[string]string) error

Setup is a no-op in index

func (*Elasticsearch) Shutdown

func (i *Elasticsearch) Shutdown() error

Shutdown provides an opportunity for the Node to clean up resources on shutdown

type IndexRequest

type IndexRequest struct {
	Index       string      `json:"index"`
	MappingType string      `json:"mapping_type"` // in ES 7.x+, leave MappingType unset and ES will use `_doc`
	DocID       string      `json:"doc_id"`
	Doc         interface{} `json:"doc"`
}

IndexRequest is the event payload type to use when passing data to the elasticsearch node.

type Metrics

type Metrics struct {
	BulkErrors                      prometheus.CounterVec
	BulkIndividualErrors            prometheus.CounterVec
	BulkProcessTime                 prometheus.Histogram
	BulkTimeouts                    prometheus.Counter
	BulkMaxRetriesReached           prometheus.Counter
	IndexErrors                     prometheus.CounterVec
	ElasticsearchConnectionFailures prometheus.Counter
	AvailableBatchRoutines          prometheus.Gauge
}

Metrics encapsulates the prometheus metrics produced by the elasticsearch indexer.

func (*Metrics) RegisterElasticIndexMetrics

func (m *Metrics) RegisterElasticIndexMetrics(min, max float64, count int)

RegisterElasticIndexMetrics initializes metrics and registers them with the prometheus client. To support user-configurable bucketing of Histogram metrics, a min, max, and count value must be supplied for generating exponential buckets