Compare commits

..

No commits in common. "fbfb1d80a89aea5c028d3d1ff1bb7540f2a54792" and "6c9647b81cbf8daf85a01b1c15a7cc23de351c27" have entirely different histories.

12 changed files with 526 additions and 805 deletions

1
.gitignore vendored
View File

@ -1,3 +1,2 @@
*.test *.test
*.swp *.swp
vendor

View File

@ -1,34 +1,35 @@
language: go language: go
addons:
apt:
packages:
- oracle-java8-set-default
go: go:
- 1.6.4 - 1.1
- 1.7.5 - 1.2
- 1.3
- 1.4.2
- 1.4.3
- 1.5
- 1.5.1
env: env:
global:
- JAVA_HOME=/usr/lib/jvm/java-8-oracle
matrix: matrix:
- ES_VERSION=1.7.5 ES_URL=https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.7.5.tar.gz - ES_VERSION=1.0.3 GROOVY_VER=2.0.0
- 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.1.2 GROOVY_VER=2.0.0
- ES_VERSION=5.2.0 ES_URL=https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.2.0.tar.gz - ES_VERSION=1.2.1 GROOVY_VER=2.2.0
- 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
before_script: before_script:
- java -version
- echo $JAVA_HOME
- mkdir ${HOME}/elasticsearch - mkdir ${HOME}/elasticsearch
- wget $ES_URL - wget https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-${ES_VERSION}.tar.gz
- tar -xzf elasticsearch-${ES_VERSION}.tar.gz -C ${HOME}/elasticsearch - tar -xzf elasticsearch-${ES_VERSION}.tar.gz -C ${HOME}/elasticsearch
- "echo 'script.inline: true' >> ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/config/elasticsearch.yml" - "echo 'script.groovy.sandbox.enabled: true' >> ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/config/elasticsearch.yml"
- ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch & - 'if [[ "${ES_VERSION}" < "1.3" ]]; then ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/plugin --install elasticsearch/elasticsearch-lang-groovy/${GROOVY_VER}; fi'
- wget --retry-connrefused http://127.0.0.1:9200/ # Wait for ES to start up - ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch >/dev/null &
install: install:
- go get github.com/Masterminds/glide - go get gopkg.in/check.v1
script: script:
- make test - make test

View File

@ -1,15 +1,10 @@
help: help:
@echo "Available targets:" @echo "Available targets:"
@echo "- test: run tests" @echo "- test: run tests"
@echo "- deps: installs dependencies with glide" @echo "- installdependencies: installs dependencies declared in dependencies.txt"
@echo "- watch: watch for changes and re-run tests"
deps: installdependencies:
glide install && mkdir -p vendor/bin && go build -o vendor/bin/ginkgo ./vendor/github.com/onsi/ginkgo/ginkgo cat dependencies.txt | xargs go get
test: installdependencies
test: deps go test -i && go test
vendor/bin/ginkgo -race -randomizeAllSpecs -r -skipPackage vendor -progress .
watch: deps
vendor/bin/ginkgo watch -race -randomizeAllSpecs -r -skipPackage vendor -progress -notify .

View File

@ -1,10 +1,3 @@
> **Note**: If you are switching to `OwnLocal/goes` from `belogik/goes` you will want to point whatever
> dependency management system you are using at the `v1.0.0` tag to maintain backward compatibility
> (or change your code to support any API changes which may exist). The `master` branch will contain
> whatever the latest released version is and the `develop` branch will contain the currently
> in-progress version. Either of these branches may contain backwards-incompatible changes from the
> `v1.0.0` tag (which contains the code as it was when the code was forked from `belogik/goes`).
Goes : a library to interact with ElasticSearch Goes : a library to interact with ElasticSearch
=============================================== ===============================================

View File

