GOCBC-652: Add support for circuit breakers for KV

Motivation
----------
In order to help reduce the load on a server component which is
already in distress we should implement client side circuit
breakers.

Changes
-------
Add circuit breaker configuration properties and send them down
to gocbcore on agent creation.

Change-Id: Ib777dd4a64e3cd1554661eb11cf5e7fe1fe4010c
Reviewed-on: http://review.couchbase.org/117059
Reviewed-by: Brett Lawson <brett19@gmail.com>
Tested-by: Charles Dixon <chvckd@gmail.com>

circuitbreaker.go

@@ -0,0 +1,18 @@
+package gocb
+
+import "time"
+
+// CircuitBreakerCallback is the callback used by the circuit breaker to determine if an error should count toward
+// the circuit breaker failure count.
+type CircuitBreakerCallback func(error) bool
+
+// CircuitBreakerConfig are the settings for configuring circuit breakers.
+type CircuitBreakerConfig struct {
+	Disabled                 bool
+	VolumeThreshold          int64
+	ErrorThresholdPercentage float64
+	SleepWindow              time.Duration
+	RollingWindow            time.Duration
+	CompletionCallback       CircuitBreakerCallback
+	CanaryTimeout            time.Duration
+}

client.go

@@ -52,6 +52,14 @@ func (c *stdClient) buildConfig() error {
 	defer c.lock.Unlock()
 
 	auth := c.cluster.auth
+	breakerCfg := c.cluster.sb.CircuitBreakerConfig
+
+	var completionCallback func(err error) bool
+	if breakerCfg.CompletionCallback != nil {
+		completionCallback = func(err error) bool {
+			return breakerCfg.CompletionCallback(maybeEnhanceKVErr(err, "", false))
+		}
+	}
 
 	config := &gocbcore.AgentConfig{
 		UserString:           Identifier(),
@@ -72,6 +80,15 @@ func (c *stdClient) buildConfig() error {
 		ZombieLoggerSampleSize: c.cluster.sb.OrphanLoggerSampleSize,
 		NoRootTraceSpans:       true,
 		Tracer:                 &requestTracerWrapper{c.cluster.sb.Tracer},
+		CircuitBreakerConfig: gocbcore.CircuitBreakerConfig{
+			Enabled:                  !breakerCfg.Disabled,
+			VolumeThreshold:          breakerCfg.VolumeThreshold,
+			ErrorThresholdPercentage: breakerCfg.ErrorThresholdPercentage,
+			SleepWindow:              breakerCfg.SleepWindow,
+			RollingWindow:            breakerCfg.RollingWindow,
+			CanaryTimeout:            breakerCfg.CanaryTimeout,
+			CompletionCallback:       completionCallback,
+		},
 	}
 
 	err := config.FromConnStr(c.cluster.connSpec().String())

cluster.go

@@ -56,6 +56,8 @@ type ClusterOptions struct {
 
 	ThresholdLoggerDisabled bool
 	ThresholdLoggingOptions *ThresholdLoggingOptions
+
+	CircuitBreakerConfig CircuitBreakerConfig
 }
 
 // ClusterCloseOptions is the set of options available when disconnecting from a Cluster.
@@ -171,6 +173,7 @@ func Connect(connStr string, opts ClusterOptions) (*Cluster, error) {
 			OrphanLoggerSampleSize: opts.OrphanLoggerSampleSize,
 			UseServerDurations:     useServerDurations,
 			Tracer:                 initialTracer,
+			CircuitBreakerConfig:   opts.CircuitBreakerConfig,
 		},
 
 		queryCache: make(map[string]*n1qlCache),

stateblock.go

@@ -47,6 +47,8 @@ type stateBlock struct {
 	OrphanLoggerSampleSize int
 
 	Tracer requestTracer
+
+	CircuitBreakerConfig CircuitBreakerConfig
 }
 
 func (sb *stateBlock) getCachedClient() client {