[Metricbeat] Support processors defined for light modules

metricbeat/docs/modules/activemq.asciidoc

@@ -35,36 +35,6 @@ metricbeat.modules:
   path: '/api/jolokia/?ignoreErrors=true&canonicalNaming=false'
   username: admin # default username
   password: admin # default password
-  processors:
-    - script:
-        lang: javascript
-        source: >
-          function process(event) {
-            var broker_memory_broker_pct = event.Get("activemq.broker.memory.broker.pct")
-            if (broker_memory_broker_pct != null) {
-              event.Put("activemq.broker.memory.broker.pct", broker_memory_broker_pct / 100.0)
-            }
-
-            var broker_memory_temp_pct = event.Get("activemq.broker.memory.temp.pct")
-            if (broker_memory_temp_pct != null) {
-              event.Put("activemq.broker.memory.temp.pct", broker_memory_temp_pct / 100.0)
-            }
-
-            var broker_memory_store_pct = event.Get("activemq.broker.memory.store.pct")
-            if (broker_memory_store_pct != null) {
-              event.Put("activemq.broker.memory.store.pct", broker_memory_store_pct / 100.0)
-            }
-
-            var queue_memory_broker_pct = event.Get("activemq.queue.memory.broker.pct")
-            if (queue_memory_broker_pct != null) {
-              event.Put("activemq.queue.memory.broker.pct", queue_memory_broker_pct / 100.0)
-            }
-
-            var topic_memory_broker_pct = event.Get("activemq.topic.memory.broker.pct")
-            if (topic_memory_broker_pct != null) {
-              event.Put("activemq.topic.memory.broker.pct", topic_memory_broker_pct / 100.0)
-            }
-          }
 ----
 
 [float]

metricbeat/docs/modules/ibmmq.asciidoc

@@ -55,26 +55,6 @@ metricbeat.modules:
   # This module uses the Prometheus collector metricset, all
   # the options for this metricset are also available here.
   metrics_path: /metrics
-
-  # The custom processor is responsible for filtering Prometheus metrics
-  # not stricly related to the IBM MQ domain, e.g. system load, process,
-  # metrics HTTP server.
-  processors:
-    - script:
-        lang: javascript
-        source: >
-          function process(event) {
-            var metrics = event.Get("prometheus.metrics");
-            Object.keys(metrics).forEach(function(key) {
-              if (!(key.match(/^ibmmq_.*$/))) {
-                event.Delete("prometheus.metrics." + key);
-              }
-            });
-            metrics = event.Get("prometheus.metrics");
-            if (Object.keys(metrics).length == 0) {
-              event.Cancel();
-            }
-          }
 ----
 
 It also supports the options described in <<module-http-config-options>>.

metricbeat/mb/lightmetricset.go

@@ -21,6 +21,7 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/processors"
 )
 
 // LightMetricSet contains the definition of a non-registered metric set
@@ -33,6 +34,7 @@ type LightMetricSet struct {
 		MetricSet string      `config:"metricset" validate:"required"`
 		Defaults  interface{} `config:"defaults"`
 	} `config:"input" validate:"required"`
+	Processors processors.PluginConfig `config:"processors"`
 }
 
 // Registration obtains a metric set registration for this light metric set, this registration