@ -19,7 +19,7 @@ var (
ES_PORT = "9200" ES_PORT = "9200"
) )
func getClient() (conn *goes.Client) { func getConnection() (conn *goes.Connection) {
h := os.Getenv("TEST_ELASTICSEARCH_HOST") h := os.Getenv("TEST_ELASTICSEARCH_HOST")
if h == "" { if h == "" {
h = ES_HOST h = ES_HOST
@ -30,13 +30,13 @@ func getClient() (conn *goes.Client) {
p = ES_PORT p = ES_PORT
} }
conn = goes.NewClient(h, p) conn = goes.NewConnection(h, p)
return return
} }
func ExampleClient_CreateIndex() { func ExampleConnection_CreateIndex() {
conn := getClient() conn := getConnection()
mapping := map[string]interface{}{ mapping := map[string]interface{}{
"settings": map[string]interface{}{ "settings": map[string]interface{}{
@ -64,8 +64,8 @@ func ExampleClient_CreateIndex() {
fmt.Printf("%s", resp) fmt.Printf("%s", resp)
} }
func ExampleClient_DeleteIndex() { func ExampleConnection_DeleteIndex() {
conn := getClient() conn := getConnection()
resp, err := conn.DeleteIndex("yourinde") resp, err := conn.DeleteIndex("yourinde")
if err != nil { if err != nil {
@ -75,8 +75,8 @@ func ExampleClient_DeleteIndex() {
fmt.Printf("%s", resp) fmt.Printf("%s", resp)
} }
func ExampleClient_RefreshIndex() { func ExampleConnection_RefreshIndex() {
conn := getClient() conn := getConnection()
resp, err := conn.RefreshIndex("yourindex") resp, err := conn.RefreshIndex("yourindex")
if err != nil { if err != nil {
@ -86,8 +86,8 @@ func ExampleClient_RefreshIndex() {
fmt.Printf("%s", resp) fmt.Printf("%s", resp)
} }
func ExampleClient_Search() { func ExampleConnection_Search() {
conn := getClient() conn := getConnection()
var query = map[string]interface{}{ var query = map[string]interface{}{
"query": map[string]interface{}{ "query": map[string]interface{}{
@ -123,8 +123,8 @@ func ExampleClient_Search() {
fmt.Printf("%s", searchResults) fmt.Printf("%s", searchResults)
} }
func ExampleClient_Index() { func ExampleConnection_Index() {
conn := getClient() conn := getConnection()
d := goes.Document{ d := goes.Document{
Index: "twitter", Index: "twitter",
@ -147,15 +147,15 @@ func ExampleClient_Index() {
fmt.Printf("%s", response) fmt.Printf("%s", response)
} }
func ExampleClient_Delete() { func ExampleConnection_Delete() {
conn := getClient() conn := getConnection()
//[create index, index document ...] //[create index, index document ...]
d := goes.Document{ d := goes.Document{
Index: "twitter", Index: "twitter",
Type: "tweet", Type: "tweet",
ID: "1", Id: "1",
Fields: map[string]interface{}{ Fields: map[string]interface{}{
"user": "foo", "user": "foo",
}, },
@ -169,15 +169,15 @@ func ExampleClient_Delete() {
fmt.Printf("%s", response) fmt.Printf("%s", response)
} }
func ExampleClient_WithHTTPClient() { func ExampleConnectionOverrideHttpClient() {
tr := &http.Transport{ tr := &http.Transport{
ResponseHeaderTimeout: 1 * time.Second, ResponseHeaderTimeout: 1 * time.Second,
} }
cl := &http.Client{ cl := &http.Client{
Transport: tr, Transport: tr,
} }
conn := getClient() conn := getConnection()
conn.WithHTTPClient(cl) conn.WithClient(cl)
fmt.Printf("%v\n", conn.Client) fmt.Printf("%v\n", conn.Client)
} }

12
glide.lock generated
View File

@ -1,12 +0,0 @@
hash: 02db47097959405b1a7e0e1e583c6fbb11c7236c450264909a4e9ac690ef4d47
updated: 2016-09-27T11:54:53.427061181-05:00
imports: []
testImports:
- name: github.com/go-check/check
version: 4f90aeace3a26ad7021961c297b22c42160c7b25
- name: github.com/onsi/ginkgo
version: 462326b1628e124b23f42e87a8f2750e3c4e2d24
subpackages:
- ginkgo
- name: github.com/onsi/gomega
version: a78ae492d53aad5a7a232d0d0462c14c400e3ee7

View File

@ -1,11 +0,0 @@
package: github.com/OwnLocal/goes
import: []
testImport:
- package: github.com/go-check/check
version: v1
- package: github.com/onsi/ginkgo
version: ^1.2.0
subpackages:
- ginkgo
- package: github.com/onsi/gomega
version: ^1.0.0

550
goes.go
View File

@ -19,144 +19,118 @@ import (
) )
const ( const (
// BulkCommandIndex specifies a bulk doc should be indexed BULK_COMMAND_INDEX = "index"
BulkCommandIndex = "index" BULK_COMMAND_DELETE = "delete"
// BulkCommandDelete specifies a bulk doc should be deleted
BulkCommandDelete = "delete"
) )
func (err *SearchError) Error() string { func (err *SearchError) Error() string {
return fmt.Sprintf("[%d] %s", err.StatusCode, err.Msg) return fmt.Sprintf("[%d] %s", err.StatusCode, err.Msg)
} }
// NewClient initiates a new client for an elasticsearch server // NewConnection initiates a new Connection to an elasticsearch server
// //
// This function is pretty useless for now but might be useful in a near future // 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. // if wee need more features like connection pooling or load balancing.
func NewClient(host string, port string) *Client { func NewConnection(host string, port string) *Connection {
return &Client{host, port, http.DefaultClient, "", "", ""} return &Connection{host, port, http.DefaultClient}
} }
// WithHTTPClient sets the http.Client to be used with the connection. Returns the original client. func (c *Connection) WithClient(cl *http.Client) *Connection {
func (c *Client) WithHTTPClient(cl *http.Client) *Client {
c.Client = cl c.Client = cl
return c 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 // CreateIndex creates a new index represented by a name and a mapping
func (c *Client) CreateIndex(name string, mapping interface{}) (*Response, error) { func (c *Connection) CreateIndex(name string, mapping interface{}) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: mapping, Query: mapping,
IndexList: []string{name}, IndexList: []string{name},
Method: "PUT", method: "PUT",
} }
return c.Do(&r) return r.Run()
} }
// DeleteIndex deletes an index represented by a name // DeleteIndex deletes an index represented by a name
func (c *Client) DeleteIndex(name string) (*Response, error) { func (c *Connection) DeleteIndex(name string) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: []string{name}, IndexList: []string{name},
Method: "DELETE", method: "DELETE",
} }
return c.Do(&r) return r.Run()
} }
// RefreshIndex refreshes an index represented by a name // RefreshIndex refreshes an index represented by a name
func (c *Client) RefreshIndex(name string) (*Response, error) { func (c *Connection) RefreshIndex(name string) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: []string{name}, IndexList: []string{name},
Method: "POST", method: "POST",
API: "_refresh", api: "_refresh",
} }
return c.Do(&r) return r.Run()
} }
// UpdateIndexSettings updates settings for existing index represented by a name and a settings // 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 // as described here: https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-update-settings.html
func (c *Client) UpdateIndexSettings(name string, settings interface{}) (*Response, error) { func (c *Connection) UpdateIndexSettings(name string, settings interface{}) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: settings, Query: settings,
IndexList: []string{name}, IndexList: []string{name},
Method: "PUT", method: "PUT",
API: "_settings", api: "_settings",
} }
return c.Do(&r) return r.Run()
} }
// Optimize an index represented by a name, extra args are also allowed please check: // 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 // http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-optimize.html#indices-optimize
func (c *Client) Optimize(indexList []string, extraArgs url.Values) (*Response, error) { func (c *Connection) Optimize(indexList []string, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: indexList, IndexList: indexList,
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
Method: "POST", method: "POST",
API: "_optimize", api: "_optimize",
}
if version, _ := c.Version(); version > "2.1" {
r.API = "_forcemerge"
} }
return c.Do(&r) return r.Run()
}
// 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 // Stats fetches statistics (_stats) for the current elasticsearch server
func (c *Client) Stats(indexList []string, extraArgs url.Values) (*Response, error) { func (c *Connection) Stats(indexList []string, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: indexList, IndexList: indexList,
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
Method: "GET", method: "GET",
API: "_stats", api: "_stats",
} }
return c.Do(&r) return r.Run()
} }
// IndexStatus fetches the status (_status) for the indices defined in // IndexStatus fetches the status (_status) for the indices defined in
// indexList. Use _all in indexList to get stats for all indices // indexList. Use _all in indexList to get stats for all indices
func (c *Client) IndexStatus(indexList []string) (*Response, error) { func (c *Connection) IndexStatus(indexList []string) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: indexList, IndexList: indexList,
Method: "GET", method: "GET",
API: "_status", api: "_status",
} }
return c.Do(&r) return r.Run()
} }
// BulkSend bulk adds multiple documents in bulk mode // Bulk adds multiple documents in bulk mode
func (c *Client) BulkSend(documents []Document) (*Response, error) { func (c *Connection) BulkSend(documents []Document) (*Response, error) {
// We do not generate a traditional JSON here (often a one liner) // We do not generate a traditional JSON here (often a one liner)
// Elasticsearch expects one line of JSON per line (EOL = \n) // Elasticsearch expects one line of JSON per line (EOL = \n)
// plus an extra \n at the very end of the document // plus an extra \n at the very end of the document
@ -175,7 +149,7 @@ func (c *Client) BulkSend(documents []Document) (*Response, error) {
// len(documents) * 2 : action + optional_sources // len(documents) * 2 : action + optional_sources
// + 1 : room for the trailing \n // + 1 : room for the trailing \n
bulkData := make([][]byte, 0, len(documents)*2+1) bulkData := make([][]byte, len(documents)*2+1)
i := 0 i := 0
for _, doc := range documents { for _, doc := range documents {
@ -183,7 +157,7 @@ func (c *Client) BulkSend(documents []Document) (*Response, error) {
doc.BulkCommand: map[string]interface{}{ doc.BulkCommand: map[string]interface{}{
"_index": doc.Index, "_index": doc.Index,
"_type": doc.Type, "_type": doc.Type,
"_id": doc.ID, "_id": doc.Id,
}, },
}) })
@ -191,7 +165,7 @@ func (c *Client) BulkSend(documents []Document) (*Response, error) {
return &Response{}, err return &Response{}, err
} }
bulkData = append(bulkData, action) bulkData[i] = action
i++ i++
if doc.Fields != nil { if doc.Fields != nil {
@ -217,211 +191,265 @@ func (c *Client) BulkSend(documents []Document) (*Response, error) {
return &Response{}, err return &Response{}, err
} }
bulkData = append(bulkData, sources) bulkData[i] = sources
i++ i++
} }
} }
// forces an extra trailing \n absolutely necessary for elasticsearch // forces an extra trailing \n absolutely necessary for elasticsearch
bulkData = append(bulkData, []byte(nil)) bulkData[len(bulkData)-1] = []byte(nil)
r := Request{ r := Request{
Method: "POST", Conn: c,
API: "_bulk", method: "POST",
BulkData: bytes.Join(bulkData, []byte("\n")), api: "_bulk",
bulkData: bytes.Join(bulkData, []byte("\n")),
} }
resp, err := c.Do(&r) return r.Run()
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
} }
// Search executes a search query against an index // Search executes a search query against an index
func (c *Client) Search(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) { func (c *Connection) Search(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: query, Query: query,
IndexList: indexList, IndexList: indexList,
TypeList: typeList, TypeList: typeList,
Method: "POST", method: "POST",
API: "_search", api: "_search",
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
} }
return c.Do(&r) return r.Run()
} }
// Count executes a count query against an index, use the Count field in the response for the result // Count executes a count query against an index, use the Count field in the response for the result
func (c *Client) Count(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) { func (c *Connection) Count(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: query, Query: query,
IndexList: indexList, IndexList: indexList,
TypeList: typeList, TypeList: typeList,
Method: "POST", method: "POST",
API: "_count", api: "_count",
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
} }
return c.Do(&r) return r.Run()
} }
//Query runs a query against an index using the provided http method. //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" //This method can be used to execute a delete by query, just pass in "DELETE"
//for the HTTP method. //for the HTTP method.
func (c *Client) Query(query interface{}, indexList []string, typeList []string, httpMethod string, extraArgs url.Values) (*Response, error) { func (c *Connection) Query(query interface{}, indexList []string, typeList []string, httpMethod string, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: query, Query: query,
IndexList: indexList, IndexList: indexList,
TypeList: typeList, TypeList: typeList,
Method: httpMethod, method: httpMethod,
API: "_query", api: "_query",
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
} }
return c.Do(&r) return r.Run()
} }
// DeleteByQuery deletes documents matching the specified query. It will return an error for ES 2.x, // Scan starts scroll over an index
// because delete by query support was removed in those versions. func (c *Connection) Scan(query interface{}, indexList []string, typeList []string, timeout string, size int) (*Response, error) {
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 := 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("search_type", "scan")
}
v.Add("scroll", timeout) v.Add("scroll", timeout)
v.Add("size", strconv.Itoa(size)) v.Add("size", strconv.Itoa(size))
r := Request{ r := Request{
Conn: c,
Query: query, Query: query,
IndexList: indexList, IndexList: indexList,
TypeList: typeList, TypeList: typeList,
Method: "POST", method: "POST",
API: "_search", api: "_search",
ExtraArgs: v, ExtraArgs: v,
} }
return c.Do(&r) return r.Run()
} }
// Scroll fetches data by scroll id // Scroll fetches data by scroll id
func (c *Client) Scroll(scrollID string, timeout string) (*Response, error) { func (c *Connection) Scroll(scrollId string, timeout string) (*Response, error) {
r := Request{
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 := url.Values{}
v.Add("scroll", timeout) v.Add("scroll", timeout)
v.Add("scroll_id", scrollID)
r.ExtraArgs = v r := Request{
Conn: c,
method: "POST",
api: "_search/scroll",
ExtraArgs: v,
Body: []byte(scrollId),
} }
return c.Do(&r) return r.Run()
} }
// Get a typed document by its id // Get a typed document by its id
func (c *Client) Get(index string, documentType string, id string, extraArgs url.Values) (*Response, error) { func (c *Connection) Get(index string, documentType string, id string, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: []string{index}, IndexList: []string{index},
Method: "GET", method: "GET",
API: documentType + "/" + id, api: documentType + "/" + id,
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
} }
return c.Do(&r) return r.Run()
} }
// Index indexes a Document // Index indexes a Document
// The extraArgs is a list of url.Values that you can send to elasticsearch as // 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. // URL arguments, for example, to control routing, ttl, version, op_type, etc.
func (c *Client) Index(d Document, extraArgs url.Values) (*Response, error) { func (c *Connection) Index(d Document, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: d.Fields, Query: d.Fields,
IndexList: []string{d.Index.(string)}, IndexList: []string{d.Index.(string)},
TypeList: []string{d.Type}, TypeList: []string{d.Type},
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
Method: "POST", method: "POST",
} }
if d.ID != nil { if d.Id != nil {
r.Method = "PUT" r.method = "PUT"
r.ID = d.ID.(string) r.id = d.Id.(string)
} }
return c.Do(&r) return r.Run()
} }
// Delete deletes a Document d // Delete deletes a Document d
// The extraArgs is a list of url.Values that you can send to elasticsearch as // The extraArgs is a list of url.Values that you can send to elasticsearch as
// URL arguments, for example, to control routing. // URL arguments, for example, to control routing.
func (c *Client) Delete(d Document, extraArgs url.Values) (*Response, error) { func (c *Connection) Delete(d Document, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: []string{d.Index.(string)}, IndexList: []string{d.Index.(string)},
TypeList: []string{d.Type}, TypeList: []string{d.Type},
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
Method: "DELETE", method: "DELETE",
ID: d.ID.(string), id: d.Id.(string),
} }
return c.Do(&r) return r.Run()
}
// Run executes an elasticsearch Request. It converts data to Json, sends the
// request and returns the Response obtained
func (req *Request) Run() (*Response, error) {
body, statusCode, err := req.run()
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 req.api == "_bulk" && esResp.Errors {
for _, item := range esResp.Items {
for _, i := range item {
if i.Error != "" {
return esResp, &SearchError{i.Error, i.Status}
}
}
}
return esResp, &SearchError{Msg: "Unknown error while bulk indexing"}
}
if esResp.Error != "" {
return esResp, &SearchError{esResp.Error, esResp.Status}
}
return esResp, nil
}
func (req *Request) run() ([]byte, uint64, error) {
postData := []byte{}
// XXX : refactor this
if len(req.Body) > 0 {
postData = req.Body
} else if req.api == "_bulk" {
postData = req.bulkData
} else {
b, err := json.Marshal(req.Query)
if err != nil {
return nil, 0, err
}
postData = b
}
reader := bytes.NewReader(postData)
newReq, err := http.NewRequest(req.method, req.Url(), reader)
if err != nil {
return nil, 0, err
}
if req.method == "POST" || req.method == "PUT" {
newReq.Header.Set("Content-Type", "application/json")
}
resp, err := req.Conn.Client.Do(newReq)
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
}
// Url builds a Request for a URL
func (r *Request) Url() string {
path := "/" + strings.Join(r.IndexList, ",")
if len(r.TypeList) > 0 {
path += "/" + strings.Join(r.TypeList, ",")
}
// XXX : for indexing documents using the normal (non bulk) API
if len(r.id) > 0 {
path += "/" + r.id
}
path += "/" + r.api
u := url.URL{
Scheme: "http",
Host: fmt.Sprintf("%s:%s", r.Conn.Host, r.Conn.Port),
Path: path,
RawQuery: r.ExtraArgs.Encode(),
}
return u.String()
} }
// Buckets returns list of buckets in aggregation // Buckets returns list of buckets in aggregation
@ -450,86 +478,85 @@ func (b Bucket) DocCount() uint64 {
func (b Bucket) Aggregation(name string) Aggregation { func (b Bucket) Aggregation(name string) Aggregation {
if agg, ok := b[name]; ok { if agg, ok := b[name]; ok {
return agg.(map[string]interface{}) return agg.(map[string]interface{})
} } else {
return Aggregation{} return Aggregation{}
} }
}
// PutMapping registers a specific mapping for one or more types in one or more indexes // PutMapping registers a specific mapping for one or more types in one or more indexes
func (c *Client) PutMapping(typeName string, mapping interface{}, indexes []string) (*Response, error) { func (c *Connection) PutMapping(typeName string, mapping interface{}, indexes []string) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: mapping, Query: mapping,
IndexList: indexes, IndexList: indexes,
Method: "PUT", method: "PUT",
API: "_mappings/" + typeName, api: "_mappings/" + typeName,
} }
return c.Do(&r) return r.Run()
} }
// GetMapping returns the mappings for the specified types func (c *Connection) GetMapping(types []string, indexes []string) (*Response, error) {
func (c *Client) GetMapping(types []string, indexes []string) (*Response, error) {
r := Request{ r := Request{
Conn: c,
IndexList: indexes, IndexList: indexes,
Method: "GET", method: "GET",
API: "_mapping/" + strings.Join(types, ","), api: "_mapping/" + strings.Join(types, ","),
} }
return c.Do(&r) return r.Run()
} }
// IndicesExist checks whether index (or indices) exist on the server // IndicesExist checks whether index (or indices) exist on the server
func (c *Client) IndicesExist(indexes []string) (bool, error) { func (c *Connection) IndicesExist(indexes []string) (bool, error) {
r := Request{ r := Request{
Conn: c,
IndexList: indexes, IndexList: indexes,
Method: "HEAD", method: "HEAD",
} }
resp, err := c.Do(&r) resp, err := r.Run()
return resp.Status == 200, err return resp.Status == 200, err
} }
// Update updates the specified document using the _update endpoint func (c *Connection) Update(d Document, query interface{}, extraArgs url.Values) (*Response, error) {
func (c *Client) Update(d Document, query interface{}, extraArgs url.Values) (*Response, error) {
r := Request{ r := Request{
Conn: c,
Query: query, Query: query,
IndexList: []string{d.Index.(string)}, IndexList: []string{d.Index.(string)},
TypeList: []string{d.Type}, TypeList: []string{d.Type},
ExtraArgs: extraArgs, ExtraArgs: extraArgs,
Method: "POST", method: "POST",
API: "_update", api: "_update",
} }
if d.ID != nil { if d.Id != nil {
r.ID = d.ID.(string) r.id = d.Id.(string)
} }
return c.Do(&r) return r.Run()
} }
// DeleteMapping deletes a mapping along with all data in the type // DeleteMapping deletes a mapping along with all data in the type
func (c *Client) DeleteMapping(typeName string, indexes []string) (*Response, error) { func (c *Connection) 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{ r := Request{
Conn: c,
IndexList: indexes, IndexList: indexes,
Method: "DELETE", method: "DELETE",
API: "_mappings/" + typeName, api: "_mappings/" + typeName,
} }
return c.Do(&r) return r.Run()
} }
func (c *Client) modifyAlias(action string, alias string, indexes []string) (*Response, error) { func (c *Connection) modifyAlias(action string, alias string, indexes []string) (*Response, error) {
command := map[string]interface{}{ command := map[string]interface{}{
"actions": make([]map[string]interface{}, 0, 1), "actions": make([]map[string]interface{}, 1),
} }
for _, index := range indexes { for _, index := range indexes {
@ -542,116 +569,35 @@ func (c *Client) modifyAlias(action string, alias string, indexes []string) (*Re
} }
r := Request{ r := Request{
Conn: c,
Query: command, Query: command,
Method: "POST", method: "POST",
API: "_aliases", api: "_aliases",
} }
return c.Do(&r) return r.Run()
} }
// AddAlias creates an alias to one or more indexes // AddAlias creates an alias to one or more indexes
func (c *Client) AddAlias(alias string, indexes []string) (*Response, error) { func (c *Connection) AddAlias(alias string, indexes []string) (*Response, error) {
return c.modifyAlias("add", alias, indexes) return c.modifyAlias("add", alias, indexes)
} }
// RemoveAlias removes an alias to one or more indexes // RemoveAlias removes an alias to one or more indexes
func (c *Client) RemoveAlias(alias string, indexes []string) (*Response, error) { func (c *Connection) RemoveAlias(alias string, indexes []string) (*Response, error) {
return c.modifyAlias("remove", alias, indexes) return c.modifyAlias("remove", alias, indexes)
} }
// AliasExists checks whether alias is defined on the server // AliasExists checks whether alias is defined on the server
func (c *Client) AliasExists(alias string) (bool, error) { func (c *Connection) AliasExists(alias string) (bool, error) {
r := Request{ r := Request{
Method: "HEAD", Conn: c,
API: "_alias/" + alias, method: "HEAD",
api: "_alias/" + alias,
} }
resp, err := c.Do(&r) resp, err := r.Run()
return resp.Status == 200, err 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)
if c.AuthUsername != "" {
req.SetBasicAuth(c.AuthUsername, c.AuthPassword)
}
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
}
c.replaceHost(req)
if c.AuthUsername != "" {
req.SetBasicAuth(c.AuthUsername, c.AuthPassword)
}
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
}

