Make kubernetes indexers/matchers pluggable (elastic#4151)

CHANGELOG.asciidoc

@@ -112,6 +112,7 @@ https://github.com/elastic/beats/compare/v5.1.1...master[Check the HEAD diff]
 - Add support for include_labels and include_annotations in kubernetes processor {pull}4043[4043]
 - Support new `index_patterns` field when loading templates for Elasticsearch >= 6.0 {pull}4056[4056]
 - Adding goimports support to make check and fmt {pull}4114[4114]
+- Make kubernetes indexers/matchers pluggable {pull}4151[4151]
 
 *Filebeat*
 

libbeat/processors/kubernetes/indexing.go

@@ -67,20 +67,20 @@ type Matchers struct {
 // Register contains Indexer and Matchers to use on pod indexing and event matching
 type Register struct {
 	sync.RWMutex
-	indexers map[string]IndexConstructor
+	indexers map[string]IndexerConstructor
 	matchers map[string]MatcherConstructor
 
 	defaultIndexerConfigs map[string]common.Config
 	defaultMatcherConfigs map[string]common.Config
 }
 
-type IndexConstructor func(config common.Config, genMeta GenMeta) (Indexer, error)
+type IndexerConstructor func(config common.Config, genMeta GenMeta) (Indexer, error)
 type MatcherConstructor func(config common.Config) (Matcher, error)
 
 // NewRegister creates and returns a new Register.
 func NewRegister() *Register {
 	return &Register{
-		indexers: make(map[string]IndexConstructor, 0),
+		indexers: make(map[string]IndexerConstructor, 0),
 		matchers: make(map[string]MatcherConstructor, 0),
 
 		defaultIndexerConfigs: make(map[string]common.Config, 0),
@@ -89,7 +89,7 @@ func NewRegister() *Register {
 }
 
 // AddIndexer to the register
-func (r *Register) AddIndexer(name string, indexer IndexConstructor) {
+func (r *Register) AddIndexer(name string, indexer IndexerConstructor) {
 	r.RWMutex.Lock()
 	defer r.RWMutex.Unlock()
 	r.indexers[name] = indexer
@@ -113,7 +113,7 @@ func (r *Register) AddDefaultMatcherConfig(name string, config common.Config) {
 }
 
 // AddIndexer to the register
-func (r *Register) GetIndexer(name string) IndexConstructor {
+func (r *Register) GetIndexer(name string) IndexerConstructor {
 	indexer, ok := r.indexers[name]
 	if ok {
 		return indexer

libbeat/processors/kubernetes/registry.go

@@ -0,0 +1,63 @@
+package kubernetes
+
+import (
+	"errors"
+	"fmt"
+
+	p "github.com/elastic/beats/libbeat/plugin"
+)
+
+var (
+	indexerKey = "libbeat.processor.kubernetes.indexer"
+	matcherKey = "libbeat.processor.kubernetes.matcher"
+)
+
+type indexerPlugin struct {
+	name        string
+	constructor IndexerConstructor
+}
+
+func IndexerPlugin(name string, c IndexerConstructor) map[string][]interface{} {
+	return p.MakePlugin(indexerKey, indexerPlugin{name, c})
+}
+
+type matcherPlugin struct {
+	name        string
+	constructor MatcherConstructor
+}
+
+func MatcherPlugin(name string, m MatcherConstructor) map[string][]interface{} {
+	return p.MakePlugin(matcherKey, matcherPlugin{name, m})
+}
+
+func init() {
+	p.MustRegisterLoader(indexerKey, func(ifc interface{}) error {
+		i, ok := ifc.(indexerPlugin)
+		if !ok {
+			return errors.New("plugin does not match output plugin type")
+		}
+
+		name := i.name
+		if Indexing.indexers[name] != nil {
+			return fmt.Errorf("indexer type %v already registered", name)
+		}
+
+		Indexing.AddIndexer(name, i.constructor)
+		return nil
+	})
+
+	p.MustRegisterLoader(matcherKey, func(ifc interface{}) error {
+		m, ok := ifc.(matcherPlugin)
+		if !ok {
+			return errors.New("plugin does not match output plugin type")
+		}
+
+		name := m.name
+		if Indexing.indexers[name] != nil {
+			return fmt.Errorf("matcher type %v already registered", name)
+		}
+
+		Indexing.AddMatcher(name, m.constructor)
+		return nil
+	})
+}