Package / arch (push) Canceled after 0s
Package / user (push) Canceled after 0s
Package / deb (push) Canceled after 0s
Package / freebsd (push) Canceled after 0s
Package / image (push) Canceled after 0s
Test / test (push) Canceled after 0s
Test / e2e (push) Canceled after 0s
lwconf: sparse stages, every problem of a part, pastes read by form
943 lines
29 KiB
Go
943 lines
29 KiB
Go
// Package compose is the configuration engine lw --tui and the website share:
|
|
// pipelines started empty, from a preset, a configuration or a pasted form,
|
|
// edited, validated, and written as a command line, environment or file that
|
|
// reads back to the same pipelines; Schema is lw --schema. It reads no file and
|
|
// links no TLS, HTTP or terminal code, so it builds as WebAssembly.
|
|
package compose
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"regexp"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"unicode"
|
|
"unicode/utf8"
|
|
|
|
"github.com/lixenwraith/logwisp/internal/config"
|
|
"github.com/lixenwraith/logwisp/internal/version"
|
|
|
|
"github.com/lixenwraith/toml"
|
|
)
|
|
|
|
// Composition is the pipelines being composed. Its flow stages and plugin
|
|
// options hold the keys set, the options in the kinds they declare, so each
|
|
// form it writes loads back to the same pipelines. LeftOut is what a pasted
|
|
// form held besides pipelines: a file's settings, other LOGWISP_ variables.
|
|
type Composition struct {
|
|
Pipelines []config.PipelineConfig
|
|
LeftOut []string
|
|
}
|
|
|
|
// Node is a part of a pipeline: a source or sink by its id, or a flow stage
|
|
// by its key in the flow table: filters with an index, format, rate_limit or
|
|
// heartbeat
|
|
type Node struct {
|
|
Role string `json:"role"`
|
|
ID string `json:"id"`
|
|
Index int `json:"index"`
|
|
}
|
|
|
|
// FromConfig starts from a loaded configuration's pipelines, its console sinks
|
|
// leaving the top-level color inherited
|
|
func FromConfig(cfg *config.Config) (*Composition, error) {
|
|
c := &Composition{}
|
|
for _, p := range cfg.Pipelines {
|
|
c.Pipelines = append(c.Pipelines, copyPipeline(p))
|
|
}
|
|
config.Uninherit(c.Pipelines, cfg.Color)
|
|
return c, c.settle()
|
|
}
|
|
|
|
// FromPreset starts from a preset's pipeline; isDir answers whether its path
|
|
// is a directory, nil off the target host (see config.IsDir)
|
|
func FromPreset(name string, values map[string]string, isDir config.IsDir) (*Composition, error) {
|
|
p, err := config.ExpandPreset(name, values, isDir)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
c := &Composition{Pipelines: []config.PipelineConfig{p}}
|
|
return c, c.settle()
|
|
}
|
|
|
|
// FromCommandLine starts from a pasted lw command line, quoted as a POSIX
|
|
// shell reads it; it takes pipeline flags only, at least one, as a composition
|
|
// holds no settings. Of a line piped into lw it reads lw's part, whose
|
|
// pipelines read standard input.
|
|
func FromCommandLine(line string, isDir config.IsDir) (*Composition, error) {
|
|
words, err := split(strings.NewReplacer("\r\n", "\n", "\r", "\n").Replace(line)) // pasted line ends
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
lw := func(word string) bool { return word == "lw" || strings.HasSuffix(word, "/lw") }
|
|
for i := len(words) - 2; i >= 0; i-- {
|
|
if words[i] == "|" && lw(words[i+1]) {
|
|
words = words[i+1:]
|
|
break
|
|
}
|
|
}
|
|
if len(words) > 0 && lw(words[0]) {
|
|
words = words[1:]
|
|
}
|
|
var specs []config.Spec
|
|
for i := 0; i < len(words); i++ {
|
|
name, value, inline := strings.Cut(words[i], "=")
|
|
if name == "-p" {
|
|
name = "--preset"
|
|
}
|
|
flag, long := strings.CutPrefix(name, "--")
|
|
if flags := config.SpecFlags(); !long || !slices.ContainsFunc(flags, func(f config.SpecFlag) bool { return f.Flag == flag }) {
|
|
var names []string
|
|
for _, f := range flags {
|
|
names = append(names, "--"+f.Flag)
|
|
}
|
|
return nil, fmt.Errorf("%s: not a pipeline flag; paste lw's %s only", words[i], strings.Join(names, ", "))
|
|
}
|
|
if !inline {
|
|
if i++; i == len(words) || strings.HasPrefix(words[i], "-") {
|
|
return nil, fmt.Errorf("%s requires a value; one that starts with '-' follows '='", name)
|
|
}
|
|
value = words[i]
|
|
}
|
|
specs = append(specs, config.Spec{Flag: flag, Value: value})
|
|
}
|
|
if len(specs) == 0 {
|
|
return nil, errors.New("no pipeline flags")
|
|
}
|
|
pipelines, err := config.SpecPipelines(specs, isDir)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
c := &Composition{Pipelines: pipelines}
|
|
return c, c.settle()
|
|
}
|
|
|
|
// settle gives every stage the values its spec writes out (FillOn), every
|
|
// plugin's options their kinds, unsetting a zero that takes its key's
|
|
// default, and every source and sink an id; the rules are Validate's. Its
|
|
// error joins a plugin's each holding a value of the wrong kind, kept as given.
|
|
func (c *Composition) settle() error {
|
|
var errs []error
|
|
for i := range c.Pipelines {
|
|
p := &c.Pipelines[i]
|
|
if p.Flow == nil {
|
|
p.Flow = &config.FlowConfig{}
|
|
}
|
|
name(len(p.PluginSources), func(j int) *string { return &p.PluginSources[j].ID },
|
|
func(j int) string { return p.PluginSources[j].Type })
|
|
name(len(p.PluginSinks), func(j int) *string { return &p.PluginSinks[j].ID },
|
|
func(j int) string { return p.PluginSinks[j].Type })
|
|
for key, stage := range stages(p.Flow) {
|
|
if !stage.IsNil() {
|
|
config.FillOn(key, stage.Interface())
|
|
}
|
|
}
|
|
for j, s := range p.PluginSources {
|
|
p.PluginSources[j].Config = orEmpty(s.Config)
|
|
if err := coerce("source", s.Type, p.PluginSources[j].Config); err != nil {
|
|
errs = append(errs, config.At(Node{Role: "source", ID: s.ID}.Path(i), err))
|
|
}
|
|
}
|
|
for j, s := range p.PluginSinks {
|
|
p.PluginSinks[j].Config = orEmpty(s.Config)
|
|
if err := coerce("sink", s.Type, p.PluginSinks[j].Config); err != nil {
|
|
errs = append(errs, config.At(Node{Role: "sink", ID: s.ID}.Path(i), err))
|
|
}
|
|
}
|
|
}
|
|
return errors.Join(errs...)
|
|
}
|
|
|
|
// coerce gives a plugin's options their kinds, then unsets a number at zero
|
|
// where zero takes the key's default: lw reads it so, and the options then
|
|
// show the default rather than a 0 that means it
|
|
func coerce(role, typ string, m map[string]any) error {
|
|
err := config.Coerce(role, typ, m)
|
|
if p, ok := config.LookupPlugin(role, typ); ok {
|
|
unsetDefaults(p.Keys(), m)
|
|
}
|
|
return err
|
|
}
|
|
|
|
func unsetDefaults(keys []config.Key, m map[string]any) {
|
|
for _, k := range keys {
|
|
switch v := m[k.Name]; v {
|
|
case int64(0), 0.0:
|
|
if k.Default != "" && k.Zero == "" {
|
|
delete(m, k.Name)
|
|
}
|
|
default:
|
|
if sub, ok := v.(map[string]any); ok {
|
|
unsetDefaults(k.Keys, sub)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// name gives the parts a file left unnamed the ids their specs would get: the
|
|
// shell forms write no empty id
|
|
func name(n int, id func(int) *string, typ func(int) string) {
|
|
var taken []string
|
|
for j := range n {
|
|
if *id(j) != "" {
|
|
taken = append(taken, *id(j))
|
|
}
|
|
}
|
|
for j := range n {
|
|
if *id(j) == "" {
|
|
*id(j) = config.FreeID(typ(j), taken)
|
|
taken = append(taken, *id(j))
|
|
}
|
|
}
|
|
}
|
|
|
|
// stages are the flow table's fields that hold one stage each, by key: a
|
|
// pointer, nil while the stage is off. The filters are a list.
|
|
func stages(f *config.FlowConfig) map[string]reflect.Value {
|
|
v, out := reflect.ValueOf(f).Elem(), map[string]reflect.Value{}
|
|
for i := range v.NumField() {
|
|
if v.Field(i).Kind() == reflect.Pointer {
|
|
key, _, _ := strings.Cut(v.Type().Field(i).Tag.Get("toml"), ",")
|
|
out[key] = v.Field(i)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func orEmpty(m map[string]any) map[string]any {
|
|
if m == nil {
|
|
return map[string]any{}
|
|
}
|
|
return m
|
|
}
|
|
|
|
func (c *Composition) pipeline(i int) (*config.PipelineConfig, error) {
|
|
if i < 0 || i >= len(c.Pipelines) {
|
|
return nil, fmt.Errorf("no pipeline %d", i)
|
|
}
|
|
return &c.Pipelines[i], nil
|
|
}
|
|
|
|
// AddPipeline appends an empty pipeline under a name no other has
|
|
func (c *Composition) AddPipeline(name string) error {
|
|
if err := c.free(name, -1); err != nil {
|
|
return err
|
|
}
|
|
c.Pipelines = append(c.Pipelines, config.PipelineConfig{Name: name, Flow: &config.FlowConfig{}})
|
|
return nil
|
|
}
|
|
|
|
// RenamePipeline gives pipeline i a name no other has
|
|
func (c *Composition) RenamePipeline(i int, name string) error {
|
|
p, err := c.pipeline(i)
|
|
if err == nil {
|
|
err = c.free(name, i)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
p.Name = name
|
|
return nil
|
|
}
|
|
|
|
// free refuses a name that is empty or held by a pipeline other than i
|
|
func (c *Composition) free(name string, i int) error {
|
|
switch j := slices.IndexFunc(c.Pipelines, func(p config.PipelineConfig) bool { return p.Name == name }); {
|
|
case name == "":
|
|
return errors.New("a pipeline needs a name; give it one")
|
|
case j >= 0 && j != i:
|
|
return fmt.Errorf("another pipeline is named %q; choose another name", name)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *Composition) RemovePipeline(i int) error {
|
|
if _, err := c.pipeline(i); err != nil {
|
|
return err
|
|
}
|
|
c.Pipelines = slices.Delete(c.Pipelines, i, i+1)
|
|
return nil
|
|
}
|
|
|
|
// Add appends a source or sink of a catalogue type under config.FreeID, or a
|
|
// filter of type include or exclude, or of the default for "", and returns
|
|
// its node
|
|
func (c *Composition) Add(pi int, role, typ string) (Node, error) {
|
|
p, err := c.pipeline(pi)
|
|
if err != nil {
|
|
return Node{}, err
|
|
}
|
|
var ids []string
|
|
switch role {
|
|
case "source":
|
|
for _, s := range p.PluginSources {
|
|
ids = append(ids, s.ID)
|
|
}
|
|
case "sink":
|
|
for _, s := range p.PluginSinks {
|
|
ids = append(ids, s.ID)
|
|
}
|
|
case "filters":
|
|
p.Flow.Filters = append(p.Flow.Filters, config.FilterConfig{Type: config.FilterType(typ)})
|
|
return Node{Role: role, Index: len(p.Flow.Filters) - 1}, nil
|
|
default:
|
|
return Node{}, fmt.Errorf("cannot add to %s", role)
|
|
}
|
|
if _, ok := config.LookupPlugin(role, typ); !ok {
|
|
return Node{}, fmt.Errorf("unknown %s type %q", role, typ)
|
|
}
|
|
n := Node{Role: role, ID: config.FreeID(typ, ids)}
|
|
if role == "source" {
|
|
p.PluginSources = append(p.PluginSources, config.PluginSourceConfig{ID: n.ID, Type: typ, Config: map[string]any{}})
|
|
} else {
|
|
p.PluginSinks = append(p.PluginSinks, config.PluginSinkConfig{ID: n.ID, Type: typ, Config: map[string]any{}})
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// Retype gives a source or sink another type of its role, in its place: a
|
|
// chosen id stays and an automatic one follows the type. Options of the same
|
|
// name and kind carry over to a type on the same network side, where they
|
|
// mean the same; across sides host, port, tls and auth do not.
|
|
func (c *Composition) Retype(pi int, n Node, typ string) (Node, error) {
|
|
p, err := c.pipeline(pi)
|
|
if err != nil {
|
|
return Node{}, err
|
|
}
|
|
to, ok := config.LookupPlugin(n.Role, typ)
|
|
if !ok {
|
|
return Node{}, fmt.Errorf("unknown %s type %q", n.Role, typ)
|
|
}
|
|
var id, old *string
|
|
var opts *map[string]any
|
|
var ids []string
|
|
switch i := index(p, n); {
|
|
case i >= 0 && n.Role == "source":
|
|
id, old, opts = &p.PluginSources[i].ID, &p.PluginSources[i].Type, &p.PluginSources[i].Config
|
|
for _, s := range p.PluginSources {
|
|
ids = append(ids, s.ID)
|
|
}
|
|
case i >= 0 && n.Role == "sink":
|
|
id, old, opts = &p.PluginSinks[i].ID, &p.PluginSinks[i].Type, &p.PluginSinks[i].Config
|
|
for _, s := range p.PluginSinks {
|
|
ids = append(ids, s.ID)
|
|
}
|
|
default:
|
|
return Node{}, n.missing()
|
|
}
|
|
from, _ := config.LookupPlugin(n.Role, *old)
|
|
kept := map[string]any{}
|
|
for _, k := range from.Keys() {
|
|
same := func(t config.Key) bool { return t.Name == k.Name && t.Kind == k.Kind }
|
|
if v, set := (*opts)[k.Name]; set && from.Side == to.Side && slices.ContainsFunc(to.Keys(), same) {
|
|
kept[k.Name] = v
|
|
}
|
|
}
|
|
suffix, auto := strings.CutPrefix(*id, *old)
|
|
if n, err := strconv.Atoi(strings.TrimPrefix(suffix, "_")); auto && (suffix == "" || suffix[0] == '_' && err == nil && n > 1) {
|
|
*id = config.FreeID(typ, slices.DeleteFunc(ids, func(s string) bool { return s == *id }))
|
|
}
|
|
*old, *opts = typ, kept
|
|
return Node{Role: n.Role, ID: *id}, nil
|
|
}
|
|
|
|
// Remove deletes a source, sink or filter, or turns a flow stage off
|
|
func (c *Composition) Remove(pi int, n Node) error {
|
|
p, err := c.pipeline(pi)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch i := index(p, n); {
|
|
case i < 0:
|
|
case n.Role == "source":
|
|
p.PluginSources = slices.Delete(p.PluginSources, i, i+1)
|
|
return nil
|
|
case n.Role == "sink":
|
|
p.PluginSinks = slices.Delete(p.PluginSinks, i, i+1)
|
|
return nil
|
|
case n.Role == "filters":
|
|
p.Flow.Filters = slices.Delete(p.Flow.Filters, i, i+1)
|
|
return nil
|
|
}
|
|
if stage, ok := stages(p.Flow)[n.Role]; ok {
|
|
stage.SetZero()
|
|
return nil
|
|
}
|
|
return n.missing()
|
|
}
|
|
|
|
// index finds a node in its list, -1 when it is in none
|
|
func index(p *config.PipelineConfig, n Node) int {
|
|
switch n.Role {
|
|
case "source":
|
|
return slices.IndexFunc(p.PluginSources, func(s config.PluginSourceConfig) bool { return s.ID == n.ID })
|
|
case "sink":
|
|
return slices.IndexFunc(p.PluginSinks, func(s config.PluginSinkConfig) bool { return s.ID == n.ID })
|
|
case "filters":
|
|
if n.Index >= 0 && n.Index < len(p.Flow.Filters) {
|
|
return n.Index
|
|
}
|
|
}
|
|
return -1
|
|
}
|
|
|
|
// Path is the path of node n's options in pipeline pi, as Locate reads it
|
|
func (n Node) Path(pi int) string {
|
|
switch n.Role {
|
|
case "source", "sink":
|
|
return fmt.Sprintf("pipelines[%d].plugin_%ss[%s].config", pi, n.Role, n.ID)
|
|
case "filters":
|
|
return fmt.Sprintf("pipelines[%d].flow.filters[%d]", pi, n.Index)
|
|
}
|
|
return fmt.Sprintf("pipelines[%d].flow.%s", pi, n.Role)
|
|
}
|
|
|
|
func (n Node) missing() error {
|
|
if n.Role == "filters" {
|
|
return fmt.Errorf("no filter %d", n.Index)
|
|
}
|
|
return fmt.Errorf("no %s %q", n.Role, n.ID)
|
|
}
|
|
|
|
// MoveFilter moves a filter to index to; entries pass filters in order
|
|
func (c *Composition) MoveFilter(pi, from, to int) error {
|
|
p, err := c.pipeline(pi)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n := len(p.Flow.Filters); from < 0 || from >= n || to < 0 || to >= n {
|
|
return fmt.Errorf("no filter %d or %d", from, to)
|
|
}
|
|
f := p.Flow.Filters[from]
|
|
p.Flow.Filters = slices.Insert(slices.Delete(p.Flow.Filters, from, from+1), to, f)
|
|
return nil
|
|
}
|
|
|
|
// Set gives a key at a dotted path its values: one, or several for a list. A
|
|
// flow stage that is off turns on. A value of the wrong kind leaves the key as
|
|
// it was, and a zero that takes the key's default unsets it; any other rule
|
|
// is Validate's.
|
|
func (c *Composition) Set(pi int, n Node, key string, values ...string) error {
|
|
return c.edit(pi, n, func(m map[string]any) error {
|
|
unset(m, strings.Split(key, "."))
|
|
for _, v := range values {
|
|
if err := config.SetOption(m, key, v); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// Unset returns a key to its default
|
|
func (c *Composition) Unset(pi int, n Node, key string) error {
|
|
return c.edit(pi, n, func(m map[string]any) error {
|
|
unset(m, strings.Split(key, "."))
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// unset deletes a dotted key, and the tables it leaves empty
|
|
func unset(m map[string]any, path []string) {
|
|
if sub, ok := m[path[0]].(map[string]any); ok && len(path) > 1 {
|
|
if unset(sub, path[1:]); len(sub) > 0 {
|
|
return
|
|
}
|
|
}
|
|
delete(m, path[0])
|
|
}
|
|
|
|
// edit changes a copy of a node's options, which replaces them when they
|
|
// decode to the kinds they declare
|
|
func (c *Composition) edit(pi int, n Node, change func(map[string]any) error) error {
|
|
p, err := c.pipeline(pi)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
plugin := func(typ string, m *map[string]any) error {
|
|
next := config.Clone(*m)
|
|
if err := change(next); err != nil {
|
|
return err
|
|
}
|
|
if err := coerce(n.Role, typ, next); err != nil {
|
|
return err
|
|
}
|
|
*m = next
|
|
return nil
|
|
}
|
|
switch i := index(p, n); {
|
|
case i < 0:
|
|
case n.Role == "source":
|
|
return plugin(p.PluginSources[i].Type, &p.PluginSources[i].Config)
|
|
case n.Role == "sink":
|
|
return plugin(p.PluginSinks[i].Type, &p.PluginSinks[i].Config)
|
|
case n.Role == "filters":
|
|
filter := &p.Flow.Filters[i]
|
|
if err := editStage(reflect.ValueOf(&filter).Elem(), "", change); err != nil {
|
|
return err
|
|
}
|
|
p.Flow.Filters[i] = *filter
|
|
return nil
|
|
}
|
|
if stage, ok := stages(p.Flow)[n.Role]; ok {
|
|
return editStage(stage, n.Role, change)
|
|
}
|
|
return n.missing()
|
|
}
|
|
|
|
// editStage edits a stage's keys as a spec sets them; the table they scan into
|
|
// replaces the stage, which turns on as a spec turns it on and holds the keys
|
|
// set, a key at its zero unset
|
|
func editStage(stage reflect.Value, key string, change func(map[string]any) error) error {
|
|
m := config.StageOn(key, nil)
|
|
if !stage.IsNil() {
|
|
m = config.Values(stage.Interface())
|
|
}
|
|
if err := change(m); err != nil {
|
|
return err
|
|
}
|
|
next := reflect.New(stage.Type().Elem())
|
|
if err := config.Scan(m, next.Interface()); err != nil {
|
|
return err
|
|
}
|
|
config.FillOn(key, next.Interface())
|
|
stage.Set(next)
|
|
return nil
|
|
}
|
|
|
|
// Options returns a copy of a node's options that are set, and whether it is
|
|
// on: a flow stage that is off has none
|
|
func (c *Composition) Options(pi int, n Node) (map[string]any, bool, error) {
|
|
p, err := c.pipeline(pi)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
switch i := index(p, n); {
|
|
case i < 0:
|
|
case n.Role == "source":
|
|
return config.Clone(p.PluginSources[i].Config), true, nil
|
|
case n.Role == "sink":
|
|
return config.Clone(p.PluginSinks[i].Config), true, nil
|
|
case n.Role == "filters":
|
|
return config.Values(&p.Flow.Filters[i]), true, nil
|
|
}
|
|
if stage, ok := stages(p.Flow)[n.Role]; ok {
|
|
if stage.IsNil() {
|
|
return nil, false, nil
|
|
}
|
|
return config.Values(stage.Interface()), true, nil
|
|
}
|
|
return nil, false, n.missing()
|
|
}
|
|
|
|
// Validate applies every rule lw applies at load: its error joins every
|
|
// problem, one a part, which Problems splits and Locate places
|
|
func (c *Composition) Validate() error { return config.ValidatePipelines(c.Pipelines) }
|
|
|
|
// Problems are the problems an error joins, in order
|
|
func Problems(err error) []error {
|
|
if joined, ok := err.(interface{ Unwrap() []error }); ok {
|
|
var out []error
|
|
for _, e := range joined.Unwrap() {
|
|
out = append(out, Problems(e)...)
|
|
}
|
|
return out
|
|
}
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
return []error{err}
|
|
}
|
|
|
|
// Locate reads where a problem lies from its path: the pipeline, the node,
|
|
// and the key in the node's options, a list's without its index; text is
|
|
// what is wrong. ok is false for a problem no pipeline holds.
|
|
func Locate(err error) (pi int, n Node, key, text string, ok bool) {
|
|
ke, _ := errors.AsType[*config.KeyError](err)
|
|
rest, found := "", false
|
|
if ke != nil {
|
|
rest, found = strings.CutPrefix(ke.Key, "pipelines[")
|
|
}
|
|
end := strings.IndexByte(rest, ']')
|
|
if !found || end < 0 {
|
|
return 0, n, "", err.Error(), false
|
|
}
|
|
if pi, err = strconv.Atoi(rest[:end]); err != nil {
|
|
return 0, n, "", ke.Error(), false
|
|
}
|
|
rest = strings.TrimPrefix(rest[end+1:], ".")
|
|
for role, list := range map[string]string{"source": "plugin_sources[", "sink": "plugin_sinks["} {
|
|
if after, ok := strings.CutPrefix(rest, list); ok {
|
|
id, more := cutID(after)
|
|
n.Role, n.ID, rest = role, id, strings.TrimPrefix(strings.TrimPrefix(more, "config"), ".")
|
|
}
|
|
}
|
|
if stage, ok := strings.CutPrefix(rest, "flow."); ok {
|
|
name, more, _ := strings.Cut(stage, ".")
|
|
name, index, _ := strings.Cut(name, "[")
|
|
n.Role, rest = name, more
|
|
n.Index, _ = strconv.Atoi(strings.TrimSuffix(index, "]"))
|
|
}
|
|
key, _, _ = strings.Cut(rest, "[")
|
|
return pi, n, key, ke.Err.Error(), true
|
|
}
|
|
|
|
// cutID cuts a plugin path at its id's ']', the first that ends the path or
|
|
// starts its type or config: ids are free text, and keys hold list indexes
|
|
func cutID(path string) (id, rest string) {
|
|
for i := range len(path) {
|
|
if after := path[i+1:]; path[i] == ']' && (after == "" || after == ".type" || after == ".config" || strings.HasPrefix(after, ".config.")) {
|
|
return path[:i], strings.TrimPrefix(after, ".")
|
|
}
|
|
}
|
|
return path, ""
|
|
}
|
|
|
|
// CommandLine writes the composition as an lw command line, a flag a line,
|
|
// quoted for a POSIX shell
|
|
func (c *Composition) CommandLine() (string, error) {
|
|
specs, err := c.shellSpecs()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
lines := []string{"lw"}
|
|
for _, s := range specs {
|
|
if strings.HasPrefix(s.Value, "-") {
|
|
lines = append(lines, quote("--"+s.Flag+"="+s.Value))
|
|
} else {
|
|
lines = append(lines, "--"+s.Flag+" "+quote(s.Value))
|
|
}
|
|
}
|
|
return strings.Join(lines, " \\\n ") + "\n", nil
|
|
}
|
|
|
|
// Environment writes the composition as LOGWISP_ variables, KEY='value' a
|
|
// line as an environment file holds them; the environment holds one pipeline
|
|
func (c *Composition) Environment() (string, error) {
|
|
if len(c.Pipelines) != 1 {
|
|
return "", fmt.Errorf("the environment holds one pipeline, not %d; choose the command or file form, or remove pipelines down to one", len(c.Pipelines))
|
|
}
|
|
specs, err := c.shellSpecs()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
flags, count := config.SpecFlags(), map[string]int{}
|
|
var b strings.Builder
|
|
for _, s := range specs {
|
|
f := flags[slices.IndexFunc(flags, func(f config.SpecFlag) bool { return f.Flag == s.Flag })]
|
|
name := f.Variable
|
|
if f.Many {
|
|
count[f.Flag]++
|
|
name += "_" + strconv.Itoa(count[f.Flag])
|
|
}
|
|
fmt.Fprintf(&b, "%s='%s'\n", name, strings.ReplaceAll(s.Value, "'", `'\''`))
|
|
}
|
|
return b.String(), nil
|
|
}
|
|
|
|
// shellSpecs are the specs of a valid composition, for a form pasted into a
|
|
// shell, which carries no control character: a terminal acts on ^U or ESC
|
|
// before the shell reads the line
|
|
func (c *Composition) shellSpecs() ([]config.Spec, error) {
|
|
if err := c.Validate(); err != nil {
|
|
return nil, err
|
|
}
|
|
specs, pipeline := config.PipelineSpecs(c.Pipelines), ""
|
|
for _, s := range specs {
|
|
if s.Flag == "pipeline" {
|
|
pipeline = s.Value
|
|
}
|
|
if strings.ContainsFunc(s.Value, func(r rune) bool { return unicode.IsControl(r) || r == utf8.RuneError }) {
|
|
return nil, fmt.Errorf("pipeline %q: a --%s value holds a control character, which only the file form carries", pipeline, s.Flag)
|
|
}
|
|
}
|
|
return specs, nil
|
|
}
|
|
|
|
// File writes the composition as a configuration file's pipelines, as lw
|
|
// --dump prints them
|
|
func (c *Composition) File() ([]byte, error) {
|
|
if err := c.Validate(); err != nil {
|
|
return nil, err
|
|
}
|
|
return toml.Marshal(map[string]any{"pipelines": c.Pipelines})
|
|
}
|
|
|
|
// Form is a way to write a composition, named for what it is, and to read it
|
|
// back to the same pipelines; looks matches text in its shape
|
|
type Form struct {
|
|
Name string `json:"name"`
|
|
Hint string `json:"hint"`
|
|
Write func(*Composition) (string, error) `json:"-"`
|
|
Read func(string, config.IsDir) (*Composition, error) `json:"-"`
|
|
looks *regexp.Regexp
|
|
}
|
|
|
|
var forms = []Form{
|
|
{"command", "an lw command line", (*Composition).CommandLine, FromCommandLine,
|
|
regexp.MustCompile(`^\s*(\S*/)?lw(\s|$)|(^|\s)(--[a-z]|-p\s)`)},
|
|
{"environment", "LOGWISP_ variables for one pipeline", (*Composition).Environment, FromEnvironment,
|
|
regexp.MustCompile(`(?m)^\s*(export\s+)?LOGWISP_[A-Z0-9_]*=`)},
|
|
{"file", "the pipelines of a configuration file", func(c *Composition) (string, error) {
|
|
data, err := c.File()
|
|
return string(data), err
|
|
}, func(text string, _ config.IsDir) (*Composition, error) { return FromFile([]byte(text)) },
|
|
regexp.MustCompile(`(?m)^\s*(\[\[?pipelines\b|pipelines\s*=)`)},
|
|
}
|
|
|
|
// Parse reads text pasted as the form. Refused, text in another form's shape
|
|
// and not in this one's is refused as that form, which reads it.
|
|
func (f Form) Parse(text string, isDir config.IsDir) (*Composition, error) {
|
|
c, err := f.Read(text, isDir)
|
|
if err == nil || f.looks.MatchString(text) {
|
|
return c, err
|
|
}
|
|
for _, g := range forms {
|
|
if g.looks.MatchString(text) {
|
|
return nil, fmt.Errorf("this is %s, not %s: read it as the %s form", g.Hint, f.Hint, g.Name)
|
|
}
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
// Forms are what a composition writes and reads: lw --tui prints one, the
|
|
// website shows them and reads them pasted
|
|
func Forms() []Form { return slices.Clone(forms) }
|
|
|
|
// FromEnvironment starts from NAME='value' lines as an environment file holds
|
|
// them, quoted as a POSIX shell reads them: '#' starts a comment line, a word
|
|
// without '=', such as export, assigns nothing, and a later assignment wins.
|
|
// LeftOut names the LOGWISP_ variables no pipeline flag has.
|
|
func FromEnvironment(text string, isDir config.IsDir) (*Composition, error) {
|
|
vars := map[string]string{}
|
|
for line := range strings.Lines(text) {
|
|
if line = strings.TrimSpace(line); line == "" || line[0] == '#' {
|
|
continue
|
|
}
|
|
words, err := split(line)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, w := range words {
|
|
if name, value, ok := strings.Cut(w, "="); ok {
|
|
vars[name] = value
|
|
}
|
|
}
|
|
}
|
|
var environ []string
|
|
for name, value := range vars {
|
|
environ = append(environ, name+"="+value)
|
|
}
|
|
pipelines, unread, err := config.EnvPipelines(environ, isDir)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
c := &Composition{Pipelines: pipelines, LeftOut: unread}
|
|
return c, c.settle()
|
|
}
|
|
|
|
// FromFile starts from a configuration file's pipelines, refusing a misspelled
|
|
// key as lw does; its settings are the host's, which LeftOut names, and its
|
|
// console sinks keep the color they name
|
|
func FromFile(data []byte) (*Composition, error) {
|
|
cfg, settings, err := config.ParseFile(data)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(cfg.Pipelines) == 0 {
|
|
return nil, errors.New("no [[pipelines]]")
|
|
}
|
|
c, err := FromConfig(&config.Config{Pipelines: cfg.Pipelines})
|
|
c.LeftOut = settings
|
|
return c, err
|
|
}
|
|
|
|
// FromJSON starts from pipelines as the website keeps them: a JSON list of
|
|
// tables in a file's shape, a misspelled key refused as a file's is, a null
|
|
// a value left unset. A value of the wrong kind returns the composition too,
|
|
// which Validate names among the rest.
|
|
func FromJSON(data []byte) (*Composition, error) {
|
|
var tables any
|
|
if err := json.Unmarshal(data, &tables); err != nil {
|
|
return nil, err
|
|
}
|
|
cfg := &config.Config{}
|
|
if err := config.Scan(map[string]any{"pipelines": dropNulls(tables)}, cfg); err != nil {
|
|
return nil, err
|
|
}
|
|
return FromConfig(cfg)
|
|
}
|
|
|
|
// dropNulls removes JSON's nulls from tables and lists, which the shell forms
|
|
// would write as text
|
|
func dropNulls(v any) any {
|
|
switch v := v.(type) {
|
|
case map[string]any:
|
|
for k, e := range v {
|
|
if e == nil {
|
|
delete(v, k)
|
|
} else {
|
|
v[k] = dropNulls(e)
|
|
}
|
|
}
|
|
case []any:
|
|
v = slices.DeleteFunc(v, func(e any) bool { return e == nil })
|
|
for i, e := range v {
|
|
v[i] = dropNulls(e)
|
|
}
|
|
return v
|
|
}
|
|
return v
|
|
}
|
|
|
|
// JSON writes the pipelines as FromJSON reads them, keyed as File writes them
|
|
func (c *Composition) JSON() ([]byte, error) {
|
|
data, err := toml.Marshal(map[string]any{"pipelines": c.Pipelines})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
root, err := toml.NewParser(data).Parse()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return json.Marshal(root["pipelines"])
|
|
}
|
|
|
|
// quote leaves a word of these characters bare, and single-quotes any other,
|
|
// or one starting with '=', which zsh expands
|
|
func quote(word string) string {
|
|
const bare = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_@%+=:,./-"
|
|
if word != "" && word[0] != '=' && strings.Trim(word, bare) == "" {
|
|
return word
|
|
}
|
|
return "'" + strings.ReplaceAll(word, "'", `'\''`) + "'"
|
|
}
|
|
|
|
// split reads words as a POSIX shell does, without expansions: quotes, '\'
|
|
// escapes, and '\' before a newline continuing the line
|
|
func split(line string) ([]string, error) {
|
|
var words []string
|
|
var word strings.Builder
|
|
in := false
|
|
for i := 0; i < len(line); i++ {
|
|
switch c := line[i]; {
|
|
case c == '\\' && i+1 < len(line):
|
|
if i++; line[i] != '\n' {
|
|
word.WriteByte(line[i])
|
|
in = true
|
|
}
|
|
case c == '\'':
|
|
end := strings.IndexByte(line[i+1:], '\'')
|
|
if end < 0 {
|
|
return nil, errors.New("unterminated '")
|
|
}
|
|
word.WriteString(line[i+1 : i+1+end])
|
|
i, in = i+1+end, true
|
|
case c == '"':
|
|
for i, in = i+1, true; ; i++ {
|
|
if i == len(line) {
|
|
return nil, errors.New(`unterminated "`)
|
|
}
|
|
if line[i] == '"' {
|
|
break
|
|
}
|
|
if line[i] == '\\' && i+1 < len(line) && strings.IndexByte("$`\"\\\n", line[i+1]) >= 0 {
|
|
if i++; line[i] == '\n' {
|
|
continue
|
|
}
|
|
}
|
|
word.WriteByte(line[i])
|
|
}
|
|
case c == ' ' || c == '\t' || c == '\n':
|
|
if in {
|
|
words, in = append(words, word.String()), false
|
|
word.Reset()
|
|
}
|
|
default:
|
|
word.WriteByte(c)
|
|
in = true
|
|
}
|
|
}
|
|
if in {
|
|
words = append(words, word.String())
|
|
}
|
|
return words, nil
|
|
}
|
|
|
|
func copyPipeline(p config.PipelineConfig) config.PipelineConfig {
|
|
f := config.FlowConfig{}
|
|
if p.Flow != nil {
|
|
f = *p.Flow
|
|
}
|
|
for _, stage := range stages(&f) {
|
|
if !stage.IsNil() {
|
|
next := reflect.New(stage.Type().Elem())
|
|
next.Elem().Set(stage.Elem())
|
|
stage.Set(next)
|
|
}
|
|
}
|
|
f.Filters = slices.Clone(f.Filters)
|
|
for i := range f.Filters {
|
|
f.Filters[i].Patterns = slices.Clone(f.Filters[i].Patterns)
|
|
}
|
|
p.Flow = &f
|
|
p.PluginSources = slices.Clone(p.PluginSources)
|
|
for i, s := range p.PluginSources {
|
|
p.PluginSources[i].Config = config.Clone(orEmpty(s.Config))
|
|
}
|
|
p.PluginSinks = slices.Clone(p.PluginSinks)
|
|
for i, s := range p.PluginSinks {
|
|
p.PluginSinks[i].Config = config.Clone(orEmpty(s.Config))
|
|
}
|
|
return p
|
|
}
|
|
|
|
// Schema is what lw --schema prints, and the website builds its catalogue
|
|
// from. A setting's flag is --KEY, a pipeline flag's --FLAG.
|
|
type Schema struct {
|
|
Version string `json:"version"`
|
|
Settings []Setting `json:"settings"`
|
|
Flags []config.SpecFlag `json:"pipeline_flags"`
|
|
Flow []config.Stage `json:"flow"`
|
|
Sources []Plugin `json:"sources"`
|
|
Sinks []Plugin `json:"sinks"`
|
|
Presets []config.Preset `json:"presets"`
|
|
Forms []Form `json:"forms"`
|
|
}
|
|
|
|
// Setting is a settings key with its environment variable
|
|
type Setting struct {
|
|
config.Key
|
|
Variable string `json:"variable"`
|
|
}
|
|
|
|
// Plugin is a catalogue row with its options
|
|
type Plugin struct {
|
|
config.Plugin
|
|
Keys []config.Key `json:"keys"`
|
|
}
|
|
|
|
func NewSchema() Schema {
|
|
s := Schema{
|
|
Version: version.Short(),
|
|
Flags: config.SpecFlags(),
|
|
Flow: config.FlowStages(),
|
|
Presets: config.Presets(),
|
|
Forms: Forms(),
|
|
}
|
|
for _, k := range config.SettingKeys() {
|
|
s.Settings = append(s.Settings, Setting{k, "LOGWISP_" + strings.ToUpper(strings.ReplaceAll(k.Name, ".", "_"))})
|
|
}
|
|
for _, p := range config.Plugins() {
|
|
if p.Role == "source" {
|
|
s.Sources = append(s.Sources, Plugin{p, p.Keys()})
|
|
} else {
|
|
s.Sinks = append(s.Sinks, Plugin{p, p.Keys()})
|
|
}
|
|
}
|
|
return s
|
|
}
|