goes/goes.go

573 lines
14 KiB
Go
Raw Normal View History

2013-06-15 14:18:48 +08:00
// Copyright 2013 Belogik. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
2013-06-16 12:56:38 +08:00
// Package goes provides an API to access Elasticsearch.
2013-06-15 14:18:48 +08:00
package goes
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"net/http"
"net/url"
2014-08-18 04:24:21 +08:00
"reflect"
2014-01-10 21:12:33 +08:00
"strconv"
2013-06-15 14:18:48 +08:00
"strings"
)
const (
2016-09-28 03:19:07 +08:00
// BulkCommandIndex specifies a bulk doc should be indexed
BulkCommandIndex = "index"
// BulkCommandDelete specifies a bulk doc should be deleted
BulkCommandDelete = "delete"
2013-06-15 14:18:48 +08:00
)
func (err *SearchError) Error() string {
return fmt.Sprintf("[%d] %s", err.StatusCode, err.Msg)
}
2016-09-28 02:56:13 +08:00
// NewClient initiates a new client for an elasticsearch server
2013-06-15 14:18:48 +08:00
//
// This function is pretty useless for now but might be useful in a near future
// if wee need more features like connection pooling or load balancing.
2016-09-28 02:56:13 +08:00
func NewClient(host string, port string) *Client {
return &Client{host, port, http.DefaultClient, ""}
2014-05-30 05:01:39 +08:00
}
2016-09-28 02:56:13 +08:00
// WithHTTPClient sets the http.Client to be used with the connection. Returns the original client.
func (c *Client) WithHTTPClient(cl *http.Client) *Client {
2014-05-30 05:01:39 +08:00
c.Client = cl
return c
2013-06-15 14:18:48 +08:00
}
// Version returns the detected version of the connected ES server
func (c *Client) Version() (string, error) {
// Use cached version if it was already fetched
if c.version != "" {
return c.version, nil
}
// Get the version if it was not cached
r := Request{Method: "GET"}
res, err := c.Do(&r)
if err != nil {
return "", err
}
if version, ok := res.Raw["version"].(map[string]interface{}); ok {
if number, ok := version["number"].(string); ok {
c.version = number
return number, nil
}
}
return "", errors.New("No version returned by ElasticSearch Server")
}
2013-06-15 14:18:48 +08:00
// CreateIndex creates a new index represented by a name and a mapping
2016-09-28 02:56:13 +08:00
func (c *Client) CreateIndex(name string, mapping interface{}) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
Query: mapping,
IndexList: []string{name},
Method: "PUT",
2013-06-15 14:18:48 +08:00
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
// DeleteIndex deletes an index represented by a name
2016-09-28 02:56:13 +08:00
func (c *Client) DeleteIndex(name string) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
IndexList: []string{name},
Method: "DELETE",
2013-06-15 14:18:48 +08:00
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
// RefreshIndex refreshes an index represented by a name
2016-09-28 02:56:13 +08:00
func (c *Client) RefreshIndex(name string) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
IndexList: []string{name},
Method: "POST",
API: "_refresh",
2013-06-15 14:18:48 +08:00
}
return c.Do(&r)
}
// UpdateIndexSettings updates settings for existing index represented by a name and a settings
// as described here: https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-update-settings.html
2016-09-28 02:56:13 +08:00
func (c *Client) UpdateIndexSettings(name string, settings interface{}) (*Response, error) {
r := Request{
Query: settings,
IndexList: []string{name},
Method: "PUT",
API: "_settings",
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
// Optimize an index represented by a name, extra args are also allowed please check:
// http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-optimize.html#indices-optimize
2016-09-28 02:56:13 +08:00
func (c *Client) Optimize(indexList []string, extraArgs url.Values) (*Response, error) {
r := Request{
IndexList: indexList,
ExtraArgs: extraArgs,
Method: "POST",
API: "_optimize",
}
return c.Do(&r)
}
2013-06-18 12:54:31 +08:00
// Stats fetches statistics (_stats) for the current elasticsearch server
2016-09-28 02:56:13 +08:00
func (c *Client) Stats(indexList []string, extraArgs url.Values) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
IndexList: indexList,
ExtraArgs: extraArgs,
Method: "GET",
API: "_stats",
2013-06-15 14:18:48 +08:00
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
2013-06-23 14:59:23 +08:00
// IndexStatus fetches the status (_status) for the indices defined in
// indexList. Use _all in indexList to get stats for all indices
2016-09-28 02:56:13 +08:00
func (c *Client) IndexStatus(indexList []string) (*Response, error) {
2013-06-23 14:59:23 +08:00
r := Request{
IndexList: indexList,
Method: "GET",
API: "_status",
2013-06-23 14:59:23 +08:00
}
return c.Do(&r)
2013-06-23 14:59:23 +08:00
}
2016-09-28 03:19:07 +08:00
// BulkSend bulk adds multiple documents in bulk mode
2016-09-28 02:56:13 +08:00
func (c *Client) BulkSend(documents []Document) (*Response, error) {
2014-07-03 20:28:37 +08:00
// We do not generate a traditional JSON here (often a one liner)
2013-06-15 14:18:48 +08:00
// Elasticsearch expects one line of JSON per line (EOL = \n)
// plus an extra \n at the very end of the document
//
// More informations about the Bulk JSON format for Elasticsearch:
//
// - http://www.elasticsearch.org/guide/reference/api/bulk.html
//
// This is quite annoying for us as we can not use the simple JSON
// Marshaler available in Run().
//
// We have to generate this special JSON by ourselves which leads to
// the code below.
//
// I know it is unreadable I must find an elegant way to fix this.
2013-11-29 18:28:37 +08:00
// len(documents) * 2 : action + optional_sources
// + 1 : room for the trailing \n
2017-02-02 06:59:14 +08:00
bulkData := make([][]byte, 0, len(documents)*2+1)
2013-11-29 18:28:37 +08:00
i := 0
2013-06-15 14:18:48 +08:00
for _, doc := range documents {
2013-11-29 18:28:37 +08:00
action, err := json.Marshal(map[string]interface{}{
2013-06-15 14:18:48 +08:00
doc.BulkCommand: map[string]interface{}{
"_index": doc.Index,
"_type": doc.Type,
2016-09-28 03:19:07 +08:00
"_id": doc.ID,
2013-06-15 14:18:48 +08:00
},
2013-11-29 18:28:37 +08:00
})
2013-06-15 14:18:48 +08:00
if err != nil {
return &Response{}, err
2013-06-15 14:18:48 +08:00
}
2017-02-02 06:59:14 +08:00
bulkData = append(bulkData, action)
2013-11-29 18:28:37 +08:00
i++
2013-06-15 14:18:48 +08:00
2014-08-18 04:24:21 +08:00
if doc.Fields != nil {
if docFields, ok := doc.Fields.(map[string]interface{}); ok {
if len(docFields) == 0 {
continue
}
} else {
typeOfFields := reflect.TypeOf(doc.Fields)
if typeOfFields.Kind() == reflect.Ptr {
typeOfFields = typeOfFields.Elem()
}
if typeOfFields.Kind() != reflect.Struct {
return &Response{}, fmt.Errorf("Document fields not in struct or map[string]interface{} format")
2014-08-18 04:24:21 +08:00
}
if typeOfFields.NumField() == 0 {
continue
}
2013-06-15 14:18:48 +08:00
}
2014-08-18 04:24:21 +08:00
sources, err := json.Marshal(doc.Fields)
2013-06-15 14:18:48 +08:00
if err != nil {
return &Response{}, err
2013-06-15 14:18:48 +08:00
}
2017-02-02 06:59:14 +08:00
bulkData = append(bulkData, sources)
2013-11-29 18:28:37 +08:00
i++
2013-06-15 14:18:48 +08:00
}
}
2013-11-29 18:28:37 +08:00
// forces an extra trailing \n absolutely necessary for elasticsearch
2017-02-02 06:59:14 +08:00
bulkData = append(bulkData, []byte(nil))
2013-11-29 18:28:37 +08:00
2013-06-15 14:18:48 +08:00
r := Request{
Method: "POST",
API: "_bulk",
BulkData: bytes.Join(bulkData, []byte("\n")),
2013-06-15 14:18:48 +08:00
}
resp, err := c.Do(&r)
if err != nil {
return resp, err
}
if resp.Errors {
for _, item := range resp.Items {
for _, i := range item {
if i.Error != "" {
return resp, &SearchError{i.Error, i.Status}
}
}
}
return resp, &SearchError{Msg: "Unknown error while bulk indexing"}
}
return resp, err
2013-06-15 14:18:48 +08:00
}
// Search executes a search query against an index
2016-09-28 02:56:13 +08:00
func (c *Client) Search(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
Query: query,
IndexList: indexList,
TypeList: typeList,
Method: "POST",
API: "_search",
ExtraArgs: extraArgs,
2013-06-15 14:18:48 +08:00
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
// Count executes a count query against an index, use the Count field in the response for the result
2016-09-28 02:56:13 +08:00
func (c *Client) Count(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
r := Request{
Query: query,
IndexList: indexList,
TypeList: typeList,
Method: "POST",
API: "_count",
ExtraArgs: extraArgs,
}
return c.Do(&r)
}
//Query runs a query against an index using the provided http method.
//This method can be used to execute a delete by query, just pass in "DELETE"
//for the HTTP method.
2016-09-28 02:56:13 +08:00
func (c *Client) Query(query interface{}, indexList []string, typeList []string, httpMethod string, extraArgs url.Values) (*Response, error) {
r := Request{
Query: query,
IndexList: indexList,
TypeList: typeList,
Method: httpMethod,
API: "_query",
ExtraArgs: extraArgs,
}
return c.Do(&r)
}
2014-01-10 21:12:33 +08:00
// Scan starts scroll over an index
2016-09-28 02:56:13 +08:00
func (c *Client) Scan(query interface{}, indexList []string, typeList []string, timeout string, size int) (*Response, error) {
2014-01-10 21:12:33 +08:00
v := url.Values{}
v.Add("search_type", "scan")
v.Add("scroll", timeout)
v.Add("size", strconv.Itoa(size))
r := Request{
Query: query,
IndexList: indexList,
TypeList: typeList,
Method: "POST",
API: "_search",
2014-01-10 21:12:33 +08:00
ExtraArgs: v,
}
return c.Do(&r)
2014-01-10 21:12:33 +08:00
}
// Scroll fetches data by scroll id
2016-09-28 03:19:07 +08:00
func (c *Client) Scroll(scrollID string, timeout string) (*Response, error) {
2014-01-10 21:12:33 +08:00
v := url.Values{}
v.Add("scroll", timeout)
r := Request{
Method: "POST",
API: "_search/scroll",
2014-01-10 21:12:33 +08:00
ExtraArgs: v,
2016-09-28 03:19:07 +08:00
Body: []byte(scrollID),
2014-01-10 21:12:33 +08:00
}
return c.Do(&r)
2014-01-10 21:12:33 +08:00
}
2013-06-15 14:18:48 +08:00
// Get a typed document by its id
2016-09-28 02:56:13 +08:00
func (c *Client) Get(index string, documentType string, id string, extraArgs url.Values) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
IndexList: []string{index},
Method: "GET",
API: documentType + "/" + id,
2013-06-15 14:18:48 +08:00
ExtraArgs: extraArgs,
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
// Index indexes a Document
// The extraArgs is a list of url.Values that you can send to elasticsearch as
// URL arguments, for example, to control routing, ttl, version, op_type, etc.
2016-09-28 02:56:13 +08:00
func (c *Client) Index(d Document, extraArgs url.Values) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
Query: d.Fields,
IndexList: []string{d.Index.(string)},
TypeList: []string{d.Type},
ExtraArgs: extraArgs,
Method: "POST",
2013-06-15 14:18:48 +08:00
}
2016-09-28 03:19:07 +08:00
if d.ID != nil {
r.Method = "PUT"
r.ID = d.ID.(string)
2013-06-15 14:18:48 +08:00
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
// Delete deletes a Document d
// The extraArgs is a list of url.Values that you can send to elasticsearch as
// URL arguments, for example, to control routing.
2016-09-28 02:56:13 +08:00
func (c *Client) Delete(d Document, extraArgs url.Values) (*Response, error) {
2013-06-15 14:18:48 +08:00
r := Request{
IndexList: []string{d.Index.(string)},
TypeList: []string{d.Type},
ExtraArgs: extraArgs,
Method: "DELETE",
ID: d.ID.(string),
2013-06-15 14:18:48 +08:00
}
return c.Do(&r)
2013-06-15 14:18:48 +08:00
}
2014-02-27 20:33:51 +08:00
// Buckets returns list of buckets in aggregation
func (a Aggregation) Buckets() []Bucket {
result := []Bucket{}
if buckets, ok := a["buckets"]; ok {
2014-07-03 20:28:55 +08:00
for _, bucket := range buckets.([]interface{}) {
2014-02-27 20:33:51 +08:00
result = append(result, bucket.(map[string]interface{}))
}
}
return result
}
// Key returns key for aggregation bucket
func (b Bucket) Key() interface{} {
return b["key"]
}
// DocCount returns count of documents in this bucket
func (b Bucket) DocCount() uint64 {
return uint64(b["doc_count"].(float64))
}
// Aggregation returns aggregation by name from bucket
2014-07-03 20:28:55 +08:00
func (b Bucket) Aggregation(name string) Aggregation {
2014-02-27 20:33:51 +08:00
if agg, ok := b[name]; ok {
return agg.(map[string]interface{})
}
2016-09-28 03:19:07 +08:00
return Aggregation{}
2014-02-27 20:33:51 +08:00
}
2014-11-11 21:37:34 +08:00
// PutMapping registers a specific mapping for one or more types in one or more indexes
2016-09-28 02:56:13 +08:00
func (c *Client) PutMapping(typeName string, mapping interface{}, indexes []string) (*Response, error) {
2014-11-11 21:37:34 +08:00
r := Request{
Query: mapping,
IndexList: indexes,
Method: "PUT",
API: "_mappings/" + typeName,
2014-11-11 21:37:34 +08:00
}
return c.Do(&r)
2014-11-11 21:37:34 +08:00
}
2014-11-11 22:39:07 +08:00
2016-09-28 03:19:07 +08:00
// GetMapping returns the mappings for the specified types
2016-09-28 02:56:13 +08:00
func (c *Client) GetMapping(types []string, indexes []string) (*Response, error) {
2014-11-13 16:22:11 +08:00
r := Request{
IndexList: indexes,
Method: "GET",
API: "_mapping/" + strings.Join(types, ","),
2014-11-13 16:22:11 +08:00
}
return c.Do(&r)
2014-11-13 16:22:11 +08:00
}
2014-11-11 22:39:07 +08:00
// IndicesExist checks whether index (or indices) exist on the server
2016-09-28 02:56:13 +08:00
func (c *Client) IndicesExist(indexes []string) (bool, error) {
2014-11-11 22:39:07 +08:00
r := Request{
IndexList: indexes,
Method: "HEAD",
2014-11-11 22:39:07 +08:00
}
resp, err := c.Do(&r)
2014-11-11 22:39:07 +08:00
return resp.Status == 200, err
}
2014-11-13 00:34:58 +08:00
2016-09-28 03:19:07 +08:00
// Update updates the specified document using the _update endpoint
2016-09-28 02:56:13 +08:00
func (c *Client) Update(d Document, query interface{}, extraArgs url.Values) (*Response, error) {
2014-11-13 00:34:58 +08:00
r := Request{
Query: query,
IndexList: []string{d.Index.(string)},
TypeList: []string{d.Type},
ExtraArgs: extraArgs,
Method: "POST",
API: "_update",
2014-11-13 00:34:58 +08:00
}
2016-09-28 03:19:07 +08:00
if d.ID != nil {
r.ID = d.ID.(string)
}
return c.Do(&r)
2014-11-13 00:34:58 +08:00
}
2014-11-25 08:41:35 +08:00
// DeleteMapping deletes a mapping along with all data in the type
2016-09-28 02:56:13 +08:00
func (c *Client) DeleteMapping(typeName string, indexes []string) (*Response, error) {
2014-11-25 08:41:35 +08:00
r := Request{
IndexList: indexes,
Method: "DELETE",
API: "_mappings/" + typeName,
2014-11-25 08:41:35 +08:00
}
return c.Do(&r)
2014-11-25 08:41:35 +08:00
}
2015-02-27 22:32:56 +08:00
2016-09-28 02:56:13 +08:00
func (c *Client) modifyAlias(action string, alias string, indexes []string) (*Response, error) {
2015-02-27 22:32:56 +08:00
command := map[string]interface{}{
2017-02-02 06:59:14 +08:00
"actions": make([]map[string]interface{}, 0, 1),
2015-02-27 22:32:56 +08:00
}
for _, index := range indexes {
command["actions"] = append(command["actions"].([]map[string]interface{}), map[string]interface{}{
action: map[string]interface{}{
2015-02-27 22:32:56 +08:00
"index": index,
"alias": alias,
},
})
}
r := Request{
Query: command,
Method: "POST",
API: "_aliases",
2015-02-27 22:32:56 +08:00
}
return c.Do(&r)
2015-02-27 22:32:56 +08:00
}
// AddAlias creates an alias to one or more indexes
2016-09-28 02:56:13 +08:00
func (c *Client) AddAlias(alias string, indexes []string) (*Response, error) {
return c.modifyAlias("add", alias, indexes)
}
// RemoveAlias removes an alias to one or more indexes
2016-09-28 02:56:13 +08:00
func (c *Client) RemoveAlias(alias string, indexes []string) (*Response, error) {
return c.modifyAlias("remove", alias, indexes)
}
2015-02-27 23:46:08 +08:00
// AliasExists checks whether alias is defined on the server
2016-09-28 02:56:13 +08:00
func (c *Client) AliasExists(alias string) (bool, error) {
2015-02-27 23:46:08 +08:00
r := Request{
Method: "HEAD",
API: "_alias/" + alias,
2015-02-27 23:46:08 +08:00
}
resp, err := c.Do(&r)
2015-02-27 23:46:08 +08:00
return resp.Status == 200, err
}
// Do runs the request returned by the requestor and returns the parsed response
func (c *Client) Do(r Requester) (*Response, error) {
req, err := r.Request()
if err != nil {
return &Response{}, err
}
req.URL.Scheme = "http"
req.URL.Host = fmt.Sprintf("%s:%s", c.Host, c.Port)
body, statusCode, err := c.doRequest(req)
esResp := &Response{Status: statusCode}
if err != nil {
return esResp, err
}
if req.Method != "HEAD" {
err = json.Unmarshal(body, &esResp)
if err != nil {
return esResp, err
}
err = json.Unmarshal(body, &esResp.Raw)
if err != nil {
return esResp, err
}
}
if len(esResp.RawError) > 0 && esResp.RawError[0] == '"' {
json.Unmarshal(esResp.RawError, &esResp.Error)
} else {
esResp.Error = string(esResp.RawError)
}
esResp.RawError = nil
if esResp.Error != "" {
return esResp, &SearchError{esResp.Error, esResp.Status}
}
return esResp, nil
}
func (c *Client) doRequest(req *http.Request) ([]byte, uint64, error) {
resp, err := c.Client.Do(req)
if err != nil {
return nil, 0, err
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, uint64(resp.StatusCode), err
}
if resp.StatusCode > 201 && resp.StatusCode < 400 {
return nil, uint64(resp.StatusCode), errors.New(string(body))
}
return body, uint64(resp.StatusCode), nil
}