|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509 |
- // Copyright 2013 Belogik. All rights reserved.
- // Use of this source code is governed by a BSD-style
- // license that can be found in the LICENSE file.
-
- package goes
-
- import (
- . "gopkg.in/check.v1"
- "net/http"
- "net/url"
- "os"
- "strings"
- "testing"
- "time"
- )
-
- var (
- ESHost = "localhost"
- ESPort = "9200"
- )
-
- // Hook up gocheck into the gotest runner.
- func Test(t *testing.T) { TestingT(t) }
-
- type GoesTestSuite struct{}
-
- var _ = Suite(&GoesTestSuite{})
-
- func (s *GoesTestSuite) SetUpTest(c *C) {
- h := os.Getenv("TEST_ELASTICSEARCH_HOST")
- if h != "" {
- ESHost = h
- }
-
- p := os.Getenv("TEST_ELASTICSEARCH_PORT")
- if p != "" {
- ESPort = p
- }
- }
-
- func (s *GoesTestSuite) TestNewClient(c *C) {
- conn := NewClient(ESHost, ESPort)
- c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, http.DefaultClient, "", "", ""})
- }
-
- func (s *GoesTestSuite) TestWithHTTPClient(c *C) {
- tr := &http.Transport{
- DisableCompression: true,
- ResponseHeaderTimeout: 1 * time.Second,
- }
- cl := &http.Client{
- Transport: tr,
- }
- conn := NewClient(ESHost, ESPort).WithHTTPClient(cl)
-
- c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, cl, "", "", ""})
- c.Assert(conn.Client.Transport.(*http.Transport).DisableCompression, Equals, true)
- c.Assert(conn.Client.Transport.(*http.Transport).ResponseHeaderTimeout, Equals, 1*time.Second)
- }
-
- func (s *GoesTestSuite) TestUrl(c *C) {
- r := Request{
- Query: "q",
- IndexList: []string{"i"},
- TypeList: []string{},
- Method: "GET",
- API: "_search",
- }
-
- c.Assert(r.URL().String(), Equals, "/i/_search")
-
- r.IndexList = []string{"a", "b"}
- c.Assert(r.URL().String(), Equals, "/a,b/_search")
-
- r.TypeList = []string{"c", "d"}
- c.Assert(r.URL().String(), Equals, "/a,b/c,d/_search")
-
- r.ExtraArgs = make(url.Values, 1)
- r.ExtraArgs.Set("version", "1")
- c.Assert(r.URL().String(), Equals, "/a,b/c,d/_search?version=1")
-
- r.ID = "1234"
- r.API = ""
- c.Assert(r.URL().String(), Equals, "/a,b/c,d/1234/?version=1")
- }
-
- func (s *GoesTestSuite) TestEsDown(c *C) {
- conn := NewClient("a.b.c.d", "1234")
-
- var query = map[string]interface{}{"query": "foo"}
-
- r := Request{
- Query: query,
- IndexList: []string{"i"},
- Method: "GET",
- API: "_search",
- }
- _, err := conn.Do(&r)
-
- c.Assert(err, ErrorMatches, ".* no such host")
- }
-
- func (s *GoesTestSuite) TestRunMissingIndex(c *C) {
- conn := NewClient(ESHost, ESPort)
-
- var query = map[string]interface{}{"query": "foo"}
-
- r := Request{
- Query: query,
- IndexList: []string{"i"},
- Method: "GET",
- API: "_search",
- }
- _, err := conn.Do(&r)
-
- c.Assert(err.Error(), Matches, "\\[40.\\] .*i.*")
- }
-
- func (s *GoesTestSuite) TestCreateIndex(c *C) {
- indexName := "testcreateindexgoes"
-
- conn := NewClient(ESHost, ESPort)
- defer conn.DeleteIndex(indexName)
-
- mapping := map[string]interface{}{
- "settings": map[string]interface{}{
- "index.number_of_shards": 1,
- "index.number_of_replicas": 0,
- },
- "mappings": map[string]interface{}{
- "_default_": map[string]interface{}{
- "_source": map[string]interface{}{
- "enabled": false,
- },
- "_all": map[string]interface{}{
- "enabled": false,
- },
- },
- },
- }
-
- resp, err := conn.CreateIndex(indexName, mapping)
-
- c.Assert(err, IsNil)
- c.Assert(resp.Acknowledged, Equals, true)
- }
-
- func (s *GoesTestSuite) TestDeleteIndexInexistantIndex(c *C) {
- conn := NewClient(ESHost, ESPort)
- resp, err := conn.DeleteIndex("foobar")
-
- c.Assert(err.Error(), Matches, "\\[404\\] .*foobar.*")
- resp.Raw = nil // Don't make us have to duplicate this.
- c.Assert(resp.Status, Equals, uint64(404))
- c.Assert(resp.Error, Matches, ".*foobar.*")
- }
-
- func (s *GoesTestSuite) TestDeleteIndexExistingIndex(c *C) {
- conn := NewClient(ESHost, ESPort)
-
- indexName := "testdeleteindexexistingindex"
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- defer conn.DeleteIndex(indexName)
-
- c.Assert(err, IsNil)
-
- resp, err := conn.DeleteIndex(indexName)
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Acknowledged: true,
- Status: 200,
- }
- resp.Raw = nil
- c.Assert(resp, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestUpdateIndexSettings(c *C) {
- conn := NewClient(ESHost, ESPort)
- indexName := "testupdateindex"
-
- // Just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- _, err = conn.UpdateIndexSettings(indexName, map[string]interface{}{
- "index": map[string]interface{}{
- "number_of_replicas": 0,
- },
- })
- c.Assert(err, IsNil)
-
- _, err = conn.DeleteIndex(indexName)
- c.Assert(err, IsNil)
- }
-
- func (s *GoesTestSuite) TestRefreshIndex(c *C) {
- conn := NewClient(ESHost, ESPort)
- indexName := "testrefreshindex"
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- _, err = conn.DeleteIndex(indexName)
- c.Assert(err, IsNil)
- }
-
- func (s *GoesTestSuite) TestOptimize(c *C) {
- conn := NewClient(ESHost, ESPort)
- indexName := "testoptimize"
-
- conn.DeleteIndex(indexName)
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- // we must wait for a bit otherwise ES crashes
- time.Sleep(1 * time.Second)
-
- response, err := conn.Optimize([]string{indexName}, url.Values{"max_num_segments": []string{"1"}})
- c.Assert(err, IsNil)
-
- c.Assert(response.All.Indices[indexName].Primaries["docs"].Count, Equals, 0)
-
- _, err = conn.DeleteIndex(indexName)
- c.Assert(err, IsNil)
- }
-
- func (s *GoesTestSuite) TestBulkSend(c *C) {
- indexName := "testbulkadd"
- docType := "tweet"
-
- tweets := []Document{
- {
- ID: "123",
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "some foo message",
- },
- },
-
- {
- ID: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "bar",
- "message": "some bar message",
- },
- },
- }
-
- conn := NewClient(ESHost, ESPort)
-
- conn.DeleteIndex(indexName)
- _, err := conn.CreateIndex(indexName, nil)
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- response, err := conn.BulkSend(tweets)
- i := Item{
- ID: "123",
- Type: docType,
- Version: 1,
- Index: indexName,
- Status: 201, //201 for indexing ( https://issues.apache.org/jira/browse/CONNECTORS-634 )
- }
- c.Assert(response.Items[0][BulkCommandIndex], Equals, i)
- c.Assert(err, IsNil)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- var query = map[string]interface{}{
- "query": map[string]interface{}{
- "match_all": map[string]interface{}{},
- },
- }
-
- searchResults, err := conn.Search(query, []string{indexName}, []string{}, url.Values{})
- c.Assert(err, IsNil)
-
- var expectedTotal uint64 = 2
- c.Assert(searchResults.Hits.Total, Equals, expectedTotal)
-
- extraDocID := ""
- checked := 0
- for _, hit := range searchResults.Hits.Hits {
- if hit.Source["user"] == "foo" {
- c.Assert(hit.ID, Equals, "123")
- checked++
- }
-
- if hit.Source["user"] == "bar" {
- c.Assert(len(hit.ID) > 0, Equals, true)
- extraDocID = hit.ID
- checked++
- }
- }
- c.Assert(checked, Equals, 2)
-
- docToDelete := []Document{
- {
- ID: "123",
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandDelete,
- },
- {
- ID: extraDocID,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandDelete,
- },
- }
-
- response, err = conn.BulkSend(docToDelete)
- i = Item{
- ID: "123",
- Type: docType,
- Version: 2,
- Index: indexName,
- Status: 200, //200 for updates
- }
- c.Assert(response.Items[0][BulkCommandDelete], Equals, i)
-
- c.Assert(err, IsNil)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- searchResults, err = conn.Search(query, []string{indexName}, []string{}, url.Values{})
- c.Assert(err, IsNil)
-
- expectedTotal = 0
- c.Assert(searchResults.Hits.Total, Equals, expectedTotal)
-
- _, err = conn.DeleteIndex(indexName)
- c.Assert(err, IsNil)
- }
-
- func (s *GoesTestSuite) TestStats(c *C) {
- conn := NewClient(ESHost, ESPort)
- indexName := "teststats"
-
- conn.DeleteIndex(indexName)
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- // we must wait for a bit otherwise ES crashes
- time.Sleep(1 * time.Second)
-
- response, err := conn.Stats([]string{indexName}, url.Values{})
- c.Assert(err, IsNil)
-
- c.Assert(response.All.Indices[indexName].Primaries["docs"].Count, Equals, 0)
-
- _, err = conn.DeleteIndex(indexName)
- c.Assert(err, IsNil)
- }
-
- func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) {
- indexName := "testindexwithfieldsinstruct"
- docType := "tweet"
- docID := "1234"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: struct {
- user string
- message string
- }{
- "foo",
- "bar",
- },
- }
-
- response, err := conn.Index(d, nil)
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Status: 201,
- Index: indexName,
- ID: docID,
- Type: docType,
- Version: 1,
- }
-
- response.Raw = nil
- response.Shards = Shard{}
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestIndexWithFieldsNotInMapOrStruct(c *C) {
- indexName := "testindexwithfieldsnotinmaporstruct"
- docType := "tweet"
- docID := "1234"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: "test",
- }
-
- _, err = conn.Index(d, nil)
- c.Assert(err, Not(IsNil))
- }
-
- func (s *GoesTestSuite) TestIndexIdDefined(c *C) {
- indexName := "testindexiddefined"
- docType := "tweet"
- docID := "1234"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "bar",
- },
- }
-
- response, err := conn.Index(d, nil)
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Status: 201,
- Index: indexName,
- ID: docID,
- Type: docType,
- Version: 1,
- }
-
- response.Raw = nil
- response.Shards = Shard{}
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestIndexIdNotDefined(c *C) {
- indexName := "testindexidnotdefined"
- docType := "tweet"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "bar",
- },
- }
-
- response, err := conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- c.Assert(response.Index, Equals, indexName)
- c.Assert(response.Type, Equals, docType)
- c.Assert(response.Version, Equals, 1)
- c.Assert(response.ID != "", Equals, true)
- }
-
- func (s *GoesTestSuite) TestDelete(c *C) {
- indexName := "testdelete"
- docType := "tweet"
- docID := "1234"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: map[string]interface{}{
- "user": "foo",
- },
- }
-
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- response, err := conn.Delete(d, url.Values{})
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Status: 200,
- Found: true,
- Index: indexName,
- Type: docType,
- ID: docID,
- // XXX : even after a DELETE the version number seems to be incremented
- Version: 2,
- }
- response.Raw = nil
- response.Shards = Shard{}
- c.Assert(response, DeepEquals, expectedResponse)
-
- response, err = conn.Delete(d, url.Values{})
- c.Assert(err, IsNil)
-
- expectedResponse = &Response{
- Status: 404,
- Found: false,
- Index: indexName,
- Type: docType,
- ID: docID,
- // XXX : even after a DELETE the version number seems to be incremented
- Version: 3,
- }
- response.Raw = nil
- response.Shards = Shard{}
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestDeleteByQuery(c *C) {
- indexName := "testdeletebyquery"
- docType := "tweet"
- docID := "1234"
-
- conn := NewClient(ESHost, ESPort)
- version, _ := conn.Version()
-
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: map[string]interface{}{
- "user": "foo",
- },
- }
-
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- query := map[string]interface{}{
- "query": map[string]interface{}{
- "bool": map[string]interface{}{
- "must": []map[string]interface{}{
- {
- "match_all": map[string]interface{}{},
- },
- },
- },
- },
- }
-
- //should be 1 doc before delete by query
- response, err := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
- c.Assert(err, IsNil)
- c.Assert(response.Hits.Total, Equals, uint64(1))
-
- response, err = conn.DeleteByQuery(query, []string{indexName}, []string{docType}, url.Values{})
-
- // There's no delete by query in ES 2.x
- if version > "2" && version < "5" {
- c.Assert(err, ErrorMatches, ".* does not support delete by query")
- return
- }
-
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Status: 200,
- Found: false,
- Index: "",
- Type: "",
- ID: "",
- Version: 0,
- }
- response.Raw = nil
- response.Shards = Shard{}
- response.Took = 0
- c.Assert(response, DeepEquals, expectedResponse)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- //should be 0 docs after delete by query
- response, err = conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
- c.Assert(err, IsNil)
- c.Assert(response.Hits.Total, Equals, uint64(0))
- }
-
- func (s *GoesTestSuite) TestGet(c *C) {
- indexName := "testget"
- docType := "tweet"
- docID := "111"
- source := map[string]interface{}{
- "f1": "foo",
- "f2": "foo",
- }
-
- conn := NewClient(ESHost, ESPort)
- version, _ := conn.Version()
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: source,
- }
-
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- response, err := conn.Get(indexName, docType, docID, url.Values{})
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Status: 200,
- Index: indexName,
- Type: docType,
- ID: docID,
- Version: 1,
- Found: true,
- Source: source,
- }
-
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
-
- expectedResponse = &Response{
- Status: 200,
- Index: indexName,
- Type: docType,
- ID: docID,
- Version: 1,
- Found: true,
- Fields: map[string]interface{}{
- "f1": []interface{}{"foo"},
- },
- }
-
- fields := make(url.Values, 1)
- // The fields param is no longer supported in ES 5.x
- if version < "5" {
- fields.Set("fields", "f1")
- } else {
- expectedResponse.Source = map[string]interface{}{"f1": "foo"}
- expectedResponse.Fields = nil
- fields.Set("_source", "f1")
- }
- response, err = conn.Get(indexName, docType, docID, fields)
- c.Assert(err, IsNil)
-
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestSearch(c *C) {
- indexName := "testsearch"
- docType := "tweet"
- docID := "1234"
- source := map[string]interface{}{
- "user": "foo",
- "message": "bar",
- }
-
- conn := NewClient(ESHost, ESPort)
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: source,
- }
-
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- // I can feel my eyes bleeding
- query := map[string]interface{}{
- "query": map[string]interface{}{
- "bool": map[string]interface{}{
- "must": []map[string]interface{}{
- {
- "match_all": map[string]interface{}{},
- },
- },
- },
- },
- }
- response, _ := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
-
- expectedHits := Hits{
- Total: 1,
- MaxScore: 1.0,
- Hits: []Hit{
- {
- Index: indexName,
- Type: docType,
- ID: docID,
- Score: 1.0,
- Source: source,
- },
- },
- }
-
- c.Assert(response.Hits, DeepEquals, expectedHits)
- }
-
- func (s *GoesTestSuite) TestCount(c *C) {
- indexName := "testcount"
- docType := "tweet"
- docID := "1234"
- source := map[string]interface{}{
- "user": "foo",
- "message": "bar",
- }
-
- conn := NewClient(ESHost, ESPort)
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: source,
- }
-
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- // I can feel my eyes bleeding
- query := map[string]interface{}{
- "query": map[string]interface{}{
- "bool": map[string]interface{}{
- "must": []map[string]interface{}{
- {
- "match_all": map[string]interface{}{},
- },
- },
- },
- },
- }
- response, _ := conn.Count(query, []string{indexName}, []string{docType}, url.Values{})
-
- c.Assert(response.Count, Equals, 1)
- }
-
- func (s *GoesTestSuite) TestIndexStatus(c *C) {
- indexName := "testindexstatus"
- conn := NewClient(ESHost, ESPort)
-
- // _status endpoint was removed in ES 2.0
- if version, _ := conn.Version(); version > "2" {
- return
- }
-
- conn.DeleteIndex(indexName)
-
- mapping := map[string]interface{}{
- "settings": map[string]interface{}{
- "index.number_of_shards": 1,
- "index.number_of_replicas": 1,
- },
- }
-
- _, err := conn.CreateIndex(indexName, mapping)
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- // gives ES some time to do its job
- time.Sleep(1 * time.Second)
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- response, err := conn.IndexStatus([]string{"testindexstatus"})
- c.Assert(err, IsNil)
-
- expectedShards := Shard{Total: 2, Successful: 1, Failed: 0}
- c.Assert(response.Shards, Equals, expectedShards)
-
- primarySizeInBytes := response.Indices[indexName].Index["primary_size_in_bytes"].(float64)
- sizeInBytes := response.Indices[indexName].Index["size_in_bytes"].(float64)
- refreshTotal := response.Indices[indexName].Refresh["total"].(float64)
-
- c.Assert(primarySizeInBytes > 0, Equals, true)
- c.Assert(sizeInBytes > 0, Equals, true)
- c.Assert(refreshTotal > 0, Equals, true)
-
- expectedIndices := map[string]IndexStatus{
- indexName: {
- Index: map[string]interface{}{
- "primary_size_in_bytes": primarySizeInBytes,
- "size_in_bytes": sizeInBytes,
- },
- Translog: map[string]uint64{
- "operations": 0,
- },
- Docs: map[string]uint64{
- "num_docs": 0,
- "max_doc": 0,
- "deleted_docs": 0,
- },
- Merges: map[string]interface{}{
- "current": float64(0),
- "current_docs": float64(0),
- "current_size_in_bytes": float64(0),
- "total": float64(0),
- "total_time_in_millis": float64(0),
- "total_docs": float64(0),
- "total_size_in_bytes": float64(0),
- },
- Refresh: map[string]interface{}{
- "total": refreshTotal,
- "total_time_in_millis": float64(0),
- },
- Flush: map[string]interface{}{
- "total": float64(0),
- "total_time_in_millis": float64(0),
- },
- },
- }
-
- c.Assert(response.Indices, DeepEquals, expectedIndices)
- }
-
- func (s *GoesTestSuite) TestScroll(c *C) {
- indexName := "testscroll"
- docType := "tweet"
-
- tweets := []Document{
- {
- ID: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "some foo message",
- },
- },
-
- {
- ID: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "bar",
- "message": "some bar message",
- },
- },
-
- {
- ID: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "another foo message",
- },
- },
- }
-
- conn := NewClient(ESHost, ESPort)
-
- mapping := map[string]interface{}{
- "settings": map[string]interface{}{
- "index.number_of_shards": 1,
- "index.number_of_replicas": 0,
- },
- }
-
- defer conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, mapping)
- c.Assert(err, IsNil)
-
- _, err = conn.BulkSend(tweets)
- c.Assert(err, IsNil)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- var query map[string]interface{}
- version, _ := conn.Version()
- if version > "5" {
- query = map[string]interface{}{
- "query": map[string]interface{}{
- "bool": map[string]interface{}{
- "filter": map[string]interface{}{
- "term": map[string]interface{}{
- "user": "foo",
- },
- },
- },
- },
- }
- } else {
- query = map[string]interface{}{
- "query": map[string]interface{}{
- "filtered": map[string]interface{}{
- "filter": map[string]interface{}{
- "term": map[string]interface{}{
- "user": "foo",
- },
- },
- },
- },
- }
- }
-
- searchResults, err := conn.Scan(query, []string{indexName}, []string{docType}, "1m", 1)
- c.Assert(err, IsNil)
- c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
-
- // Versions < 5.x don't include results in the initial response
- if version < "5" {
- searchResults, err = conn.Scroll(searchResults.ScrollID, "1m")
- c.Assert(err, IsNil)
- }
-
- // some data in first chunk
- c.Assert(searchResults.Hits.Total, Equals, uint64(2))
- c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
- c.Assert(len(searchResults.Hits.Hits), Equals, 1)
-
- searchResults, err = conn.Scroll(searchResults.ScrollID, "1m")
- c.Assert(err, IsNil)
-
- // more data in second chunk
- c.Assert(searchResults.Hits.Total, Equals, uint64(2))
- c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
- c.Assert(len(searchResults.Hits.Hits), Equals, 1)
-
- searchResults, err = conn.Scroll(searchResults.ScrollID, "1m")
- c.Assert(err, IsNil)
-
- // nothing in third chunk
- c.Assert(searchResults.Hits.Total, Equals, uint64(2))
- c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
- c.Assert(len(searchResults.Hits.Hits), Equals, 0)
- }
-
- func (s *GoesTestSuite) TestAggregations(c *C) {
- indexName := "testaggs"
- docType := "tweet"
-
- tweets := []Document{
- {
- ID: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "some foo message",
- "age": 25,
- },
- },
-
- {
- ID: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "bar",
- "message": "some bar message",
- "age": 30,
- },
- },
-
- {
- ID: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BulkCommandIndex,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "another foo message",
- },
- },
- }
-
- conn := NewClient(ESHost, ESPort)
-
- mapping := map[string]interface{}{
- "settings": map[string]interface{}{
- "index.number_of_shards": 1,
- "index.number_of_replicas": 0,
- },
- "mappings": map[string]interface{}{
- docType: map[string]interface{}{
- "properties": map[string]interface{}{
- "user": map[string]interface{}{
- "type": "string",
- "index": "not_analyzed",
- },
- },
- },
- },
- }
-
- defer conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, mapping)
- c.Assert(err, IsNil)
-
- _, err = conn.BulkSend(tweets)
- c.Assert(err, IsNil)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- query := map[string]interface{}{
- "aggs": map[string]interface{}{
- "user": map[string]interface{}{
- "terms": map[string]interface{}{
- "field": "user",
- "order": map[string]interface{}{
- "_term": "asc",
- },
- },
- "aggs": map[string]interface{}{
- "age": map[string]interface{}{
- "stats": map[string]interface{}{
- "field": "age",
- },
- },
- },
- },
- "age": map[string]interface{}{
- "stats": map[string]interface{}{
- "field": "age",
- },
- },
- },
- }
-
- resp, _ := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
-
- user, ok := resp.Aggregations["user"]
- c.Assert(ok, Equals, true)
-
- c.Assert(len(user.Buckets()), Equals, 2)
- c.Assert(user.Buckets()[0].Key(), Equals, "bar")
- c.Assert(user.Buckets()[1].Key(), Equals, "foo")
-
- barAge := user.Buckets()[0].Aggregation("age")
- c.Assert(barAge["count"], Equals, 1.0)
- c.Assert(barAge["sum"], Equals, 30.0)
-
- fooAge := user.Buckets()[1].Aggregation("age")
- c.Assert(fooAge["count"], Equals, 1.0)
- c.Assert(fooAge["sum"], Equals, 25.0)
-
- age, ok := resp.Aggregations["age"]
- c.Assert(ok, Equals, true)
-
- c.Assert(age["count"], Equals, 2.0)
- c.Assert(age["sum"], Equals, 25.0+30.0)
- }
-
- func (s *GoesTestSuite) TestPutMapping(c *C) {
- indexName := "testputmapping"
- docType := "tweet"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "bar",
- },
- }
-
- response, err := conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- mapping := map[string]interface{}{
- "tweet": map[string]interface{}{
- "properties": map[string]interface{}{
- "count": map[string]interface{}{
- "type": "integer",
- "index": "not_analyzed",
- "store": true,
- },
- },
- },
- }
- response, err = conn.PutMapping("tweet", mapping, []string{indexName})
- c.Assert(err, IsNil)
-
- c.Assert(response.Acknowledged, Equals, true)
- c.Assert(response.TimedOut, Equals, false)
- }
-
- func (s *GoesTestSuite) TestIndicesExist(c *C) {
- indices := []string{"testindicesexist"}
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indices[0])
-
- exists, err := conn.IndicesExist(indices)
- c.Assert(exists, Equals, false)
-
- _, err = conn.CreateIndex(indices[0], map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indices[0])
- time.Sleep(200 * time.Millisecond)
-
- exists, _ = conn.IndicesExist(indices)
- c.Assert(exists, Equals, true)
-
- indices = append(indices, "nonexistent")
- exists, _ = conn.IndicesExist(indices)
- c.Assert(exists, Equals, false)
- }
-
- func (s *GoesTestSuite) TestUpdate(c *C) {
- indexName := "testupdate"
- docType := "tweet"
- docID := "1234"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "bar",
- "counter": 1,
- },
- }
-
- extraArgs := make(url.Values, 1)
- response, err := conn.Index(d, extraArgs)
- c.Assert(err, IsNil)
- time.Sleep(200 * time.Millisecond)
-
- expectedResponse := &Response{
- Status: 201,
- Index: indexName,
- ID: docID,
- Type: docType,
- Version: 1,
- }
-
- response.Raw = nil
- response.Shards.Successful = 0
- response.Shards.Total = 0
- c.Assert(response, DeepEquals, expectedResponse)
-
- // Now that we have an ordinary document indexed, try updating it
- var query map[string]interface{}
- if version, _ := conn.Version(); version > "5" {
- query = map[string]interface{}{
- "script": map[string]interface{}{
- "inline": "ctx._source.counter += params.count",
- "lang": "painless",
- "params": map[string]interface{}{
- "count": 5,
- },
- },
- "upsert": map[string]interface{}{
- "message": "candybar",
- "user": "admin",
- "counter": 1,
- },
- }
- } else {
- query = map[string]interface{}{
- "script": "ctx._source.counter += count",
- "lang": "groovy",
- "params": map[string]interface{}{
- "count": 5,
- },
- "upsert": map[string]interface{}{
- "message": "candybar",
- "user": "admin",
- "counter": 1,
- },
- }
- }
-
- response, err = conn.Update(d, query, extraArgs)
-
- if err != nil && strings.Contains(err.(*SearchError).Msg, "dynamic scripting") {
- c.Skip("Scripting is disabled on server, skipping this test")
- return
- }
- time.Sleep(200 * time.Millisecond)
-
- c.Assert(err, Equals, nil)
-
- response, err = conn.Get(indexName, docType, docID, url.Values{})
- c.Assert(err, Equals, nil)
- c.Assert(response.Source["counter"], Equals, float64(6))
- c.Assert(response.Source["user"], Equals, "foo")
- c.Assert(response.Source["message"], Equals, "bar")
-
- // Test another document, non-existent
- docID = "555"
- d.ID = docID
- response, err = conn.Update(d, query, extraArgs)
- c.Assert(err, Equals, nil)
- time.Sleep(200 * time.Millisecond)
-
- response, err = conn.Get(indexName, docType, docID, url.Values{})
- c.Assert(err, Equals, nil)
- c.Assert(response.Source["user"], Equals, "admin")
- c.Assert(response.Source["message"], Equals, "candybar")
- }
-
- func (s *GoesTestSuite) TestGetMapping(c *C) {
- indexName := "testmapping"
- docType := "tweet"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- time.Sleep(300 * time.Millisecond)
-
- response, err := conn.GetMapping([]string{docType}, []string{indexName})
- c.Assert(err, Equals, nil)
- c.Assert(len(response.Raw), Equals, 0)
-
- d := Document{
- Index: indexName,
- Type: docType,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "bar",
- },
- }
-
- response, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
- time.Sleep(200 * time.Millisecond)
-
- response, err = conn.GetMapping([]string{docType}, []string{indexName})
- c.Assert(err, Equals, nil)
- c.Assert(len(response.Raw), Not(Equals), 0)
- }
-
- func (s *GoesTestSuite) TestDeleteMapping(c *C) {
- indexName := "testdeletemapping"
- docType := "tweet"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "bar",
- },
- }
-
- response, err := conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- mapping := map[string]interface{}{
- "tweet": map[string]interface{}{
- "properties": map[string]interface{}{
- "count": map[string]interface{}{
- "type": "integer",
- "index": "not_analyzed",
- "store": true,
- },
- },
- },
- }
- response, err = conn.PutMapping("tweet", mapping, []string{indexName})
- c.Assert(err, IsNil)
- time.Sleep(200 * time.Millisecond)
-
- response, err = conn.DeleteMapping("tweet", []string{indexName})
- if version, _ := conn.Version(); version > "2" {
- c.Assert(err, ErrorMatches, ".*not supported.*")
- return
- }
- c.Assert(err, IsNil)
-
- c.Assert(response.Acknowledged, Equals, true)
- c.Assert(response.TimedOut, Equals, false)
- }
-
- func (s *GoesTestSuite) TestAddAlias(c *C) {
- aliasName := "testAlias"
- indexName := "testalias_1"
- docType := "testDoc"
- docID := "1234"
- source := map[string]interface{}{
- "user": "foo",
- "message": "bar",
- }
-
- conn := NewClient(ESHost, ESPort)
- defer conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: source,
- }
-
- // Index data
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- // Add alias
- _, err = conn.AddAlias(aliasName, []string{indexName})
- c.Assert(err, IsNil)
-
- // Get document via alias
- response, err := conn.Get(aliasName, docType, docID, url.Values{})
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Status: 200,
- Index: indexName,
- Type: docType,
- ID: docID,
- Version: 1,
- Found: true,
- Source: source,
- }
-
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestRemoveAlias(c *C) {
- aliasName := "testAlias"
- indexName := "testalias_1"
- docType := "testDoc"
- docID := "1234"
- source := map[string]interface{}{
- "user": "foo",
- "message": "bar",
- }
-
- conn := NewClient(ESHost, ESPort)
- defer conn.DeleteIndex(indexName)
-
- _, err := conn.CreateIndex(indexName, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indexName)
-
- d := Document{
- Index: indexName,
- Type: docType,
- ID: docID,
- Fields: source,
- }
-
- // Index data
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- // Add alias
- _, err = conn.AddAlias(aliasName, []string{indexName})
- c.Assert(err, IsNil)
-
- // Remove alias
- _, err = conn.RemoveAlias(aliasName, []string{indexName})
- c.Assert(err, IsNil)
-
- // Get document via alias
- _, err = conn.Get(aliasName, docType, docID, url.Values{})
- c.Assert(err.Error(), Matches, "\\[404\\] .*"+aliasName+".*")
- }
-
- func (s *GoesTestSuite) TestAliasExists(c *C) {
- index := "testaliasexist_1"
- alias := "testaliasexists"
-
- conn := NewClient(ESHost, ESPort)
- // just in case
- conn.DeleteIndex(index)
-
- exists, err := conn.AliasExists(alias)
- c.Assert(exists, Equals, false)
-
- _, err = conn.CreateIndex(index, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(index)
- time.Sleep(200 * time.Millisecond)
-
- _, err = conn.AddAlias(alias, []string{index})
- c.Assert(err, IsNil)
- time.Sleep(200 * time.Millisecond)
- defer conn.RemoveAlias(alias, []string{index})
-
- exists, _ = conn.AliasExists(alias)
- c.Assert(exists, Equals, true)
- }
|