Merge pull request #9 from OwnLocal/es_2_and_5

ES 2 and 5

.travis.yml

@@ -1,27 +1,31 @@
 language: go
 
+addons:
+  apt:
+    packages:
+      - oracle-java8-set-default
+
 go:
-  - 1.5.4
+  - 1.6.4
-  - 1.6.3
+  - 1.7.5
-  - 1.7.1
 
 env:
   global:
-    - GO15VENDOREXPERIMENT=1
+    - JAVA_HOME=/usr/lib/jvm/java-8-oracle
   matrix:
-    - ES_VERSION=1.3.4
+    - ES_VERSION=1.7.5 ES_URL=https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.7.5.tar.gz
-    - ES_VERSION=1.4.4
+    - 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=1.5.2
+    - ES_VERSION=5.2.0 ES_URL=https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.2.0.tar.gz
-    - ES_VERSION=1.6.0
-    - ES_VERSION=1.7.0
 
 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"
+  - "echo 'script.inline: true' >> ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/config/elasticsearch.yml"
-  - ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch >/dev/null &
+  - ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch &
-  - sleep 10 # Wait for ES to start up
+  - wget --retry-connrefused http://127.0.0.1:9200/ # Wait for ES to start up
 
 install:
   - go get github.com/Masterminds/glide

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{}
+	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),
+
+	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"
+	c.replaceHost(req)
-	req.URL.Host = fmt.Sprintf("%s:%s", c.Host, c.Port)
 
 	body, statusCode, err := c.doRequest(req)
 	esResp := &Response{Status: statusCode}

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)
+	response, err := conn.Index(d, nil)
-	extraArgs.Set("ttl", "86400000")
-	response, err := conn.Index(d, extraArgs)
 	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)
+	_, err = conn.Index(d, nil)
-	extraArgs.Set("ttl", "86400000")
-	_, err = conn.Index(d, extraArgs)
 	c.Assert(err, Not(IsNil))
 }
 
@@ -455,9 +462,7 @@ func (s *GoesTestSuite) TestIndexIdDefined(c *C) {
 		},
 	}
 
-	extraArgs := make(url.Values, 1)
+	response, err := conn.Index(d, nil)
-	extraArgs.Set("ttl", "86400000")
-	response, err := conn.Index(d, extraArgs)
 	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,7 +959,22 @@ func (s *GoesTestSuite) TestScroll(c *C) {
 	_, err = conn.RefreshIndex(indexName)
 	c.Assert(err, IsNil)
 
-	query := map[string]interface{}{
+	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{}{
@@ -935,13 +985,17 @@ func (s *GoesTestSuite) TestScroll(c *C) {
 				},
 			},
 		}
+	}
 
-	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")
+	// 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,10 +1242,29 @@ 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{}{
+	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{}{
@@ -1193,8 +1276,10 @@ func (s *GoesTestSuite) TestUpdate(c *C) {
 				"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) {

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" {

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