Make it possible to chain Beat processors in the script processor

libbeat/processors/actions/add_fields.go

@@ -36,12 +36,13 @@ const FieldsKey = "fields"
 
 func init() {
 	processors.RegisterPlugin("add_fields",
-		configChecked(createAddFields,
+		configChecked(CreateAddFields,
 			requireFields(FieldsKey),
 			allowedFields(FieldsKey, "target", "when")))
 }
 
-func createAddFields(c *common.Config) (processors.Processor, error) {
+// CreateAddFields constructs an add_fields processor from config.
+func CreateAddFields(c *common.Config) (processors.Processor, error) {
 	config := struct {
 		Fields common.MapStr `config:"fields" validate:"required"`
 		Target *string       `config:"target"`

libbeat/processors/actions/copy_fields.go

@@ -40,13 +40,14 @@ type copyFieldsConfig struct {
 
 func init() {
 	processors.RegisterPlugin("copy_fields",
-		configChecked(newCopyFields,
+		configChecked(NewCopyFields,
 			requireFields("fields"),
 		),
 	)
 }
 
-func newCopyFields(c *common.Config) (processors.Processor, error) {
+// NewCopyFields returns a new copy_fields processor.
+func NewCopyFields(c *common.Config) (processors.Processor, error) {
 	config := copyFieldsConfig{
 		IgnoreMissing: false,
 		FailOnError:   true,

libbeat/processors/actions/rename.go

@@ -45,11 +45,12 @@ type fromTo struct {
 
 func init() {
 	processors.RegisterPlugin("rename",
-		configChecked(newRenameFields,
+		configChecked(NewRenameFields,
 			requireFields("fields")))
 }
 
-func newRenameFields(c *common.Config) (processors.Processor, error) {
+// NewRenameFields returns a new rename processor.
+func NewRenameFields(c *common.Config) (processors.Processor, error) {
 	config := renameFieldsConfig{
 		IgnoreMissing: false,
 		FailOnError:   true,

libbeat/processors/actions/truncate_fields.go

@@ -48,14 +48,15 @@ type truncater func(*truncateFields, []byte) ([]byte, bool, error)
 
 func init() {
 	processors.RegisterPlugin("truncate_fields",
-		configChecked(newTruncateFields,
+		configChecked(NewTruncateFields,
 			requireFields("fields"),
 			mutuallyExclusiveRequiredFields("max_bytes", "max_characters"),
 		),
 	)
 }
 
-func newTruncateFields(c *common.Config) (processors.Processor, error) {
+// NewTruncateFields returns a new truncate_fields processor.
+func NewTruncateFields(c *common.Config) (processors.Processor, error) {
 	var config truncateFieldsConfig
 	err := c.Unpack(&config)
 	if err != nil {

libbeat/processors/script/javascript/module/processor/chain.go

@@ -0,0 +1,183 @@
+// 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 processor
+
+import (
+	"github.com/dop251/goja"
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/processors"
+	"github.com/elastic/beats/libbeat/processors/script/javascript"
+)
+
+// chainBuilder builds a new processor chain.
+type chainBuilder struct {
+	chain
+	runtime *goja.Runtime
+	this    *goja.Object
+}
+
+// newChainBuilder returns a javascript constructor that constructs a
+// chainBuilder.
+func newChainBuilder(runtime *goja.Runtime) func(call goja.ConstructorCall) *goja.Object {
+	return func(call goja.ConstructorCall) *goja.Object {
+		if len(call.Arguments) > 0 {
+			panic(runtime.NewGoError(errors.New("Chain accepts no arguments")))
+		}
+
+		c := &chainBuilder{runtime: runtime, this: call.This}
+		for name, fn := range constructors {
+			c.this.Set(name, c.makeBuilderFunc(fn))
+		}
+		call.This.Set("Add", c.Add)
+		call.This.Set("Build", c.Build)
+
+		return nil
+	}
+}
+
+// makeBuilderFunc returns a javascript function that constructs a new native
+// beat processor and adds it to the chain.
+func (b *chainBuilder) makeBuilderFunc(constructor processors.Constructor) func(call goja.FunctionCall) goja.Value {
+	return func(call goja.FunctionCall) goja.Value {
+		return b.addProcessor(constructor, call)
+	}
+}
+
+// addProcessor constructs a new native beat processor and adds it to the chain.
+func (b *chainBuilder) addProcessor(constructor processors.Constructor, call goja.FunctionCall) *goja.Object {
+	p, err := newNativeProcessor(constructor, call)
+	if err != nil {
+		panic(b.runtime.NewGoError(err))
+	}
+
+	b.procs = append(b.procs, p)
+	return b.this
+}
+
+// Add adds a processor to the chain. It requires one argument that can be
+// either a beat processor or a javascript function.
+func (b *chainBuilder) Add(call goja.FunctionCall) goja.Value {
+	a0 := call.Argument(0)
+	if goja.IsUndefined(a0) {
+		panic(b.runtime.NewGoError(errors.New("Add requires a processor object parameter, but got undefined")))
+	}
+
+	switch v := a0.Export().(type) {
+	case *beatProcessor:
+		b.procs = append(b.procs, v.p)
+	case func(goja.FunctionCall) goja.Value:
+		b.procs = append(b.procs, &jsProcessor{fn: v})
+	default:
+		panic(b.runtime.NewGoError(errors.Errorf("arg0 must be a processor object, but got %T", a0.Export())))
+	}
+
+	return b.this
+}
+
+// Build returns a processor consisting of the previously added processors.
+func (b *chainBuilder) Build(call goja.FunctionCall) goja.Value {
+	if len(b.procs) == 0 {
+		b.runtime.NewGoError(errors.New("no processors have been added to the chain"))
+	}
+
+	p := &beatProcessor{b.runtime, &b.chain}
+	o := b.runtime.NewObject()
+	o.Set("Run", p.Run)
+	return o
+}
+
+type gojaCall interface {
+	Argument(idx int) goja.Value
+}
+
+type jsFunction func(call goja.FunctionCall) goja.Value
+
+type processor interface {
+	run(event javascript.Event) error
+}
+
+// jsProcessor is a javascript function that accepts the event as a parameter.
+type jsProcessor struct {
+	fn   jsFunction
+	call goja.FunctionCall
+}
+
+func (p *jsProcessor) run(event javascript.Event) error {
+	p.call.Arguments = p.call.Arguments[0:]
+	p.call.Arguments = append(p.call.Arguments, event.JSObject())
+	p.fn(p.call)
+	return nil
+}
+
+// nativeProcessor is a normal Beat processor.
+type nativeProcessor struct {
+	processors.Processor
+}
+
+func newNativeProcessor(constructor processors.Constructor, call gojaCall) (processor, error) {
+	var config *common.Config
+
+	if a0 := call.Argument(0); !goja.IsUndefined(a0) {
+		var err error
+		config, err = common.NewConfigFrom(a0.Export())
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		// No config so use an empty config.
+		config = common.NewConfig()
+	}
+
+	p, err := constructor(config)
+	if err != nil {
+		return nil, err
+	}
+	return &nativeProcessor{p}, nil
+}
+
+func (p *nativeProcessor) run(event javascript.Event) error {
+	out, err := p.Processor.Run(event.Wrapped())
+	if err != nil {
+		return err
+	}
+	if out == nil {
+		event.Cancel()
+	}
+	return nil
+}
+
+// chain is a list of processors to run serially to process an event.
+type chain struct {
+	procs []processor
+}
+
+func (c *chain) run(event javascript.Event) error {
+	for _, p := range c.procs {
+		if event.IsCancelled() {
+			return nil
+		}
+
+		if err := p.run(event); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}