elasticsearch

README

Simple Elasticsearch 6.x API for Golang

Features

• Document

  • Insert document
  • Update document
  • Delete document
  • Get document

• Query

  • Update documents by query
  • Delete documents by query
  • Get documents by query (paging)
  • Get documents by query (scroll)

• Bulk

  • Insert documents

• Index

  • Delete index
  • Refresh index
  • Add Template
  • Delete Template

• Aggregate

  • Term Aggregate (Get most frequent values of a field) Terms Aggregation
  • Range Aggregate (Get min- and max-value of a field) Range Aggregation
  • Cardinality Aggregate (Get unique count of a field) Cardinality Aggregation
  • Composite Term Aggregate (Get all values of a field) Composite Aggregation

• Other

  • Connection test
  • Health status Cluster Health
  • Optional debug logs

Tested with Elasticsearch 6.1.1

Documentation

Index

• Constants
• func SetLogger(l Logger)
• type Bucket
• type Client
  • func Open(baseURL string) (*Client, error)
  • func (c *Client) AddRepository(name string, location string) error
  • func (c *Client) AddSnapshot(repositoryName string, snapshotName string) error
  • func (c *Client) AddTemplate(id string, template map[string]interface{}) error
  • func (c *Client) CardinalityAggregate(index, doctype string, query map[string]interface{}, field string) (int64, error)
  • func (c *Client) CompositeAggregate(index, doctype string, query map[string]interface{}, field string) ([]*Bucket, error)
  • func (c *Client) DateHistogramAggregate(index, doctype string, query map[string]interface{}, field string, ...) ([]*Bucket, error)
  • func (c *Client) DeleteDocument(index, doctype, id string, refresh Refresh) error
  • func (c *Client) DeleteDocuments(index, doctype string, query map[string]interface{}, refresh Refresh) error
  • func (c *Client) DeleteIndex(index string) error
  • func (c *Client) DeleteTemplate(id string) error
  • func (c *Client) GetDocument(index, doctype, id string) (map[string]interface{}, error)
  • func (c *Client) GetDocuments(index, doctype string, query map[string]interface{}, from int64, size int64, ...) ([]map[string]interface{}, int64, error)
  • func (c *Client) Health() (string, error)
  • func (c *Client) InsertDocument(index, doctype, id string, document map[string]interface{}, refresh Refresh) error
  • func (c *Client) InsertDocuments(index string, doctype string, docs map[string]map[string]interface{}) (map[string]error, error)
  • func (c *Client) Ping() error
  • func (c *Client) RangeAggregate(index, doctype string, query map[string]interface{}, field string) (float64, float64, error)
  • func (c *Client) Refresh(index string) error
  • func (c *Client) ScrollDocuments(index, doctype string, query map[string]interface{}, ...) error
  • func (c *Client) TermAggregate(index, doctype string, query map[string]interface{}, ...) (TermAggregationResults, error)
  • func (c *Client) UpdateDocument(index, doctype, id string, painlessScript string, ...) error
  • func (c *Client) UpdateDocuments(index, doctype string, query map[string]interface{}, painlessScript string, ...) error
• type DateHistogramInterval
• type Logger
• type Order
  • func (o *Order) MarshalJSON() ([]byte, error)
• type Refresh
• type TermAggregation
  • func (t *TermAggregation) MarshalJSON() ([]byte, error)
• type TermAggregationResult
• type TermAggregationResults
• type TermAggregations
  • func NewTermAggregations(aggs []*TermAggregation) TermAggregations

Examples

• SetLogger

Constants

const (
	DateHistogramIntervalYear   = "year"
	DateHistogramIntervalMonth  = "month"
	DateHistogramIntervalDay    = "day"
	DateHistogramIntervalHour   = "hour"
	DateHistogramIntervalMinute = "minute"
	DateHistogramIntervalSecond = "second"
	DateHistogramIntervalAuto   = "auto"
)

const (
	StatusGreen  = "green"
	StatusYellow = "yellow"
	StatusRed    = "red"
)

Status constants for Elasticsearch health

Functions

func SetLogger

func SetLogger(l Logger)

SetLogger changes the logger for the elasticsearch package. By default, all messages are discarded.

Example

package main

import "fmt"

type FmtLogger struct {
	Debug bool
}

func (f *FmtLogger) Infof(format string, a ...interface{}) {
	fmt.Printf(format, a...)
}

