Update sarama to master + SASL fix (elastic#2798) (elastic#2804)

* Update sarama to master + SASL fix

- based on v1.10.1
- master:
  - fixes bug in in v0.10 protocol if compression is enabled (default)
  - introduces broker metrics
- SASL/plain included

* Add kafka 0.10.0.1 version string to config

* Hook up sarama kafka metrics

CHANGELOG.asciidoc

@@ -30,6 +30,8 @@ https://github.com/elastic/beats/compare/v5.0.0-rc1...5.0[Check the HEAD diff]
 *Affecting all Beats*
 
 - Fix kafka output re-trying batches with too large events. {issue}2735[2735]
+- Fix kafka output protocol error if `version: 0.10` is configured. {issue}2651[2651]
+- Fix kafka output connection closed by broker on SASL/PLAIN. {issue}2717[2717]
 
 *Metricbeat*
 

glide.yaml

@@ -59,7 +59,10 @@ import:
 - package: github.com/miekg/dns
   version: 5d001d020961ae1c184f9f8152fdc73810481677
 - package: github.com/Shopify/sarama
-  version: v1.10.0
+  version: fix/sasl-handshake
+  repo: https://github.com/urso/sarama
+- package: github.com/rcrowley/go-metrics
+  version: ab2277b1c5d15c3cba104e9cbddbdfc622df5ad8
 - package: github.com/klauspost/crc32
   version: v1.0
 - package: github.com/golang/snappy

libbeat/outputs/kafka/kafka.go

@@ -4,9 +4,12 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/Shopify/sarama"
+	metrics "github.com/rcrowley/go-metrics"
+	"github.com/rcrowley/go-metrics/exp"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/op"
@@ -35,11 +38,26 @@ const (
 	defaultMaxWaitRetry = 60 * time.Second
 )
 
+var kafkaMetricsRegistryInstance metrics.Registry
+
 func init() {
 	sarama.Logger = kafkaLogger{}
+
+	reg := metrics.NewPrefixedRegistry("libbeat.kafka.")
+
+	// Note: registers /debug/metrics handler for displaying all expvar counters
+	exp.Exp(reg)
+	kafkaMetricsRegistryInstance = reg
+
 	outputs.RegisterOutputPlugin("kafka", New)
 }
 
+var kafkaMetricsOnce sync.Once
+
+func kafkaMetricsRegistry() metrics.Registry {
+	return kafkaMetricsRegistryInstance
+}
+
 var debugf = logp.MakeDebug("kafka")
 
 var (
@@ -71,8 +89,9 @@ var (
 		"0.9":     sarama.V0_9_0_1,
 
 		"0.10.0.0": sarama.V0_10_0_0,
-		"0.10.0":   sarama.V0_10_0_0,
-		"0.10":     sarama.V0_10_0_0,
+		"0.10.0.1": sarama.V0_10_0_1,
+		"0.10.0":   sarama.V0_10_0_1,
+		"0.10":     sarama.V0_10_0_1,
 	}
 )
 
@@ -313,5 +332,6 @@ func newKafkaConfig(config *kafkaConfig) (*sarama.Config, error) {
 	}
 	k.Version = version
 
+	k.MetricRegistry = kafkaMetricsRegistry()
 	return k, nil
 }