Documentation
¶
Index ¶
- func DamperFlow(ctx *engine.Context) engine.Flow
- func DumpFlow(ctx *engine.Context) engine.Flow
- func FlattenFlow(ctx *engine.Context) engine.Flow
- func MergeFlow(ctx *engine.Context) engine.Flow
- func SelectFlow(ctx *engine.Context) engine.Flow
- func SetFieldFlow(ctx *engine.Context) engine.Flow
- func SetFieldNameFlow(ctx *engine.Context) engine.Flow
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func MergeFlow ¶
Example ¶
package main
import (
"time"
"github.com/OutOfBedlam/tine/engine"
_ "github.com/OutOfBedlam/tine/plugin/codec/csv"
_ "github.com/OutOfBedlam/tine/plugin/codec/json"
_ "github.com/OutOfBedlam/tine/plugin/flows/base"
_ "github.com/OutOfBedlam/tine/plugin/inlets/exec"
_ "github.com/OutOfBedlam/tine/plugin/inlets/file"
_ "github.com/OutOfBedlam/tine/plugin/outlets/file"
)
func main() {
// This example demonstrates how to use the merge flow.
dsl := `
[log]
level = "warn"
[[inlets.file]]
data = [
"a,1",
]
format = "csv"
[[inlets.exec]]
commands = ["echo", "hello world"]
trim_space = true
[[flows.merge]]
wait_limit = "1s"
[[outlets.file]]
path = "-"
format = "json"
`
// Make the output time deterministic. so we can compare it.
// This line is not needed in production code.
engine.Now = func() time.Time { return time.Unix(1721954797, 0) }
// Create a new engine.
pipeline, err := engine.New(engine.WithConfig(dsl))
if err != nil {
panic(err)
}
// Run the engine.
if err := pipeline.Run(); err != nil {
panic(err)
}
}
Output: [{"_ts":1721954797,"exec.stdout":"hello world","file.0":"a","file.1":"1"}]
func SelectFlow ¶ added in v0.0.2
Example ¶
package main
import (
"time"
"github.com/OutOfBedlam/tine/engine"
_ "github.com/OutOfBedlam/tine/plugin/codec/csv"
_ "github.com/OutOfBedlam/tine/plugin/codec/json"
_ "github.com/OutOfBedlam/tine/plugin/flows/base"
_ "github.com/OutOfBedlam/tine/plugin/inlets/file"
_ "github.com/OutOfBedlam/tine/plugin/outlets/file"
)
func main() {
dsl := `
[log]
level = "warn"
[[inlets.file]]
data = [
"a,1",
"b,2",
"c,3",
]
format = "csv"
[[flows.select]]
includes = ["**", "not-exist", "#_ts", "1"]
[[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) }
// Create a new pipeline
pipeline, err := engine.New(engine.WithConfig(dsl))
if err != nil {
panic(err)
}
// Run the pipeline
if err := pipeline.Run(); err != nil {
panic(err)
}
}
Output: file,1721954798,a,1,,1721954798,1 file,1721954799,b,2,,1721954799,2 file,1721954800,c,3,,1721954800,3
func SetFieldFlow ¶
Example ¶
package main
import (
"time"
"github.com/OutOfBedlam/tine/engine"
_ "github.com/OutOfBedlam/tine/plugin/codec/csv"
_ "github.com/OutOfBedlam/tine/plugin/codec/json"
_ "github.com/OutOfBedlam/tine/plugin/flows/base"
_ "github.com/OutOfBedlam/tine/plugin/inlets/file"
_ "github.com/OutOfBedlam/tine/plugin/outlets/file"
)
func main() {
dsl := `
[log]
level = "warn"
[[inlets.file]]
data = [
"a,1",
"b,2",
"c,3",
]
format = "csv"
[[outlets.file]]
path = "-"
format = "json"
`
// Make the output timestamp deterministic, so we can compare it
// This line is required only for testing
engine.Now = func() time.Time { return time.Unix(1721954797, 0) }
// Create a new pipeline
pipeline, err := engine.New(engine.WithConfig(dsl))
if err != nil {
panic(err)
}
// Run the pipeline
if err := pipeline.Run(); err != nil {
panic(err)
}
}
Output: [{"0":"a","1":"1","_in":"file","_ts":1721954797},{"0":"b","1":"2","_in":"file","_ts":1721954797},{"0":"c","1":"3","_in":"file","_ts":1721954797}]
func SetFieldNameFlow ¶
Example ¶
package main
import (
"os"
"time"
"github.com/OutOfBedlam/tine/engine"
_ "github.com/OutOfBedlam/tine/plugin/codec/csv"
_ "github.com/OutOfBedlam/tine/plugin/flows/base"
_ "github.com/OutOfBedlam/tine/plugin/inlets/args"
_ "github.com/OutOfBedlam/tine/plugin/outlets/file"
)
func main() {
// This example demonstrates how to use the exec inlet to run a command and
dsl := `
[log]
level = "warn"
[[inlets.args]]
[[flows.set_field_name]]
prefix = "pre_"
suffix = "_suf"
[[flows.flatten]]
[[outlets.file]]
path = "-"
format = "csv"
`
// Make the output timestamp deterministic, so we can compare it
// This line is required only for testing
engine.Now = func() time.Time { return time.Unix(1721954797, 0) }
// Build pipeline
pipeline, err := engine.New(engine.WithConfig(dsl))
if err != nil {
panic(err)
}
// Simulate the command line arguments
os.Args = []string{"command", "command-arg", "--", "msg=hello world"}
// Run the pipeline
if err := pipeline.Run(); err != nil {
panic(err)
}
}
Output: pre_msg_suf,hello world
Types ¶
This section is empty.
Click to show internal directories.
Click to hide internal directories.