func (f *FmtLogger) Debugf(format string, a ...interface{}) {
	if f.Debug {
		fmt.Printf(format, a...)
	}
}

func (f *FmtLogger) DebugMode() bool {
	return f.Debug
}

func main() {
	log.Infof("Hello, this message will be discarded.\n")
	log.Debugf("DEBUG: Hello, this message will be discarded.\n")
	f := &FmtLogger{}
	SetLogger(f)
	log.Infof("Hello, this message will be shown.\n")
	log.Debugf("DEBUG: Hello, this message will be discarded.\n")
	f.Debug = true
	log.Infof("Hello, this message will be shown.\n")
	log.Debugf("DEBUG: Hello, this message will be shown.\n")
}

Output:

Hello, this message will be shown.
Hello, this message will be shown.
DEBUG: Hello, this message will be shown.

Types

type Bucket

type Bucket struct {
	Key   interface{} `json:"key"`
	Count int         `json:"doc_count"`
}

Bucket contains how often a specific key was found in a term aggregation.

type Client

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

Client is the api client for Elasticsearch.

func Open

func Open(baseURL string) (*Client, error)

Open creates a new Client instance based on a baseURL. This function does not test the connection. Use Ping() for connection tests.

func (*Client) AddRepository

func (c *Client) AddRepository(name string, location string) error

AddRepository adds a new repository for snapshots. Ensure that a location is configured in /etc/elasticsearch/elasticsearch.yml before calling this function.

func (*Client) AddSnapshot

func (c *Client) AddSnapshot(repositoryName string, snapshotName string) error

AddSnapshot adds a new snapshot in a specified repository. Ensure that the repository exists before calling this function.

func (*Client) AddTemplate

func (c *Client) AddTemplate(id string, template map[string]interface{}) error

AddTemplate adds a new template to Elasticsearch.

func (*Client) CardinalityAggregate

func (c *Client) CardinalityAggregate(index, doctype string, query map[string]interface{}, field string) (int64, error)

CardinalityAggregate returns the unique count of a specific field in a specific index. A query is optional.

func (*Client) CompositeAggregate

func (c *Client) CompositeAggregate(index, doctype string, query map[string]interface{}, field string) ([]*Bucket, error)

func (*Client) DateHistogramAggregate

func (c *Client) DateHistogramAggregate(index, doctype string, query map[string]interface{}, field string, interval DateHistogramInterval, buckets int) ([]*Bucket, error)

func (*Client) DeleteDocument

func (c *Client) DeleteDocument(index, doctype, id string, refresh Refresh) error

DeleteDocument deletes a specific document in a specific index.

func (*Client) DeleteDocuments

func (c *Client) DeleteDocuments(index, doctype string, query map[string]interface{}, refresh Refresh) error

DeleteDocuments deletes multiple documents in a specific index. A query is optional.

func (*Client) DeleteIndex

func (c *Client) DeleteIndex(index string) error

DeleteIndex deletes a whole index.

func (*Client) DeleteTemplate

func (c *Client) DeleteTemplate(id string) error

DeleteTemplate deletes a template.

func (*Client) GetDocument

func (c *Client) GetDocument(index, doctype, id string) (map[string]interface{}, error)

GetDocument returns the document in a specific index and a specific id.

func (*Client) GetDocuments

func (c *Client) GetDocuments(index, doctype string, query map[string]interface{}, from int64, size int64, order *Order) ([]map[string]interface{}, int64, error)

GetDocuments returns multiple documents in a specific index. Order and Query are optional. A offset and size have to be defined. The offset+size have to be lower than 10.000, otherwise Elasticsearch returns an error. If you want to get more than 10.000, use ScrollDocuments instead.

func (*Client) Health

func (c *Client) Health() (string, error)

Health returns the health status of Elasticsearch (green, yellow, red).

func (*Client) InsertDocument

func (c *Client) InsertDocument(index, doctype, id string, document map[string]interface{}, refresh Refresh) error

InsertDocument inserts a document in a specific index. If the id already exists, the old document will be replaced. If refresh is set to false, the result will be returned immediately. If refresh is set to true, elasticsearch waits until all changes were done. If multiple inserts are done and all changes have to be done before continuing, set refresh to false and call Refresh() after.

func (*Client) InsertDocuments

func (c *Client) InsertDocuments(index string, doctype string, docs map[string]map[string]interface{}) (map[string]error, error)