metricbeat/mb/lightmodules.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/processors"
 )
 
 const (
@@ -150,6 +151,19 @@ type lightModuleConfig struct {
 	MetricSets []string `config:"metricsets"`
 }
 
+// ProcessorsForMetricSet returns processors defined for the light metricset.
+func (s *LightModulesSource) ProcessorsForMetricSet(r *Register, moduleName string, metricSetName string) (*processors.Processors, error) {
+	module, err := s.loadModule(r, moduleName)
+	if err != nil {
+		return nil, errors.Wrapf(err, "reading processors for metricset '%s' in module '%s' failed", metricSetName, moduleName)
+	}
+	metricSet, ok := module.MetricSets[metricSetName]
+	if !ok {
+		return nil, fmt.Errorf("unknown metricset '%s' in module '%s'", metricSetName, moduleName)
+	}
+	return processors.New(metricSet.Processors)
+}
+
 // LightModule contains the definition of a light module
 type LightModule struct {
 	Name       string

metricbeat/mb/lightmodules_test.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
+	_ "github.com/elastic/beats/libbeat/processors/add_id"
 )
 
 // TestLightModulesAsModuleSource checks that registry correctly lists
@@ -302,6 +303,31 @@ func TestNewModulesCallModuleFactory(t *testing.T) {
 	assert.True(t, called, "module factory must be called if registered")
 }
 
+func TestProcessorsForMetricSet_UnknownModule(t *testing.T) {
+	r := NewRegister()
+	source := NewLightModulesSource("testdata/lightmodules")
+	procs, err := source.ProcessorsForMetricSet(r, "nonexisting", "fake")
+	require.Error(t, err)
+	require.Nil(t, procs)
+}
+
+func TestProcessorsForMetricSet_UnknownMetricSet(t *testing.T) {
+	r := NewRegister()
+	source := NewLightModulesSource("testdata/lightmodules")
+	procs, err := source.ProcessorsForMetricSet(r, "unpack", "nonexisting")
+	require.Error(t, err)
+	require.Nil(t, procs)
+}
+
+func TestProcessorsForMetricSet_ProcessorsRead(t *testing.T) {
+	r := NewRegister()
+	source := NewLightModulesSource("testdata/lightmodules")
+	procs, err := source.ProcessorsForMetricSet(r, "unpack", "withprocessors")
+	require.NoError(t, err)
+	require.NotNil(t, procs)
+	require.Len(t, procs.List, 1)
+}
+
 type metricSetWithOption struct {
 	BaseMetricSet
 	Option string

metricbeat/mb/module/connector.go

@@ -18,6 +18,8 @@
 package module
 
 import (
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/fmtstr"
@@ -47,6 +49,10 @@ type connectorConfig struct {
 	common.EventMetadata `config:",inline"` // Fields and tags to add to events.
 }
 
+type metricSetRegister interface {
+	ProcessorsForMetricSet(moduleName, metricSetName string) (*processors.Processors, error)
+}
+
 func NewConnector(
 	beatInfo beat.Info, pipeline beat.Pipeline,
 	c *common.Config, dynFields *common.MapStrPointer,
@@ -70,6 +76,27 @@ func NewConnector(
 	}, nil
 }
 
+// UseMetricSetProcessors appends processors defined in metricset configuration to the connector properties.
+func (c *Connector) UseMetricSetProcessors(r metricSetRegister, moduleName, metricSetName string) error {
+	metricSetProcessors, err := r.ProcessorsForMetricSet(moduleName, metricSetName)
+	if err != nil {
+		return errors.Wrapf(err, "reading metricset processors failed (module: %s, metricset: %s)",
+			moduleName, metricSetName)
+	}
+
+	if metricSetProcessors == nil || len(metricSetProcessors.List) == 0 {
+		return nil // no processors are defined
+	}
+
+	procs := processors.NewList(nil)
+	procs.AddProcessors(*metricSetProcessors)
+	for _, p := range c.processors.List {
+		procs.AddProcessor(p)
+	}
+	c.processors = procs
+	return nil
+}
+
 func (c *Connector) Connect() (beat.Client, error) {
 	return c.pipeline.ConnectWith(beat.ClientConfig{
 		Processing: beat.ProcessingConfig{

metricbeat/mb/module/connector_test.go

@@ -18,13 +18,16 @@
 package module
 
 import (
+	"errors"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/processors"
 )
 
 func TestProcessorsForConfig(t *testing.T) {
@@ -91,6 +94,44 @@ func TestProcessorsForConfig(t *testing.T) {
 	}
 }
 
+type fakeMetricSetRegister struct {
+	success bool
+}
+
+func (fmsr *fakeMetricSetRegister) ProcessorsForMetricSet(moduleName, metricSetName string) (*processors.Processors, error) {
+	if !fmsr.success {
+		return nil, errors.New("failure")
+	}
+
+	procs := new(processors.Processors)
+	procs.List = []processors.Processor{nil, nil}
+	return procs, nil
+}
+
+func TestUseMetricSetProcessors_ReadingProcessorsFailed(t *testing.T) {
+	r := new(fakeMetricSetRegister)
+
+	var connector Connector
+	err := connector.UseMetricSetProcessors(r, "module", "metricset")
+	require.Error(t, err)
+	require.Nil(t, connector.processors)
+}
+
+func TestUseMetricSetProcessors_ReadingProcessorsSucceeded(t *testing.T) {
+	r := &fakeMetricSetRegister{
+		success: true,
+	}
+
+	connector := Connector{
+		processors: &processors.Processors{
+			List: []processors.Processor{},
+		},
+	}
+	err := connector.UseMetricSetProcessors(r, "module", "metricset")
+	require.NoError(t, err)
+	require.Len(t, connector.processors.List, 2)
+}
+
 // Helper function to convert from YML input string to an unpacked
 // connectorConfig
 func connectorConfigFromString(s string) (connectorConfig, error) {

metricbeat/mb/module/factory.go

@@ -18,8 +18,6 @@
 package module
 
 import (
-	"github.com/joeshaw/multierror"
-
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/cfgfile"
 	"github.com/elastic/beats/libbeat/common"
@@ -43,28 +41,36 @@ func NewFactory(beatInfo beat.Info, options ...Option) *Factory {
 
 // Create creates a new metricbeat module runner reporting events to the passed pipeline.
 func (r *Factory) Create(p beat.Pipeline, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) {
-	var errs multierror.Errors
-
-	connector, err := NewConnector(r.beatInfo, p, c, meta)
-	if err != nil {
-		errs = append(errs, err)
-	}
-	w, err := NewWrapper(c, mb.Registry, r.options...)
+	module, metricSets, err := mb.NewModule(c, mb.Registry)
 	if err != nil {
-		errs = append(errs, err)
-	}
-
-	if err := errs.Err(); err != nil {
 		return nil, err
 	}
 
-	client, err := connector.Connect()
-	if err != nil {
-		return nil, err
+	var runners []Runner
+	for _, metricSet := range metricSets {
+		wrapper, err := NewWrapperForMetricSet(module, metricSet, r.options...)
+		if err != nil {
+			return nil, err
+		}
+
+		connector, err := NewConnector(r.beatInfo, p, c, meta)
+		if err != nil {
+			return nil, err
+		}
+
+		err = connector.UseMetricSetProcessors(mb.Registry, module.Name(), metricSet.Name())
+		if err != nil {
+			return nil, err
+		}
+
+		client, err := connector.Connect()
+		if err != nil {
+			return nil, err
+		}
+		runners = append(runners, NewRunner(client, wrapper))
 	}
 
-	mr := NewRunner(client, w)
-	return mr, nil
+	return newRunnerGroup(runners), nil
 }
 
 // CheckConfig checks if a config is valid or not

metricbeat/mb/module/runner_group.go

@@ -0,0 +1,62 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package module
+
+import (
+	"strings"
+	"sync"
+)
+
+type runnerGroup struct {
+	Runner
+
+	runners []Runner
+
+	startOnce sync.Once
+	stopOnce  sync.Once
+}
+
+func newRunnerGroup(runners []Runner) Runner {
+	return &runnerGroup{
+		runners: runners,
+	}
+}
+
+func (rg *runnerGroup) Start() {
+	rg.startOnce.Do(func() {
+		for _, runner := range rg.runners {
+			runner.Start()
+		}
+	})
+}
+
+func (rg *runnerGroup) Stop() {
+	rg.stopOnce.Do(func() {
+		for _, runner := range rg.runners {
+			runner.Stop()
+		}
+	})
+}
+
+func (rg *runnerGroup) String() string {
+	var entries []string
+	for _, runner := range rg.runners {
+		entries = append(entries, runner.String())
+	}
+	return "RunnerGroup: " + strings.Join(entries, ", ")
+}