v5

Documentation

Index

• type BulkService
  • func Bulk(client *Elasticsearch, timeout time.Duration, actions int) *BulkService
  • func (e BulkService) Add(index, tzpe string, msg interface{})
  • func (e BulkService) CommitRequired(actions int, bulkSize int) bool
  • func (e BulkService) Do(ctx context.Context) (interface{}, int, bool, error)
  • func (e BulkService) Errors(bulkResponse interface{}) []map[int]string
  • func (e BulkService) EstimatedSizeInBytes() int64
  • func (e BulkService) NumberOfActions() int
  • func (e BulkService) ResendRequests(bulkResponse []*elastic.BulkResponseItem, statusCode ...int)
• type Elasticsearch
  • func NewClient(address, username, password string, timeout time.Duration, sniff bool, ...) (*Elasticsearch, error)
  • func (e *Elasticsearch) Add(index, tzpe string, msg interface{}) error
  • func (e *Elasticsearch) Close() error
  • func (e *Elasticsearch) Flush() error
  • func (e *Elasticsearch) Log(ctx context.Context, index, tzpe string, msg interface{}) error
  • func (e *Elasticsearch) NewBulkProcessorService(ctx context.Context, workers, actions, size int, ...) error
  • func (e *Elasticsearch) Stop()
  • func (e *Elasticsearch) Version() int
• type Index
• type MyRetrier
  • func NewMyRetrier(timeout time.Duration) *MyRetrier
  • func (r *MyRetrier) Retry(ctx context.Context, retry int, req *http.Request, resp *http.Response, ...) (time.Duration, bool, error)
• type Payload

Types

type BulkService

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

BulkService ...

func Bulk

func Bulk(client *Elasticsearch, timeout time.Duration, actions int) *BulkService

Bulk creates a service

func (BulkService) Add

func (e BulkService) Add(index, tzpe string, msg interface{})

Add adds bulkable requests, i.e. BulkIndexRequest, BulkUpdateRequest, and/or BulkDeleteRequest.

func (BulkService) CommitRequired

func (e BulkService) CommitRequired(actions int, bulkSize int) bool

CommitRequired returns true if the service has to commit its bulk requests. This can be either because the number of actions or the estimated size in bytes is larger than specified in the BulkProcessorService.

func (BulkService) Do

func (e BulkService) Do(ctx context.Context) (interface{}, int, bool, error)

Do sends the batched requests to Elasticsearch. Note that, when successful, you can reuse the BulkService for the next batch as the list of bulk requests is cleared on success.

{
  "took":3,
  "errors":false,
  "items":[{
    "index":{
      "_index":"index1",
      "_type":"tweet",
      "_id":"1",
      "_version":3,
      "status":201
    }
  }
}

func (BulkService) Errors

func (e BulkService) Errors(bulkResponse interface{}) []map[int]string

Errors parses a BulkResponse and returns the reason of the failure requests

{
	"error" : {
	  "root_cause" : [
		{
		  "type" : "illegal_argument_exception",
		  "reason" : "Failed to parse int parameter [size] with value [surprise_me]"
		}
	  ],
	  "type" : "illegal_argument_exception",
	  "reason" : "Failed to parse int parameter [size] with value [surprise_me]",
	  "caused_by" : {
		"type" : "number_format_exception",
		"reason" : "For input string: \"surprise_me\""
	  }
	},
	"status" : 400
  }

func (BulkService) EstimatedSizeInBytes

func (e BulkService) EstimatedSizeInBytes() int64

EstimatedSizeInBytes returns the estimated size of all bulkable requests added via Add.

func (BulkService) NumberOfActions

func (e BulkService) NumberOfActions() int

NumberOfActions returns the number of bulkable requests that need to be sent to Elasticsearch on the next batch.

func (BulkService) ResendRequests

func (e BulkService) ResendRequests(bulkResponse []*elastic.BulkResponseItem, statusCode ...int)

ResendRequests helps dealing with bulk rejections https://www.elastic.co/guide/en/elasticsearch/guide/current/_monitoring_individual_nodes.html

type Elasticsearch

type Elasticsearch struct {
	*elastic.Client
	*elastic.BulkProcessor
	*elastic.BulkProcessorService
}

Elasticsearch ...

func NewClient

func NewClient(address, username, password string, timeout time.Duration, sniff bool, insecure bool) (*Elasticsearch, error)

NewClient ...

func (*Elasticsearch) Add

func (e *Elasticsearch) Add(index, tzpe string, msg interface{}) error

func (*Elasticsearch) Close

func (e *Elasticsearch) Close() error

func (*Elasticsearch) Flush

func (e *Elasticsearch) Flush() error

func (*Elasticsearch) Log

func (e *Elasticsearch) Log(ctx context.Context, index, tzpe string, msg interface{}) error

Log sends log messages to elasticsearch

func (*Elasticsearch) NewBulkProcessorService

func (e *Elasticsearch) NewBulkProcessorService(ctx context.Context, workers, actions, size int, flushInterval, timeout time.Duration, stats bool, log *logrus.Entry) error

func (*Elasticsearch) Stop

func (e *Elasticsearch) Stop()

Stop stops the background processes that the client is running, i.e. sniffing the cluster periodically and running health checks on the nodes.

func (*Elasticsearch) Version

func (e *Elasticsearch) Version() int

Version reports the client version

type Index

type Index struct {
	ID    string `json:"_id"`
	Index string `json:"_index"`
	Type  string `json:"_type"`
}

type MyRetrier

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

MyRetrier ...

func NewMyRetrier

func NewMyRetrier(timeout time.Duration) *MyRetrier

NewMyRetrier ...

func (*MyRetrier) Retry

func (r *MyRetrier) Retry(ctx context.Context, retry int, req *http.Request, resp *http.Response, err error) (time.Duration, bool, error)

Retry ...

type Payload

type Payload struct {
	Index `json:"index"`
}