View File

@ -1,13 +0,0 @@
package goes_test
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"testing"
)
func TestGoes(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "Goes Suite")
}

File diff suppressed because it is too large Load Diff

View File

@ -1,113 +0,0 @@
package goes
import (
"bytes"
"encoding/json"
"io/ioutil"
"net/http"
"net/url"
"strings"
)
// Requester implements Request which builds an HTTP request for Elasticsearch
type Requester interface {
// Request should set the URL and Body (if needed). The host of the URL will be overwritten by the client.
Request() (*http.Request, error)
}
// Request holds a single request to elasticsearch
type Request struct {
// A search query
Query interface{}
// Which index to search into
IndexList []string
// Which type to search into
TypeList []string
// HTTP Method to user (GET, POST ...)
Method string
// Which api keyword (_search, _bulk, etc) to use
API string
// Bulk data
BulkData []byte
// Request body
Body []byte
// A list of extra URL arguments
ExtraArgs url.Values
// Used for the id field when indexing a document
ID string
// Auth username
AuthUsername string
// Auth password
AuthPassword string
}
// URL builds a URL for a Request
func (req *Request) URL() *url.URL {
var path string
if len(req.IndexList) > 0 {
path = "/" + strings.Join(req.IndexList, ",")
}
if len(req.TypeList) > 0 {
path += "/" + strings.Join(req.TypeList, ",")
}
// XXX : for indexing documents using the normal (non bulk) API
if len(req.ID) > 0 {
path += "/" + req.ID
}
path += "/" + req.API
u := url.URL{
//Scheme: "http",
//Host: fmt.Sprintf("%s:%s", req.Conn.Host, req.Conn.Port),
Path: path,
RawQuery: req.ExtraArgs.Encode(),
}
return &u
}
// Request generates an http.Request based on the contents of the Request struct
func (req *Request) Request() (*http.Request, error) {
postData := []byte{}
// XXX : refactor this
if len(req.Body) > 0 {
postData = req.Body
} else if req.API == "_bulk" {
postData = req.BulkData
} else if req.Query != nil {
b, err := json.Marshal(req.Query)
if err != nil {
return nil, err
}
postData = b
}
newReq, err := http.NewRequest(req.Method, "", nil)
if err != nil {
return nil, err
}
newReq.URL = req.URL()
newReq.Body = ioutil.NopCloser(bytes.NewReader(postData))
newReq.ContentLength = int64(len(postData))
if req.Method == "POST" || req.Method == "PUT" {
newReq.Header.Set("Content-Type", "application/json")
}
return newReq, nil
}
var _ Requester = (*Request)(nil)

