Merge (experimental) BulkProcessor

Bulk processor is a service that accepts bulk requests and commits them
in the background, based on policies. It slightly resembles the
BulkProcessor of the Java API [1], [2].

Notice that it is still experimental. Use with care! Feedback welcome.

The current state of bulk processor is documented in the Wiki:
https://github.com/olivere/elastic/wiki/BulkProcessor

[1] https://github.com/elastic/elasticsearch/blob/2.x/core/src/main/java/org/elasticsearch/action/bulk/BulkProcessor.java
[2] https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs-bulk-processor.html

README.md

@@ -404,6 +404,11 @@ Thanks a lot for the great folks working hard on
 and
 [Go](http://www.golang.org/).
 
+Elastic uses portions of the
+[uritemplates](https://github.com/jtacoma/uritemplates) library
+by Joshua Tacoma and
+[backoff](https://github.com/cenkalti/backoff) by Cenk Altı.
+
 ## LICENSE
 
 MIT-LICENSE. See [LICENSE](http://olivere.mit-license.org/)

backoff/LICENSE

@@ -0,0 +1,22 @@
+Portions of this code rely on this LICENSE:
+
+The MIT License (MIT)
+
+Copyright (c) 2014 Cenk Altı
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

backoff/backoff.go

@@ -0,0 +1,159 @@
+// Copyright 2012-2016 Oliver Eilhard. All rights reserved.
+// Use of this source code is governed by a MIT-license.
+// See http://olivere.mit-license.org/license.txt for details.
+
+package backoff
+
+import (
+	"math"
+	"math/rand"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Backoff is an interface for different types of backoff algorithms.
+type Backoff interface {
+	Next() time.Duration
+	Reset()
+}
+
+// Stop is used as a signal to indicate that no more retries should be made.
+const Stop time.Duration = -1
+
+// -- Simple Backoff --
+
+// SimpleBackoff takes a list of fixed values for backoff intervals.
+// Each call to Next returns the next value from that fixed list.
+// After each value is returned, subsequent calls to Next will only return
+// the last element. The caller may specify if the values are "jittered".
+type SimpleBackoff struct {
+	sync.Mutex
+	ticks  []int
+	index  int
+	jitter bool
+	stop   bool
+}
+
+// NewSimpleBackoff creates a SimpleBackoff algorithm with the specified
+// list of fixed intervals in milliseconds.
+func NewSimpleBackoff(ticks ...int) *SimpleBackoff {
+	return &SimpleBackoff{
+		ticks:  ticks,
+		index:  0,
+		jitter: false,
+		stop:   false,
+	}
+}
+
+// Jitter, when set, randomizes to return a value of [0.5*value .. 1.5*value].
+func (b *SimpleBackoff) Jitter(doJitter bool) *SimpleBackoff {
+	b.Lock()
+	defer b.Unlock()
+	b.jitter = doJitter
+	return b
+}
+
+// SendStop, when enables, makes Next to return Stop once
+// the list of values is exhausted.
+func (b *SimpleBackoff) SendStop(doStop bool) *SimpleBackoff {
+	b.Lock()
+	defer b.Unlock()
+	b.stop = doStop
+	return b
+}
+
+// Next returns the next wait interval.
+func (b *SimpleBackoff) Next() time.Duration {
+	b.Lock()
+	defer b.Unlock()
+
+	i := b.index
+	if i >= len(b.ticks) {
+		if b.stop {
+			return Stop
+		}
+		i = len(b.ticks) - 1
+		b.index = i
+	} else {
+		b.index++
+	}
+
+	ms := b.ticks[i]
+	if b.jitter {
+		ms = jitter(ms)
+	}
+	return time.Duration(ms) * time.Millisecond
+}
+
+// Reset resets SimpleBackoff.
+func (b *SimpleBackoff) Reset() {
+	b.Lock()
+	b.index = 0
+	b.Unlock()
+}
+
+// jitter randomizes the interval to return a value of [0.5*millis .. 1.5*millis].
+func jitter(millis int) int {
+	if millis <= 0 {
+		return 0
+	}
+	return millis/2 + rand.Intn(millis)
+}
+
+// -- Exponential --
+
+// ExponentialBackoff implements the simple exponential backoff described by
+// Douglas Thain at http://dthain.blogspot.de/2009/02/exponential-backoff-in-distributed.html.
+type ExponentialBackoff struct {
+	sync.Mutex
+	t    float64 // initial timeout (in msec)
+	f    float64 // exponential factor (e.g. 2)
+	m    float64 // maximum timeout (in msec)
+	n    int64   // number of retries
+	stop bool    // indicates whether Next should send "Stop" whan max timeout is reached
+}
+
+// NewExponentialBackoff returns a ExponentialBackoff backoff policy.
+// Use initialTimeout to set the first/minimal interval
+// and maxTimeout to set the maximum wait interval.
+func NewExponentialBackoff(initialTimeout, maxTimeout time.Duration) *ExponentialBackoff {
+	return &ExponentialBackoff{
+		t:    float64(int64(initialTimeout / time.Millisecond)),
+		f:    2.0,
+		m:    float64(int64(maxTimeout / time.Millisecond)),
+		n:    0,
+		stop: false,
+	}
+}
+
+// SendStop, when enables, makes Next to return Stop once
+// the maximum timeout is reached.
+func (b *ExponentialBackoff) SendStop(doStop bool) *ExponentialBackoff {
+	b.Lock()
+	defer b.Unlock()
+	b.stop = doStop
+	return b
+}
+
+// Next returns the next wait interval.
+func (t *ExponentialBackoff) Next() time.Duration {
+	t.Lock()
+	defer t.Unlock()
+
+	n := float64(atomic.AddInt64(&t.n, 1))
+	r := 1.0 + rand.Float64() // random number in [1..2]
+	m := math.Min(r*t.t*math.Pow(t.f, n), t.m)
+	if t.stop && m >= t.m {
+		return Stop
+	}
+	d := time.Duration(int64(m)) * time.Millisecond
+	return d
+}
+
+// Reset resets the backoff policy so that it can be reused.
+func (t *ExponentialBackoff) Reset() {
+	t.Lock()
+	t.n = 0
+	t.Unlock()
+}

backoff/backoff_test.go

@@ -0,0 +1,146 @@
+// Copyright 2012-2016 Oliver Eilhard. All rights reserved.
+// Use of this source code is governed by a MIT-license.
+// See http://olivere.mit-license.org/license.txt for details.
+
+package backoff
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+)
+
+func TestSimpleBackoff(t *testing.T) {
+	b := NewSimpleBackoff(1, 2, 7)
+
+	if got, want := b.Next(), time.Duration(1)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+	if got, want := b.Next(), time.Duration(2)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+	if got, want := b.Next(), time.Duration(7)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+	if got, want := b.Next(), time.Duration(7)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+
+	b.Reset()
+
+	if got, want := b.Next(), time.Duration(1)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+	if got, want := b.Next(), time.Duration(2)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+	if got, want := b.Next(), time.Duration(7)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+	if got, want := b.Next(), time.Duration(7)*time.Millisecond; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+}
+
+func TestSimpleBackoffWithStop(t *testing.T) {
+	b := NewSimpleBackoff(1, 2, 7).SendStop(true)
+
+	// It should eventually return Stop (-1) after some loops.
+	var last time.Duration
+	for i := 0; i < 10; i++ {
+		last = b.Next()
+		if last == Stop {
+			break
+		}
+	}
+	if got, want := last, Stop; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+
+	b.Reset()
+
+	// It should eventually return Stop (-1) after some loops.
+	for i := 0; i < 10; i++ {
+		last = b.Next()
+		if last == Stop {
+			break
+		}
+	}
+	if got, want := last, Stop; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+}
+
+func TestExponentialBackoff(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+
+	min := time.Duration(8) * time.Millisecond
+	max := time.Duration(256) * time.Millisecond
+	b := NewExponentialBackoff(min, max)
+
+	between := func(value time.Duration, a, b int) bool {
+		x := int(value / time.Millisecond)
+		return a <= x && x <= b
+	}
+
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+
+	b.Reset()
+
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+	if got := b.Next(); !between(got, 8, 256) {
+		t.Errorf("expected [%v..%v]; got: %v", 8, 256, got)
+	}
+}
+
+func TestExponentialBackoffWithStop(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+
+	min := time.Duration(8) * time.Millisecond
+	max := time.Duration(256) * time.Millisecond
+	b := NewExponentialBackoff(min, max).SendStop(true)
+
+	// It should eventually return Stop (-1) after some loops.
+	var last time.Duration
+	for i := 0; i < 10; i++ {
+		last = b.Next()
+		if last == Stop {
+			break
+		}
+	}
+	if got, want := last, Stop; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+
+	b.Reset()
+
+	// It should eventually return Stop (-1) after some loops.
+	for i := 0; i < 10; i++ {
+		last = b.Next()
+		if last == Stop {
+			break
+		}
+	}
+	if got, want := last, Stop; got != want {
+		t.Errorf("expected %v; got: %v", want, got)
+	}
+}

backoff/retry.go

@@ -0,0 +1,53 @@
+// Copyright 2012-2016 Oliver Eilhard. All rights reserved.
+// Use of this source code is governed by a MIT-license.
+// See http://olivere.mit-license.org/license.txt for details.
+
+// This file is (c) 2014 Cenk Altı and governed by the MIT license.
+// See https://github.com/cenkalti/backoff for original source.
+
+package backoff
+
+import "time"
+
+// An Operation is executing by Retry() or RetryNotify().
+// The operation will be retried using a backoff policy if it returns an error.
+type Operation func() error
+
+// Notify is a notify-on-error function. It receives an operation error and
+// backoff delay if the operation failed (with an error).
+//
+// NOTE that if the backoff policy stated to stop retrying,
+// the notify function isn't called.
+type Notify func(error, time.Duration)
+
+// Retry the function f until it does not return error or BackOff stops.
+// f is guaranteed to be run at least once.
+// It is the caller's responsibility to reset b after Retry returns.
+//
+// Retry sleeps the goroutine for the duration returned by BackOff after a
+// failed operation returns.
+func Retry(o Operation, b Backoff) error { return RetryNotify(o, b, nil) }
+
+// RetryNotify calls notify function with the error and wait duration
+// for each failed attempt before sleep.
+func RetryNotify(operation Operation, b Backoff, notify Notify) error {
+	var err error
+	var next time.Duration
+
+	b.Reset()
+	for {
+		if err = operation(); err == nil {
+			return nil
+		}
+
+		if next = b.Next(); next == Stop {
+			return err
+		}
+
+		if notify != nil {
+			notify(err, next)
+		}
+
+		time.Sleep(next)
+	}
+}