forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add support for the bulk API on indexing (elastic#10)
This implementation uses a channel to pass the bulk messages. The caller needs to write the operations to the channel, close the channel and call the Bulk method.
- Loading branch information
1 parent
978034c
commit ae75cd7
Showing
8 changed files
with
468 additions
and
70 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| package elasticsearch | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
| "fmt" | ||
| "io/ioutil" | ||
| "net/http" | ||
| "time" | ||
|
|
||
| "github.com/elastic/libbeat/common" | ||
| ) | ||
|
|
||
| type BulkMsg struct { | ||
| Ts time.Time | ||
| Event common.MapStr | ||
| } | ||
|
|
||
| func (es *Elasticsearch) Bulk(index string, doc_type string, | ||
| params map[string]string, body chan interface{}) (*QueryResult, error) { | ||
|
|
||
| path, err := MakePath(index, doc_type, "_bulk") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| var buf bytes.Buffer | ||
| enc := json.NewEncoder(&buf) | ||
| for obj := range body { | ||
| enc.Encode(obj) | ||
| } | ||
|
|
||
| url := es.Url + path | ||
| if len(params) > 0 { | ||
| url = url + "?" + UrlEncode(params) | ||
| } | ||
|
|
||
| req, err := http.NewRequest("POST", url, &buf) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| resp, err := es.client.Do(req) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| defer resp.Body.Close() | ||
| obj, err := ioutil.ReadAll(resp.Body) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| var result QueryResult | ||
| err = json.Unmarshal(obj, &result) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| if resp.StatusCode > 299 { | ||
| return &result, fmt.Errorf("ES returned an error: %s", resp.Status) | ||
| } | ||
| return &result, err | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,158 @@ | ||
| package elasticsearch | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "os" | ||
| "testing" | ||
|
|
||
| "github.com/elastic/libbeat/logp" | ||
| ) | ||
|
|
||
| func TestBulk(t *testing.T) { | ||
| if testing.Verbose() { | ||
| logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"elasticsearch"}) | ||
| } | ||
| if testing.Short() { | ||
| t.Skip("Skipping in short mode, because it requires Elasticsearch") | ||
| } | ||
| es := NewElasticsearch("http://localhost:9200") | ||
| index := fmt.Sprintf("packetbeat-unittest-%d", os.Getpid()) | ||
|
|
||
| ops := []map[string]interface{}{ | ||
| map[string]interface{}{ | ||
| "index": map[string]interface{}{ | ||
| "_index": index, | ||
| "_type": "type1", | ||
| "_id": "1", | ||
| }, | ||
| }, | ||
| map[string]interface{}{ | ||
| "field1": "value1", | ||
| }, | ||
| } | ||
|
|
||
| body := make(chan interface{}, 10) | ||
| for _, op := range ops { | ||
| body <- op | ||
| } | ||
| close(body) | ||
|
|
||
| params := map[string]string{ | ||
| "refresh": "true", | ||
| } | ||
| _, err := es.Bulk(index, "type1", params, body) | ||
| if err != nil { | ||
| t.Errorf("Bulk() returned error: %s", err) | ||
| } | ||
|
|
||
| params = map[string]string{ | ||
| "q": "field1:value1", | ||
| } | ||
| result, err := es.SearchUri(index, "type1", params) | ||
| if err != nil { | ||
| t.Errorf("SearchUri() returns an error: %s", err) | ||
| } | ||
| if result.Hits.Total != 1 { | ||
| t.Errorf("Wrong number of search results: %d", result.Hits.Total) | ||
| } | ||
|
|
||
| _, err = es.Delete(index, "", "", nil) | ||
| if err != nil { | ||
| t.Errorf("Delete() returns error: %s", err) | ||
| } | ||
| } | ||
|
|
||
| func TestBulkMoreOperations(t *testing.T) { | ||
| if testing.Verbose() { | ||
| logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"elasticsearch"}) | ||
| } | ||
| if testing.Short() { | ||
| t.Skip("Skipping in short mode, because it requires Elasticsearch") | ||
| } | ||
| es := NewElasticsearch("http://localhost:9200") | ||
| index := fmt.Sprintf("packetbeat-unittest-%d", os.Getpid()) | ||
|
|
||
| ops := []map[string]interface{}{ | ||
| map[string]interface{}{ | ||
| "index": map[string]interface{}{ | ||
| "_index": index, | ||
| "_type": "type1", | ||
| "_id": "1", | ||
| }, | ||
| }, | ||
| map[string]interface{}{ | ||
| "field1": "value1", | ||
| }, | ||
| map[string]interface{}{ | ||
| "delete": map[string]interface{}{ | ||
| "_index": index, | ||
| "_type": "type1", | ||
| "_id": "2", | ||
| }, | ||
| }, | ||
| map[string]interface{}{ | ||
| "create": map[string]interface{}{ | ||
| "_index": index, | ||
| "_type": "type1", | ||
| "_id": "3", | ||
| }, | ||
| }, | ||
| map[string]interface{}{ | ||
| "field1": "value3", | ||
| }, | ||
| map[string]interface{}{ | ||
| "update": map[string]interface{}{ | ||
| "_id": "1", | ||
| "_index": index, | ||
| "_type": "type1", | ||
| }, | ||
| }, | ||
| map[string]interface{}{ | ||
| "doc": map[string]interface{}{ | ||
| "field2": "value2", | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| body := make(chan interface{}, 10) | ||
| for _, op := range ops { | ||
| body <- op | ||
| } | ||
| close(body) | ||
|
|
||
| params := map[string]string{ | ||
| "refresh": "true", | ||
| } | ||
| resp, err := es.Bulk(index, "type1", params, body) | ||
| if err != nil { | ||
| t.Errorf("Bulk() returned error: %s [%s]", err, resp) | ||
| return | ||
| } | ||
|
|
||
| params = map[string]string{ | ||
| "q": "field1:value3", | ||
| } | ||
| result, err := es.SearchUri(index, "type1", params) | ||
| if err != nil { | ||
| t.Errorf("SearchUri() returns an error: %s", err) | ||
| } | ||
| if result.Hits.Total != 1 { | ||
| t.Errorf("Wrong number of search results: %d", result.Hits.Total) | ||
| } | ||
|
|
||
| params = map[string]string{ | ||
| "q": "field2:value2", | ||
| } | ||
| result, err = es.SearchUri(index, "type1", params) | ||
| if err != nil { | ||
| t.Errorf("SearchUri() returns an error: %s", err) | ||
| } | ||
| if result.Hits.Total != 1 { | ||
| t.Errorf("Wrong number of search results: %d", result.Hits.Total) | ||
| } | ||
|
|
||
| _, err = es.Delete(index, "", "", nil) | ||
| if err != nil { | ||
| t.Errorf("Delete() returns error: %s", err) | ||
| } | ||
| } |
Oops, something went wrong.