View File

@ -5,12 +5,12 @@
package goes package goes
import ( import (
"encoding/json"
"net/http" "net/http"
"net/url"
) )
// Client represents a connection to elasticsearch // Represents a Connection object to elasticsearch
type Client struct { type Connection struct {
// The host to connect to // The host to connect to
Host string Host string
@ -20,22 +20,45 @@ type Client struct {
// Client is the http client used to make requests, allowing settings things // Client is the http client used to make requests, allowing settings things
// such as timeouts etc // such as timeouts etc
Client *http.Client Client *http.Client
// Detected version of ES
version string
// user name for http basic auth
AuthUsername string `json:"username"`
// pass word for http basic auth
AuthPassword string `json:"password"`
} }
// Response holds an elasticsearch response // Represents a Request to elasticsearch
type Request struct {
// Which connection will be used
Conn *Connection
// A search query
Query interface{}
// Which index to search into
IndexList []string
// Which type to search into
TypeList []string
// HTTP Method to user (GET, POST ...)
method string
// Which api keyword (_search, _bulk, etc) to use
api string
// Bulk data
bulkData []byte
// Request body
Body []byte
// A list of extra URL arguments
ExtraArgs url.Values
// Used for the id field when indexing a document
id string
}
// Represents a Response from elasticsearch
type Response struct { type Response struct {
Acknowledged bool Acknowledged bool
Error string Error string
RawError json.RawMessage `json:"error"`
Errors bool Errors bool
Status uint64 Status uint64
Took uint64 Took uint64
@ -43,7 +66,7 @@ type Response struct {
Shards Shard `json:"_shards"` Shards Shard `json:"_shards"`
Hits Hits Hits Hits
Index string `json:"_index"` Index string `json:"_index"`
ID string `json:"_id"` Id string `json:"_id"`
Type string `json:"_type"` Type string `json:"_type"`
Version int `json:"_version"` Version int `json:"_version"`
Found bool Found bool
@ -63,77 +86,75 @@ type Response struct {
Indices map[string]IndexStatus Indices map[string]IndexStatus
// Scroll id for iteration // Scroll id for iteration
ScrollID string `json:"_scroll_id"` ScrollId string `json:"_scroll_id"`
Aggregations map[string]Aggregation `json:"aggregations,omitempty"` Aggregations map[string]Aggregation `json:"aggregations,omitempty"`
Raw map[string]interface{} Raw map[string]interface{}
} }
// Aggregation holds the aggregation portion of an ES response // Represents an aggregation from response
type Aggregation map[string]interface{} type Aggregation map[string]interface{}
// Bucket represents a bucket for aggregation // Represents a bucket for aggregation
type Bucket map[string]interface{} type Bucket map[string]interface{}
// Document holds a document to send to elasticsearch // Represents a document to send to elasticsearch
type Document struct { type Document struct {
// XXX : interface as we can support nil values // XXX : interface as we can support nil values
Index interface{} Index interface{}
Type string Type string
ID interface{} Id interface{}
BulkCommand string BulkCommand string
Fields interface{} Fields interface{}
} }
// Item holds an item from the "items" field in a _bulk response // Represents the "items" field in a _bulk response
type Item struct { type Item struct {
Type string `json:"_type"` Type string `json:"_type"`
ID string `json:"_id"` Id string `json:"_id"`
Index string `json:"_index"` Index string `json:"_index"`
Version int `json:"_version"` Version int `json:"_version"`
Error string `json:"error"` Error string `json:"error"`
Status uint64 `json:"status"` Status uint64 `json:"status"`
} }
// All represents the "_all" field when calling the _stats API // Represents the "_all" field when calling the _stats API
// This is minimal but this is what I only need // This is minimal but this is what I only need
type All struct { type All struct {
Indices map[string]StatIndex `json:"indices"` Indices map[string]StatIndex `json:"indices"`
Primaries map[string]StatPrimary `json:"primaries"` Primaries map[string]StatPrimary `json:"primaries"`
} }
// StatIndex contains stats for a specific index
type StatIndex struct { type StatIndex struct {
Primaries map[string]StatPrimary `json:"primaries"` Primaries map[string]StatPrimary `json:"primaries"`
} }
// StatPrimary contains stats for a primary index
type StatPrimary struct { type StatPrimary struct {
// primary/docs: // primary/docs:
Count int Count int
Deleted int Deleted int
} }
// Shard holds the "shard" struct as returned by elasticsearch // Represents the "shard" struct as returned by elasticsearch
type Shard struct { type Shard struct {
Total uint64 Total uint64
Successful uint64 Successful uint64
Failed uint64 Failed uint64
} }
// Hit holds a hit returned by a search // Represent a hit returned by a search
type Hit struct { type Hit struct {
Index string `json:"_index"` Index string `json:"_index"`
Type string `json:"_type"` Type string `json:"_type"`
ID string `json:"_id"` Id string `json:"_id"`
Score float64 `json:"_score"` Score float64 `json:"_score"`
Source map[string]interface{} `json:"_source"` Source map[string]interface{} `json:"_source"`
Highlight map[string]interface{} `json:"highlight"` Highlight map[string]interface{} `json:"highlight"`
Fields map[string]interface{} `json:"fields"` Fields map[string]interface{} `json:"fields"`
} }
// Hits holds the hits structure as returned by elasticsearch // Represent the hits structure as returned by elasticsearch
type Hits struct { type Hits struct {
Total uint64 Total uint64
// max_score may contain the "null" value // max_score may contain the "null" value
@ -141,13 +162,12 @@ type Hits struct {
Hits []Hit Hits []Hit
} }
// SearchError holds errors returned from an ES search
type SearchError struct { type SearchError struct {
Msg string Msg string
StatusCode uint64 StatusCode uint64
} }
// IndexStatus holds the status for a given index for the _status command // Represent the status for a given index for the _status command
type IndexStatus struct { type IndexStatus struct {
// XXX : problem, int will be marshaled to a float64 which seems logical // XXX : problem, int will be marshaled to a float64 which seems logical
// XXX : is it better to use strings even for int values or to keep // XXX : is it better to use strings even for int values or to keep