InsertDocuments bulk imports multiple documents into a specific index. Use the document id as key for the 'docs' map. If an error for a specific document occurs, the error will be returned in a map with the document id as key. If an error occurs that regards to all documents, this function will return an error.

func (*Client) Ping

func (c *Client) Ping() error

Ping is the connection test for the Elasticsearch client.

func (*Client) RangeAggregate

func (c *Client) RangeAggregate(index, doctype string, query map[string]interface{}, field string) (float64, float64, error)

RangeAggregate returns the min- and max-value for a specific field in a specific index. A query is optional.

func (*Client) Refresh

func (c *Client) Refresh(index string) error

Refresh refreshs a index. Useful if multiple updates or inserts were done without refresh = true.

func (*Client) ScrollDocuments

func (c *Client) ScrollDocuments(index, doctype string, query map[string]interface{}, docs chan map[string]interface{}) error

ScrollDocuments is the more performant solution to get lots of documents in a specific index. A query is optional. This function will return always all found documents without an order into the 'docs' channel. Ensure that this function is called as a go routine!

func (*Client) TermAggregate

func (c *Client) TermAggregate(index, doctype string, query map[string]interface{}, aggregations TermAggregations) (TermAggregationResults, error)

TermAggregate term aggregates in a specific index. A query is optional.

func (*Client) UpdateDocument

func (c *Client) UpdateDocument(index, doctype, id string, painlessScript string, params map[string]interface{}, refresh Refresh) error

UpdateDocument runs a update script on a specific index and a specific id. It's recommended to use parameterized update scripts and pass the parameters in 'params'. Then elasticsearch has to compile the script only once. Elasticsearch will also return an error, if to many different scripts are executed in a small time interval.

func (*Client) UpdateDocuments

func (c *Client) UpdateDocuments(index, doctype string, query map[string]interface{}, painlessScript string, params map[string]interface{}, refresh Refresh) error

UpdateDocuments runs an update script on multiple documents in a specific index. A query is optional. It's recommended to use parameterized update scripts and pass the parameters in 'params'. Then elasticsearch has to compile the script only once. Elasticsearch will also return an error, if to many different scripts are executed in a small time interval.

type DateHistogramInterval

type DateHistogramInterval string

type Logger

type Logger interface {
	Infof(format string, a ...interface{})
	Debugf(format string, a ...interface{})
	DebugMode() bool
}

Logger defines, that the logger for this package needs to bring the function 'Printf' with it.

type Order

type Order struct {
	Field string
	Order string // asc or desc
}

Order can be used to define the order of the Elasticsearch result.

func (*Order) MarshalJSON

func (o *Order) MarshalJSON() ([]byte, error)

MarshalJSON is the interface implementation for json Marshaler

type Refresh

type Refresh string

Refresh parameter for most requests, default should be RefreshFalse, but if changes have to be done immediately, then you should use RefreshTrue or RefreshWaitFor, see: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-refresh.html

const (
	// RefreshTrue refreshes the relevant primary and replica shards immediately
	RefreshTrue Refresh = "true"
	// RefreshWaitFor does not force a refresh, instead it waits for the next refresh specified by 'index.refresh_interval'
	RefreshWaitFor Refresh = "wait_for"
	// RefreshFalse is the default behaviour, does not refresh anything and is the fastest solution
	RefreshFalse Refresh = "false"
)

type TermAggregation

type TermAggregation struct {
	Field string
	Size  int
}

TermAggregation term aggregates for the specified field. The higher the size is, the more accurate are the results.

func (*TermAggregation) MarshalJSON

func (t *TermAggregation) MarshalJSON() ([]byte, error)

MarshalJSON is the interface implementation for json Marshaler

type TermAggregationResult

type TermAggregationResult struct {
	Buckets []Bucket `json:"buckets"`
}

TermAggregationResult contains the result of the TermAggregation.

type TermAggregationResults

type TermAggregationResults map[string]TermAggregationResult

TermAggregationResults is a list of TermAggregationResult

type TermAggregations

type TermAggregations map[string]*TermAggregation

TermAggregations is a list of TermAggregation, allowing multiple aggregations with one request.

func NewTermAggregations

func NewTermAggregations(aggs []*TermAggregation) TermAggregations

NewTermAggregations takes a list of TermAggregation and returns a TermAggregations