From c763ceea953328ade31def00e6eed101511ebffa Mon Sep 17 00:00:00 2001
From: urso <steffen.siering@elastic.co>
Date: Mon, 24 Apr 2017 12:03:03 +0200
Subject: [PATCH] Fix race in go-metrics adapater

Fix race in go-metrics adapter, if registry is updated concurrently from
multiple go routines.

(cherry picked from commit 7eaaec3b900a39cae5987aa611ae994153bb9bd6)
---
 CHANGELOG.asciidoc                       |  1 +
 libbeat/monitoring/adapter/go-metrics.go | 25 ++++++++++++++++++++++--
 2 files changed, 24 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 31d6ca6ba2d..e2bb9951f7f 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -96,6 +96,7 @@ https://github.com/elastic/beats/compare/v5.3.0...v5.3.1[View commits]
 *Affecting all Beats*
 
 - Fix panic when testing regex-AST to match against date patterns. {issue}3889[3889]
+- Fix panic due to race condition in kafka output. {pull}4098[4098]
 
 *Filebeat*
 
diff --git a/libbeat/monitoring/adapter/go-metrics.go b/libbeat/monitoring/adapter/go-metrics.go
index 2d0b94188ac..143bb0bf6ea 100644
--- a/libbeat/monitoring/adapter/go-metrics.go
+++ b/libbeat/monitoring/adapter/go-metrics.go
@@ -3,6 +3,7 @@ package adapter
 import (
 	"fmt"
 	"reflect"
+	"sync"
 
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/monitoring"
@@ -21,6 +22,8 @@ import (
 //       It's recommended to not mix go-metrics with other metrics types
 //       in the same namespace.
 type GoMetricsRegistry struct {
+	mutex sync.Mutex
+
 	reg     *monitoring.Registry
 	filters *metricFilters
 
@@ -80,6 +83,12 @@ func (r *GoMetricsRegistry) find(name string) interface{} {
 //       It's recommended to not mix go-metrics with other metrics types in one
 //       namespace.
 func (r *GoMetricsRegistry) Get(name string) interface{} {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return r.get(name)
+}
+
+func (r *GoMetricsRegistry) get(name string) interface{} {
 	m := r.find(name)
 	if m == nil {
 		return r.shadow.Get(name)
@@ -95,7 +104,10 @@ func (r *GoMetricsRegistry) Get(name string) interface{} {
 // GetOrRegister retries an existing metric via `Get` or registers a new one
 // if the metric is unknown. For lazy instantiation metric can be a function.
 func (r *GoMetricsRegistry) GetOrRegister(name string, metric interface{}) interface{} {
-	v := r.Get(name)
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+
+	v := r.get(name)
 	if v != nil {
 		return v
 	}
@@ -106,7 +118,10 @@ func (r *GoMetricsRegistry) GetOrRegister(name string, metric interface{}) inter
 // Register adds a new metric.
 // An error is returned if the metric is already known.
 func (r *GoMetricsRegistry) Register(name string, metric interface{}) error {
-	if r.Get(name) != nil {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+
+	if r.get(name) != nil {
 		return fmt.Errorf("metric '%v' already registered", name)
 	}
 
@@ -139,6 +154,9 @@ func (r *GoMetricsRegistry) RunHealthchecks() {}
 
 // Unregister removes a metric.
 func (r *GoMetricsRegistry) Unregister(name string) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+
 	st := r.rmState(name)
 	r.reg.Remove(st.name)
 	r.shadow.Unregister(name)
@@ -146,6 +164,9 @@ func (r *GoMetricsRegistry) Unregister(name string) {
 
 // UnregisterAll calls `Clear` on the underlying monitoring.Registry
 func (r *GoMetricsRegistry) UnregisterAll() {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+
 	r.shadow.UnregisterAll()
 	err := r.reg.Clear()
 	if err != nil {