You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

1427 lines
32 KiB

  1. // Copyright 2013 Belogik. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. package goes
  5. import (
  6. "net/http"
  7. "net/url"
  8. "os"
  9. "strings"
  10. "testing"
  11. "time"
  12. . "github.com/go-check/check"
  13. )
  14. var (
  15. ESHost = "localhost"
  16. ESPort = "9200"
  17. )
  18. // Hook up gocheck into the gotest runner.
  19. func Test(t *testing.T) { TestingT(t) }
  20. type GoesTestSuite struct{}
  21. var _ = Suite(&GoesTestSuite{})
  22. func (s *GoesTestSuite) SetUpTest(c *C) {
  23. h := os.Getenv("TEST_ELASTICSEARCH_HOST")
  24. if h != "" {
  25. ESHost = h
  26. }
  27. p := os.Getenv("TEST_ELASTICSEARCH_PORT")
  28. if p != "" {
  29. ESPort = p
  30. }
  31. }
  32. func (s *GoesTestSuite) TestNewClient(c *C) {
  33. conn := NewClient(ESHost, ESPort)
  34. c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, http.DefaultClient})
  35. }
  36. func (s *GoesTestSuite) TestWithHTTPClient(c *C) {
  37. tr := &http.Transport{
  38. DisableCompression: true,
  39. ResponseHeaderTimeout: 1 * time.Second,
  40. }
  41. cl := &http.Client{
  42. Transport: tr,
  43. }
  44. conn := NewClient(ESHost, ESPort).WithHTTPClient(cl)
  45. c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, cl})
  46. c.Assert(conn.Client.Transport.(*http.Transport).DisableCompression, Equals, true)
  47. c.Assert(conn.Client.Transport.(*http.Transport).ResponseHeaderTimeout, Equals, 1*time.Second)
  48. }
  49. func (s *GoesTestSuite) TestUrl(c *C) {
  50. conn := NewClient(ESHost, ESPort)
  51. r := Request{
  52. Conn: conn,
  53. Query: "q",
  54. IndexList: []string{"i"},
  55. TypeList: []string{},
  56. method: "GET",
  57. api: "_search",
  58. }
  59. c.Assert(r.URL(), Equals, "http://"+ESHost+":"+ESPort+"/i/_search")
  60. r.IndexList = []string{"a", "b"}
  61. c.Assert(r.URL(), Equals, "http://"+ESHost+":"+ESPort+"/a,b/_search")
  62. r.TypeList = []string{"c", "d"}
  63. c.Assert(r.URL(), Equals, "http://"+ESHost+":"+ESPort+"/a,b/c,d/_search")
  64. r.ExtraArgs = make(url.Values, 1)
  65. r.ExtraArgs.Set("version", "1")
  66. c.Assert(r.URL(), Equals, "http://"+ESHost+":"+ESPort+"/a,b/c,d/_search?version=1")
  67. r.id = "1234"
  68. r.api = ""
  69. c.Assert(r.URL(), Equals, "http://"+ESHost+":"+ESPort+"/a,b/c,d/1234/?version=1")
  70. }
  71. func (s *GoesTestSuite) TestEsDown(c *C) {
  72. conn := NewClient("a.b.c.d", "1234")
  73. var query = map[string]interface{}{"query": "foo"}
  74. r := Request{
  75. Conn: conn,
  76. Query: query,
  77. IndexList: []string{"i"},
  78. method: "GET",
  79. api: "_search",
  80. }
  81. _, err := r.Run()
  82. c.Assert(err, ErrorMatches, ".* no such host")
  83. }
  84. func (s *GoesTestSuite) TestRunMissingIndex(c *C) {
  85. conn := NewClient(ESHost, ESPort)
  86. var query = map[string]interface{}{"query": "foo"}
  87. r := Request{
  88. Conn: conn,
  89. Query: query,
  90. IndexList: []string{"i"},
  91. method: "GET",
  92. api: "_search",
  93. }
  94. _, err := r.Run()
  95. c.Assert(err.Error(), Equals, "[404] IndexMissingException[[i] missing]")
  96. }
  97. func (s *GoesTestSuite) TestCreateIndex(c *C) {
  98. indexName := "testcreateindexgoes"
  99. conn := NewClient(ESHost, ESPort)
  100. defer conn.DeleteIndex(indexName)
  101. mapping := map[string]interface{}{
  102. "settings": map[string]interface{}{
  103. "index.number_of_shards": 1,
  104. "index.number_of_replicas": 0,
  105. },
  106. "mappings": map[string]interface{}{
  107. "_default_": map[string]interface{}{
  108. "_source": map[string]interface{}{
  109. "enabled": false,
  110. },
  111. "_all": map[string]interface{}{
  112. "enabled": false,
  113. },
  114. },
  115. },
  116. }
  117. resp, err := conn.CreateIndex(indexName, mapping)
  118. c.Assert(err, IsNil)
  119. c.Assert(resp.Acknowledged, Equals, true)
  120. }
  121. func (s *GoesTestSuite) TestDeleteIndexInexistantIndex(c *C) {
  122. conn := NewClient(ESHost, ESPort)
  123. resp, err := conn.DeleteIndex("foobar")
  124. c.Assert(err.Error(), Equals, "[404] IndexMissingException[[foobar] missing]")
  125. resp.Raw = nil // Don't make us have to duplicate this.
  126. c.Assert(resp, DeepEquals, &Response{Status: 404, Error: "IndexMissingException[[foobar] missing]"})
  127. }
  128. func (s *GoesTestSuite) TestDeleteIndexExistingIndex(c *C) {
  129. conn := NewClient(ESHost, ESPort)
  130. indexName := "testdeleteindexexistingindex"
  131. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  132. c.Assert(err, IsNil)
  133. resp, err := conn.DeleteIndex(indexName)
  134. c.Assert(err, IsNil)
  135. expectedResponse := &Response{
  136. Acknowledged: true,
  137. Status: 200,
  138. }
  139. resp.Raw = nil
  140. c.Assert(resp, DeepEquals, expectedResponse)
  141. }
  142. func (s *GoesTestSuite) TestUpdateIndexSettings(c *C) {
  143. conn := NewClient(ESHost, ESPort)
  144. indexName := "testupdateindex"
  145. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  146. c.Assert(err, IsNil)
  147. _, err = conn.UpdateIndexSettings(indexName, map[string]interface{}{
  148. "index": map[string]interface{}{
  149. "number_of_replicas": 0,
  150. },
  151. })
  152. c.Assert(err, IsNil)
  153. _, err = conn.DeleteIndex(indexName)
  154. c.Assert(err, IsNil)
  155. }
  156. func (s *GoesTestSuite) TestRefreshIndex(c *C) {
  157. conn := NewClient(ESHost, ESPort)
  158. indexName := "testrefreshindex"
  159. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  160. c.Assert(err, IsNil)
  161. _, err = conn.RefreshIndex(indexName)
  162. c.Assert(err, IsNil)
  163. _, err = conn.DeleteIndex(indexName)
  164. c.Assert(err, IsNil)
  165. }
  166. func (s *GoesTestSuite) TestOptimize(c *C) {
  167. conn := NewClient(ESHost, ESPort)
  168. indexName := "testoptimize"
  169. conn.DeleteIndex(indexName)
  170. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  171. c.Assert(err, IsNil)
  172. // we must wait for a bit otherwise ES crashes
  173. time.Sleep(1 * time.Second)
  174. response, err := conn.Optimize([]string{indexName}, url.Values{"max_num_segments": []string{"1"}})
  175. c.Assert(err, IsNil)
  176. c.Assert(response.All.Indices[indexName].Primaries["docs"].Count, Equals, 0)
  177. _, err = conn.DeleteIndex(indexName)
  178. c.Assert(err, IsNil)
  179. }
  180. func (s *GoesTestSuite) TestBulkSend(c *C) {
  181. indexName := "testbulkadd"
  182. docType := "tweet"
  183. tweets := []Document{
  184. {
  185. ID: "123",
  186. Index: indexName,
  187. Type: docType,
  188. BulkCommand: BulkCommandIndex,
  189. Fields: map[string]interface{}{
  190. "user": "foo",
  191. "message": "some foo message",
  192. },
  193. },
  194. {
  195. ID: nil,
  196. Index: indexName,
  197. Type: docType,
  198. BulkCommand: BulkCommandIndex,
  199. Fields: map[string]interface{}{
  200. "user": "bar",
  201. "message": "some bar message",
  202. },
  203. },
  204. }
  205. conn := NewClient(ESHost, ESPort)
  206. conn.DeleteIndex(indexName)
  207. _, err := conn.CreateIndex(indexName, nil)
  208. c.Assert(err, IsNil)
  209. response, err := conn.BulkSend(tweets)
  210. i := Item{
  211. ID: "123",
  212. Type: docType,
  213. Version: 1,
  214. Index: indexName,
  215. Status: 201, //201 for indexing ( https://issues.apache.org/jira/browse/CONNECTORS-634 )
  216. }
  217. c.Assert(response.Items[0][BulkCommandIndex], Equals, i)
  218. c.Assert(err, IsNil)
  219. _, err = conn.RefreshIndex(indexName)
  220. c.Assert(err, IsNil)
  221. var query = map[string]interface{}{
  222. "query": map[string]interface{}{
  223. "match_all": map[string]interface{}{},
  224. },
  225. }
  226. searchResults, err := conn.Search(query, []string{indexName}, []string{}, url.Values{})
  227. c.Assert(err, IsNil)
  228. var expectedTotal uint64 = 2
  229. c.Assert(searchResults.Hits.Total, Equals, expectedTotal)
  230. extraDocID := ""
  231. checked := 0
  232. for _, hit := range searchResults.Hits.Hits {
  233. if hit.Source["user"] == "foo" {
  234. c.Assert(hit.ID, Equals, "123")
  235. checked++
  236. }
  237. if hit.Source["user"] == "bar" {
  238. c.Assert(len(hit.ID) > 0, Equals, true)
  239. extraDocID = hit.ID
  240. checked++
  241. }
  242. }
  243. c.Assert(checked, Equals, 2)
  244. docToDelete := []Document{
  245. {
  246. ID: "123",
  247. Index: indexName,
  248. Type: docType,
  249. BulkCommand: BulkCommandDelete,
  250. },
  251. {
  252. ID: extraDocID,
  253. Index: indexName,
  254. Type: docType,
  255. BulkCommand: BulkCommandDelete,
  256. },
  257. }
  258. response, err = conn.BulkSend(docToDelete)
  259. i = Item{
  260. ID: "123",
  261. Type: docType,
  262. Version: 2,
  263. Index: indexName,
  264. Status: 200, //200 for updates
  265. }
  266. c.Assert(response.Items[0][BulkCommandDelete], Equals, i)
  267. c.Assert(err, IsNil)
  268. _, err = conn.RefreshIndex(indexName)
  269. c.Assert(err, IsNil)
  270. searchResults, err = conn.Search(query, []string{indexName}, []string{}, url.Values{})
  271. c.Assert(err, IsNil)
  272. expectedTotal = 0
  273. c.Assert(searchResults.Hits.Total, Equals, expectedTotal)
  274. _, err = conn.DeleteIndex(indexName)
  275. c.Assert(err, IsNil)
  276. }
  277. func (s *GoesTestSuite) TestStats(c *C) {
  278. conn := NewClient(ESHost, ESPort)
  279. indexName := "teststats"
  280. conn.DeleteIndex(indexName)
  281. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  282. c.Assert(err, IsNil)
  283. // we must wait for a bit otherwise ES crashes
  284. time.Sleep(1 * time.Second)
  285. response, err := conn.Stats([]string{indexName}, url.Values{})
  286. c.Assert(err, IsNil)
  287. c.Assert(response.All.Indices[indexName].Primaries["docs"].Count, Equals, 0)
  288. _, err = conn.DeleteIndex(indexName)
  289. c.Assert(err, IsNil)
  290. }
  291. func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) {
  292. indexName := "testindexwithfieldsinstruct"
  293. docType := "tweet"
  294. docID := "1234"
  295. conn := NewClient(ESHost, ESPort)
  296. // just in case
  297. conn.DeleteIndex(indexName)
  298. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  299. c.Assert(err, IsNil)
  300. defer conn.DeleteIndex(indexName)
  301. d := Document{
  302. Index: indexName,
  303. Type: docType,
  304. ID: docID,
  305. Fields: struct {
  306. user string
  307. message string
  308. }{
  309. "foo",
  310. "bar",
  311. },
  312. }
  313. extraArgs := make(url.Values, 1)
  314. extraArgs.Set("ttl", "86400000")
  315. response, err := conn.Index(d, extraArgs)
  316. c.Assert(err, IsNil)
  317. expectedResponse := &Response{
  318. Status: 201,
  319. Index: indexName,
  320. ID: docID,
  321. Type: docType,
  322. Version: 1,
  323. }
  324. response.Raw = nil
  325. c.Assert(response, DeepEquals, expectedResponse)
  326. }
  327. func (s *GoesTestSuite) TestIndexWithFieldsNotInMapOrStruct(c *C) {
  328. indexName := "testindexwithfieldsnotinmaporstruct"
  329. docType := "tweet"
  330. docID := "1234"
  331. conn := NewClient(ESHost, ESPort)
  332. // just in case
  333. conn.DeleteIndex(indexName)
  334. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  335. c.Assert(err, IsNil)
  336. defer conn.DeleteIndex(indexName)
  337. d := Document{
  338. Index: indexName,
  339. Type: docType,
  340. ID: docID,
  341. Fields: "test",
  342. }
  343. extraArgs := make(url.Values, 1)
  344. extraArgs.Set("ttl", "86400000")
  345. _, err = conn.Index(d, extraArgs)
  346. c.Assert(err, Not(IsNil))
  347. }
  348. func (s *GoesTestSuite) TestIndexIdDefined(c *C) {
  349. indexName := "testindexiddefined"
  350. docType := "tweet"
  351. docID := "1234"
  352. conn := NewClient(ESHost, ESPort)
  353. // just in case
  354. conn.DeleteIndex(indexName)
  355. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  356. c.Assert(err, IsNil)
  357. defer conn.DeleteIndex(indexName)
  358. d := Document{
  359. Index: indexName,
  360. Type: docType,
  361. ID: docID,
  362. Fields: map[string]interface{}{
  363. "user": "foo",
  364. "message": "bar",
  365. },
  366. }
  367. extraArgs := make(url.Values, 1)
  368. extraArgs.Set("ttl", "86400000")
  369. response, err := conn.Index(d, extraArgs)
  370. c.Assert(err, IsNil)
  371. expectedResponse := &Response{
  372. Status: 201,
  373. Index: indexName,
  374. ID: docID,
  375. Type: docType,
  376. Version: 1,
  377. }
  378. response.Raw = nil
  379. c.Assert(response, DeepEquals, expectedResponse)
  380. }
  381. func (s *GoesTestSuite) TestIndexIdNotDefined(c *C) {
  382. indexName := "testindexidnotdefined"
  383. docType := "tweet"
  384. conn := NewClient(ESHost, ESPort)
  385. // just in case
  386. conn.DeleteIndex(indexName)
  387. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  388. c.Assert(err, IsNil)
  389. defer conn.DeleteIndex(indexName)
  390. d := Document{
  391. Index: indexName,
  392. Type: docType,
  393. Fields: map[string]interface{}{
  394. "user": "foo",
  395. "message": "bar",
  396. },
  397. }
  398. response, err := conn.Index(d, url.Values{})
  399. c.Assert(err, IsNil)
  400. c.Assert(response.Index, Equals, indexName)
  401. c.Assert(response.Type, Equals, docType)
  402. c.Assert(response.Version, Equals, 1)
  403. c.Assert(response.ID != "", Equals, true)
  404. }
  405. func (s *GoesTestSuite) TestDelete(c *C) {
  406. indexName := "testdelete"
  407. docType := "tweet"
  408. docID := "1234"
  409. conn := NewClient(ESHost, ESPort)
  410. // just in case
  411. conn.DeleteIndex(indexName)
  412. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  413. c.Assert(err, IsNil)
  414. defer conn.DeleteIndex(indexName)
  415. d := Document{
  416. Index: indexName,
  417. Type: docType,
  418. ID: docID,
  419. Fields: map[string]interface{}{
  420. "user": "foo",
  421. },
  422. }
  423. _, err = conn.Index(d, url.Values{})
  424. c.Assert(err, IsNil)
  425. response, err := conn.Delete(d, url.Values{})
  426. c.Assert(err, IsNil)
  427. expectedResponse := &Response{
  428. Status: 200,
  429. Found: true,
  430. Index: indexName,
  431. Type: docType,
  432. ID: docID,
  433. // XXX : even after a DELETE the version number seems to be incremented
  434. Version: 2,
  435. }
  436. response.Raw = nil
  437. c.Assert(response, DeepEquals, expectedResponse)
  438. response, err = conn.Delete(d, url.Values{})
  439. c.Assert(err, IsNil)
  440. expectedResponse = &Response{
  441. Status: 404,
  442. Found: false,
  443. Index: indexName,
  444. Type: docType,
  445. ID: docID,
  446. // XXX : even after a DELETE the version number seems to be incremented
  447. Version: 3,
  448. }
  449. response.Raw = nil
  450. c.Assert(response, DeepEquals, expectedResponse)
  451. }
  452. func (s *GoesTestSuite) TestDeleteByQuery(c *C) {
  453. indexName := "testdeletebyquery"
  454. docType := "tweet"
  455. docID := "1234"
  456. conn := NewClient(ESHost, ESPort)
  457. // just in case
  458. conn.DeleteIndex(indexName)
  459. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  460. c.Assert(err, IsNil)
  461. defer conn.DeleteIndex(indexName)
  462. d := Document{
  463. Index: indexName,
  464. Type: docType,
  465. ID: docID,
  466. Fields: map[string]interface{}{
  467. "user": "foo",
  468. },
  469. }
  470. _, err = conn.Index(d, url.Values{})
  471. c.Assert(err, IsNil)
  472. _, err = conn.RefreshIndex(indexName)
  473. c.Assert(err, IsNil)
  474. query := map[string]interface{}{
  475. "query": map[string]interface{}{
  476. "bool": map[string]interface{}{
  477. "must": []map[string]interface{}{
  478. {
  479. "match_all": map[string]interface{}{},
  480. },
  481. },
  482. },
  483. },
  484. }
  485. //should be 1 doc before delete by query
  486. response, err := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
  487. c.Assert(err, IsNil)
  488. c.Assert(response.Hits.Total, Equals, uint64(1))
  489. response, err = conn.Query(query, []string{indexName}, []string{docType}, "DELETE", url.Values{})
  490. c.Assert(err, IsNil)
  491. expectedResponse := &Response{
  492. Status: 200,
  493. Found: false,
  494. Index: "",
  495. Type: "",
  496. ID: "",
  497. Version: 0,
  498. }
  499. response.Raw = nil
  500. c.Assert(response, DeepEquals, expectedResponse)
  501. //should be 0 docs after delete by query
  502. response, err = conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
  503. c.Assert(err, IsNil)
  504. c.Assert(response.Hits.Total, Equals, uint64(0))
  505. }
  506. func (s *GoesTestSuite) TestGet(c *C) {
  507. indexName := "testget"
  508. docType := "tweet"
  509. docID := "111"
  510. source := map[string]interface{}{
  511. "f1": "foo",
  512. "f2": "foo",
  513. }
  514. conn := NewClient(ESHost, ESPort)
  515. conn.DeleteIndex(indexName)
  516. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  517. c.Assert(err, IsNil)
  518. defer conn.DeleteIndex(indexName)
  519. d := Document{
  520. Index: indexName,
  521. Type: docType,
  522. ID: docID,
  523. Fields: source,
  524. }
  525. _, err = conn.Index(d, url.Values{})
  526. c.Assert(err, IsNil)
  527. response, err := conn.Get(indexName, docType, docID, url.Values{})
  528. c.Assert(err, IsNil)
  529. expectedResponse := &Response{
  530. Status: 200,
  531. Index: indexName,
  532. Type: docType,
  533. ID: docID,
  534. Version: 1,
  535. Found: true,
  536. Source: source,
  537. }
  538. response.Raw = nil
  539. c.Assert(response, DeepEquals, expectedResponse)
  540. fields := make(url.Values, 1)
  541. fields.Set("fields", "f1")
  542. response, err = conn.Get(indexName, docType, docID, fields)
  543. c.Assert(err, IsNil)
  544. expectedResponse = &Response{
  545. Status: 200,
  546. Index: indexName,
  547. Type: docType,
  548. ID: docID,
  549. Version: 1,
  550. Found: true,
  551. Fields: map[string]interface{}{
  552. "f1": []interface{}{"foo"},
  553. },
  554. }
  555. response.Raw = nil
  556. c.Assert(response, DeepEquals, expectedResponse)
  557. }
  558. func (s *GoesTestSuite) TestSearch(c *C) {
  559. indexName := "testsearch"
  560. docType := "tweet"
  561. docID := "1234"
  562. source := map[string]interface{}{
  563. "user": "foo",
  564. "message": "bar",
  565. }
  566. conn := NewClient(ESHost, ESPort)
  567. conn.DeleteIndex(indexName)
  568. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  569. c.Assert(err, IsNil)
  570. defer conn.DeleteIndex(indexName)
  571. d := Document{
  572. Index: indexName,
  573. Type: docType,
  574. ID: docID,
  575. Fields: source,
  576. }
  577. _, err = conn.Index(d, url.Values{})
  578. c.Assert(err, IsNil)
  579. _, err = conn.RefreshIndex(indexName)
  580. c.Assert(err, IsNil)
  581. // I can feel my eyes bleeding
  582. query := map[string]interface{}{
  583. "query": map[string]interface{}{
  584. "bool": map[string]interface{}{
  585. "must": []map[string]interface{}{
  586. {
  587. "match_all": map[string]interface{}{},
  588. },
  589. },
  590. },
  591. },
  592. }
  593. response, _ := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
  594. expectedHits := Hits{
  595. Total: 1,
  596. MaxScore: 1.0,
  597. Hits: []Hit{
  598. {
  599. Index: indexName,
  600. Type: docType,
  601. ID: docID,
  602. Score: 1.0,
  603. Source: source,
  604. },
  605. },
  606. }
  607. c.Assert(response.Hits, DeepEquals, expectedHits)
  608. }
  609. func (s *GoesTestSuite) TestCount(c *C) {
  610. indexName := "testcount"
  611. docType := "tweet"
  612. docID := "1234"
  613. source := map[string]interface{}{
  614. "user": "foo",
  615. "message": "bar",
  616. }
  617. conn := NewClient(ESHost, ESPort)
  618. conn.DeleteIndex(indexName)
  619. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  620. c.Assert(err, IsNil)
  621. defer conn.DeleteIndex(indexName)
  622. d := Document{
  623. Index: indexName,
  624. Type: docType,
  625. ID: docID,
  626. Fields: source,
  627. }
  628. _, err = conn.Index(d, url.Values{})
  629. c.Assert(err, IsNil)
  630. _, err = conn.RefreshIndex(indexName)
  631. c.Assert(err, IsNil)
  632. // I can feel my eyes bleeding
  633. query := map[string]interface{}{
  634. "query": map[string]interface{}{
  635. "bool": map[string]interface{}{
  636. "must": []map[string]interface{}{
  637. {
  638. "match_all": map[string]interface{}{},
  639. },
  640. },
  641. },
  642. },
  643. }
  644. response, _ := conn.Count(query, []string{indexName}, []string{docType}, url.Values{})
  645. c.Assert(response.Count, Equals, 1)
  646. }
  647. func (s *GoesTestSuite) TestIndexStatus(c *C) {
  648. indexName := "testindexstatus"
  649. conn := NewClient(ESHost, ESPort)
  650. conn.DeleteIndex(indexName)
  651. mapping := map[string]interface{}{
  652. "settings": map[string]interface{}{
  653. "index.number_of_shards": 1,
  654. "index.number_of_replicas": 1,
  655. },
  656. }
  657. _, err := conn.CreateIndex(indexName, mapping)
  658. c.Assert(err, IsNil)
  659. defer conn.DeleteIndex(indexName)
  660. // gives ES some time to do its job
  661. time.Sleep(1 * time.Second)
  662. _, err = conn.RefreshIndex(indexName)
  663. c.Assert(err, IsNil)
  664. response, err := conn.IndexStatus([]string{"testindexstatus"})
  665. c.Assert(err, IsNil)
  666. expectedShards := Shard{Total: 2, Successful: 1, Failed: 0}
  667. c.Assert(response.Shards, Equals, expectedShards)
  668. primarySizeInBytes := response.Indices[indexName].Index["primary_size_in_bytes"].(float64)
  669. sizeInBytes := response.Indices[indexName].Index["size_in_bytes"].(float64)
  670. refreshTotal := response.Indices[indexName].Refresh["total"].(float64)
  671. c.Assert(primarySizeInBytes > 0, Equals, true)
  672. c.Assert(sizeInBytes > 0, Equals, true)
  673. c.Assert(refreshTotal > 0, Equals, true)
  674. expectedIndices := map[string]IndexStatus{
  675. indexName: {
  676. Index: map[string]interface{}{
  677. "primary_size_in_bytes": primarySizeInBytes,
  678. "size_in_bytes": sizeInBytes,
  679. },
  680. Translog: map[string]uint64{
  681. "operations": 0,
  682. },
  683. Docs: map[string]uint64{
  684. "num_docs": 0,
  685. "max_doc": 0,
  686. "deleted_docs": 0,
  687. },
  688. Merges: map[string]interface{}{
  689. "current": float64(0),
  690. "current_docs": float64(0),
  691. "current_size_in_bytes": float64(0),
  692. "total": float64(0),
  693. "total_time_in_millis": float64(0),
  694. "total_docs": float64(0),
  695. "total_size_in_bytes": float64(0),
  696. },
  697. Refresh: map[string]interface{}{
  698. "total": refreshTotal,
  699. "total_time_in_millis": float64(0),
  700. },
  701. Flush: map[string]interface{}{
  702. "total": float64(0),
  703. "total_time_in_millis": float64(0),
  704. },
  705. },
  706. }
  707. c.Assert(response.Indices, DeepEquals, expectedIndices)
  708. }
  709. func (s *GoesTestSuite) TestScroll(c *C) {
  710. indexName := "testscroll"
  711. docType := "tweet"
  712. tweets := []Document{
  713. {
  714. ID: nil,
  715. Index: indexName,
  716. Type: docType,
  717. BulkCommand: BulkCommandIndex,
  718. Fields: map[string]interface{}{
  719. "user": "foo",
  720. "message": "some foo message",
  721. },
  722. },
  723. {
  724. ID: nil,
  725. Index: indexName,
  726. Type: docType,
  727. BulkCommand: BulkCommandIndex,
  728. Fields: map[string]interface{}{
  729. "user": "bar",
  730. "message": "some bar message",
  731. },
  732. },
  733. {
  734. ID: nil,
  735. Index: indexName,
  736. Type: docType,
  737. BulkCommand: BulkCommandIndex,
  738. Fields: map[string]interface{}{
  739. "user": "foo",
  740. "message": "another foo message",
  741. },
  742. },
  743. }
  744. conn := NewClient(ESHost, ESPort)
  745. mapping := map[string]interface{}{
  746. "settings": map[string]interface{}{
  747. "index.number_of_shards": 1,
  748. "index.number_of_replicas": 0,
  749. },
  750. }
  751. defer conn.DeleteIndex(indexName)
  752. _, err := conn.CreateIndex(indexName, mapping)
  753. c.Assert(err, IsNil)
  754. _, err = conn.BulkSend(tweets)
  755. c.Assert(err, IsNil)
  756. _, err = conn.RefreshIndex(indexName)
  757. c.Assert(err, IsNil)
  758. query := map[string]interface{}{
  759. "query": map[string]interface{}{
  760. "filtered": map[string]interface{}{
  761. "filter": map[string]interface{}{
  762. "term": map[string]interface{}{
  763. "user": "foo",
  764. },
  765. },
  766. },
  767. },
  768. }
  769. scan, err := conn.Scan(query, []string{indexName}, []string{docType}, "1m", 1)
  770. c.Assert(err, IsNil)
  771. c.Assert(len(scan.ScrollID) > 0, Equals, true)
  772. searchResults, err := conn.Scroll(scan.ScrollID, "1m")
  773. c.Assert(err, IsNil)
  774. // some data in first chunk
  775. c.Assert(searchResults.Hits.Total, Equals, uint64(2))
  776. c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
  777. c.Assert(len(searchResults.Hits.Hits), Equals, 1)
  778. searchResults, err = conn.Scroll(searchResults.ScrollID, "1m")
  779. c.Assert(err, IsNil)
  780. // more data in second chunk
  781. c.Assert(searchResults.Hits.Total, Equals, uint64(2))
  782. c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
  783. c.Assert(len(searchResults.Hits.Hits), Equals, 1)
  784. searchResults, err = conn.Scroll(searchResults.ScrollID, "1m")
  785. c.Assert(err, IsNil)
  786. // nothing in third chunk
  787. c.Assert(searchResults.Hits.Total, Equals, uint64(2))
  788. c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
  789. c.Assert(len(searchResults.Hits.Hits), Equals, 0)
  790. }
  791. func (s *GoesTestSuite) TestAggregations(c *C) {
  792. indexName := "testaggs"
  793. docType := "tweet"
  794. tweets := []Document{
  795. {
  796. ID: nil,
  797. Index: indexName,
  798. Type: docType,
  799. BulkCommand: BulkCommandIndex,
  800. Fields: map[string]interface{}{
  801. "user": "foo",
  802. "message": "some foo message",
  803. "age": 25,
  804. },
  805. },
  806. {
  807. ID: nil,
  808. Index: indexName,
  809. Type: docType,
  810. BulkCommand: BulkCommandIndex,
  811. Fields: map[string]interface{}{
  812. "user": "bar",
  813. "message": "some bar message",
  814. "age": 30,
  815. },
  816. },
  817. {
  818. ID: nil,
  819. Index: indexName,
  820. Type: docType,
  821. BulkCommand: BulkCommandIndex,
  822. Fields: map[string]interface{}{
  823. "user": "foo",
  824. "message": "another foo message",
  825. },
  826. },
  827. }
  828. conn := NewClient(ESHost, ESPort)
  829. mapping := map[string]interface{}{
  830. "settings": map[string]interface{}{
  831. "index.number_of_shards": 1,
  832. "index.number_of_replicas": 0,
  833. },
  834. }
  835. defer conn.DeleteIndex(indexName)
  836. _, err := conn.CreateIndex(indexName, mapping)
  837. c.Assert(err, IsNil)
  838. _, err = conn.BulkSend(tweets)
  839. c.Assert(err, IsNil)
  840. _, err = conn.RefreshIndex(indexName)
  841. c.Assert(err, IsNil)
  842. query := map[string]interface{}{
  843. "aggs": map[string]interface{}{
  844. "user": map[string]interface{}{
  845. "terms": map[string]interface{}{
  846. "field": "user",
  847. "order": map[string]interface{}{
  848. "_term": "asc",
  849. },
  850. },
  851. "aggs": map[string]interface{}{
  852. "age": map[string]interface{}{
  853. "stats": map[string]interface{}{
  854. "field": "age",
  855. },
  856. },
  857. },
  858. },
  859. "age": map[string]interface{}{
  860. "stats": map[string]interface{}{
  861. "field": "age",
  862. },
  863. },
  864. },
  865. }
  866. resp, _ := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
  867. user, ok := resp.Aggregations["user"]
  868. c.Assert(ok, Equals, true)
  869. c.Assert(len(user.Buckets()), Equals, 2)
  870. c.Assert(user.Buckets()[0].Key(), Equals, "bar")
  871. c.Assert(user.Buckets()[1].Key(), Equals, "foo")
  872. barAge := user.Buckets()[0].Aggregation("age")
  873. c.Assert(barAge["count"], Equals, 1.0)
  874. c.Assert(barAge["sum"], Equals, 30.0)
  875. fooAge := user.Buckets()[1].Aggregation("age")
  876. c.Assert(fooAge["count"], Equals, 1.0)
  877. c.Assert(fooAge["sum"], Equals, 25.0)
  878. age, ok := resp.Aggregations["age"]
  879. c.Assert(ok, Equals, true)
  880. c.Assert(age["count"], Equals, 2.0)
  881. c.Assert(age["sum"], Equals, 25.0+30.0)
  882. }
  883. func (s *GoesTestSuite) TestPutMapping(c *C) {
  884. indexName := "testputmapping"
  885. docType := "tweet"
  886. conn := NewClient(ESHost, ESPort)
  887. // just in case
  888. conn.DeleteIndex(indexName)
  889. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  890. c.Assert(err, IsNil)
  891. defer conn.DeleteIndex(indexName)
  892. d := Document{
  893. Index: indexName,
  894. Type: docType,
  895. Fields: map[string]interface{}{
  896. "user": "foo",
  897. "message": "bar",
  898. },
  899. }
  900. response, err := conn.Index(d, url.Values{})
  901. c.Assert(err, IsNil)
  902. mapping := map[string]interface{}{
  903. "tweet": map[string]interface{}{
  904. "properties": map[string]interface{}{
  905. "count": map[string]interface{}{
  906. "type": "integer",
  907. "index": "not_analyzed",
  908. "store": true,
  909. },
  910. },
  911. },
  912. }
  913. response, err = conn.PutMapping("tweet", mapping, []string{indexName})
  914. c.Assert(err, IsNil)
  915. c.Assert(response.Acknowledged, Equals, true)
  916. c.Assert(response.TimedOut, Equals, false)
  917. }
  918. func (s *GoesTestSuite) TestIndicesExist(c *C) {
  919. indices := []string{"testindicesexist"}
  920. conn := NewClient(ESHost, ESPort)
  921. // just in case
  922. conn.DeleteIndex(indices[0])
  923. exists, err := conn.IndicesExist(indices)
  924. c.Assert(exists, Equals, false)
  925. _, err = conn.CreateIndex(indices[0], map[string]interface{}{})
  926. c.Assert(err, IsNil)
  927. defer conn.DeleteIndex(indices[0])
  928. time.Sleep(200 * time.Millisecond)
  929. exists, _ = conn.IndicesExist(indices)
  930. c.Assert(exists, Equals, true)
  931. indices = append(indices, "nonexistent")
  932. exists, _ = conn.IndicesExist(indices)
  933. c.Assert(exists, Equals, false)
  934. }
  935. func (s *GoesTestSuite) TestUpdate(c *C) {
  936. indexName := "testupdate"
  937. docType := "tweet"
  938. docID := "1234"
  939. conn := NewClient(ESHost, ESPort)
  940. // just in case
  941. conn.DeleteIndex(indexName)
  942. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  943. c.Assert(err, IsNil)
  944. defer conn.DeleteIndex(indexName)
  945. d := Document{
  946. Index: indexName,
  947. Type: docType,
  948. ID: docID,
  949. Fields: map[string]interface{}{
  950. "user": "foo",
  951. "message": "bar",
  952. "counter": 1,
  953. },
  954. }
  955. extraArgs := make(url.Values, 1)
  956. response, err := conn.Index(d, extraArgs)
  957. c.Assert(err, IsNil)
  958. time.Sleep(200 * time.Millisecond)
  959. expectedResponse := &Response{
  960. Status: 201,
  961. Index: indexName,
  962. ID: docID,
  963. Type: docType,
  964. Version: 1,
  965. }
  966. response.Raw = nil
  967. c.Assert(response, DeepEquals, expectedResponse)
  968. // Now that we have an ordinary document indexed, try updating it
  969. query := map[string]interface{}{
  970. "script": "ctx._source.counter += count",
  971. "lang": "groovy",
  972. "params": map[string]interface{}{
  973. "count": 5,
  974. },
  975. "upsert": map[string]interface{}{
  976. "message": "candybar",
  977. "user": "admin",
  978. "counter": 1,
  979. },
  980. }
  981. response, err = conn.Update(d, query, extraArgs)
  982. if err != nil && strings.Contains(err.(*SearchError).Msg, "dynamic scripting") {
  983. c.Skip("Scripting is disabled on server, skipping this test")
  984. return
  985. }
  986. time.Sleep(200 * time.Millisecond)
  987. c.Assert(err, Equals, nil)
  988. response, err = conn.Get(indexName, docType, docID, url.Values{})
  989. c.Assert(err, Equals, nil)
  990. c.Assert(response.Source["counter"], Equals, float64(6))
  991. c.Assert(response.Source["user"], Equals, "foo")
  992. c.Assert(response.Source["message"], Equals, "bar")
  993. // Test another document, non-existent
  994. docID = "555"
  995. d.ID = docID
  996. response, err = conn.Update(d, query, extraArgs)
  997. c.Assert(err, Equals, nil)
  998. time.Sleep(200 * time.Millisecond)
  999. response, err = conn.Get(indexName, docType, docID, url.Values{})
  1000. c.Assert(err, Equals, nil)
  1001. c.Assert(response.Source["user"], Equals, "admin")
  1002. c.Assert(response.Source["message"], Equals, "candybar")
  1003. }
  1004. func (s *GoesTestSuite) TestGetMapping(c *C) {
  1005. indexName := "testmapping"
  1006. docType := "tweet"
  1007. conn := NewClient(ESHost, ESPort)
  1008. // just in case
  1009. conn.DeleteIndex(indexName)
  1010. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  1011. c.Assert(err, IsNil)
  1012. defer conn.DeleteIndex(indexName)
  1013. time.Sleep(300 * time.Millisecond)
  1014. response, err := conn.GetMapping([]string{docType}, []string{indexName})
  1015. c.Assert(err, Equals, nil)
  1016. c.Assert(len(response.Raw), Equals, 0)
  1017. d := Document{
  1018. Index: indexName,
  1019. Type: docType,
  1020. Fields: map[string]interface{}{
  1021. "user": "foo",
  1022. "message": "bar",
  1023. },
  1024. }
  1025. response, err = conn.Index(d, url.Values{})
  1026. c.Assert(err, IsNil)
  1027. time.Sleep(200 * time.Millisecond)
  1028. response, err = conn.GetMapping([]string{docType}, []string{indexName})
  1029. c.Assert(err, Equals, nil)
  1030. c.Assert(len(response.Raw), Not(Equals), 0)
  1031. }
  1032. func (s *GoesTestSuite) TestDeleteMapping(c *C) {
  1033. indexName := "testdeletemapping"
  1034. docType := "tweet"
  1035. conn := NewClient(ESHost, ESPort)
  1036. // just in case
  1037. conn.DeleteIndex(indexName)
  1038. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  1039. c.Assert(err, IsNil)
  1040. defer conn.DeleteIndex(indexName)
  1041. d := Document{
  1042. Index: indexName,
  1043. Type: docType,
  1044. Fields: map[string]interface{}{
  1045. "user": "foo",
  1046. "message": "bar",
  1047. },
  1048. }
  1049. response, err := conn.Index(d, url.Values{})
  1050. c.Assert(err, IsNil)
  1051. mapping := map[string]interface{}{
  1052. "tweet": map[string]interface{}{
  1053. "properties": map[string]interface{}{
  1054. "count": map[string]interface{}{
  1055. "type": "integer",
  1056. "index": "not_analyzed",
  1057. "store": true,
  1058. },
  1059. },
  1060. },
  1061. }
  1062. response, err = conn.PutMapping("tweet", mapping, []string{indexName})
  1063. c.Assert(err, IsNil)
  1064. time.Sleep(200 * time.Millisecond)
  1065. response, err = conn.DeleteMapping("tweet", []string{indexName})
  1066. c.Assert(err, IsNil)
  1067. c.Assert(response.Acknowledged, Equals, true)
  1068. c.Assert(response.TimedOut, Equals, false)
  1069. }
  1070. func (s *GoesTestSuite) TestAddAlias(c *C) {
  1071. aliasName := "testAlias"
  1072. indexName := "testalias_1"
  1073. docType := "testDoc"
  1074. docID := "1234"
  1075. source := map[string]interface{}{
  1076. "user": "foo",
  1077. "message": "bar",
  1078. }
  1079. conn := NewClient(ESHost, ESPort)
  1080. defer conn.DeleteIndex(indexName)
  1081. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  1082. c.Assert(err, IsNil)
  1083. defer conn.DeleteIndex(indexName)
  1084. d := Document{
  1085. Index: indexName,
  1086. Type: docType,
  1087. ID: docID,
  1088. Fields: source,
  1089. }
  1090. // Index data
  1091. _, err = conn.Index(d, url.Values{})
  1092. c.Assert(err, IsNil)
  1093. // Add alias
  1094. _, err = conn.AddAlias(aliasName, []string{indexName})
  1095. c.Assert(err, IsNil)
  1096. // Get document via alias
  1097. response, err := conn.Get(aliasName, docType, docID, url.Values{})
  1098. c.Assert(err, IsNil)
  1099. expectedResponse := &Response{
  1100. Status: 200,
  1101. Index: indexName,
  1102. Type: docType,
  1103. ID: docID,
  1104. Version: 1,
  1105. Found: true,
  1106. Source: source,
  1107. }
  1108. response.Raw = nil
  1109. c.Assert(response, DeepEquals, expectedResponse)
  1110. }
  1111. func (s *GoesTestSuite) TestRemoveAlias(c *C) {
  1112. aliasName := "testAlias"
  1113. indexName := "testalias_1"
  1114. docType := "testDoc"
  1115. docID := "1234"
  1116. source := map[string]interface{}{
  1117. "user": "foo",
  1118. "message": "bar",
  1119. }
  1120. conn := NewClient(ESHost, ESPort)
  1121. defer conn.DeleteIndex(indexName)
  1122. _, err := conn.CreateIndex(indexName, map[string]interface{}{})
  1123. c.Assert(err, IsNil)
  1124. defer conn.DeleteIndex(indexName)
  1125. d := Document{
  1126. Index: indexName,
  1127. Type: docType,
  1128. ID: docID,
  1129. Fields: source,
  1130. }
  1131. // Index data
  1132. _, err = conn.Index(d, url.Values{})
  1133. c.Assert(err, IsNil)
  1134. // Add alias
  1135. _, err = conn.AddAlias(aliasName, []string{indexName})
  1136. c.Assert(err, IsNil)
  1137. // Remove alias
  1138. _, err = conn.RemoveAlias(aliasName, []string{indexName})
  1139. c.Assert(err, IsNil)
  1140. // Get document via alias
  1141. _, err = conn.Get(aliasName, docType, docID, url.Values{})
  1142. c.Assert(err.Error(), Equals, "[404] IndexMissingException[["+aliasName+"] missing]")
  1143. }
  1144. func (s *GoesTestSuite) TestAliasExists(c *C) {
  1145. index := "testaliasexist_1"
  1146. alias := "testaliasexists"
  1147. conn := NewClient(ESHost, ESPort)
  1148. // just in case
  1149. conn.DeleteIndex(index)
  1150. exists, err := conn.AliasExists(alias)
  1151. c.Assert(exists, Equals, false)
  1152. _, err = conn.CreateIndex(index, map[string]interface{}{})
  1153. c.Assert(err, IsNil)
  1154. defer conn.DeleteIndex(index)
  1155. time.Sleep(200 * time.Millisecond)
  1156. _, err = conn.AddAlias(alias, []string{index})
  1157. c.Assert(err, IsNil)
  1158. time.Sleep(200 * time.Millisecond)
  1159. defer conn.RemoveAlias(alias, []string{index})
  1160. exists, _ = conn.AliasExists(alias)
  1161. c.Assert(exists, Equals, true)
  1162. }