exec

Documentation

Index

• func ExecFlow(ctx *engine.Context) engine.Flow
• func ExecInlet(ctx *engine.Context) engine.Inlet
• type ExecDriver
  • func New(ctx *engine.Context) *ExecDriver
  • func (ed *ExecDriver) Close() error
  • func (ed *ExecDriver) Interval() time.Duration
  • func (ed *ExecDriver) Open() error
  • func (ed *ExecDriver) Parallelism() int
  • func (ed *ExecDriver) Process(inputRecord engine.Record, next func([]engine.Record, error))

Examples

• ExecFlow
• ExecFlow (Sample)
• ExecInlet
• ExecInlet (Helloworld)

Functions

func ExecFlow

func ExecFlow(ctx *engine.Context) engine.Flow

Example

package main

import (
	"time"

	"github.com/OutOfBedlam/tine/engine"
	_ "github.com/OutOfBedlam/tine/plugins/base"
	_ "github.com/OutOfBedlam/tine/plugins/exec"
)

func main() {
	// This example demonstrates how to use the exec inlet to run a command and
	dsl := `
	[[inlets.file]]
		data = [
			"a,1",
			"b,2",
		]
		format = "csv"
	[[flows.exec]]
		commands = ["sh", "-c", "echo hello $FOO $FIELD_0 $FIELD_1 $TAG__IN $TAG__TS"]
		environments = ["FOO=BAR"]
		trim_space = true
		ignore_error = true
	[[flows.select]]
		includes= ["#_ts", "stdout"]
	[[outlets.file]]
		path = "-"
		format = "json"
	`
	// Make the output timestamp deterministic, so we can compare it
	// This line is required only for testing
	count := int64(0)
	engine.Now = func() time.Time { count++; return time.Unix(1721954797+count, 0) }
	// Build pipeline
	pipeline, err := engine.New(engine.WithConfig(dsl))
	if err != nil {
		panic(err)
	}
	if err := pipeline.Run(); err != nil {
		panic(err)
	}
}

Output:

{"_ts":1721954798,"stdout":"hello BAR a 1 file 2024-07-26T00:46:38Z"}
{"_ts":1721954799,"stdout":"hello BAR b 2 file 2024-07-26T00:46:39Z"}

Example (Sample)

package main

import (
	"time"

	"github.com/OutOfBedlam/tine/engine"
	_ "github.com/OutOfBedlam/tine/plugins/base"
	_ "github.com/OutOfBedlam/tine/plugins/exec"
)

func main() {
	// This example demonstrates how to use the exec inlet to run a command and
	dsl := `
	[[inlets.file]]
		data = [
			"a,1",
			"b,2",
		]
		format = "csv"
	[[flows.exec]]
		commands = ["sh", "-c", "echo hello $FOO $FIELD_0 $FIELD_1"]
		environments = ["FOO=BAR"]
		trim_space = true
		ignore_error = true
		stdout_field = "output"
	[[flows.select]]
		includes= ["#_ts", "*"]
	[[outlets.file]]
		path = "-"
		format = "json"
	`
	// Make the output timestamp deterministic, so we can compare it
	// This line is required only for testing
	count := int64(0)
	engine.Now = func() time.Time { count++; return time.Unix(1721954797+count, 0) }
	// Build pipeline
	pipeline, err := engine.New(engine.WithConfig(dsl))
	if err != nil {
		panic(err)
	}
	if err := pipeline.Run(); err != nil {
		panic(err)
	}
}

Output:

{"_ts":1721954798,"output":"hello BAR a 1"}
{"_ts":1721954799,"output":"hello BAR b 2"}

func ExecInlet

func ExecInlet(ctx *engine.Context) engine.Inlet

Example

package main

import (
	"time"

	"github.com/OutOfBedlam/tine/engine"
	_ "github.com/OutOfBedlam/tine/plugins/base"
	_ "github.com/OutOfBedlam/tine/plugins/exec"
)

func main() {
	// This example demonstrates how to use the exec inlet to run a command and
	dsl := `
	[[inlets.exec]]
		commands = ["sh", "-c", "echo hello world $FOO"]
		environments = ["FOO=BAR"]
		trim_space = true
		count = 1
		ignore_error = true
	[[flows.select]]
		includes= ["#_ts", "stdout"]
	[[outlets.file]]
		path = "-"
		format = "csv"
	`
	// Make the output timestamp deterministic, so we can compare it
	// This line is required only for testing
	count := int64(0)
	engine.Now = func() time.Time { count++; return time.Unix(1721954797+count, 0) }
	// Build pipeline
	pipeline, err := engine.New(engine.WithConfig(dsl))
	if err != nil {
		panic(err)
	}
	if err := pipeline.Run(); err != nil {
		panic(err)
	}
}

Output:

1721954798,hello world BAR

Example (Helloworld)

package main

import (
	"time"

	"github.com/OutOfBedlam/tine/engine"
	_ "github.com/OutOfBedlam/tine/plugins/base"
	_ "github.com/OutOfBedlam/tine/plugins/exec"
)

func main() {
	// This example demonstrates how to use the exec inlet to run a command and
	dsl := `
	[[inlets.exec]]
		commands = ["echo", "hello", "world"]
		trim_space = true
		count = 1
		ignore_error = true
	[[flows.select]]
		includes= ["#_ts", "stdout"]
	[[outlets.file]]
		path = "-"
		format = "csv"
	`
	// Make the output timestamp deterministic, so we can compare it
	// This line is required only for testing
	count := int64(0)
	engine.Now = func() time.Time { count++; return time.Unix(1721954797+count, 0) }
	// Build pipeline
	pipeline, err := engine.New(engine.WithConfig(dsl))
	if err != nil {
		panic(err)
	}
	if err := pipeline.Run(); err != nil {
		panic(err)
	}
}

Output:

1721954798,hello world

Types

type ExecDriver

type ExecDriver struct {
	// contains filtered or unexported fields
}

func New

func New(ctx *engine.Context) *ExecDriver

func (*ExecDriver) Close

func (ed *ExecDriver) Close() error

func (*ExecDriver) Interval

func (ed *ExecDriver) Interval() time.Duration

func (*ExecDriver) Open

func (ed *ExecDriver) Open() error

func (*ExecDriver) Parallelism

func (ed *ExecDriver) Parallelism() int

func (*ExecDriver) Process

func (ed *ExecDriver) Process(inputRecord engine.Record, next func([]engine.Record, error))