Compare commits
No commits in common. "fbfb1d80a89aea5c028d3d1ff1bb7540f2a54792" and "6c9647b81cbf8daf85a01b1c15a7cc23de351c27" have entirely different histories.
fbfb1d80a8
...
6c9647b81c
1
.gitignore
vendored
1
.gitignore
vendored
@ -1,3 +1,2 @@
|
|||||||
*.test
|
*.test
|
||||||
*.swp
|
*.swp
|
||||||
vendor
|
|
||||||
|
39
.travis.yml
39
.travis.yml
@ -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
|
||||||
|
15
Makefile
15
Makefile
@ -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 .
|
|
||||||
|
@ -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
|
||||||
===============================================
|
===============================================
|
||||||
|
|
@ -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
12
glide.lock
generated
@ -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
|
|
11
glide.yaml
11
glide.yaml
@ -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
|
|
556
goes.go
556
goes.go
@ -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()
|
v.Add("search_type", "scan")
|
||||||
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("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) {
|
||||||
|
v := url.Values{}
|
||||||
|
v.Add("scroll", timeout)
|
||||||
|
|
||||||
r := Request{
|
r := Request{
|
||||||
Method: "POST",
|
Conn: c,
|
||||||
API: "_search/scroll",
|
method: "POST",
|
||||||
|
api: "_search/scroll",
|
||||||
|
ExtraArgs: v,
|
||||||
|
Body: []byte(scrollId),
|
||||||
}
|
}
|
||||||
|
|
||||||
if version, err := c.Version(); err != nil {
|
return r.Run()
|
||||||
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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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
|
|
||||||
}
|
|
||||||
|
@ -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")
|
|
||||||
}
|
|
446
goes_test.go
446
goes_test.go
File diff suppressed because it is too large
Load Diff
113
request.go
113
request.go
@ -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)
|
|
82
structs.go
82
structs.go
@ -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
|
||||||
|
Loading…
Reference in New Issue
Block a user