diff --git a/.travis.yml b/.travis.yml index 2b494e8..123cbbc 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,27 +1,31 @@ language: go +addons: + apt: + packages: + - oracle-java8-set-default + go: - - 1.5.4 - - 1.6.3 - - 1.7.1 + - 1.6.4 + - 1.7.5 env: global: - - GO15VENDOREXPERIMENT=1 + - JAVA_HOME=/usr/lib/jvm/java-8-oracle matrix: - - ES_VERSION=1.3.4 - - ES_VERSION=1.4.4 - - ES_VERSION=1.5.2 - - ES_VERSION=1.6.0 - - ES_VERSION=1.7.0 + - ES_VERSION=1.7.5 ES_URL=https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.7.5.tar.gz + - ES_VERSION=2.4.4 ES_URL=https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.4.4/elasticsearch-2.4.4.tar.gz + - ES_VERSION=5.2.0 ES_URL=https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.2.0.tar.gz before_script: + - java -version + - echo $JAVA_HOME - mkdir ${HOME}/elasticsearch - - wget https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-${ES_VERSION}.tar.gz + - wget $ES_URL - tar -xzf elasticsearch-${ES_VERSION}.tar.gz -C ${HOME}/elasticsearch - - "echo 'script.groovy.sandbox.enabled: true' >> ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/config/elasticsearch.yml" - - ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch >/dev/null & - - sleep 10 # Wait for ES to start up + - "echo 'script.inline: true' >> ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/config/elasticsearch.yml" + - ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch & + - wget --retry-connrefused http://127.0.0.1:9200/ # Wait for ES to start up install: - go get github.com/Masterminds/glide diff --git a/goes.go b/goes.go index 94b3fe4..5ff3d1c 100644 --- a/goes.go +++ b/goes.go @@ -34,7 +34,7 @@ func (err *SearchError) Error() string { // 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. func NewClient(host string, port string) *Client { - return &Client{host, port, http.DefaultClient} + return &Client{host, port, http.DefaultClient, ""} } // WithHTTPClient sets the http.Client to be used with the connection. Returns the original client. @@ -43,6 +43,28 @@ func (c *Client) WithHTTPClient(cl *http.Client) *Client { return c } +// 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") +} + // CreateIndex creates a new index represented by a name and a mapping func (c *Client) CreateIndex(name string, mapping interface{}) (*Response, error) { r := Request{ @@ -97,10 +119,18 @@ func (c *Client) Optimize(indexList []string, extraArgs url.Values) (*Response, Method: "POST", API: "_optimize", } + if version, _ := c.Version(); version > "2.1" { + r.API = "_forcemerge" + } return c.Do(&r) } +// ForceMerge is the same as Optimize, but matches the naming of the endpoint as of ES 2.1.0 +func (c *Client) ForceMerge(indexList []string, extraArgs url.Values) (*Response, error) { + return c.Optimize(indexList, extraArgs) +} + // Stats fetches statistics (_stats) for the current elasticsearch server func (c *Client) Stats(indexList []string, extraArgs url.Values) (*Response, error) { r := Request{ @@ -145,7 +175,7 @@ func (c *Client) BulkSend(documents []Document) (*Response, error) { // len(documents) * 2 : action + optional_sources // + 1 : room for the trailing \n - bulkData := make([][]byte, len(documents)*2+1) + bulkData := make([][]byte, 0, len(documents)*2+1) i := 0 for _, doc := range documents { @@ -161,7 +191,7 @@ func (c *Client) BulkSend(documents []Document) (*Response, error) { return &Response{}, err } - bulkData[i] = action + bulkData = append(bulkData, action) i++ if doc.Fields != nil { @@ -187,13 +217,13 @@ func (c *Client) BulkSend(documents []Document) (*Response, error) { return &Response{}, err } - bulkData[i] = sources + bulkData = append(bulkData, sources) i++ } } // forces an extra trailing \n absolutely necessary for elasticsearch - bulkData[len(bulkData)-1] = []byte(nil) + bulkData = append(bulkData, []byte(nil)) r := Request{ Method: "POST", @@ -264,10 +294,49 @@ func (c *Client) Query(query interface{}, indexList []string, typeList []string, return c.Do(&r) } -// Scan starts scroll over an index +// DeleteByQuery deletes documents matching the specified query. It will return an error for ES 2.x, +// because delete by query support was removed in those versions. +func (c *Client) DeleteByQuery(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) { + version, err := c.Version() + if err != nil { + return nil, err + } + if version > "2" && version < "5" { + return nil, errors.New("ElasticSearch 2.x does not support delete by query") + } + + r := Request{ + Query: query, + IndexList: indexList, + TypeList: typeList, + Method: "DELETE", + API: "_query", + ExtraArgs: extraArgs, + } + + if version > "5" { + r.API = "_delete_by_query" + r.Method = "POST" + } + + return c.Do(&r) +} + +// Scan starts scroll over an index. +// For ES versions < 5.x, it uses search_type=scan; for 5.x it uses sort=_doc. This means that data +// will be returned in the initial response for 5.x versions, but not for older versions. Code +// wishing to be compatible with both should be written to handle either case. func (c *Client) Scan(query interface{}, indexList []string, typeList []string, timeout string, size int) (*Response, error) { v := url.Values{} - v.Add("search_type", "scan") + version, err := c.Version() + if err != nil { + return nil, err + } + if version > "5" { + v.Add("sort", "_doc") + } else { + v.Add("search_type", "scan") + } v.Add("scroll", timeout) v.Add("size", strconv.Itoa(size)) @@ -285,14 +354,24 @@ func (c *Client) Scan(query interface{}, indexList []string, typeList []string, // Scroll fetches data by scroll id func (c *Client) Scroll(scrollID string, timeout string) (*Response, error) { - v := url.Values{} - v.Add("scroll", timeout) - r := Request{ - Method: "POST", - API: "_search/scroll", - ExtraArgs: v, - Body: []byte(scrollID), + Method: "POST", + API: "_search/scroll", + } + + if version, err := c.Version(); err != nil { + return nil, err + } else if version > "2" { + r.Body, err = json.Marshal(map[string]string{"scroll": timeout, "scroll_id": scrollID}) + if err != nil { + return nil, err + } + } else { + v := url.Values{} + v.Add("scroll", timeout) + v.Add("scroll_id", scrollID) + + r.ExtraArgs = v } return c.Do(&r) @@ -433,6 +512,11 @@ func (c *Client) Update(d Document, query interface{}, extraArgs url.Values) (*R // DeleteMapping deletes a mapping along with all data in the type func (c *Client) DeleteMapping(typeName string, indexes []string) (*Response, error) { + if version, err := c.Version(); err != nil { + return nil, err + } else if version > "2" { + return nil, errors.New("Deletion of mappings is not supported in ES 2.x and above.") + } r := Request{ IndexList: indexes, @@ -445,7 +529,7 @@ func (c *Client) DeleteMapping(typeName string, indexes []string) (*Response, er func (c *Client) modifyAlias(action string, alias string, indexes []string) (*Response, error) { command := map[string]interface{}{ - "actions": make([]map[string]interface{}, 1), + "actions": make([]map[string]interface{}, 0, 1), } for _, index := range indexes { @@ -489,14 +573,28 @@ func (c *Client) AliasExists(alias string) (bool, error) { return resp.Status == 200, err } +func (c *Client) replaceHost(req *http.Request) { + req.URL.Scheme = "http" + req.URL.Host = fmt.Sprintf("%s:%s", c.Host, c.Port) +} + +// DoRaw Does the provided requeset and returns the raw bytes and the status code of the response +func (c *Client) DoRaw(r Requester) ([]byte, uint64, error) { + req, err := r.Request() + if err != nil { + return nil, 0, err + } + c.replaceHost(req) + return c.doRequest(req) +} + // 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) + c.replaceHost(req) body, statusCode, err := c.doRequest(req) esResp := &Response{Status: statusCode} diff --git a/goes_test.go b/goes_test.go index 1abe586..0050a36 100644 --- a/goes_test.go +++ b/goes_test.go @@ -41,7 +41,7 @@ func (s *GoesTestSuite) SetUpTest(c *C) { func (s *GoesTestSuite) TestNewClient(c *C) { conn := NewClient(ESHost, ESPort) - c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, http.DefaultClient}) + c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, http.DefaultClient, ""}) } func (s *GoesTestSuite) TestWithHTTPClient(c *C) { @@ -54,7 +54,7 @@ func (s *GoesTestSuite) TestWithHTTPClient(c *C) { } conn := NewClient(ESHost, ESPort).WithHTTPClient(cl) - c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, cl}) + c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, cl, ""}) c.Assert(conn.Client.Transport.(*http.Transport).DisableCompression, Equals, true) c.Assert(conn.Client.Transport.(*http.Transport).ResponseHeaderTimeout, Equals, 1*time.Second) } @@ -114,7 +114,7 @@ func (s *GoesTestSuite) TestRunMissingIndex(c *C) { } _, err := conn.Do(&r) - c.Assert(err.Error(), Equals, "[404] IndexMissingException[[i] missing]") + c.Assert(err.Error(), Matches, "\\[40.\\] .*i.*") } func (s *GoesTestSuite) TestCreateIndex(c *C) { @@ -150,9 +150,10 @@ func (s *GoesTestSuite) TestDeleteIndexInexistantIndex(c *C) { conn := NewClient(ESHost, ESPort) resp, err := conn.DeleteIndex("foobar") - c.Assert(err.Error(), Equals, "[404] IndexMissingException[[foobar] missing]") + c.Assert(err.Error(), Matches, "\\[404\\] .*foobar.*") resp.Raw = nil // Don't make us have to duplicate this. - c.Assert(resp, DeepEquals, &Response{Status: 404, Error: "IndexMissingException[[foobar] missing]"}) + c.Assert(resp.Status, Equals, uint64(404)) + c.Assert(resp.Error, Matches, ".*foobar.*") } func (s *GoesTestSuite) TestDeleteIndexExistingIndex(c *C) { @@ -161,6 +162,7 @@ func (s *GoesTestSuite) TestDeleteIndexExistingIndex(c *C) { indexName := "testdeleteindexexistingindex" _, err := conn.CreateIndex(indexName, map[string]interface{}{}) + defer conn.DeleteIndex(indexName) c.Assert(err, IsNil) @@ -179,8 +181,12 @@ func (s *GoesTestSuite) TestUpdateIndexSettings(c *C) { conn := NewClient(ESHost, ESPort) indexName := "testupdateindex" + // Just in case + conn.DeleteIndex(indexName) + _, err := conn.CreateIndex(indexName, map[string]interface{}{}) c.Assert(err, IsNil) + defer conn.DeleteIndex(indexName) _, err = conn.UpdateIndexSettings(indexName, map[string]interface{}{ "index": map[string]interface{}{ @@ -199,6 +205,7 @@ func (s *GoesTestSuite) TestRefreshIndex(c *C) { _, err := conn.CreateIndex(indexName, map[string]interface{}{}) c.Assert(err, IsNil) + defer conn.DeleteIndex(indexName) _, err = conn.RefreshIndex(indexName) c.Assert(err, IsNil) @@ -214,6 +221,7 @@ func (s *GoesTestSuite) TestOptimize(c *C) { conn.DeleteIndex(indexName) _, err := conn.CreateIndex(indexName, map[string]interface{}{}) c.Assert(err, IsNil) + defer conn.DeleteIndex(indexName) // we must wait for a bit otherwise ES crashes time.Sleep(1 * time.Second) @@ -260,6 +268,7 @@ func (s *GoesTestSuite) TestBulkSend(c *C) { conn.DeleteIndex(indexName) _, err := conn.CreateIndex(indexName, nil) c.Assert(err, IsNil) + defer conn.DeleteIndex(indexName) response, err := conn.BulkSend(tweets) i := Item{ @@ -350,6 +359,7 @@ func (s *GoesTestSuite) TestStats(c *C) { conn.DeleteIndex(indexName) _, err := conn.CreateIndex(indexName, map[string]interface{}{}) c.Assert(err, IsNil) + defer conn.DeleteIndex(indexName) // we must wait for a bit otherwise ES crashes time.Sleep(1 * time.Second) @@ -389,9 +399,7 @@ func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) { }, } - extraArgs := make(url.Values, 1) - extraArgs.Set("ttl", "86400000") - response, err := conn.Index(d, extraArgs) + response, err := conn.Index(d, nil) c.Assert(err, IsNil) expectedResponse := &Response{ @@ -403,6 +411,7 @@ func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) { } response.Raw = nil + response.Shards = Shard{} c.Assert(response, DeepEquals, expectedResponse) } @@ -426,9 +435,7 @@ func (s *GoesTestSuite) TestIndexWithFieldsNotInMapOrStruct(c *C) { Fields: "test", } - extraArgs := make(url.Values, 1) - extraArgs.Set("ttl", "86400000") - _, err = conn.Index(d, extraArgs) + _, err = conn.Index(d, nil) c.Assert(err, Not(IsNil)) } @@ -455,9 +462,7 @@ func (s *GoesTestSuite) TestIndexIdDefined(c *C) { }, } - extraArgs := make(url.Values, 1) - extraArgs.Set("ttl", "86400000") - response, err := conn.Index(d, extraArgs) + response, err := conn.Index(d, nil) c.Assert(err, IsNil) expectedResponse := &Response{ @@ -469,6 +474,7 @@ func (s *GoesTestSuite) TestIndexIdDefined(c *C) { } response.Raw = nil + response.Shards = Shard{} c.Assert(response, DeepEquals, expectedResponse) } @@ -540,6 +546,7 @@ func (s *GoesTestSuite) TestDelete(c *C) { Version: 2, } response.Raw = nil + response.Shards = Shard{} c.Assert(response, DeepEquals, expectedResponse) response, err = conn.Delete(d, url.Values{}) @@ -555,6 +562,7 @@ func (s *GoesTestSuite) TestDelete(c *C) { Version: 3, } response.Raw = nil + response.Shards = Shard{} c.Assert(response, DeepEquals, expectedResponse) } @@ -564,6 +572,8 @@ func (s *GoesTestSuite) TestDeleteByQuery(c *C) { docID := "1234" conn := NewClient(ESHost, ESPort) + version, _ := conn.Version() + // just in case conn.DeleteIndex(indexName) @@ -603,7 +613,13 @@ func (s *GoesTestSuite) TestDeleteByQuery(c *C) { c.Assert(err, IsNil) c.Assert(response.Hits.Total, Equals, uint64(1)) - response, err = conn.Query(query, []string{indexName}, []string{docType}, "DELETE", url.Values{}) + response, err = conn.DeleteByQuery(query, []string{indexName}, []string{docType}, url.Values{}) + + // There's no delete by query in ES 2.x + if version > "2" && version < "5" { + c.Assert(err, ErrorMatches, ".* does not support delete by query") + return + } c.Assert(err, IsNil) @@ -616,8 +632,13 @@ func (s *GoesTestSuite) TestDeleteByQuery(c *C) { Version: 0, } response.Raw = nil + response.Shards = Shard{} + response.Took = 0 c.Assert(response, DeepEquals, expectedResponse) + _, err = conn.RefreshIndex(indexName) + c.Assert(err, IsNil) + //should be 0 docs after delete by query response, err = conn.Search(query, []string{indexName}, []string{docType}, url.Values{}) c.Assert(err, IsNil) @@ -634,6 +655,7 @@ func (s *GoesTestSuite) TestGet(c *C) { } conn := NewClient(ESHost, ESPort) + version, _ := conn.Version() conn.DeleteIndex(indexName) _, err := conn.CreateIndex(indexName, map[string]interface{}{}) @@ -666,11 +688,6 @@ func (s *GoesTestSuite) TestGet(c *C) { response.Raw = nil c.Assert(response, DeepEquals, expectedResponse) - fields := make(url.Values, 1) - fields.Set("fields", "f1") - response, err = conn.Get(indexName, docType, docID, fields) - c.Assert(err, IsNil) - expectedResponse = &Response{ Status: 200, Index: indexName, @@ -683,6 +700,18 @@ func (s *GoesTestSuite) TestGet(c *C) { }, } + fields := make(url.Values, 1) + // The fields param is no longer supported in ES 5.x + if version < "5" { + fields.Set("fields", "f1") + } else { + expectedResponse.Source = map[string]interface{}{"f1": "foo"} + expectedResponse.Fields = nil + fields.Set("_source", "f1") + } + response, err = conn.Get(indexName, docType, docID, fields) + c.Assert(err, IsNil) + response.Raw = nil c.Assert(response, DeepEquals, expectedResponse) } @@ -796,6 +825,12 @@ func (s *GoesTestSuite) TestCount(c *C) { func (s *GoesTestSuite) TestIndexStatus(c *C) { indexName := "testindexstatus" conn := NewClient(ESHost, ESPort) + + // _status endpoint was removed in ES 2.0 + if version, _ := conn.Version(); version > "2" { + return + } + conn.DeleteIndex(indexName) mapping := map[string]interface{}{ @@ -924,24 +959,43 @@ func (s *GoesTestSuite) TestScroll(c *C) { _, err = conn.RefreshIndex(indexName) c.Assert(err, IsNil) - query := map[string]interface{}{ - "query": map[string]interface{}{ - "filtered": map[string]interface{}{ - "filter": map[string]interface{}{ - "term": map[string]interface{}{ - "user": "foo", + var query map[string]interface{} + version, _ := conn.Version() + if version > "5" { + query = map[string]interface{}{ + "query": map[string]interface{}{ + "bool": map[string]interface{}{ + "filter": map[string]interface{}{ + "term": map[string]interface{}{ + "user": "foo", + }, }, }, }, - }, + } + } else { + query = map[string]interface{}{ + "query": map[string]interface{}{ + "filtered": map[string]interface{}{ + "filter": map[string]interface{}{ + "term": map[string]interface{}{ + "user": "foo", + }, + }, + }, + }, + } } - scan, err := conn.Scan(query, []string{indexName}, []string{docType}, "1m", 1) + searchResults, err := conn.Scan(query, []string{indexName}, []string{docType}, "1m", 1) c.Assert(err, IsNil) - c.Assert(len(scan.ScrollID) > 0, Equals, true) + c.Assert(len(searchResults.ScrollID) > 0, Equals, true) - searchResults, err := conn.Scroll(scan.ScrollID, "1m") - c.Assert(err, IsNil) + // Versions < 5.x don't include results in the initial response + if version < "5" { + searchResults, err = conn.Scroll(searchResults.ScrollID, "1m") + c.Assert(err, IsNil) + } // some data in first chunk c.Assert(searchResults.Hits.Total, Equals, uint64(2)) @@ -1013,6 +1067,16 @@ func (s *GoesTestSuite) TestAggregations(c *C) { "index.number_of_shards": 1, "index.number_of_replicas": 0, }, + "mappings": map[string]interface{}{ + docType: map[string]interface{}{ + "properties": map[string]interface{}{ + "user": map[string]interface{}{ + "type": "string", + "index": "not_analyzed", + }, + }, + }, + }, } defer conn.DeleteIndex(indexName) @@ -1178,23 +1242,44 @@ func (s *GoesTestSuite) TestUpdate(c *C) { } response.Raw = nil + response.Shards.Successful = 0 + response.Shards.Total = 0 c.Assert(response, DeepEquals, expectedResponse) // Now that we have an ordinary document indexed, try updating it - query := map[string]interface{}{ - "script": "ctx._source.counter += count", - "lang": "groovy", - "params": map[string]interface{}{ - "count": 5, - }, - "upsert": map[string]interface{}{ - "message": "candybar", - "user": "admin", - "counter": 1, - }, + var query map[string]interface{} + if version, _ := conn.Version(); version > "5" { + query = map[string]interface{}{ + "script": map[string]interface{}{ + "inline": "ctx._source.counter += params.count", + "lang": "painless", + "params": map[string]interface{}{ + "count": 5, + }, + }, + "upsert": map[string]interface{}{ + "message": "candybar", + "user": "admin", + "counter": 1, + }, + } + } else { + query = map[string]interface{}{ + "script": "ctx._source.counter += count", + "lang": "groovy", + "params": map[string]interface{}{ + "count": 5, + }, + "upsert": map[string]interface{}{ + "message": "candybar", + "user": "admin", + "counter": 1, + }, + } } response, err = conn.Update(d, query, extraArgs) + if err != nil && strings.Contains(err.(*SearchError).Msg, "dynamic scripting") { c.Skip("Scripting is disabled on server, skipping this test") return @@ -1298,6 +1383,10 @@ func (s *GoesTestSuite) TestDeleteMapping(c *C) { time.Sleep(200 * time.Millisecond) response, err = conn.DeleteMapping("tweet", []string{indexName}) + if version, _ := conn.Version(); version > "2" { + c.Assert(err, ErrorMatches, ".*not supported.*") + return + } c.Assert(err, IsNil) c.Assert(response.Acknowledged, Equals, true) @@ -1392,7 +1481,7 @@ func (s *GoesTestSuite) TestRemoveAlias(c *C) { // Get document via alias _, err = conn.Get(aliasName, docType, docID, url.Values{}) - c.Assert(err.Error(), Equals, "[404] IndexMissingException[["+aliasName+"] missing]") + c.Assert(err.Error(), Matches, "\\[404\\] .*"+aliasName+".*") } func (s *GoesTestSuite) TestAliasExists(c *C) { diff --git a/request.go b/request.go index bf8093d..9886865 100644 --- a/request.go +++ b/request.go @@ -82,7 +82,7 @@ func (req *Request) Request() (*http.Request, error) { postData = req.Body } else if req.API == "_bulk" { postData = req.BulkData - } else { + } else if req.Query != nil { b, err := json.Marshal(req.Query) if err != nil { return nil, err @@ -90,14 +90,12 @@ func (req *Request) Request() (*http.Request, error) { postData = b } - reader := ioutil.NopCloser(bytes.NewReader(postData)) - newReq, err := http.NewRequest(req.Method, "", nil) if err != nil { return nil, err } newReq.URL = req.URL() - newReq.Body = reader + newReq.Body = ioutil.NopCloser(bytes.NewReader(postData)) newReq.ContentLength = int64(len(postData)) if req.Method == "POST" || req.Method == "PUT" { diff --git a/structs.go b/structs.go index ca08edd..75cdc5d 100644 --- a/structs.go +++ b/structs.go @@ -20,6 +20,9 @@ type Client struct { // Client is the http client used to make requests, allowing settings things // such as timeouts etc Client *http.Client + + // Detected version of ES + version string } // Response holds an elasticsearch response