From 3c0b179f212fd9267992010349f792ba08e808b1 Mon Sep 17 00:00:00 2001 From: "Burgett, Gordon" Date: Thu, 6 Aug 2015 12:51:35 -0500 Subject: [PATCH 1/5] Improved error handling for bulk indexing to provide more information to the error channel --- lib/baserequest.go | 10 ++++-- lib/corebulk.go | 40 ++++++++++++++++++++--- lib/corebulk_test.go | 75 ++++++++++++++++++++++++++++++++++++++++++++ lib/request.go | 13 +++----- 4 files changed, 123 insertions(+), 15 deletions(-) diff --git a/lib/baserequest.go b/lib/baserequest.go index e44ee9e9..fac71d27 100644 --- a/lib/baserequest.go +++ b/lib/baserequest.go @@ -64,10 +64,14 @@ func (c *Conn) DoCommand(method string, url string, args map[string]interface{}, httpStatusCode, body, err = req.Do(&response) if err != nil { - return body, err + if httpStatusCode == -1 { + //connection error like *url.Error + return body, err + } + //error reading response body, or something else after we obtained an HTTP status code + return body, ESError{time.Now(), fmt.Sprintf("Error [%v] Status [%v]", err, httpStatusCode), httpStatusCode} } if httpStatusCode > 304 { - jsonErr := json.Unmarshal(body, &response) if jsonErr == nil { if res_err, ok := response["error"]; ok { @@ -75,7 +79,7 @@ func (c *Conn) DoCommand(method string, url string, args map[string]interface{}, return body, ESError{time.Now(), fmt.Sprintf("Error [%s] Status [%v]", res_err, status), httpStatusCode} } } - return body, jsonErr + return body, ESError{time.Now(), fmt.Sprintf("Error [%v] Status [%v]", jsonErr, httpStatusCode), httpStatusCode} } return body, nil } diff --git a/lib/corebulk.go b/lib/corebulk.go index 5010f843..9d5bcc2a 100644 --- a/lib/corebulk.go +++ b/lib/corebulk.go @@ -38,6 +38,15 @@ type ErrorBuffer struct { Buf *bytes.Buffer } +// An error implementation which contains the actual items in the bulk indexing response. +type BulkIndexingError struct { + Items []map[string]interface{} +} + +func (e BulkIndexingError) Error() string{ + return fmt.Sprintf("Bulk Insertion Error. Failed item count [%d]", len(e.Items)) +} + // A bulk indexer creates goroutines, and channels for connecting and sending data // to elasticsearch in bulk, using buffers. type BulkIndexer struct { @@ -188,7 +197,7 @@ func (b *BulkIndexer) startHttpSender() { go func() { for buf := range b.sendBuf { b.sendWg.Add(1) - // Copy for the potential re-send. + // Copy so we can put the buffer on the error channel, or potentially re-send it. bufCopy := bytes.NewBuffer(buf.Bytes()) err := b.Sender(buf) @@ -197,7 +206,10 @@ func (b *BulkIndexer) startHttpSender() { // 2. Retry then return error and let runner decide // 3. Retry, then log to disk? retry later? if err != nil { + buf = bufCopy if b.RetryForSeconds > 0 { + //copy again so we can keep the original buffer for the error channel. + bufCopy := bytes.NewBuffer(buf.Bytes()) time.Sleep(time.Second * time.Duration(b.RetryForSeconds)) err = b.Sender(bufCopy) if err == nil { @@ -331,7 +343,7 @@ func (b *BulkIndexer) Send(buf *bytes.Buffer) error { type responseStruct struct { Took int64 `json:"took"` Errors bool `json:"errors"` - Items []map[string]interface{} `json:"items"` + Items *json.RawMessage `json:"items"` } response := responseStruct{} @@ -345,10 +357,30 @@ func (b *BulkIndexer) Send(buf *bytes.Buffer) error { // check for response errors, bulk insert will give 200 OK but then include errors in response jsonErr := json.Unmarshal(body, &response) if jsonErr == nil { + //unmarshal the errors only if we need to if response.Errors { - b.numErrors += uint64(len(response.Items)) - return fmt.Errorf("Bulk Insertion Error. Failed item count [%d]", len(response.Items)) + var items []map[string]interface{} + jsonErr = json.Unmarshal([]byte(*response.Items), &items) + if jsonErr != nil { + return jsonErr + } + + for _, item := range items { + for _, body := range item { + body, ok := body.(map[string]interface{}) + if !ok { + continue + } + if status, ok := body["status"]; ok && status.(float64) > 304 { + b.numErrors++ + } + } + } + + return BulkIndexingError{items} } + } else { + return jsonErr } return nil } diff --git a/lib/corebulk_test.go b/lib/corebulk_test.go index 3a6b615e..33bee118 100644 --- a/lib/corebulk_test.go +++ b/lib/corebulk_test.go @@ -24,6 +24,7 @@ import ( "github.com/araddon/gou" "github.com/bmizerany/assert" + "strings" ) // go test -bench=".*" @@ -91,6 +92,7 @@ func TestBulkIndexerBasic(t *testing.T) { <-time.After(time.Millisecond * 10) // we need to wait for doc to hit send channel // this will test to ensure that Flush actually catches a doc indexer.Flush() + <-time.After(time.Millisecond * 1) // we need to wait for the httpSender to read from the send buffer totalBytesSent = totalBytesSent - len(*eshost) assert.T(t, err == nil, fmt.Sprintf("Should have nil error =%v", err)) assert.T(t, len(buffers) == 2, fmt.Sprintf("Should have another buffer ct=%d", len(buffers))) @@ -102,6 +104,79 @@ func TestBulkIndexerBasic(t *testing.T) { indexer.Stop() } +func TestBulkIndexerErrors(t *testing.T){ + testIndex := "users" + InitTests(true) + c := NewTestConn() + + c.DeleteIndex(testIndex) + + sent := make(chan struct{}, 1) + + indexer := c.NewBulkIndexer(3) + indexer.Start() + indexer.Sender = func(buf *bytes.Buffer) error { + // log.Printf("buffer:%s", string(buf.Bytes())) + ret := indexer.Send(buf) + sent<-struct{}{} + return ret + } + errch := make(chan *ErrorBuffer, 1) + indexer.ErrorChannel = errch + + date := time.Unix(1257894000, 0) + + data := map[string]interface{}{ + "name": "smurfettes", + "age": 21, + "date": "today", + } + data2 := map[string]interface{}{ + "name": "smurfs", + "age": "this is not an int", + "date": "yesterday", + } + + _, err := c.DoCommand("PUT", fmt.Sprintf("/%s", testIndex), nil, + `{ + "mappings": { + "user": { + "properties": { + "age": { "type": "integer" } + } + } + } + }`) + assert.T(t, err == nil, fmt.Sprintf("Should not return an error: %v", err)) + + //act + err = indexer.Index(testIndex, "user", "1", "", &date, data, true) + err2 := indexer.Index(testIndex, "user", "2", "", &date, data2, true) + + <-sent + //assert + assert.T(t, err == nil, fmt.Sprintf("Should not return an error: %v", err)) + assert.T(t, err2 == nil, fmt.Sprintf("Should not return an error: %v", err2)) + time.Sleep(1 * time.Microsecond) + assert.T(t, indexer.NumErrors() == 1, fmt.Sprintf("Should have recorded 1 error but saw: %d", indexer.NumErrors())) + + errBuf := <- errch + bulkErr, ok := errBuf.Err.(BulkIndexingError) + assert.T(t, ok, fmt.Sprintf("Error should have been a BulkIndexingError but was %T", errBuf.Err)) + + assert.T(t, len(bulkErr.Items) == 2, fmt.Sprintf("Expected 2 items in response but got: %d", len(bulkErr.Items))) + status1 := int(bulkErr.Items[1]["index"].(map[string]interface{})["status"].(float64)) + assert.T(t, status1 == 400, fmt.Sprintf("Expected second item to have status 400 but was %d", status1)) + status0 := int(bulkErr.Items[0]["index"].(map[string]interface{})["status"].(float64)) + assert.T(t, status0 == 201, fmt.Sprintf("Expected first item to have status 201 but was %d", status0)) + + lines := strings.Split(errBuf.Buf.String(), "\n") + assert.T(t, lines[0] == `{"index":{"_index":"users","_type":"user","_id":"1","_timestamp":"1257894000000","refresh":true}}`, fmt.Sprintf("Expected index header but got: %s", lines[0])) + assert.T(t, lines[1] == `{"age":21,"date":"today","name":"smurfettes"}`, fmt.Sprintf("Expected document but got: %s", lines[1])) + assert.T(t, lines[2] == `{"index":{"_index":"users","_type":"user","_id":"2","_timestamp":"1257894000000","refresh":true}}`, fmt.Sprintf("Expected index header but got: %s", lines[2])) + assert.T(t, lines[3] == `{"age":"this is not an int","date":"yesterday","name":"smurfs"}`, fmt.Sprintf("Expected document but got: %s", lines[3])) +} + // currently broken in drone.io func XXXTestBulkUpdate(t *testing.T) { var ( diff --git a/lib/request.go b/lib/request.go index b4ac4b34..13e15902 100644 --- a/lib/request.go +++ b/lib/request.go @@ -68,8 +68,8 @@ func (r *Request) SetBody(body io.Reader) { func (r *Request) Do(v interface{}) (int, []byte, error) { response, bodyBytes, err := r.DoResponse(v) - if err != nil { - return -1, nil, err + if response == nil { + return -1, bodyBytes, err } return response.StatusCode, bodyBytes, err } @@ -91,18 +91,15 @@ func (r *Request) DoResponse(v interface{}) (*http.Response, []byte, error) { bodyBytes, err := ioutil.ReadAll(res.Body) if err != nil { - return nil, nil, err + return res, nil, err } if res.StatusCode == 404 { - return nil, bodyBytes, RecordNotFound + return res, bodyBytes, RecordNotFound } if res.StatusCode > 304 && v != nil { - jsonErr := json.Unmarshal(bodyBytes, v) - if jsonErr != nil { - return nil, nil, jsonErr - } + err = json.Unmarshal(bodyBytes, v) } return res, bodyBytes, err } From 169c06b8a5ca477d96548c72f4f61453dfe92c1d Mon Sep 17 00:00:00 2001 From: "Burgett, Gordon" Date: Tue, 18 Aug 2015 15:37:46 -0500 Subject: [PATCH 2/5] Added ability to specify version on bulk updates --- lib/baserequest.go | 12 +-- lib/corebulk.go | 72 ++++++++++++---- lib/corebulk_test.go | 194 +++++++++++++++++++++++++++++++++++++++---- lib/coretest_test.go | 2 +- 4 files changed, 243 insertions(+), 37 deletions(-) diff --git a/lib/baserequest.go b/lib/baserequest.go index fac71d27..c96ec90b 100644 --- a/lib/baserequest.go +++ b/lib/baserequest.go @@ -53,12 +53,14 @@ func (c *Conn) DoCommand(method string, url string, args map[string]interface{}, // Copy request body for tracer if c.RequestTracer != nil { - requestBody, err := ioutil.ReadAll(req.Body) - if err != nil { - return body, err + requestBody := []byte{} + if req.Body != nil { + requestBody, err = ioutil.ReadAll(req.Body) + if err != nil { + return body, err + } + req.SetBody(bytes.NewReader(requestBody)) } - - req.SetBody(bytes.NewReader(requestBody)) c.RequestTracer(req.Method, req.URL.String(), string(requestBody)) } diff --git a/lib/corebulk.go b/lib/corebulk.go index 9d5bcc2a..f6e6c4ad 100644 --- a/lib/corebulk.go +++ b/lib/corebulk.go @@ -43,10 +43,38 @@ type BulkIndexingError struct { Items []map[string]interface{} } -func (e BulkIndexingError) Error() string{ +func (e BulkIndexingError) Error() string { return fmt.Sprintf("Bulk Insertion Error. Failed item count [%d]", len(e.Items)) } +type DocVersion struct { + // The version to assign to the document + Version int64 + + // The Version Type to assign to the document + VersionType VersionType +} + +// An enum representing the various allowed version types. +// see https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-index_.html#_version_types +type VersionType string + +const ( + INTERNAL VersionType = "internal" + EXTERNAL VersionType = "external" + EXTERNAL_GT VersionType = "external_gt" + EXTERNAL_GTE VersionType = "external_gte" + FORCE VersionType = "force" +) + +// Creates a DocVersion struct with the given value and this version type. +func (t VersionType) V(v int64) *DocVersion { + return &DocVersion{ + Version: v, + VersionType: t, + } +} + // A bulk indexer creates goroutines, and channels for connecting and sending data // to elasticsearch in bulk, using buffers. type BulkIndexer struct { @@ -293,9 +321,9 @@ func (b *BulkIndexer) shutdown() { // The index bulk API adds or updates a typed JSON document to a specific index, making it searchable. // it operates by buffering requests, and ocassionally flushing to elasticsearch // http://www.elasticsearch.org/guide/reference/api/bulk.html -func (b *BulkIndexer) Index(index string, _type string, id, ttl string, date *time.Time, data interface{}, refresh bool) error { +func (b *BulkIndexer) Index(index string, _type string, id, ttl string, date *time.Time, version *DocVersion, data interface{}, refresh bool) error { //{ "index" : { "_index" : "test", "_type" : "type1", "_id" : "1" } } - by, err := WriteBulkBytes("index", index, _type, id, ttl, date, data, refresh) + by, err := WriteBulkBytes("index", index, _type, id, ttl, date, version, data, refresh) if err != nil { return err } @@ -303,9 +331,9 @@ func (b *BulkIndexer) Index(index string, _type string, id, ttl string, date *ti return nil } -func (b *BulkIndexer) Update(index string, _type string, id, ttl string, date *time.Time, data interface{}, refresh bool) error { +func (b *BulkIndexer) Update(index string, _type string, id, ttl string, date *time.Time, version *DocVersion, data interface{}, refresh bool) error { //{ "index" : { "_index" : "test", "_type" : "type1", "_id" : "1" } } - by, err := WriteBulkBytes("update", index, _type, id, ttl, date, data, refresh) + by, err := WriteBulkBytes("update", index, _type, id, ttl, date, version, data, refresh) if err != nil { return err } @@ -313,8 +341,15 @@ func (b *BulkIndexer) Update(index string, _type string, id, ttl string, date *t return nil } -func (b *BulkIndexer) Delete(index, _type, id string, refresh bool) { - queryLine := fmt.Sprintf("{\"delete\":{\"_index\":%q,\"_type\":%q,\"_id\":%q,\"refresh\":%t}}\n", index, _type, id, refresh) +func (b *BulkIndexer) Delete(index, _type, id string, version *DocVersion, refresh bool) { + verStr := "" + if version != nil { + verStr = fmt.Sprintf(",\"_version\":%d", version.Version) + if version.VersionType != "" { + verStr = verStr + ",\"_version_type\":\"" + string(version.VersionType) + "\"" + } + } + queryLine := fmt.Sprintf("{\"delete\":{\"_index\":%q,\"_type\":%q,\"_id\":%q%s,\"refresh\":%t}}\n", index, _type, id, verStr, refresh) b.bulkChannel <- []byte(queryLine) return } @@ -323,10 +358,10 @@ func (b *BulkIndexer) UpdateWithWithScript(index string, _type string, id, ttl s var data map[string]interface{} = make(map[string]interface{}) data["script"] = script - return b.Update(index, _type, id, ttl, date, data, refresh) + return b.Update(index, _type, id, ttl, date, nil, data, refresh) } -func (b *BulkIndexer) UpdateWithPartialDoc(index string, _type string, id, ttl string, date *time.Time, partialDoc interface{}, upsert bool, refresh bool) error { +func (b *BulkIndexer) UpdateWithPartialDoc(index string, _type string, id, ttl string, date *time.Time, version *DocVersion, partialDoc interface{}, upsert bool, refresh bool) error { var data map[string]interface{} = make(map[string]interface{}) @@ -334,16 +369,16 @@ func (b *BulkIndexer) UpdateWithPartialDoc(index string, _type string, id, ttl s if upsert { data["doc_as_upsert"] = true } - return b.Update(index, _type, id, ttl, date, data, refresh) + return b.Update(index, _type, id, ttl, date, version, data, refresh) } // This does the actual send of a buffer, which has already been formatted // into bytes of ES formatted bulk data func (b *BulkIndexer) Send(buf *bytes.Buffer) error { type responseStruct struct { - Took int64 `json:"took"` - Errors bool `json:"errors"` - Items *json.RawMessage `json:"items"` + Took int64 `json:"took"` + Errors bool `json:"errors"` + Items *json.RawMessage `json:"items"` } response := responseStruct{} @@ -387,7 +422,7 @@ func (b *BulkIndexer) Send(buf *bytes.Buffer) error { // Given a set of arguments for index, type, id, data create a set of bytes that is formatted for bulkd index // http://www.elasticsearch.org/guide/reference/api/bulk.html -func WriteBulkBytes(op string, index string, _type string, id, ttl string, date *time.Time, data interface{}, refresh bool) ([]byte, error) { +func WriteBulkBytes(op string, index string, _type string, id, ttl string, date *time.Time, version *DocVersion, data interface{}, refresh bool) ([]byte, error) { // only index and update are currently supported if op != "index" && op != "update" { return nil, errors.New(fmt.Sprintf("Operation '%s' is not yet supported", op)) @@ -420,6 +455,15 @@ func WriteBulkBytes(op string, index string, _type string, id, ttl string, date buf.WriteString(strconv.FormatInt(date.UnixNano()/1e6, 10)) buf.WriteString(`"`) } + if version != nil { + buf.WriteString(`,"_version":`) + buf.WriteString(strconv.FormatInt(version.Version, 10)) + if version.VersionType != "" { + buf.WriteString(`,"_version_type":"`) + buf.WriteString(string(version.VersionType)) + buf.WriteString(`"`) + } + } if refresh { buf.WriteString(`,"refresh":true`) } diff --git a/lib/corebulk_test.go b/lib/corebulk_test.go index 33bee118..58a5b2f4 100644 --- a/lib/corebulk_test.go +++ b/lib/corebulk_test.go @@ -76,7 +76,7 @@ func TestBulkIndexerBasic(t *testing.T) { "date": "yesterday", } - err := indexer.Index(testIndex, "user", "1", "", &date, data, true) + err := indexer.Index(testIndex, "user", "1", "", &date, nil, data, true) waitFor(func() bool { return len(buffers) > 0 @@ -88,7 +88,7 @@ func TestBulkIndexerBasic(t *testing.T) { expectedBytes := 144 assert.T(t, totalBytesSent == expectedBytes, fmt.Sprintf("Should have sent %v bytes but was %v", expectedBytes, totalBytesSent)) - err = indexer.Index(testIndex, "user", "2", "", nil, data, true) + err = indexer.Index(testIndex, "user", "2", "", nil, nil, data, true) <-time.After(time.Millisecond * 10) // we need to wait for doc to hit send channel // this will test to ensure that Flush actually catches a doc indexer.Flush() @@ -104,7 +104,7 @@ func TestBulkIndexerBasic(t *testing.T) { indexer.Stop() } -func TestBulkIndexerErrors(t *testing.T){ +func TestBulkIndexerErrors(t *testing.T) { testIndex := "users" InitTests(true) c := NewTestConn() @@ -118,7 +118,7 @@ func TestBulkIndexerErrors(t *testing.T){ indexer.Sender = func(buf *bytes.Buffer) error { // log.Printf("buffer:%s", string(buf.Bytes())) ret := indexer.Send(buf) - sent<-struct{}{} + sent <- struct{}{} return ret } errch := make(chan *ErrorBuffer, 1) @@ -128,7 +128,7 @@ func TestBulkIndexerErrors(t *testing.T){ data := map[string]interface{}{ "name": "smurfettes", - "age": 21, + "age": 21, "date": "today", } data2 := map[string]interface{}{ @@ -138,7 +138,7 @@ func TestBulkIndexerErrors(t *testing.T){ } _, err := c.DoCommand("PUT", fmt.Sprintf("/%s", testIndex), nil, - `{ + `{ "mappings": { "user": { "properties": { @@ -150,8 +150,8 @@ func TestBulkIndexerErrors(t *testing.T){ assert.T(t, err == nil, fmt.Sprintf("Should not return an error: %v", err)) //act - err = indexer.Index(testIndex, "user", "1", "", &date, data, true) - err2 := indexer.Index(testIndex, "user", "2", "", &date, data2, true) + err = indexer.Index(testIndex, "user", "1", "", &date, nil, data, true) + err2 := indexer.Index(testIndex, "user", "2", "", &date, nil, data2, true) <-sent //assert @@ -160,7 +160,7 @@ func TestBulkIndexerErrors(t *testing.T){ time.Sleep(1 * time.Microsecond) assert.T(t, indexer.NumErrors() == 1, fmt.Sprintf("Should have recorded 1 error but saw: %d", indexer.NumErrors())) - errBuf := <- errch + errBuf := <-errch bulkErr, ok := errBuf.Err.(BulkIndexingError) assert.T(t, ok, fmt.Sprintf("Error should have been a BulkIndexingError but was %T", errBuf.Err)) @@ -209,7 +209,7 @@ func XXXTestBulkUpdate(t *testing.T) { data := map[string]interface{}{ "script": "ctx._source.count += 2", } - err = indexer.Update("users", "user", "5", "", &date, data, true) + err = indexer.Update("users", "user", "5", "", &date, nil, data, true) // So here's the deal. Flushing does seem to work, you just have to give the // channel a moment to recieve the message ... // <- time.After(time.Millisecond * 20) @@ -255,9 +255,9 @@ func TestBulkSmallBatch(t *testing.T) { indexer.Start() <-time.After(time.Millisecond * 20) - indexer.Index("users", "user", "2", "", &date, data, true) - indexer.Index("users", "user", "3", "", &date, data, true) - indexer.Index("users", "user", "4", "", &date, data, true) + indexer.Index("users", "user", "2", "", &date, nil, data, true) + indexer.Index("users", "user", "3", "", &date, nil, data, true) + indexer.Index("users", "user", "4", "", &date, nil, data, true) <-time.After(time.Millisecond * 200) // indexer.Flush() indexer.Stop() @@ -279,7 +279,7 @@ func TestBulkDelete(t *testing.T) { indexer.Start() - indexer.Delete("fake", "fake_type", "1", true) + indexer.Delete("fake", "fake_type", "1", nil, true) indexer.Flush() indexer.Stop() @@ -306,7 +306,7 @@ func XXXTestBulkErrors(t *testing.T) { for i := 0; i < 20; i++ { date := time.Unix(1257894000, 0) data := map[string]interface{}{"name": "smurfs", "age": 22, "date": date} - indexer.Index("users", "user", strconv.Itoa(i), "", &date, data, true) + indexer.Index("users", "user", strconv.Itoa(i), "", &date, nil, data, true) } }() var errBuf *ErrorBuffer @@ -321,6 +321,166 @@ func XXXTestBulkErrors(t *testing.T) { indexer.Stop() } +func TestBulkVersioning_Internal(t *testing.T) { + testIndex := "users" + var ( + buffers = make([]*bytes.Buffer, 0) + totalBytesSent int + messageSets int + ) + + InitTests(true) + c := NewTestConn() + c.RequestTracer = func(method, url, body string) { + t.Logf("%s %s HTTP/1.1\n%s", method, url, body) + } + + c.DeleteIndex(testIndex) + + indexer := c.NewBulkIndexer(3) + indexer.Sender = func(buf *bytes.Buffer) error { + messageSets += 1 + totalBytesSent += buf.Len() + buffers = append(buffers, buf) + // log.Printf("buffer:%s", string(buf.Bytes())) + return indexer.Send(buf) + } + errCh := make(chan *ErrorBuffer) + indexer.ErrorChannel = errCh + indexer.Start() + + date := time.Unix(1257894000, 0) + data := map[string]interface{}{ + "name": "smurfs", + "age": 22, + "date": "yesterday", + } + + indexer.Index(testIndex, "user", "1", "", &date, nil, data, true) + + //act + data["extra"] = "1" + indexer.Index(testIndex, "user", "1", "", &date, &DocVersion{Version: 1}, data, true) + + indexer.Update(testIndex, "user", "1", "", &date, &DocVersion{Version: 2, VersionType: "internal"}, map[string]interface{}{ + "doc": map[string]interface{}{ + "updated": true, + }, + }, true) + + data["extra"] = "3" + indexer.Delete(testIndex, "user", "1", &DocVersion{Version: 7}, true) + + //assert + errBuf := <-errCh + + bulkErr, ok := errBuf.Err.(BulkIndexingError) + assert.T(t, ok, fmt.Sprintf("Expected bulk indexing error but was: %T\n\t%v", errBuf.Err, errBuf.Err)) + + js, _ := json.MarshalIndent(bulkErr.Items, "", " ") + t.Logf("Items:%s", string(js)) + + assert.T(t, getStatus(0, bulkErr.Items) == 201, "First should be created") + assert.T(t, getVersion(0, bulkErr.Items) == int64(1), "First should have version 1") + assert.T(t, getStatus(1, bulkErr.Items) == 200, "Should be reindexed with version 2") + assert.T(t, getVersion(1, bulkErr.Items) == int64(2), "Should be reindexed with version 2") + assert.T(t, getStatus(2, bulkErr.Items) == 200, "Should be updated with version 3") + assert.T(t, getVersion(2, bulkErr.Items) == int64(3), "Should be updated with version 3") + assert.T(t, getStatus(3, bulkErr.Items) == 409, "Should fail to delete due to version conflict") + assert.T(t, getError(3, bulkErr.Items) == "VersionConflictEngineException[[users][2] [user][1]: version conflict, current [3], provided [7]]") + +} + +func TestBulkVersioning_External(t *testing.T) { + testIndex := "users" + var ( + buffers = make([]*bytes.Buffer, 0) + totalBytesSent int + messageSets int + ) + + InitTests(true) + c := NewTestConn() + c.RequestTracer = func(method, url, body string) { + t.Logf("%s %s HTTP/1.1\n%s", method, url, body) + } + + c.DeleteIndex(testIndex) + + indexer := c.NewBulkIndexer(3) + indexer.Sender = func(buf *bytes.Buffer) error { + messageSets += 1 + totalBytesSent += buf.Len() + buffers = append(buffers, buf) + // log.Printf("buffer:%s", string(buf.Bytes())) + return indexer.Send(buf) + } + errCh := make(chan *ErrorBuffer) + indexer.ErrorChannel = errCh + indexer.Start() + + date := time.Unix(1257894000, 0) + data := map[string]interface{}{ + "name": "smurfs", + "age": 22, + "date": "yesterday", + } + + now := time.Now().Unix() + indexer.Index(testIndex, "user", "1", "", &date, EXTERNAL.V(now), data, true) + + //act + data["extra"] = "1" + indexer.Index(testIndex, "user", "1", "", &date, EXTERNAL_GT.V(now), data, true) + + data["extra"] = "2" + indexer.Index(testIndex, "user", "1", "", &date, EXTERNAL_GT.V(now+2), data, true) + + data["extra"] = "3" + indexer.Delete(testIndex, "user", "1", EXTERNAL_GT.V(now-1), true) + + //assert + errBuf := <-errCh + + bulkErr, ok := errBuf.Err.(BulkIndexingError) + assert.T(t, ok, fmt.Sprintf("Expected bulk indexing error but was: %T\n\t%v", errBuf.Err, errBuf.Err)) + + js, _ := json.MarshalIndent(bulkErr.Items, "", " ") + t.Logf("Items:%s", string(js)) + + assert.T(t, getStatus(0, bulkErr.Items) == 201, "First should be created") + assert.T(t, getVersion(0, bulkErr.Items) == now, "First should have version now") + assert.T(t, getStatus(1, bulkErr.Items) == 409, "Should fail to reindex with same version") + assert.T(t, getStatus(2, bulkErr.Items) == 200, "Should be updated with version now+2") + assert.T(t, getVersion(2, bulkErr.Items) == now+2, "Should be updated with version now+2") + assert.T(t, getStatus(3, bulkErr.Items) == 409, "Should fail to delete due to version conflict") + +} + +func getStatus(index int, items []map[string]interface{}) int { + for _, doc := range items[index] { + doc := doc.(map[string]interface{}) + return int(doc["status"].(float64)) + } + panic(fmt.Sprintf("no properties in %v", items[index])) +} + +func getVersion(index int, items []map[string]interface{}) int64 { + for _, doc := range items[index] { + doc := doc.(map[string]interface{}) + return int64(doc["_version"].(float64)) + } + panic(fmt.Sprintf("no properties in %v", items[index])) +} + +func getError(index int, items []map[string]interface{}) string { + for _, doc := range items[index] { + doc := doc.(map[string]interface{}) + return doc["error"].(string) + } + panic(fmt.Sprintf("no properties in %v", items[index])) +} + /* BenchmarkSend 18:33:00 bulk_test.go:131: Sent 1 messages in 0 sets totaling 0 bytes 18:33:00 bulk_test.go:131: Sent 100 messages in 1 sets totaling 145889 bytes @@ -346,7 +506,7 @@ func BenchmarkSend(b *testing.B) { about := make([]byte, 1000) rand.Read(about) data := map[string]interface{}{"name": "smurfs", "age": 22, "date": time.Unix(1257894000, 0), "about": about} - indexer.Index("users", "user", strconv.Itoa(i), "", nil, data, true) + indexer.Index("users", "user", strconv.Itoa(i), "", nil, nil, data, true) } log.Printf("Sent %d messages in %d sets totaling %d bytes \n", b.N, sets, totalBytes) if indexer.NumErrors() != 0 { @@ -380,7 +540,7 @@ func BenchmarkSendBytes(b *testing.B) { return indexer.Send(buf) } for i := 0; i < b.N; i++ { - indexer.Index("users", "user", strconv.Itoa(i), "", nil, body, true) + indexer.Index("users", "user", strconv.Itoa(i), "", nil, nil, body, true) } log.Printf("Sent %d messages in %d sets totaling %d bytes \n", b.N, sets, totalBytes) if indexer.NumErrors() != 0 { diff --git a/lib/coretest_test.go b/lib/coretest_test.go index 9af08bce..61aa9c2d 100644 --- a/lib/coretest_test.go +++ b/lib/coretest_test.go @@ -179,7 +179,7 @@ func LoadTestData() { log.Println("HM, already exists? ", ge.Url) } docsm[id] = true - indexer.Index(testIndex, ge.Type, id, "", &ge.Created, line, true) + indexer.Index(testIndex, ge.Type, id, "", &ge.Created, nil, line, true) docCt++ } else { log.Println("ERROR? ", string(line)) From c4f0122042537f04c393e534ef9ff2de99f021b1 Mon Sep 17 00:00:00 2001 From: Levi Corcoran Date: Wed, 30 Sep 2015 12:01:19 -0500 Subject: [PATCH 3/5] Fixed data races; corrected import paths to reference the zaphod-concur fork properly --- client.go | 2 +- lib/corebulk.go | 24 +++++---------- lib/corebulk_test.go | 66 ++++++++++++++++++++++++++++++----------- lib/coreexample_test.go | 7 +++-- tutorial/start_1.go | 3 +- 5 files changed, 63 insertions(+), 39 deletions(-) diff --git a/client.go b/client.go index 0ad3f2eb..cd743a8b 100644 --- a/client.go +++ b/client.go @@ -19,7 +19,7 @@ import ( "log" "time" - elastigo "github.com/mattbaird/elastigo/lib" + elastigo "github.com/zaphod-concur/elastigo/lib" ) var ( diff --git a/lib/corebulk.go b/lib/corebulk.go index f6e6c4ad..bb3d46ca 100644 --- a/lib/corebulk.go +++ b/lib/corebulk.go @@ -19,6 +19,7 @@ import ( "io" "strconv" "sync" + "sync/atomic" "time" ) @@ -128,7 +129,7 @@ type BulkIndexer struct { } func (b *BulkIndexer) NumErrors() uint64 { - return b.numErrors + return atomic.LoadUint64(&b.numErrors) } func (c *Conn) NewBulkIndexer(maxConns int) *BulkIndexer { @@ -192,16 +193,6 @@ func (b *BulkIndexer) Stop() { } } -// Make a channel that will close when the given WaitGroup is done. -func wgChan(wg *sync.WaitGroup) <-chan interface{} { - ch := make(chan interface{}) - go func() { - wg.Wait() - close(ch) - }() - return ch -} - func (b *BulkIndexer) PendingDocuments() int { return b.docCt } @@ -222,9 +213,10 @@ func (b *BulkIndexer) startHttpSender() { // in theory, the whole set will cause a backup all the way to IndexBulk if // we have consumed all maxConns for i := 0; i < b.maxConns; i++ { + b.sendWg.Add(1) go func() { + defer b.sendWg.Done() for buf := range b.sendBuf { - b.sendWg.Add(1) // Copy so we can put the buffer on the error channel, or potentially re-send it. bufCopy := bytes.NewBuffer(buf.Bytes()) err := b.Sender(buf) @@ -242,7 +234,6 @@ func (b *BulkIndexer) startHttpSender() { err = b.Sender(bufCopy) if err == nil { // Successfully re-sent with no error - b.sendWg.Done() continue } } @@ -250,7 +241,6 @@ func (b *BulkIndexer) startHttpSender() { b.ErrorChannel <- &ErrorBuffer{err, buf} } } - b.sendWg.Done() } }() } @@ -315,7 +305,7 @@ func (b *BulkIndexer) shutdown() { close(b.timerDoneChan) close(b.sendBuf) close(b.bulkChannel) - <-wgChan(b.sendWg) + b.sendWg.Wait() } // The index bulk API adds or updates a typed JSON document to a specific index, making it searchable. @@ -386,7 +376,7 @@ func (b *BulkIndexer) Send(buf *bytes.Buffer) error { body, err := b.conn.DoCommand("POST", "/_bulk", nil, buf) if err != nil { - b.numErrors += 1 + atomic.AddUint64(&b.numErrors, 1) return err } // check for response errors, bulk insert will give 200 OK but then include errors in response @@ -407,7 +397,7 @@ func (b *BulkIndexer) Send(buf *bytes.Buffer) error { continue } if status, ok := body["status"]; ok && status.(float64) > 304 { - b.numErrors++ + atomic.AddUint64(&b.numErrors, 1) } } } diff --git a/lib/corebulk_test.go b/lib/corebulk_test.go index 58a5b2f4..77ac669b 100644 --- a/lib/corebulk_test.go +++ b/lib/corebulk_test.go @@ -19,17 +19,42 @@ import ( "fmt" "log" "strconv" + "sync" "testing" "time" + "strings" + "github.com/araddon/gou" "github.com/bmizerany/assert" - "strings" ) // go test -bench=".*" // go test -bench="Bulk" +type sharedBuffer struct { + mu sync.Mutex + Buffer []*bytes.Buffer +} + +func NewSharedBuffer() *sharedBuffer { + return &sharedBuffer{ + Buffer: make([]*bytes.Buffer, 0), + } +} + +func (b *sharedBuffer) Append(buf *bytes.Buffer) { + b.mu.Lock() + defer b.mu.Unlock() + b.Buffer = append(b.Buffer, buf) +} + +func (b *sharedBuffer) Length() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.Buffer) +} + func init() { flag.Parse() if testing.Verbose() { @@ -49,7 +74,7 @@ func closeInt(a, b int) bool { func TestBulkIndexerBasic(t *testing.T) { testIndex := "users" var ( - buffers = make([]*bytes.Buffer, 0) + buffers = NewSharedBuffer() totalBytesSent int messageSets int ) @@ -63,8 +88,8 @@ func TestBulkIndexerBasic(t *testing.T) { indexer.Sender = func(buf *bytes.Buffer) error { messageSets += 1 totalBytesSent += buf.Len() - buffers = append(buffers, buf) - // log.Printf("buffer:%s", string(buf.Bytes())) + buffers.Append(buf) + //log.Printf("buffer:%s", string(buf.Bytes())) return indexer.Send(buf) } indexer.Start() @@ -79,23 +104,26 @@ func TestBulkIndexerBasic(t *testing.T) { err := indexer.Index(testIndex, "user", "1", "", &date, nil, data, true) waitFor(func() bool { - return len(buffers) > 0 + return buffers.Length() > 0 }, 5) // part of request is url, so lets factor that in //totalBytesSent = totalBytesSent - len(*eshost) - assert.T(t, len(buffers) == 1, fmt.Sprintf("Should have sent one operation but was %d", len(buffers))) + assert.T(t, buffers.Length() == 1, fmt.Sprintf("Should have sent one operation but was %d", buffers.Length())) assert.T(t, indexer.NumErrors() == 0 && err == nil, fmt.Sprintf("Should not have any errors. NumErrors: %v, err: %v", indexer.NumErrors(), err)) expectedBytes := 144 assert.T(t, totalBytesSent == expectedBytes, fmt.Sprintf("Should have sent %v bytes but was %v", expectedBytes, totalBytesSent)) err = indexer.Index(testIndex, "user", "2", "", nil, nil, data, true) - <-time.After(time.Millisecond * 10) // we need to wait for doc to hit send channel + waitFor(func() bool { + return buffers.Length() > 1 + }, 5) + // this will test to ensure that Flush actually catches a doc indexer.Flush() <-time.After(time.Millisecond * 1) // we need to wait for the httpSender to read from the send buffer totalBytesSent = totalBytesSent - len(*eshost) assert.T(t, err == nil, fmt.Sprintf("Should have nil error =%v", err)) - assert.T(t, len(buffers) == 2, fmt.Sprintf("Should have another buffer ct=%d", len(buffers))) + assert.T(t, buffers.Length() == 2, fmt.Sprintf("Should have another buffer ct=%d", buffers.Length())) assert.T(t, indexer.NumErrors() == 0, fmt.Sprintf("Should not have any errors %d", indexer.NumErrors())) expectedBytes = 250 // with refresh @@ -114,13 +142,13 @@ func TestBulkIndexerErrors(t *testing.T) { sent := make(chan struct{}, 1) indexer := c.NewBulkIndexer(3) - indexer.Start() indexer.Sender = func(buf *bytes.Buffer) error { // log.Printf("buffer:%s", string(buf.Bytes())) ret := indexer.Send(buf) sent <- struct{}{} return ret } + indexer.Start() errch := make(chan *ErrorBuffer, 1) indexer.ErrorChannel = errch @@ -180,7 +208,7 @@ func TestBulkIndexerErrors(t *testing.T) { // currently broken in drone.io func XXXTestBulkUpdate(t *testing.T) { var ( - buffers = make([]*bytes.Buffer, 0) + buffers = NewSharedBuffer() totalBytesSent int messageSets int ) @@ -192,7 +220,7 @@ func XXXTestBulkUpdate(t *testing.T) { indexer.Sender = func(buf *bytes.Buffer) error { messageSets += 1 totalBytesSent += buf.Len() - buffers = append(buffers, buf) + buffers.Append(buf) return indexer.Send(buf) } indexer.Start() @@ -216,7 +244,7 @@ func XXXTestBulkUpdate(t *testing.T) { // indexer.Flush() waitFor(func() bool { - return len(buffers) > 0 + return buffers.Length() > 0 }, 5) indexer.Stop() @@ -267,13 +295,15 @@ func TestBulkSmallBatch(t *testing.T) { func TestBulkDelete(t *testing.T) { InitTests(true) - + var lock sync.Mutex c := NewTestConn() indexer := c.NewBulkIndexer(1) sentBytes := []byte{} indexer.Sender = func(buf *bytes.Buffer) error { + lock.Lock() sentBytes = append(sentBytes, buf.Bytes()...) + lock.Unlock() return nil } @@ -284,7 +314,9 @@ func TestBulkDelete(t *testing.T) { indexer.Flush() indexer.Stop() + lock.Lock() sent := string(sentBytes) + lock.Unlock() expected := `{"delete":{"_index":"fake","_type":"fake_type","_id":"1","refresh":true}} ` @@ -324,7 +356,7 @@ func XXXTestBulkErrors(t *testing.T) { func TestBulkVersioning_Internal(t *testing.T) { testIndex := "users" var ( - buffers = make([]*bytes.Buffer, 0) + buffers = NewSharedBuffer() totalBytesSent int messageSets int ) @@ -341,7 +373,7 @@ func TestBulkVersioning_Internal(t *testing.T) { indexer.Sender = func(buf *bytes.Buffer) error { messageSets += 1 totalBytesSent += buf.Len() - buffers = append(buffers, buf) + buffers.Append(buf) // log.Printf("buffer:%s", string(buf.Bytes())) return indexer.Send(buf) } @@ -394,7 +426,7 @@ func TestBulkVersioning_Internal(t *testing.T) { func TestBulkVersioning_External(t *testing.T) { testIndex := "users" var ( - buffers = make([]*bytes.Buffer, 0) + buffers = NewSharedBuffer() totalBytesSent int messageSets int ) @@ -411,7 +443,7 @@ func TestBulkVersioning_External(t *testing.T) { indexer.Sender = func(buf *bytes.Buffer) error { messageSets += 1 totalBytesSent += buf.Len() - buffers = append(buffers, buf) + buffers.Append(buf) // log.Printf("buffer:%s", string(buf.Bytes())) return indexer.Send(buf) } diff --git a/lib/coreexample_test.go b/lib/coreexample_test.go index 231ebd2f..ba3085aa 100644 --- a/lib/coreexample_test.go +++ b/lib/coreexample_test.go @@ -14,8 +14,9 @@ package elastigo_test import ( "bytes" "fmt" - elastigo "github.com/mattbaird/elastigo/lib" "strconv" + + elastigo "github.com/zaphod-concur/elastigo/lib" ) // The simplest usage of background bulk indexing @@ -24,7 +25,7 @@ func ExampleBulkIndexer_simple() { indexer := c.NewBulkIndexerErrors(10, 60) indexer.Start() - indexer.Index("twitter", "user", "1", "", nil, `{"name":"bob"}`, true) + indexer.Index("twitter", "user", "1", "", nil, nil, `{"name":"bob"}`, true) indexer.Stop() } @@ -45,7 +46,7 @@ func ExampleBulkIndexer_responses() { } indexer.Start() for i := 0; i < 20; i++ { - indexer.Index("twitter", "user", strconv.Itoa(i), "", nil, `{"name":"bob"}`, true) + indexer.Index("twitter", "user", strconv.Itoa(i), "", nil, nil, `{"name":"bob"}`, true) } indexer.Stop() } diff --git a/tutorial/start_1.go b/tutorial/start_1.go index db807f66..232219fa 100644 --- a/tutorial/start_1.go +++ b/tutorial/start_1.go @@ -14,9 +14,10 @@ package main import ( "flag" "fmt" - elastigo "github.com/mattbaird/elastigo/lib" "log" "os" + + elastigo "github.com/zaphod-concur/elastigo/lib" ) var ( From 47ee816807dfda3ca0142366626540c279eddb5c Mon Sep 17 00:00:00 2001 From: Levi Corcoran Date: Mon, 12 Oct 2015 15:18:06 -0500 Subject: [PATCH 4/5] Added PendingSends mechanism to track any buffers that need to be sent, until they are fully sent and any errors are reported, to ensure proper shutdown sequencing --- lib/corebulk.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/lib/corebulk.go b/lib/corebulk.go index bb3d46ca..afc95bcd 100644 --- a/lib/corebulk.go +++ b/lib/corebulk.go @@ -126,6 +126,9 @@ type BulkIndexer struct { mu sync.Mutex // Wait Group for the http sends sendWg *sync.WaitGroup + + // numPendingSends tracks queued buffers pending 'send' in sendBuf + numPendingSends int64 } func (b *BulkIndexer) NumErrors() uint64 { @@ -197,6 +200,10 @@ func (b *BulkIndexer) PendingDocuments() int { return b.docCt } +func (b *BulkIndexer) PendingSends() int64 { + return atomic.LoadInt64(&b.numPendingSends) +} + // Flush all current documents to ElasticSearch func (b *BulkIndexer) Flush() { b.mu.Lock() @@ -232,15 +239,12 @@ func (b *BulkIndexer) startHttpSender() { bufCopy := bytes.NewBuffer(buf.Bytes()) time.Sleep(time.Second * time.Duration(b.RetryForSeconds)) err = b.Sender(bufCopy) - if err == nil { - // Successfully re-sent with no error - continue - } } - if b.ErrorChannel != nil { + if err != nil && b.ErrorChannel != nil { b.ErrorChannel <- &ErrorBuffer{err, buf} } } + atomic.AddInt64(&b.numPendingSends, -1) } }() } @@ -294,6 +298,7 @@ func (b *BulkIndexer) startDocChannel() { func (b *BulkIndexer) send(buf *bytes.Buffer) { //b2 := *b.buf + atomic.AddInt64(&b.numPendingSends, 1) b.sendBuf <- buf b.buf = new(bytes.Buffer) // b.buf.Reset() From 8b1c9d62b6682b3075a69c43e5a206b6466f5d7b Mon Sep 17 00:00:00 2001 From: Levi Corcoran Date: Thu, 29 Oct 2015 15:27:07 -0500 Subject: [PATCH 5/5] fixed race condition on PendingDocuments() docCt access --- lib/corebulk.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/corebulk.go b/lib/corebulk.go index 54853d75..ea76f53a 100644 --- a/lib/corebulk.go +++ b/lib/corebulk.go @@ -202,6 +202,8 @@ func (b *BulkIndexer) Stop() { } func (b *BulkIndexer) PendingDocuments() int { + b.mu.Lock() + defer b.mu.Unlock() return b.docCt }