Merge pull request #30 from Jeffail/feature/badger-buffer

Feature/badger buffer

Gopkg.toml

@@ -2,51 +2,3 @@
   non-go = true
   go-tests = true
   unused-packages = true
-
-[[constraint]]
-  name = "github.com/Shopify/sarama"
-  version = "1.15.0"
-
-[[constraint]]
-  branch = "master"
-  name = "github.com/edsrzf/mmap-go"
-
-[[constraint]]
-  name = "github.com/go-mangos/mangos"
-  version = "1.2.0"
-
-[[constraint]]
-  name = "github.com/nats-io/go-nats"
-  version = "1.3.0"
-
-[[constraint]]
-  name = "github.com/nsqio/go-nsq"
-  version = "1.0.7"
-
-[[constraint]]
-  branch = "master"
-  name = "github.com/pebbe/zmq4"
-
-[[constraint]]
-  branch = "master"
-  name = "github.com/streadway/amqp"
-
-[[constraint]]
-  branch = "v2"
-  name = "gopkg.in/yaml.v2"
-
-[[constraint]]
-  name = "github.com/go-redis/redis"
-  version = "6.7.5"
-
-[[constraint]]
-  name = "github.com/satori/go.uuid"
-  version = "1.2.0"
-
-[[constraint]]
-  name = "github.com/gorilla/mux"
-  version = "1.6.1"
-
-[[constraint]]
-  name = "github.com/eclipse/paho.mqtt.golang"
-  version = "1.1.0"

config/everything.yaml

@@ -341,6 +341,10 @@ output:
   processors: []
 buffer:
   type: none
+  badger:
+    directory: ""
+    sync_writes: true
+    gc_interval_ms: 1000
   memory:
     limit: 524288000
   mmap_file:

lib/buffer/badger.go

@@ -0,0 +1,60 @@
+// Copyright (c) 2018 Ashley Jeffs
+//
+// 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.
+
+package buffer
+
+import (
+	"github.com/Jeffail/benthos/lib/buffer/parallel"
+	"github.com/Jeffail/benthos/lib/util/service/log"
+	"github.com/Jeffail/benthos/lib/util/service/metrics"
+)
+
+//------------------------------------------------------------------------------
+
+func init() {
+	Constructors["badger"] = TypeSpec{
+		constructor: NewBadger,
+		description: `
+The badger buffer type uses a [badger](https://github.com/dgraph-io/badger) db
+in order to persist messages to disk as a key/value store. The benefit of this
+method is that unlike the mmap_file approach this buffer can be emptied by
+parallel consumers.
+
+Note that throughput can be significantly improved by disabling 'sync_writes',
+but this comes at the cost of delivery guarantees under crashes.
+
+This buffer has stronger delivery guarantees and higher throughput across
+brokered outputs (except for the fan_out pattern) at the cost of lower single
+output throughput.`,
+	}
+}
+
+//------------------------------------------------------------------------------
+
+// NewBadger creates a buffer around a badger k/v db.
+func NewBadger(config Config, log log.Modular, stats metrics.Type) (Type, error) {
+	b, err := parallel.NewBadger(config.Badger)
+	if err != nil {
+		return nil, err
+	}
+	return NewParallelWrapper(config, b, log, stats), nil
+}
+
+//------------------------------------------------------------------------------

lib/buffer/constructor.go

@@ -26,6 +26,7 @@ import (
 	"sort"
 	"strings"
 
+	"github.com/Jeffail/benthos/lib/buffer/parallel"
 	"github.com/Jeffail/benthos/lib/buffer/single"
 	"github.com/Jeffail/benthos/lib/types"
 	"github.com/Jeffail/benthos/lib/util/service/log"
@@ -48,6 +49,7 @@ var Constructors = map[string]TypeSpec{}
 // Config is the all encompassing configuration struct for all input types.
 type Config struct {
 	Type   string                  `json:"type" yaml:"type"`
+	Badger parallel.BadgerConfig   `json:"badger" yaml:"badger"`
 	Memory single.MemoryConfig     `json:"memory" yaml:"memory"`
 	Mmap   single.MmapBufferConfig `json:"mmap_file" yaml:"mmap_file"`
 	None   struct{}                `json:"none" yaml:"none"`
@@ -57,6 +59,7 @@ type Config struct {
 func NewConfig() Config {
 	return Config{
 		Type:   "none",
+		Badger: parallel.NewBadgerConfig(),
 		Memory: single.NewMemoryConfig(),
 		Mmap:   single.NewMmapBufferConfig(),
 		None:   struct{}{},
@@ -85,6 +88,44 @@ func SanitiseConfig(conf Config) (interface{}, error) {
 
 //------------------------------------------------------------------------------
 
+var header = "This document was generated with `benthos --list-buffers`" + `
+
+Buffers can solve a number of typical streaming problems and are worth
+considering if you face circumstances similar to the following:
+
+- Input sources can periodically spike beyond the capacity of your output sinks.
+- You have more outputs than inputs and wish to distribute messages across them
+  in order to maximize overall throughput.
+- Your input source needs occasional protection against back pressure from your
+  sink, e.g. during restarts. Please keep in mind that all buffers have an
+  eventual limit.
+
+If you believe that a problem you have would be solved by a buffer the next step
+is to choose an implementation based on the throughput and delivery guarantees
+you need. In order to help here are some simplified tables outlining the
+different options and their qualities:
+
+#### Performance
+
+| Type      | Throughput | Consumers | Capacity |
+| --------- | ---------- | --------- | -------- |
+| Memory    | Highest    | Parallel  | RAM      |
+| Mmap File | High       | Single    | Disk     |
+| Badger    | Medium     | Parallel  | Disk     |
+
+#### Delivery Guarantees
+
+| Type      | On Restart | On Crash  | On Disk Corruption |
+| --------- | ---------- | --------- | ------------------ |
+| Memory    | Lost       | Lost      | Lost               |
+| Mmap File | Persisted  | Lost      | Lost               |
+| Badger    | Persisted  | Persisted | Lost               |
+
+Please note that the badger buffer can be set to disable synchronous writes.
+This removes the guarantee of message persistence after a crash, but brings
+performance on par with the mmap file buffer. This can make it the faster
+overall disk persisted buffer when writing to multiple outputs.`
+
 // Descriptions returns a formatted string of collated descriptions of each type.
 func Descriptions() string {
 	// Order our buffer types alphabetically
@@ -98,7 +139,7 @@ func Descriptions() string {
 	buf.WriteString("BUFFERS\n")
 	buf.WriteString(strings.Repeat("=", 7))
 	buf.WriteString("\n\n")
-	buf.WriteString("This document has been generated with `benthos --list-buffers`.")
+	buf.WriteString(header)
 	buf.WriteString("\n\n")
 
 	// Append each description