Remove unneeded locks in add_kubernetes_metadata

libbeat/processors/add_kubernetes_metadata/indexers.go

@@ -19,7 +19,6 @@ package add_kubernetes_metadata
 
 import (
 	"fmt"
-	"sync"
 
 	"github.com/elastic/beats/v7/libbeat/common/kubernetes/metadata"
 
@@ -55,7 +54,6 @@ type MetadataIndex struct {
 }
 
 type Indexers struct {
-	sync.RWMutex
 	indexers []Indexer
 }
 
@@ -91,8 +89,6 @@ func NewIndexers(configs PluginConfig, metaGen metadata.MetaGen) *Indexers {
 // GetIndexes returns the composed index list from all registered indexers
 func (i *Indexers) GetIndexes(pod *kubernetes.Pod) []string {
 	var indexes []string
-	i.RLock()
-	defer i.RUnlock()
 	for _, indexer := range i.indexers {
 		for _, i := range indexer.GetIndexes(pod) {
 			indexes = append(indexes, i)
@@ -104,8 +100,6 @@ func (i *Indexers) GetIndexes(pod *kubernetes.Pod) []string {
 // GetMetadata returns the composed metadata list from all registered indexers
 func (i *Indexers) GetMetadata(pod *kubernetes.Pod) []MetadataIndex {
 	var metadata []MetadataIndex
-	i.RLock()
-	defer i.RUnlock()
 	for _, indexer := range i.indexers {
 		for _, m := range indexer.GetMetadata(pod) {
 			metadata = append(metadata, m)
@@ -116,8 +110,6 @@ func (i *Indexers) GetMetadata(pod *kubernetes.Pod) []MetadataIndex {
 
 // Empty returns true if indexers list is empty
 func (i *Indexers) Empty() bool {
-	i.RLock()
-	defer i.RUnlock()
 	if len(i.indexers) == 0 {
 		return true
 	}

libbeat/processors/add_kubernetes_metadata/indexing.go

@@ -50,30 +50,36 @@ func NewRegister() *Register {
 
 // AddIndexer to the register
 func (r *Register) AddIndexer(name string, indexer IndexerConstructor) {
-	r.RWMutex.Lock()
-	defer r.RWMutex.Unlock()
+	r.Lock()
+	defer r.Unlock()
 	r.indexers[name] = indexer
 }
 
 // AddMatcher to the register
 func (r *Register) AddMatcher(name string, matcher MatcherConstructor) {
-	r.RWMutex.Lock()
-	defer r.RWMutex.Unlock()
+	r.Lock()
+	defer r.Unlock()
 	r.matchers[name] = matcher
 }
 
 // AddIndexer to the register
 func (r *Register) AddDefaultIndexerConfig(name string, config common.Config) {
+	r.Lock()
+	defer r.Unlock()
 	r.defaultIndexerConfigs[name] = config
 }
 
 // AddMatcher to the register
 func (r *Register) AddDefaultMatcherConfig(name string, config common.Config) {
+	r.Lock()
+	defer r.Unlock()
 	r.defaultMatcherConfigs[name] = config
 }
 
 // AddIndexer to the register
 func (r *Register) GetIndexer(name string) IndexerConstructor {
+	r.RLock()
+	defer r.RUnlock()
 	indexer, ok := r.indexers[name]
 	if ok {
 		return indexer
@@ -84,6 +90,8 @@ func (r *Register) GetIndexer(name string) IndexerConstructor {
 
 // AddMatcher to the register
 func (r *Register) GetMatcher(name string) MatcherConstructor {
+	r.RLock()
+	defer r.RUnlock()
 	matcher, ok := r.matchers[name]
 	if ok {
 		return matcher
@@ -92,10 +100,30 @@ func (r *Register) GetMatcher(name string) MatcherConstructor {
 	}
 }
 
-func (r *Register) GetDefaultIndexerConfigs() map[string]common.Config {
-	return r.defaultIndexerConfigs
+// GetDefaultIndexerConfigs obtains the plugin configuration for the default indexer
+// configurations registered
+func (r *Register) GetDefaultIndexerConfigs() PluginConfig {
+	r.RLock()
+	defer r.RUnlock()
+
+	configs := make(PluginConfig, 0, len(r.defaultIndexerConfigs))
+	for key, cfg := range r.defaultIndexerConfigs {
+		configs = append(configs, map[string]common.Config{key: cfg})
+	}
+
+	return configs
 }
 
-func (r *Register) GetDefaultMatcherConfigs() map[string]common.Config {
-	return r.defaultMatcherConfigs
+// GetDefaultMatcherConfigs obtains the plugin configuration for the default matcher
+// configurations registered
+func (r *Register) GetDefaultMatcherConfigs() PluginConfig {
+	r.RLock()
+	defer r.RUnlock()
+
+	configs := make(PluginConfig, 0, len(r.defaultMatcherConfigs))
+	for key, cfg := range r.defaultMatcherConfigs {
+		configs = append(configs, map[string]common.Config{key: cfg})
+	}
+
+	return configs
 }

libbeat/processors/add_kubernetes_metadata/kubernetes.go

@@ -100,20 +100,12 @@ func New(cfg *common.Config) (processors.Processor, error) {
 
 	//Load default indexer configs
 	if config.DefaultIndexers.Enabled == true {
-		Indexing.RLock()
-		for key, cfg := range Indexing.GetDefaultIndexerConfigs() {
-			config.Indexers = append(config.Indexers, map[string]common.Config{key: cfg})
-		}
-		Indexing.RUnlock()
+		config.Indexers = Indexing.GetDefaultIndexerConfigs()
 	}
 
 	//Load default matcher configs
 	if config.DefaultMatchers.Enabled == true {
-		Indexing.RLock()
-		for key, cfg := range Indexing.GetDefaultMatcherConfigs() {
-			config.Matchers = append(config.Matchers, map[string]common.Config{key: cfg})
-		}
-		Indexing.RUnlock()
+		config.Matchers = Indexing.GetDefaultMatcherConfigs()
 	}
 
 	processor := &kubernetesAnnotator{

libbeat/processors/add_kubernetes_metadata/matchers.go

@@ -19,7 +19,6 @@ package add_kubernetes_metadata
 
 import (
 	"fmt"
-	"sync"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -43,7 +42,6 @@ type Matcher interface {
 }
 
 type Matchers struct {
-	sync.RWMutex
 	matchers []Matcher
 }
 
@@ -76,8 +74,6 @@ func NewMatchers(configs PluginConfig) *Matchers {
 
 // MetadataIndex returns the index string for the first matcher from the Registry returning one
 func (m *Matchers) MetadataIndex(event common.MapStr) string {
-	m.RLock()
-	defer m.RUnlock()
 	for _, matcher := range m.matchers {
 		index := matcher.MetadataIndex(event)
 		if index != "" {
@@ -90,8 +86,6 @@ func (m *Matchers) MetadataIndex(event common.MapStr) string {
 }
 
 func (m *Matchers) Empty() bool {
-	m.RLock()
-	defer m.RUnlock()
 	if len(m.matchers) == 0 {
 		return true
 	}