Add ability to use existing processors from script processor (#11260)

* Add JS require with modules This adds a `require` function to the JS runtime that scripts can call to import "modules". Some standard modules that are added in this PR are - `processor` - You can construct beat processors in JS (e.g. `new processor.Dissect({...})`). - `console` - You can write to beat's logger. - `path` - You can parse win32 and posix paths. The `processor` module supports constructing: - `AddCloudMetadata` - `AddDockerMetadata` - `AddHostMetadata` - `AddKubernetesMetadata` - `AddLocale` - `AddProcessMetadata` - `CommunityID` - `DecodeJSONFields` - `Dissect` - `DNS` * Add github.com/dop251/goja_nodejs to vendor
elastic · Mar 21, 2019 · d36eb5a · d36eb5a
1 parent 02d29c6
commit d36eb5a
Show file tree

Hide file tree

Showing 37 changed files with 1,405 additions and 43 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -256,7 +256,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add output test to kafka output {pull}10834[10834]
 - Add ip fields to default_field in Elasticsearch template. {pull}11035[11035]
 - Gracefully shut down on SIGHUP {pull}10704[10704]
-- Add `script` processor that supports using Javascript to process events. {pull}10850[10850]
+- Add `script` processor that supports using Javascript to process events. {pull}10850[10850] {pull}11260[11260]
 
 *Auditbeat*
 

diff --git a/NOTICE.txt b/NOTICE.txt
@@ -504,6 +504,26 @@ WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEM
 COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
 OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
+--------------------------------------------------------------------
+Dependency: github.com/dop251/goja_nodejs
+Revision: adff31b136e6b7e044712aab7236a775e1bd085e
+License type (autodetected): MIT
+./vendor/github.com/dop251/goja_nodejs/LICENSE:
+--------------------------------------------------------------------
+Copyright (c) 2016 Dmitry Panov
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the Software without restriction, including without limitation
+the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
 --------------------------------------------------------------------
 Dependency: github.com/dustin/go-humanize
 Revision: 259d2a102b871d17f30e3cd9881a642961a1e486

diff --git a/libbeat/processors/actions/decode_json_fields.go b/libbeat/processors/actions/decode_json_fields.go
@@ -58,12 +58,13 @@ var debug = logp.MakeDebug("filters")
 
 func init() {
  processors.RegisterPlugin("decode_json_fields",
- configChecked(newDecodeJSONFields,
+ configChecked(NewDecodeJSONFields,
  requireFields("fields"),
  allowedFields("fields", "max_depth", "overwrite_keys", "process_array", "target", "when")))
 }
 
-func newDecodeJSONFields(c *common.Config) (processors.Processor, error) {
+// NewDecodeJSONFields construct a new decode_json_fields processor.
+func NewDecodeJSONFields(c *common.Config) (processors.Processor, error) {
  config := defaultConfig
 
  err := c.Unpack(&config)

diff --git a/libbeat/processors/actions/decode_json_fields_test.go b/libbeat/processors/actions/decode_json_fields_test.go
@@ -202,7 +202,7 @@ func TestTargetRootOption(t *testing.T) {
 func getActualValue(t *testing.T, config *common.Config, input common.MapStr) common.MapStr {
  logp.TestingSetup()
 
- p, err := newDecodeJSONFields(config)
+ p, err := NewDecodeJSONFields(config)
  if err != nil {
  logp.Err("Error initializing decode_json_fields")
  t.Fatal(err)

diff --git a/libbeat/processors/add_cloud_metadata/add_cloud_metadata.go b/libbeat/processors/add_cloud_metadata/add_cloud_metadata.go
@@ -49,7 +49,7 @@ var debugf = logp.MakeDebug("filters")
 
 // init registers the add_cloud_metadata processor.
 func init() {
- processors.RegisterPlugin("add_cloud_metadata", newCloudMetadata)
+ processors.RegisterPlugin("add_cloud_metadata", New)
 }
 
 type schemaConv func(m map[string]interface{}) common.MapStr
@@ -299,7 +299,8 @@ func setupFetchers(c *common.Config) ([]*metadataFetcher, error) {
  return fetchers, nil
 }
 
-func newCloudMetadata(c *common.Config) (processors.Processor, error) {
+// New constructs a new add_cloud_metadata processor.
+func New(c *common.Config) (processors.Processor, error) {
  config := struct {
  Timeout time.Duration `config:"timeout"` // Amount of time to wait for responses from the metadata services.
  }{

diff --git a/libbeat/processors/add_cloud_metadata/provider_alibaba_cloud_test.go b/libbeat/processors/add_cloud_metadata/provider_alibaba_cloud_test.go
@@ -62,7 +62,7 @@ func TestRetrieveAlibabaCloudMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- p, err := newCloudMetadata(config)
+ p, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/add_cloud_metadata/provider_aws_ec2_test.go b/libbeat/processors/add_cloud_metadata/provider_aws_ec2_test.go
@@ -70,7 +70,7 @@ func TestRetrieveAWSMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- p, err := newCloudMetadata(config)
+ p, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/add_cloud_metadata/provider_azure_vm_test.go b/libbeat/processors/add_cloud_metadata/provider_azure_vm_test.go
@@ -67,7 +67,7 @@ func TestRetrieveAzureMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- p, err := newCloudMetadata(config)
+ p, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/add_cloud_metadata/provider_digital_ocean_test.go b/libbeat/processors/add_cloud_metadata/provider_digital_ocean_test.go
@@ -101,7 +101,7 @@ func TestRetrieveDigitalOceanMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- p, err := newCloudMetadata(config)
+ p, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/add_cloud_metadata/provider_google_gce_test.go b/libbeat/processors/add_cloud_metadata/provider_google_gce_test.go
@@ -140,7 +140,7 @@ func TestRetrieveGCEMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- p, err := newCloudMetadata(config)
+ p, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/add_cloud_metadata/provider_openstack_nova_test.go b/libbeat/processors/add_cloud_metadata/provider_openstack_nova_test.go
@@ -66,7 +66,7 @@ func TestRetrieveOpenstackNovaMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- p, err := newCloudMetadata(config)
+ p, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/add_cloud_metadata/provider_tencent_cloud_test.go b/libbeat/processors/add_cloud_metadata/provider_tencent_cloud_test.go
@@ -62,7 +62,7 @@ func TestRetrieveQCloudMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- p, err := newCloudMetadata(config)
+ p, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/add_docker_metadata/add_docker_metadata.go b/libbeat/processors/add_docker_metadata/add_docker_metadata.go
@@ -48,7 +48,7 @@ const (
 var processCgroupPaths = cgroup.ProcessCgroupPaths
 
 func init() {
- processors.RegisterPlugin(processorName, newDockerMetadataProcessor)
+ processors.RegisterPlugin(processorName, New)
 }
 
 type addDockerMetadata struct {
@@ -63,7 +63,8 @@ type addDockerMetadata struct {
  dedot bool // If set to true, replace dots in labels with `_`.
 }
 
-func newDockerMetadataProcessor(cfg *common.Config) (processors.Processor, error) {
+// New constructs a new add_docker_metadata processor.
+func New(cfg *common.Config) (processors.Processor, error) {
  return buildDockerMetadataProcessor(cfg, docker.NewWatcher)
 }
 

diff --git a/libbeat/processors/add_host_metadata/add_host_metadata.go b/libbeat/processors/add_host_metadata/add_host_metadata.go
@@ -36,7 +36,7 @@ import (
 )
 
 func init() {
- processors.RegisterPlugin("add_host_metadata", newHostMetadataProcessor)
+ processors.RegisterPlugin("add_host_metadata", New)
 }
 
 type addHostMetadata struct {
@@ -53,7 +53,8 @@ const (
  processorName = "add_host_metadata"
 )
 
-func newHostMetadataProcessor(cfg *common.Config) (processors.Processor, error) {
+// New constructs a new add_host_metadata processor.
+func New(cfg *common.Config) (processors.Processor, error) {
  config := defaultConfig()
  if err := cfg.Unpack(&config); err != nil {
  return nil, errors.Wrapf(err, "fail to unpack the %v configuration", processorName)

diff --git a/libbeat/processors/add_host_metadata/add_host_metadata_test.go b/libbeat/processors/add_host_metadata/add_host_metadata_test.go
@@ -40,7 +40,7 @@ func TestConfigDefault(t *testing.T) {
  testConfig, err := common.NewConfigFrom(map[string]interface{}{})
  assert.NoError(t, err)
 
- p, err := newHostMetadataProcessor(testConfig)
+ p, err := New(testConfig)
  switch runtime.GOOS {
  case "windows", "darwin", "linux":
  assert.NoError(t, err)
@@ -83,7 +83,7 @@ func TestConfigNetInfoEnabled(t *testing.T) {
  })
  assert.NoError(t, err)
 
- p, err := newHostMetadataProcessor(testConfig)
+ p, err := New(testConfig)
  switch runtime.GOOS {
  case "windows", "darwin", "linux":
  assert.NoError(t, err)
@@ -129,7 +129,7 @@ func TestConfigName(t *testing.T) {
  testConfig, err := common.NewConfigFrom(config)
  assert.NoError(t, err)
 
- p, err := newHostMetadataProcessor(testConfig)
+ p, err := New(testConfig)
  require.NoError(t, err)
 
  newEvent, err := p.Run(event)
@@ -163,7 +163,7 @@ func TestConfigGeoEnabled(t *testing.T) {
  testConfig, err := common.NewConfigFrom(config)
  assert.NoError(t, err)
 
- p, err := newHostMetadataProcessor(testConfig)
+ p, err := New(testConfig)
  require.NoError(t, err)
 
  newEvent, err := p.Run(event)
@@ -192,7 +192,7 @@ func TestPartialGeo(t *testing.T) {
  testConfig, err := common.NewConfigFrom(config)
  assert.NoError(t, err)
 
- p, err := newHostMetadataProcessor(testConfig)
+ p, err := New(testConfig)
  require.NoError(t, err)
 
  newEvent, err := p.Run(event)
@@ -233,7 +233,7 @@ func TestGeoLocationValidation(t *testing.T) {
  })
  require.NoError(t, err)
 
- _, err = newHostMetadataProcessor(conf)
+ _, err = New(conf)
 
  if location.valid {
  require.NoError(t, err)

diff --git a/libbeat/processors/add_kubernetes_metadata/kubernetes.go b/libbeat/processors/add_kubernetes_metadata/kubernetes.go
@@ -41,7 +41,7 @@ type kubernetesAnnotator struct {
 }
 
 func init() {
- processors.RegisterPlugin("add_kubernetes_metadata", newKubernetesAnnotator)
+ processors.RegisterPlugin("add_kubernetes_metadata", New)
 
  // Register default indexers
  Indexing.AddIndexer(PodNameIndexerName, NewPodNameIndexer)
@@ -52,7 +52,8 @@ func init() {
  Indexing.AddMatcher(FieldFormatMatcherName, NewFieldFormatMatcher)
 }
 
-func newKubernetesAnnotator(cfg *common.Config) (processors.Processor, error) {
+// New constructs a new add_kubernetes_metadata processor.
+func New(cfg *common.Config) (processors.Processor, error) {
  config := defaultKubernetesAnnotatorConfig()
 
  err := cfg.Unpack(&config)

diff --git a/libbeat/processors/add_locale/add_locale.go b/libbeat/processors/add_locale/add_locale.go
@@ -52,10 +52,11 @@ func (t TimezoneFormat) String() string {
 }
 
 func init() {
- processors.RegisterPlugin("add_locale", newAddLocale)
+ processors.RegisterPlugin("add_locale", New)
 }
 
-func newAddLocale(c *common.Config) (processors.Processor, error) {
+// New constructs a new add_locale processor.
+func New(c *common.Config) (processors.Processor, error) {
  config := struct {
  Format string `config:"format"`
  }{

diff --git a/libbeat/processors/add_locale/add_locale_test.go b/libbeat/processors/add_locale/add_locale_test.go
@@ -87,7 +87,7 @@ func TestTimezoneFormat(t *testing.T) {
 func getActualValue(t *testing.T, config *common.Config, input common.MapStr) common.MapStr {
  logp.TestingSetup()
 
- p, err := newAddLocale(config)
+ p, err := New(config)
  if err != nil {
  logp.Err("Error initializing add_locale")
  t.Fatal(err)
@@ -102,7 +102,7 @@ func BenchmarkConstruct(b *testing.B) {
 
  input := common.MapStr{}
 
- p, err := newAddLocale(testConfig)
+ p, err := New(testConfig)
  if err != nil {
  b.Fatal(err)
  }

diff --git a/libbeat/processors/add_process_metadata/add_process_metadata.go b/libbeat/processors/add_process_metadata/add_process_metadata.go
@@ -71,10 +71,11 @@ type processMetadataProvider interface {
 }
 
 func init() {
- processors.RegisterPlugin(processorName, newProcessMetadataProcessor)
+ processors.RegisterPlugin(processorName, New)
 }
 
-func newProcessMetadataProcessor(cfg *common.Config) (processors.Processor, error) {
+// New constructs a new add_process_metadata processor.
+func New(cfg *common.Config) (processors.Processor, error) {
  return newProcessMetadataProcessorWithProvider(cfg, &procCache)
 }
 

diff --git a/libbeat/processors/add_process_metadata/add_process_metadata_test.go b/libbeat/processors/add_process_metadata/add_process_metadata_test.go
@@ -430,7 +430,7 @@ func TestSelf(t *testing.T) {
  if err != nil {
  t.Fatal(err)
  }
- proc, err := newProcessMetadataProcessor(config)
+ proc, err := New(config)
  if err != nil {
  t.Fatal(err)
  }
@@ -463,7 +463,7 @@ func TestBadProcess(t *testing.T) {
  if err != nil {
  t.Fatal(err)
  }
- proc, err := newProcessMetadataProcessor(config)
+ proc, err := New(config)
  if err != nil {
  t.Fatal(err)
  }

diff --git a/libbeat/processors/dissect/processor.go b/libbeat/processors/dissect/processor.go
@@ -34,10 +34,11 @@ type processor struct {
 }
 
 func init() {
- processors.RegisterPlugin("dissect", newProcessor)
+ processors.RegisterPlugin("dissect", NewProcessor)
 }
 
-func newProcessor(c *common.Config) (processors.Processor, error) {
+// NewProcessor constructs a new dissect processor.
+func NewProcessor(c *common.Config) (processors.Processor, error) {
  config := defaultConfig
  err := c.Unpack(&config)
  if err != nil {

diff --git a/libbeat/processors/dissect/processor_test.go b/libbeat/processors/dissect/processor_test.go
@@ -84,7 +84,7 @@ func TestProcessor(t *testing.T) {
  return
  }
 
- processor, err := newProcessor(c)
+ processor, err := NewProcessor(c)
  if !assert.NoError(t, err) {
  return
  }
@@ -113,7 +113,7 @@ func TestFieldDoesntExist(t *testing.T) {
  return
  }
 
- processor, err := newProcessor(c)
+ processor, err := NewProcessor(c)
  if !assert.NoError(t, err) {
  return
  }
@@ -163,7 +163,7 @@ func TestFieldAlreadyExist(t *testing.T) {
  return
  }
 
- processor, err := newProcessor(c)
+ processor, err := NewProcessor(c)
  if !assert.NoError(t, err) {
  return
  }
@@ -187,7 +187,7 @@ func TestErrorFlagging(t *testing.T) {
  return
  }
 
- processor, err := newProcessor(c)
+ processor, err := NewProcessor(c)
  if !assert.NoError(t, err) {
  return
  }
@@ -216,7 +216,7 @@ func TestErrorFlagging(t *testing.T) {
  return
  }
 
- processor, err := newProcessor(c)
+ processor, err := NewProcessor(c)
  if !assert.NoError(t, err) {
  return
  }

diff --git a/libbeat/processors/dns/dns.go b/libbeat/processors/dns/dns.go
@@ -39,7 +39,7 @@ const logName = "processor.dns"
 var instanceID = atomic.MakeUint32(0)
 
 func init() {
- processors.RegisterPlugin("dns", newDNSProcessor)
+ processors.RegisterPlugin("dns", New)
 }
 
 type processor struct {
@@ -48,7 +48,8 @@ type processor struct {
  log *logp.Logger
 }
 
-func newDNSProcessor(cfg *common.Config) (processors.Processor, error) {
+// New constructs a new DNS processor.
+func New(cfg *common.Config) (processors.Processor, error) {
  c := defaultConfig
  if err := cfg.Unpack(&c); err != nil {
  return nil, errors.Wrap(err, "fail to unpack the dns configuration")