v0.29.0 the configuration engine, and lw --schema
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
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
This commit is contained in:
1 parent
e8e5f05e47
commit
dfd8a68988
28 files changed
+1381
-240
No files matched your search
@@ -79,6 +79,10 @@ change touches.
|
||||
`config.Decode`, which applies the defaults, rules and pure checks and
|
||||
refuses unknown keys at any depth; they keep only I/O. Rules that read
|
||||
nothing belong in a `Check` method, so `ValidateConfig` sees them offline.
|
||||
- `internal/compose` is the engine `lw --tui` and the website's WebAssembly
|
||||
share: it edits pipelines and writes the command line, environment and file
|
||||
forms, each loading back to the same pipelines; it reads no file and links
|
||||
no TLS, HTTP or terminal code (its test runs `go list -deps`).
|
||||
- `cmd/lw/cli.go` owns the command line: lw's own flags, and the commands
|
||||
table whose groups (`auth`, `tls`, `preset`, `config`) share one flag
|
||||
mechanism; `config.Load` takes the parsed `config.Args`. Presets are rows of
|
||||
|
||||
+1
-8
@@ -135,14 +135,7 @@ func checkConfig(cfg *config.Config) int {
|
||||
// dumpConfig prints the effective configuration, flags and environment
|
||||
// resolved into it, as a file lw -c reads back.
|
||||
func dumpConfig(cfg *config.Config, w io.Writer) int {
|
||||
// Snapshot gave each console sink the top-level color: keep it inherited
|
||||
for _, p := range cfg.Pipelines {
|
||||
for _, s := range p.PluginSinks {
|
||||
if s.Type == "console" && s.Config["color"] == cfg.Color {
|
||||
delete(s.Config, "color")
|
||||
}
|
||||
}
|
||||
}
|
||||
config.Uninherit(cfg.Pipelines, cfg.Color)
|
||||
data, err := toml.Marshal(cfg)
|
||||
if err == nil {
|
||||
_, err = w.Write(data)
|
||||
|
||||
+7
-3
@@ -58,7 +58,7 @@ var shorts = map[string]struct {
|
||||
|
||||
// switches are lw's own options that take no value and are no setting, so no
|
||||
// file or environment variable turns one on
|
||||
var switches = []string{"check", "dump", "version"}
|
||||
var switches = []string{"check", "dump", "version", "schema"}
|
||||
|
||||
// usageError is a command-line mistake: exit status 2 rather than 1
|
||||
type usageError string
|
||||
@@ -241,7 +241,7 @@ func parseCommandLine(argv []string) (inv invocation, err error) {
|
||||
case arg == "--color":
|
||||
// Bare, it means always: config takes no bare flag for a string key
|
||||
inv.load.Overrides = append(inv.load.Overrides, "--color="+cmp.Or(next(&i), "always"))
|
||||
case strings.HasPrefix(name, "--") && slices.Contains(specFlags, name[2:]):
|
||||
case strings.HasPrefix(name, "--") && slices.ContainsFunc(specFlags, isFlag(name[2:])):
|
||||
if !inline {
|
||||
value = next(&i)
|
||||
}
|
||||
@@ -271,6 +271,10 @@ func parseCommandLine(argv []string) (inv invocation, err error) {
|
||||
return inv, err
|
||||
}
|
||||
|
||||
func isFlag(name string) func(config.SpecFlag) bool {
|
||||
return func(f config.SpecFlag) bool { return f.Flag == name }
|
||||
}
|
||||
|
||||
// presetCommands makes each preset a subcommand whose flags are its keys
|
||||
func presetCommands() []subcommand {
|
||||
var subs []subcommand
|
||||
@@ -294,7 +298,7 @@ func presetCommands() []subcommand {
|
||||
for name, v := range values {
|
||||
set[name] = *v
|
||||
}
|
||||
pipeline, err := config.ExpandPreset(p.Name, set)
|
||||
pipeline, err := config.ExpandPreset(p.Name, set, config.HostIsDir)
|
||||
if err != nil {
|
||||
return usageError(err.Error())
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"flag"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"regexp"
|
||||
@@ -12,7 +13,9 @@ import (
|
||||
"testing"
|
||||
|
||||
shipped "github.com/lixenwraith/logwisp/config"
|
||||
"github.com/lixenwraith/logwisp/internal/compose"
|
||||
"github.com/lixenwraith/logwisp/internal/config"
|
||||
"github.com/lixenwraith/logwisp/internal/plugin"
|
||||
"github.com/lixenwraith/logwisp/internal/testutil"
|
||||
"github.com/lixenwraith/logwisp/internal/tlsx"
|
||||
)
|
||||
@@ -187,6 +190,123 @@ func TestDumpReadsBack(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The command line, file and environment a composition writes, read by a
|
||||
// POSIX shell and loaded as lw loads them, give its pipelines back: every
|
||||
// preset, a second sink, a file's stage without its defaults, an integer for
|
||||
// a number, an edge and aggregator pair, and values to escape and quote. The
|
||||
// environment holds one pipeline.
|
||||
func TestEveryFormLoadsTheSamePipelines(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
empty := filepath.Join(dir, "empty.toml")
|
||||
if err := os.WriteFile(empty, nil, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
users, pass := filepath.Join(dir, "users.toml"), filepath.Join(dir, "it's.pass")
|
||||
values := map[string]map[string]string{
|
||||
"pipe": {"format": "txt"},
|
||||
"tail": {"path": dir},
|
||||
"serve": {"path": filepath.Join(dir, "*.log"), "listen": "127.0.0.1:15841", "tls": "self", "users": users, "allow": "127.0.0.0/8", "deny": "127.0.0.2"},
|
||||
"edge": {"path": dir + "/", "to": "127.0.0.1:15842", "transport": "http", "pin": "sha256//" + strings.Repeat("A", 43) + "=",
|
||||
"user": `edge,01=a\b`, "password_file": pass},
|
||||
"aggregator": {"listen": "127.0.0.1:15843", "users": users, "out": filepath.Join(dir, "out dir"), "allow": "10.0.0.0/8,127.0.0.1"},
|
||||
}
|
||||
compositions := map[string]*compose.Composition{}
|
||||
for _, p := range config.Presets() {
|
||||
c, err := compose.FromPreset(p.Name, values[p.Name], config.HostIsDir)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", p.Name, err)
|
||||
}
|
||||
compositions[p.Name] = c
|
||||
}
|
||||
if _, err := compositions["tail"].Add(0, "sink", "null"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
written := filepath.Join(dir, "written.toml")
|
||||
if err := os.WriteFile(written, []byte(`[[pipelines]]
|
||||
name = "written"
|
||||
flow.rate_limit.rate = 5
|
||||
plugin_sources = [{id = "in", type = "null"}]
|
||||
plugin_sinks = [{id = "out", type = "http", config = {port = 15844, acl = {requests_per_second_per_client = 2}}}]
|
||||
`), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c, err := compose.FromConfig(loadTestConfig(t, written))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
compositions["written"] = c
|
||||
pair := &compose.Composition{Pipelines: append(compositions["edge"].Pipelines, compositions["aggregator"].Pipelines...)}
|
||||
filter, err := pair.Add(0, "filters", "exclude")
|
||||
if err == nil {
|
||||
err = pair.Set(0, filter, "patterns", `password=\S{8,64}`, `it's "quoted"`)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
compositions["pair"] = pair
|
||||
shell := func(script string) []string {
|
||||
out, err := exec.Command("sh", "-c", script).Output()
|
||||
if err != nil {
|
||||
t.Fatalf("sh: %v\n%s", err, script)
|
||||
}
|
||||
return strings.Split(strings.TrimSuffix(string(out), "\x00"), "\x00")
|
||||
}
|
||||
for name, c := range compositions {
|
||||
line, err := c.CommandLine()
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", name, err)
|
||||
}
|
||||
words := shell("set --" + strings.TrimPrefix(line, "lw") + `printf '%s\0' "$@"`)
|
||||
loaded := []*config.Config{loadTestConfig(t, empty, words...)}
|
||||
file := filepath.Join(dir, name+".toml")
|
||||
data, err := c.File()
|
||||
if err == nil {
|
||||
err = os.WriteFile(file, data, 0o600)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
loaded = append(loaded, loadTestConfig(t, file))
|
||||
env, err := c.Environment()
|
||||
if len(c.Pipelines) > 1 {
|
||||
if err == nil {
|
||||
t.Errorf("%s: the environment took %d pipelines", name, len(c.Pipelines))
|
||||
}
|
||||
} else {
|
||||
var names []string
|
||||
for _, line := range strings.Split(strings.TrimSpace(env), "\n") {
|
||||
names = append(names, `"$`+line[:strings.IndexByte(line, '=')]+`"`)
|
||||
}
|
||||
testutil.ClearEnvPrefix(t, "LOGWISP_")
|
||||
for i, value := range shell(env + "printf '%s\\0' " + strings.Join(names, " ")) {
|
||||
t.Setenv(strings.Trim(names[i], `"$`), value)
|
||||
}
|
||||
loaded = append(loaded, loadWithEnvironment(t, empty))
|
||||
}
|
||||
for _, cfg := range loaded {
|
||||
back, err := compose.FromConfig(cfg)
|
||||
if err != nil || !reflect.DeepEqual(back, c) {
|
||||
t.Errorf("%s: loads back as\n%+v %v\nwant %+v\nfrom %s\n%s\n%s", name, back, err, c, line, data, env)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// lw --schema lists the sources and sinks lw builds
|
||||
func TestSchemaListsWhatLwBuilds(t *testing.T) {
|
||||
s := compose.NewSchema()
|
||||
for _, p := range s.Sources {
|
||||
if _, ok := plugin.GetSource(p.Type); !ok {
|
||||
t.Errorf("source %s has no factory", p.Type)
|
||||
}
|
||||
}
|
||||
for _, p := range s.Sinks {
|
||||
if _, ok := plugin.GetSink(p.Type); !ok {
|
||||
t.Errorf("sink %s has no factory", p.Type)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Every preset expands to plugins that build, as lw --check builds them
|
||||
func TestEveryPresetBuilds(t *testing.T) {
|
||||
testLogger(t)
|
||||
|
||||
@@ -77,8 +77,8 @@ func (s shell) script() []byte {
|
||||
// options maps each long option lw itself takes to its short letter, if any
|
||||
func options() map[string]string {
|
||||
options := map[string]string{}
|
||||
for _, long := range config.SpecFlags() {
|
||||
options[long] = ""
|
||||
for _, f := range config.SpecFlags() {
|
||||
options[f.Flag] = ""
|
||||
}
|
||||
for long := range config.Settings() {
|
||||
options[long] = ""
|
||||
@@ -99,7 +99,7 @@ func options() map[string]string {
|
||||
func completionModel() (top []compFlag, cmds []compCommand, presets string) {
|
||||
settings, specFlags, shortOf := config.Settings(), config.SpecFlags(), options()
|
||||
for _, long := range slices.Sorted(maps.Keys(shortOf)) {
|
||||
f := compFlag{long: long, short: shortOf[long], value: "text", repeat: slices.Contains(specFlags, long)}
|
||||
f := compFlag{long: long, short: shortOf[long], value: "text", repeat: slices.ContainsFunc(specFlags, isFlag(long))}
|
||||
if v, ok := valueOf[long]; ok {
|
||||
f.value = v
|
||||
} else if settings[long] || slices.Contains(switches, long) {
|
||||
|
||||
+3
-1
@@ -19,7 +19,7 @@ Usage:
|
||||
lw -t|--check [options] Build every pipeline and plugin, report, exit
|
||||
lw --dump [options] Print the effective configuration as TOML, exit
|
||||
lw help | -h | --help
|
||||
lw -V | --version
|
||||
lw -V | --version | --schema
|
||||
|
||||
Commands (lw COMMAND -h lists its subcommands):
|
||||
%s
|
||||
@@ -37,6 +37,8 @@ after '='; short options do not combine. lw takes options only.
|
||||
(on a terminal; default), always (a bare
|
||||
--color) or never
|
||||
-V, --version Print the version and exit
|
||||
--schema Print every setting, flag, plugin option and
|
||||
preset as JSON, exit
|
||||
-h, --help Print this help and exit
|
||||
-- End the options
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
@@ -9,6 +10,7 @@ import (
|
||||
"reflect"
|
||||
"syscall"
|
||||
|
||||
"github.com/lixenwraith/logwisp/internal/compose"
|
||||
"github.com/lixenwraith/logwisp/internal/config"
|
||||
"github.com/lixenwraith/logwisp/internal/core"
|
||||
"github.com/lixenwraith/logwisp/internal/version"
|
||||
@@ -33,6 +35,14 @@ func main() {
|
||||
case inv.on["version"]:
|
||||
fmt.Println(version.String())
|
||||
os.Exit(0)
|
||||
case inv.on["schema"]:
|
||||
out := json.NewEncoder(os.Stdout)
|
||||
out.SetIndent("", " ")
|
||||
if err := out.Encode(compose.NewSchema()); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Error: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
// --- 1. Initial setup ---
|
||||
|
||||
@@ -20,6 +20,11 @@ func testLogger(t *testing.T) {
|
||||
func loadTestConfig(t *testing.T, path string, args ...string) *config.Config {
|
||||
t.Helper()
|
||||
testutil.ClearEnvPrefix(t, "LOGWISP_")
|
||||
return loadWithEnvironment(t, path, args...)
|
||||
}
|
||||
|
||||
func loadWithEnvironment(t *testing.T, path string, args ...string) *config.Config {
|
||||
t.Helper()
|
||||
inv, err := parseCommandLine(append([]string{"-c", path, "--quiet", "--status_reporter=false"}, args...))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -202,6 +202,7 @@ _arguments -S \
|
||||
'*'{-p=,--preset=}':preset:(pipe tail serve edge aggregator)' \
|
||||
'(-q --quiet)'{-q,--quiet} \
|
||||
'*--rate-limit=:value:_default' \
|
||||
'--schema' \
|
||||
'*--sink=:value:_default' \
|
||||
'*--source=:value:_default' \
|
||||
'--status_reporter' \
|
||||
|
||||
@@ -10,7 +10,7 @@ _lw() {
|
||||
elif [[ $prev == = ]]; then prev=${COMP_WORDS[i-2]}
|
||||
fi
|
||||
case ${COMP_WORDS[1]}:$i in
|
||||
*:1) words='--auto_reload --check -t --color --config -c --dump --filter --format --heartbeat --help -h --logging.console.target --logging.file.directory --logging.file.max_size_mb --logging.file.max_total_size_mb --logging.file.name --logging.file.retention_hours --logging.format --logging.level --logging.output --logging.sanitization --pipeline --preset -p --quiet -q --rate-limit --sink --source --status_reporter --version -V auth tls preset config help' ;;
|
||||
*:1) words='--auto_reload --check -t --color --config -c --dump --filter --format --heartbeat --help -h --logging.console.target --logging.file.directory --logging.file.max_size_mb --logging.file.max_total_size_mb --logging.file.name --logging.file.retention_hours --logging.format --logging.level --logging.output --logging.sanitization --pipeline --preset -p --quiet -q --rate-limit --schema --sink --source --status_reporter --version -V auth tls preset config help' ;;
|
||||
auth:2) words='add-user remove-user token stream' ;;
|
||||
auth:*)
|
||||
case ${COMP_WORDS[2]} in
|
||||
@@ -95,7 +95,7 @@ _lw() {
|
||||
--preset|-p) mapfile -t COMPREPLY < <(compgen -W 'pipe tail serve edge aggregator' -- "$cur"); return ;;
|
||||
--config|-c|--filter|--format|--heartbeat|--logging.console.target|--logging.file.max_size_mb|--logging.file.max_total_size_mb|--logging.file.name|--logging.file.retention_hours|--logging.format|--logging.level|--logging.output|--logging.sanitization|--pipeline|--rate-limit|--sink|--source) COMPREPLY=(); return ;;
|
||||
esac
|
||||
words='--auto_reload --check -t --color --config -c --dump --filter --format --heartbeat --help -h --logging.console.target --logging.file.directory --logging.file.max_size_mb --logging.file.max_total_size_mb --logging.file.name --logging.file.retention_hours --logging.format --logging.level --logging.output --logging.sanitization --pipeline --preset -p --quiet -q --rate-limit --sink --source --status_reporter --version -V' ;;
|
||||
words='--auto_reload --check -t --color --config -c --dump --filter --format --heartbeat --help -h --logging.console.target --logging.file.directory --logging.file.max_size_mb --logging.file.max_total_size_mb --logging.file.name --logging.file.retention_hours --logging.format --logging.level --logging.output --logging.sanitization --pipeline --preset -p --quiet -q --rate-limit --schema --sink --source --status_reporter --version -V' ;;
|
||||
esac
|
||||
mapfile -t COMPREPLY < <(compgen -W "$words" -- "$cur")
|
||||
}
|
||||
|
||||
@@ -37,6 +37,7 @@ complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset
|
||||
complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset; and not __lw_in config' -l preset -o p -x -a 'pipe tail serve edge aggregator'
|
||||
complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset; and not __lw_in config' -l quiet -o q
|
||||
complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset; and not __lw_in config' -l rate-limit -r -F
|
||||
complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset; and not __lw_in config' -l schema
|
||||
complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset; and not __lw_in config' -l sink -r -F
|
||||
complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset; and not __lw_in config' -l source -r -F
|
||||
complete -c lw -n 'not __lw_in auth; and not __lw_in tls; and not __lw_in preset; and not __lw_in config' -l status_reporter
|
||||
|
||||
+4
-1
@@ -11,7 +11,7 @@ lw tls COMMAND [flags]
|
||||
lw preset NAME [flags]
|
||||
lw config init [--out FILE]
|
||||
lw help | -h | --help
|
||||
lw -V | --version
|
||||
lw -V | --version | --schema
|
||||
```
|
||||
|
||||
Without a configuration file or pipeline options, lw is a filter: stdin to
|
||||
@@ -66,6 +66,9 @@ pipelines have [their own flags](#pipelines):
|
||||
exit; see [Usage Patterns](#usage-patterns)
|
||||
- `--dump`: print the effective configuration, flags, environment and presets
|
||||
resolved into it, as TOML that `-c` reads back, and exit
|
||||
- `--schema`: print every setting, pipeline flag, flow stage, plugin option
|
||||
and preset as one JSON document (kinds, defaults, rules, help), and exit;
|
||||
tools build their option lists from it
|
||||
- `--color [WHEN]`, WHEN `auto`, `always` or `never`: level names in color on
|
||||
console sinks, default `auto` (a terminal, `NO_COLOR` unset, `TERM` not
|
||||
`dumb`); bare, `always`. A console sink's own `color` wins
|
||||
|
||||
@@ -50,8 +50,8 @@ Top-level keys, each as type and default:
|
||||
set their own: `"auto"` (a terminal, `NO_COLOR` unset, `TERM` not `dumb`),
|
||||
`"always"` or `"never"`; see the [console sink](sinks.md#console).
|
||||
|
||||
`--version`, `--check` and `--dump` print and exit; they are not persistent
|
||||
settings.
|
||||
`--version`, `--check`, `--dump` and `--schema` print and exit; they are not
|
||||
persistent settings.
|
||||
|
||||
Note that `status_reporter` writes at DEBUG level, so it produces nothing unless
|
||||
`logging.level = "debug"`.
|
||||
|
||||
@@ -24,7 +24,7 @@
|
||||
.Ar command
|
||||
.Op Ar flags
|
||||
.Nm
|
||||
.Fl V | Fl \-version
|
||||
.Fl V | Fl \-version | Fl \-schema
|
||||
.Nm
|
||||
.Cm help | Fl h | Fl \-help
|
||||
.Sh DESCRIPTION
|
||||
@@ -146,6 +146,10 @@ Print the effective configuration, with the options, environment and
|
||||
presets resolved into it, as a file
|
||||
.Fl c
|
||||
reads back, and exit.
|
||||
.It Fl \-schema
|
||||
Print every setting, pipeline flag, flow stage, plugin option and preset as
|
||||
one JSON document, with kinds, defaults, rules and help, and exit; tools build
|
||||
their option lists from it.
|
||||
.It Fl q , Fl \-quiet
|
||||
Turn off the log of
|
||||
.Nm
|
||||
|
||||
+10
-36
@@ -92,42 +92,23 @@ What exists:
|
||||
offline and names the refused key's path.
|
||||
- `internal/config` already compiles for `GOOS=js`; it links no net/http,
|
||||
crypto/tls or terminal code.
|
||||
- The engine, `internal/compose`: starts empty, from a preset (whose path
|
||||
`config.IsDir` resolves, nil off the host), a loaded configuration or a
|
||||
pasted command line; adds, removes and moves sources, sinks and filters,
|
||||
sets and unsets keys, validates; writes a command line, environment
|
||||
variables or a file, each loading back to the same pipelines. Its test
|
||||
fails when it links TLS, HTTP, process, signal or terminal code.
|
||||
- `lw --schema`: settings, pipeline flags, flow stages in order, the tls,
|
||||
auth and acl tables, sources, sinks and presets, as one JSON document.
|
||||
- `github.com/lixenwraith/terminal` (pinned) and its `tui` package: regions,
|
||||
layout, boxes, List, Tree, TabBar, StatusBar, TextField, Form, Modal,
|
||||
ConfirmDialog, scrolling, mouse. `github.com/lixenwraith/color`: RGB,
|
||||
blending, RGBTo256, RGBTo16.
|
||||
|
||||
What is missing:
|
||||
- A host-independent preset: `fileSource` (`preset.go`) decides directory or
|
||||
pattern with `os.Stat`, so off the target host `path=/var/log` expands to
|
||||
directory `/var`, pattern `log`.
|
||||
- Emitters: only parsers exist for the three forms.
|
||||
|
||||
Steps:
|
||||
1. Schema, in `internal/config`: done.
|
||||
2. Engine, `internal/compose`.
|
||||
- Imports internal/config, internal/version and lixenwraith/toml only; a
|
||||
test fails when `go list -deps` shows net, crypto/tls, os/exec, os/signal,
|
||||
x/term or the auth, log and terminal modules.
|
||||
- Starts: empty; a preset, with a `stat` function that is nil off the host
|
||||
(the UI then asks "directory or file pattern"); a loaded configuration; a
|
||||
pasted command line (wrapping `cliSpecs` and `buildPipelines`).
|
||||
- Operations: add or remove a source, sink or filter; set or unset an
|
||||
option; move a filter; validate.
|
||||
- Emitters, one form per call:
|
||||
- Command line: `--pipeline NAME`, `id=`, the rate limit policy and
|
||||
heartbeat `enabled` always explicit (their command-line defaults differ
|
||||
from the file's); `--key=value` when a value starts with `-`; `,` `=`
|
||||
`\` escaped; POSIX-quoted.
|
||||
- Environment: `LOGWISP_SOURCE_1` and the rest, single-quoted
|
||||
`KEY='value'` lines; an error for more than one pipeline, which the
|
||||
environment form cannot hold.
|
||||
- File: the TOML `lw --dump` prints for those pipelines.
|
||||
3. `lw --schema`, a parser switch like `--help` (not a Config field, so no
|
||||
file or environment can trigger it). It prints one JSON document: version;
|
||||
settings with kind, default, enum, flag and variable; the pipeline flags;
|
||||
flow stage order; the tls, auth and acl tables; sources, sinks, presets.
|
||||
The site generates its catalogue from it.
|
||||
2. Engine, `internal/compose`: done.
|
||||
3. `lw --schema`: done. The site generates its catalogue from it.
|
||||
4. In `lixenwraith/terminal`, released and pinned before the TUI:
|
||||
- Full-screen 8- and 16-color output (SGR 30-37, 90-97, 40-47) and the
|
||||
default colors 39/49; it emits only 256-color and truecolor today.
|
||||
@@ -191,13 +172,6 @@ Steps:
|
||||
eval, so the page needs only `'wasm-unsafe-eval'` in its `script-src`.
|
||||
|
||||
Verify:
|
||||
- A Go test in `cmd/lw`: for every preset, an edge and aggregator pair,
|
||||
escaped values and two pipelines, the pipelines loaded from the emitted
|
||||
command line, file and environment (one pipeline) are equal. It runs in an
|
||||
empty directory with the `LOGWISP_` environment cleared, since a discovered
|
||||
file would add settings.
|
||||
- A schema test: every option has a kind, every enum values, every catalogue
|
||||
row a factory; `TestEveryPluginRejectsUnknownKeys` still passes.
|
||||
- CI builds `cmd/lwconf` for `GOOS=js` and runs a node test of its calls.
|
||||
- The diagram rendered into a cell buffer at 80x24 and 64x24, per tier; a
|
||||
scripted session on a pseudo-terminal for Run and each output.
|
||||
@@ -7,13 +7,13 @@ require (
|
||||
github.com/lixenwraith/color v0.1.3
|
||||
github.com/lixenwraith/config v0.2.2
|
||||
github.com/lixenwraith/log v0.1.11
|
||||
github.com/lixenwraith/terminal v0.1.9
|
||||
github.com/lixenwraith/terminal v0.2.0
|
||||
github.com/lixenwraith/toml v0.1.3
|
||||
golang.org/x/term v0.46.0
|
||||
golang.org/x/term v0.47.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/golang-jwt/jwt/v5 v5.3.1 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
golang.org/x/crypto v0.58.0 // indirect
|
||||
golang.org/x/sys v0.49.0 // indirect
|
||||
)
|
||||
@@ -8,13 +8,13 @@ github.com/lixenwraith/config v0.2.2 h1:X+l3Tf8Rbk49cipYBJU7/pfdw5wDwD4hRxtFH7rZ
|
||||
github.com/lixenwraith/config v0.2.2/go.mod h1:yfJQwKS+/6Dg70GIoHeYc0eYztOPETpqe6nUIbakcyg=
|
||||
github.com/lixenwraith/log v0.1.11 h1:dKWLnaR8dhXxj4l/aU5XZiFIsr8geaJQbOf8m7SiKhU=
|
||||
github.com/lixenwraith/log v0.1.11/go.mod h1:EE8i5nSAW2yZBTytKuhdgehmgzrby6kW1k5DyH1SvBg=
|
||||
github.com/lixenwraith/terminal v0.1.9 h1:WPxLdqdzZoqtPgTb86oEwE66lF1+iPsJ/gpfo6h/rik=
|
||||
github.com/lixenwraith/terminal v0.1.9/go.mod h1:f9Ntk+xKBS2LWQEe6gk/1TOpG22HH0scTe4wlPhEy44=
|
||||
github.com/lixenwraith/terminal v0.2.0 h1:Xwsg8LoAeV7E//JLIFmEBTTV9e03jpze3EUAOdqS6P8=
|
||||
github.com/lixenwraith/terminal v0.2.0/go.mod h1:f9Ntk+xKBS2LWQEe6gk/1TOpG22HH0scTe4wlPhEy44=
|
||||
github.com/lixenwraith/toml v0.1.3 h1:mhnNVLNogdaqjGd3eC4rV9ZqY3j1Sv+DK1erZ+SPFeI=
|
||||
github.com/lixenwraith/toml v0.1.3/go.mod h1:itcT/Vt5mwmX9KVGCMsWZr6vbU/JgZ24wV/I0J5cMt0=
|
||||
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
|
||||
golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc=
|
||||
golang.org/x/crypto v0.58.0 h1:COkYLr4k7nDI4r6QtMML2AfzeIswWHAJthLS6K/M/54=
|
||||
golang.org/x/crypto v0.58.0/go.mod h1:Xh+kl5A+M0gMiWDytH2Hhr5Z7WbNCpv/eB/EFgfSk+w=
|
||||
golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA=
|
||||
golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/term v0.47.0 h1:cpyOwQkr2Nzxe2p6KnrTG7Ei2wtzjsuSjhpXXepZ/BM=
|
||||
golang.org/x/term v0.47.0/go.mod h1:efAfAuG6RKABvg1ArGebRkvX6LXul14SM67o2e4mUCY=
|
||||
@@ -0,0 +1,571 @@
|
||||
// Package compose is the configuration engine lw --tui and the website share:
|
||||
// pipelines started empty, from a preset, a configuration or a command line,
|
||||
// edited, validated, and written as a command line, environment or file. It
|
||||
// reads no file and links no network or terminal code, so it builds as WebAssembly.
|
||||
package compose
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"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 hold their
|
||||
// defaults and its plugin options the kinds they declare, so each form it
|
||||
// writes loads back to the same pipelines.
|
||||
type Composition struct {
|
||||
Pipelines []config.PipelineConfig
|
||||
}
|
||||
|
||||
// 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
|
||||
ID string
|
||||
Index int
|
||||
}
|
||||
|
||||
// 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, as a composition holds no settings
|
||||
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
|
||||
}
|
||||
if len(words) > 0 && (words[0] == "lw" || strings.HasSuffix(words[0], "/lw")) {
|
||||
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 !long || !slices.ContainsFunc(config.SpecFlags(), func(f config.SpecFlag) bool { return f.Flag == flag }) {
|
||||
return nil, fmt.Errorf("%s: not a pipeline flag", words[i])
|
||||
}
|
||||
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})
|
||||
}
|
||||
pipelines, err := config.SpecPipelines(specs, isDir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c := &Composition{Pipelines: pipelines}
|
||||
return c, c.settle()
|
||||
}
|
||||
|
||||
// settle gives every flow stage its defaults and every plugin's options their
|
||||
// kinds; the rules are Validate's
|
||||
func (c *Composition) settle() error {
|
||||
for i := range c.Pipelines {
|
||||
p := &c.Pipelines[i]
|
||||
if p.Flow == nil {
|
||||
p.Flow = &config.FlowConfig{}
|
||||
}
|
||||
for _, stage := range stages(p.Flow) {
|
||||
if !stage.IsNil() {
|
||||
config.Fill(stage.Interface())
|
||||
}
|
||||
}
|
||||
for j := range p.Flow.Filters {
|
||||
config.Fill(&p.Flow.Filters[j])
|
||||
}
|
||||
for j, s := range p.PluginSources {
|
||||
p.PluginSources[j].Config = orEmpty(s.Config)
|
||||
if err := config.Coerce("source", s.Type, p.PluginSources[j].Config); err != nil {
|
||||
return fmt.Errorf("source %s: %w", s.ID, err)
|
||||
}
|
||||
}
|
||||
for j, s := range p.PluginSinks {
|
||||
p.PluginSinks[j].Config = orEmpty(s.Config)
|
||||
if err := config.Coerce("sink", s.Type, p.PluginSinks[j].Config); err != nil {
|
||||
return fmt.Errorf("sink %s: %w", s.ID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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 name == "" || slices.ContainsFunc(c.Pipelines, func(p config.PipelineConfig) bool { return p.Name == name }) {
|
||||
return fmt.Errorf("pipeline %q: a name no other pipeline has", name)
|
||||
}
|
||||
c.Pipelines = append(c.Pipelines, config.PipelineConfig{Name: name, Flow: &config.FlowConfig{}})
|
||||
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, 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":
|
||||
f := config.FilterConfig{Type: config.FilterType(typ)}
|
||||
config.Fill(&f)
|
||||
p.Flow.Filters = append(p.Flow.Filters, f)
|
||||
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
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
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; 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 := config.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, 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
|
||||
func editStage(stage reflect.Value, change func(map[string]any) error) error {
|
||||
m := map[string]any{}
|
||||
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.Fill(next.Interface())
|
||||
stage.Set(next)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Validate applies every rule lw applies at load, naming the refused key's path
|
||||
func (c *Composition) Validate() error { return config.ValidatePipelines(c.Pipelines) }
|
||||
|
||||
// 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", 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})
|
||||
}
|
||||
|
||||
// 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"`
|
||||
Tables map[string][]config.Key `json:"tables"`
|
||||
Sources []Plugin `json:"sources"`
|
||||
Sinks []Plugin `json:"sinks"`
|
||||
Presets []config.Preset `json:"presets"`
|
||||
}
|
||||
|
||||
// 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(),
|
||||
Tables: map[string][]config.Key{
|
||||
"tls": config.KeysOf[config.TLSOptions](),
|
||||
"auth": config.KeysOf[config.AuthOptions](),
|
||||
"acl": config.KeysOf[config.ACLOptions](),
|
||||
},
|
||||
Presets: config.Presets(),
|
||||
}
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package compose
|
||||
|
||||
import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The engine builds as WebAssembly for the website: it links no TLS, HTTP,
|
||||
// process, signal or terminal code, nor lw's log, auth or terminal modules
|
||||
func TestEngineLinksNoHostCode(t *testing.T) {
|
||||
cmd := exec.Command("go", "list", "-deps", ".")
|
||||
cmd.Env = append(os.Environ(), "GOOS=js", "GOARCH=wasm")
|
||||
out, err := cmd.CombinedOutput()
|
||||
if err != nil {
|
||||
t.Fatalf("go list: %v\n%s", err, out)
|
||||
}
|
||||
for _, dep := range strings.Fields(string(out)) {
|
||||
module, mine := strings.CutPrefix(dep, "github.com/lixenwraith/")
|
||||
if slices.Contains([]string{"crypto/tls", "net/http", "os/exec", "os/signal", "golang.org/x/term"}, dep) ||
|
||||
mine && slices.Contains([]string{"log", "auth", "terminal"}, strings.Split(module, "/")[0]) {
|
||||
t.Errorf("links %s", dep)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A pasted command line, its line ends LF or CRLF, starts the composition that
|
||||
// wrote it: escaped and quoted values, nested keys, lists, moved filters,
|
||||
// stages whose flags' defaults differ from a file's, a value starting with '-'
|
||||
func TestPastedCommandLineStartsTheSameComposition(t *testing.T) {
|
||||
c := &Composition{}
|
||||
must := func(err error) {
|
||||
t.Helper()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
must(c.AddPipeline("edge"))
|
||||
src, err := c.Add(0, "source", "file")
|
||||
must(err)
|
||||
must(c.Set(0, src, "directory", "/var/log/my app"))
|
||||
must(c.Set(0, src, "pattern", "it's *.log"))
|
||||
sink, err := c.Add(0, "sink", "tcp_chain")
|
||||
must(err)
|
||||
for key, value := range map[string]string{"host": "agg.example", "port": "9000", "tls.enabled": "true",
|
||||
"tls.pin_sha256": "sha256//" + strings.Repeat("A", 43) + "=", "auth.type": "scram",
|
||||
"auth.username": `edge,01=a\b\`, "auth.password_file": "/etc/lw/edge.pass"} {
|
||||
must(c.Set(0, sink, key, value))
|
||||
}
|
||||
exclude, err := c.Add(0, "filters", "exclude")
|
||||
must(err)
|
||||
must(c.Set(0, exclude, "patterns", `password=\S{8,64}`, `"quoted"`))
|
||||
_, err = c.Add(0, "filters", "include")
|
||||
must(err)
|
||||
must(c.MoveFilter(0, 1, 0))
|
||||
must(c.Set(0, Node{Role: "rate_limit"}, "rate", "100"))
|
||||
must(c.Set(0, Node{Role: "heartbeat"}, "interval_ms", "5000"))
|
||||
must(c.Set(0, Node{Role: "heartbeat"}, "enabled", "true"))
|
||||
must(c.AddPipeline("-b"))
|
||||
for _, role := range []string{"source", "sink"} {
|
||||
_, err = c.Add(1, role, "null")
|
||||
must(err)
|
||||
}
|
||||
must(c.Set(1, Node{Role: "heartbeat"}, "interval_ms", "2000"))
|
||||
line, err := c.CommandLine()
|
||||
must(err)
|
||||
for _, pasted := range []string{line, strings.ReplaceAll(line, "\n", "\r\n")} {
|
||||
got, err := FromCommandLine(pasted, nil)
|
||||
if err != nil || !reflect.DeepEqual(got, c) {
|
||||
t.Fatalf("%q\n got %+v %v\nwant %+v", pasted, got, err, c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The command line and environment carry no control character, which a
|
||||
// terminal acts on before the shell reads the paste; the file carries it
|
||||
func TestShellFormsRefuseControlCharacters(t *testing.T) {
|
||||
c, err := FromPreset("tail", map[string]string{"path": "/srv/a\x15b/"}, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, line := c.CommandLine()
|
||||
_, env := c.Environment()
|
||||
if _, err := c.File(); err != nil || line == nil || env == nil {
|
||||
t.Fatalf("command line %v, environment %v, file %v", line, env, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Set refuses a value of the wrong kind and leaves the key as it was; Unset
|
||||
// returns a key to its default, dropping a table it leaves empty
|
||||
func TestSetKeepsAKeyItCannotTake(t *testing.T) {
|
||||
c := &Composition{}
|
||||
if err := c.AddPipeline("p"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n, _ := c.Add(0, "sink", "tcp")
|
||||
beat := Node{Role: "heartbeat"}
|
||||
for _, err := range []error{c.Set(0, n, "port", "8080"), c.Set(0, n, "tls.enabled", "true"),
|
||||
c.Unset(0, n, "tls.enabled"), c.Set(0, beat, "interval_ms", "5000"), c.Unset(0, beat, "interval_ms")} {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := c.Set(0, n, "port", "http"); err == nil || !strings.HasPrefix(err.Error(), "port: ") {
|
||||
t.Fatalf("port http: %v", err)
|
||||
}
|
||||
if m, beat := c.Pipelines[0].PluginSinks[0].Config, c.Pipelines[0].Flow.Heartbeat; !reflect.DeepEqual(m, map[string]any{"port": int64(8080)}) ||
|
||||
beat.IntervalMS != 1000 {
|
||||
t.Fatalf("options %v, heartbeat %+v", m, beat)
|
||||
}
|
||||
}
|
||||
+18
-28
@@ -4,15 +4,13 @@ package config
|
||||
|
||||
// Config is the top-level configuration structure for the LogWisp application
|
||||
type Config struct {
|
||||
Quiet bool `toml:"quiet"`
|
||||
Quiet bool `toml:"quiet" help:"silence lw's own log and notices; pipeline output still flows"`
|
||||
|
||||
// Console sinks without their own color: "auto" is a terminal with
|
||||
// NO_COLOR unset and TERM not dumb
|
||||
Color string `toml:"color" lw:"enum=auto|always|never"`
|
||||
// "auto" is a terminal with NO_COLOR unset and TERM not dumb
|
||||
Color string `toml:"color" lw:"enum=auto|always|never" help:"level names in color on console sinks without their own"`
|
||||
|
||||
// Runtime behavior flags
|
||||
StatusReporter bool `toml:"status_reporter"`
|
||||
ConfigAutoReload bool `toml:"auto_reload"`
|
||||
StatusReporter bool `toml:"status_reporter" help:"log pipeline statistics every 30 s at debug; without a file, off"`
|
||||
ConfigAutoReload bool `toml:"auto_reload" help:"reload when the file changes"`
|
||||
|
||||
// Selected file path is runtime metadata, never a configurable override.
|
||||
ConfigFile string `toml:"-"`
|
||||
@@ -26,10 +24,10 @@ type Config struct {
|
||||
|
||||
// LogConfig represents the logging configuration for the LogWisp application itself
|
||||
type LogConfig struct {
|
||||
Output string `toml:"output" lw:"enum=file|stdout|stderr|split|all|none"`
|
||||
Level string `toml:"level" lw:"enum=debug|info|warn|error"`
|
||||
Format string `toml:"format" lw:"enum=raw|txt|json,zero=the log default"`
|
||||
Sanitization string `toml:"sanitization" lw:"enum=raw|json|txt|shell,zero=the log default"` // console output
|
||||
Output string `toml:"output" lw:"enum=file|stdout|stderr|split|all|none" help:"where lw's own log goes; split: warn and error to stderr"`
|
||||
Level string `toml:"level" lw:"enum=debug|info|warn|error" help:"lowest level logged; without a file, warn"`
|
||||
Format string `toml:"format" lw:"enum=raw|txt|json,zero=the log default" help:"format of lw's own log"`
|
||||
Sanitization string `toml:"sanitization" lw:"enum=raw|json|txt|shell,zero=the log default" help:"how the console log escapes control characters"`
|
||||
|
||||
// File output settings (when Output includes "file" or "all")
|
||||
File *LogFileConfig `toml:"file"`
|
||||
@@ -40,25 +38,16 @@ type LogConfig struct {
|
||||
|
||||
// LogFileConfig defines settings for file-based application logging
|
||||
type LogFileConfig struct {
|
||||
// Directory for log files
|
||||
Directory string `toml:"directory"`
|
||||
|
||||
// Base name for log files
|
||||
Name string `toml:"name"`
|
||||
|
||||
// Maximum size per log file in MB
|
||||
MaxSizeMB int64 `toml:"max_size_mb"`
|
||||
|
||||
// Maximum total size of all logs in MB
|
||||
MaxTotalSizeMB int64 `toml:"max_total_size_mb"`
|
||||
|
||||
// Log retention in hours (0 = disabled)
|
||||
RetentionHours float64 `toml:"retention_hours"`
|
||||
Directory string `toml:"directory" lw:"hint=dir" help:"where file and all write; the limits below delete old .log files in it"`
|
||||
Name string `toml:"name" help:"base name of the files"`
|
||||
MaxSizeMB int64 `toml:"max_size_mb" help:"size at which a file rotates"`
|
||||
MaxTotalSizeMB int64 `toml:"max_total_size_mb" help:"size of all .log files there, beyond which the oldest go"`
|
||||
RetentionHours float64 `toml:"retention_hours" help:"hours a .log file is kept; 0 keeps them"`
|
||||
}
|
||||
|
||||
// LogConsoleConfig defines settings for console-based application logging
|
||||
type LogConsoleConfig struct {
|
||||
Target string `toml:"target" lw:"enum=stdout|stderr|split"`
|
||||
Target string `toml:"target" lw:"enum=stdout|stderr|split" help:"accepted without effect: logging.output chooses the console"`
|
||||
}
|
||||
|
||||
// --- Pipeline ---
|
||||
@@ -74,12 +63,13 @@ type PipelineConfig struct {
|
||||
|
||||
// --- Flow ---
|
||||
|
||||
// FlowConfig consolidates all processing stages between sources and sinks
|
||||
// FlowConfig holds the stages between sources and sinks, in the order entries
|
||||
// pass them; the heartbeat runs beside them
|
||||
type FlowConfig struct {
|
||||
Heartbeat *HeartbeatConfig `toml:"heartbeat"`
|
||||
RateLimit *RateLimitConfig `toml:"rate_limit"`
|
||||
Filters []FilterConfig `toml:"filters,omitempty"`
|
||||
Format *FormatConfig `toml:"format"`
|
||||
Heartbeat *HeartbeatConfig `toml:"heartbeat"`
|
||||
}
|
||||
|
||||
// --- Flow stages ---
|
||||
|
||||
+16
-10
@@ -35,15 +35,9 @@ type Args struct {
|
||||
// Settings maps each --path setting Load takes to whether it is a switch, which
|
||||
// takes no separate value; the parser must not leave config to guess.
|
||||
func Settings() map[string]bool {
|
||||
c := lconfig.New()
|
||||
if err := c.RegisterStructWithTags("", defaults(), lconfig.FormatTOML); err != nil {
|
||||
panic(err) // Config is static: every run fails or none does
|
||||
}
|
||||
settings := map[string]bool{}
|
||||
for path, value := range c.GetRegisteredPathsWithDefaults() {
|
||||
if _, tables := value.([]PipelineConfig); !tables { // the pipeline flags set those
|
||||
_, settings[path] = value.(bool)
|
||||
}
|
||||
for _, k := range SettingKeys() {
|
||||
settings[k.Name] = k.Kind == "bool"
|
||||
}
|
||||
return settings
|
||||
}
|
||||
@@ -59,7 +53,7 @@ func Load(args Args) (*Manager, error) {
|
||||
if len(specs) == 0 {
|
||||
specs = envPipelineSpecs()
|
||||
}
|
||||
if _, err := buildPipelines(specs); err != nil {
|
||||
if _, err := buildPipelines(specs, HostIsDir); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m := &Manager{path: configPath, explicit: isExplicit, specs: specs}
|
||||
@@ -132,12 +126,24 @@ func (m *Manager) Snapshot() (*Config, error) {
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
// Uninherit removes the color Snapshot gave each console sink, so a dump or a
|
||||
// composition leaves it inherited
|
||||
func Uninherit(pipelines []PipelineConfig, color string) {
|
||||
for _, p := range pipelines {
|
||||
for _, s := range p.PluginSinks {
|
||||
if s.Type == "console" && s.Config["color"] == color {
|
||||
delete(s.Config, "color")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// usePipelines replaces the file's or default pipelines with the spec ones.
|
||||
func (m *Manager) usePipelines(cfg *Config) error {
|
||||
if len(m.specs) == 0 {
|
||||
return nil
|
||||
}
|
||||
pipelines, err := buildPipelines(m.specs)
|
||||
pipelines, err := buildPipelines(m.specs, HostIsDir)
|
||||
cfg.Pipelines = pipelines
|
||||
return err
|
||||
}
|
||||
|
||||
+66
-31
@@ -2,6 +2,7 @@ package config
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"net"
|
||||
@@ -16,16 +17,34 @@ import (
|
||||
// a pipeline with it. Later flags add sources, sinks and filters, or replace
|
||||
// its format, rate limit and heartbeat.
|
||||
type Preset struct {
|
||||
Name, Summary string
|
||||
Params []PresetParam
|
||||
build func(p *PipelineConfig, v map[string]string) error
|
||||
Name string `json:"name"`
|
||||
Summary string `json:"summary"`
|
||||
Params []PresetParam `json:"keys"`
|
||||
build func(p *PipelineConfig, v map[string]string, isDir IsDir) error
|
||||
}
|
||||
|
||||
// PresetParam is one key. A list takes several values, ',' between them or
|
||||
// the key repeated; any other key takes one.
|
||||
type PresetParam struct {
|
||||
Name, Default, Help string
|
||||
Required, List bool
|
||||
Name string `json:"name"`
|
||||
Default string `json:"default,omitempty"`
|
||||
Help string `json:"help"`
|
||||
Required bool `json:"required,omitempty"`
|
||||
List bool `json:"list,omitempty"`
|
||||
}
|
||||
|
||||
// IsDir tells whether a preset's path is a directory, whose files it follows,
|
||||
// or a file pattern. Off the target host it is the user's answer, or nil: a
|
||||
// path ending in '/' is then a directory, one with a glob a pattern, and any
|
||||
// other fails with ErrPathKind.
|
||||
type IsDir func(path string) (bool, error)
|
||||
|
||||
var ErrPathKind = errors.New("path: a directory or a file pattern?")
|
||||
|
||||
// HostIsDir asks this host: a path that does not exist is a pattern
|
||||
func HostIsDir(path string) (bool, error) {
|
||||
fi, err := os.Stat(path)
|
||||
return err == nil && fi.IsDir(), nil
|
||||
}
|
||||
|
||||
// Keys several presets share
|
||||
@@ -50,7 +69,7 @@ var (
|
||||
var presets = []Preset{
|
||||
{"pipe", "Standard input to standard output, line for line: lw without a file",
|
||||
[]PresetParam{formatParam("raw")},
|
||||
func(p *PipelineConfig, v map[string]string) error {
|
||||
func(p *PipelineConfig, v map[string]string, isDir IsDir) error {
|
||||
stdio := pipeDefault("")
|
||||
p.PluginSources, p.PluginSinks = stdio.PluginSources, stdio.PluginSinks
|
||||
p.Flow.Format = &FormatConfig{Type: v["format"]}
|
||||
@@ -58,10 +77,11 @@ var presets = []Preset{
|
||||
}},
|
||||
{"tail", "Follow files to standard output, like tail -F",
|
||||
[]PresetParam{{Name: "path", Required: true, Help: "file, directory (its files) or glob to follow"}, fromParam, formatParam("raw")},
|
||||
func(p *PipelineConfig, v map[string]string) error {
|
||||
p.PluginSources, p.PluginSinks = pathOrStdin(v), pipeDefault("").PluginSinks
|
||||
func(p *PipelineConfig, v map[string]string, isDir IsDir) error {
|
||||
sources, err := pathOrStdin(v, isDir)
|
||||
p.PluginSources, p.PluginSinks = sources, pipeDefault("").PluginSinks
|
||||
p.Flow.Format = &FormatConfig{Type: v["format"]}
|
||||
return nil
|
||||
return err
|
||||
}},
|
||||
{"serve", "Serve files or standard input live: a browser viewer at /, SSE at /stream",
|
||||
slices.Concat([]PresetParam{pathParam, fromParam, formatParam("json"),
|
||||
@@ -71,8 +91,12 @@ var presets = []Preset{
|
||||
{Name: "proxy", List: true, Help: "addresses or CIDRs of the TLS-terminating proxies browsers come through; ',' between them"},
|
||||
{Name: "viewer", Default: "false", Help: "true: the login page and viewer for users, behind proxy; without users the viewer is always on"},
|
||||
}, aclParams, tlsParams),
|
||||
func(p *PipelineConfig, v map[string]string) error {
|
||||
p.PluginSources = pathOrStdin(v)
|
||||
func(p *PipelineConfig, v map[string]string, isDir IsDir) error {
|
||||
sources, err := pathOrStdin(v, isDir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.PluginSources = sources
|
||||
sink, err := listener(v, "off")
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -115,8 +139,12 @@ var presets = []Preset{
|
||||
{Name: "key", Help: "client key file"},
|
||||
{Name: "node", Help: "origin label; default: this host's name"},
|
||||
},
|
||||
func(p *PipelineConfig, v map[string]string) error {
|
||||
p.PluginSources = pathOrStdin(v)
|
||||
func(p *PipelineConfig, v map[string]string, isDir IsDir) error {
|
||||
sources, err := pathOrStdin(v, isDir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.PluginSources = sources
|
||||
typ, err := chainType(v["transport"])
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -159,7 +187,7 @@ var presets = []Preset{
|
||||
{Name: "out", Help: "directory for the received entries; default: standard output"},
|
||||
formatParam("json"),
|
||||
}, aclParams, tlsParams),
|
||||
func(p *PipelineConfig, v map[string]string) error {
|
||||
func(p *PipelineConfig, v map[string]string, isDir IsDir) error {
|
||||
typ, err := chainType(v["transport"])
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -197,18 +225,18 @@ var presets = []Preset{
|
||||
func Presets() []Preset { return slices.Clone(presets) }
|
||||
|
||||
// ExpandPreset builds the pipeline a preset makes of values, as --preset does
|
||||
func ExpandPreset(name string, values map[string]string) (PipelineConfig, error) {
|
||||
func ExpandPreset(name string, values map[string]string, isDir IsDir) (PipelineConfig, error) {
|
||||
p := PipelineConfig{Flow: &FlowConfig{}}
|
||||
each := map[string][]string{}
|
||||
for k, v := range values {
|
||||
each[k] = []string{v}
|
||||
}
|
||||
err := applyPreset(&p, name, each)
|
||||
err := applyPreset(&p, name, each, isDir)
|
||||
return p, err
|
||||
}
|
||||
|
||||
// applyPreset fills defaults and checks keys before the preset's build
|
||||
func applyPreset(p *PipelineConfig, name string, values map[string][]string) error {
|
||||
func applyPreset(p *PipelineConfig, name string, values map[string][]string, isDir IsDir) error {
|
||||
i := slices.IndexFunc(presets, func(r Preset) bool { return r.Name == name })
|
||||
if i < 0 {
|
||||
var names []string
|
||||
@@ -239,24 +267,31 @@ func applyPreset(p *PipelineConfig, name string, values map[string][]string) err
|
||||
}
|
||||
}
|
||||
p.Name = cmp.Or(p.Name, name)
|
||||
return r.build(p, v)
|
||||
return r.build(p, v, isDir)
|
||||
}
|
||||
|
||||
func pathOrStdin(v map[string]string) []PluginSourceConfig {
|
||||
if v["path"] == "" {
|
||||
return pipeDefault("").PluginSources
|
||||
// pathOrStdin follows path: a directory's files, or the files a glob or a
|
||||
// file name matches in its directory; without one, standard input
|
||||
func pathOrStdin(v map[string]string, isDir IsDir) ([]PluginSourceConfig, error) {
|
||||
path := v["path"]
|
||||
if path == "" {
|
||||
return pipeDefault("").PluginSources, nil
|
||||
}
|
||||
return []PluginSourceConfig{fileSource(v["path"], v["from"])}
|
||||
}
|
||||
|
||||
// fileSource follows path: a directory's files, or the files a glob or a
|
||||
// file name matches in its directory.
|
||||
func fileSource(path, from string) PluginSourceConfig {
|
||||
dir, pattern := path, "*"
|
||||
if fi, err := os.Stat(path); err != nil || !fi.IsDir() {
|
||||
dir, pattern = filepath.Dir(path), filepath.Base(path)
|
||||
dir, err := strings.HasSuffix(path, "/"), error(nil)
|
||||
switch {
|
||||
case isDir != nil:
|
||||
dir, err = isDir(path)
|
||||
case !dir && !strings.ContainsAny(path, `*?[\`):
|
||||
err = ErrPathKind
|
||||
}
|
||||
return PluginSourceConfig{ID: "file", Type: "file", Config: map[string]any{"directory": dir, "pattern": pattern, "from": from}}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
directory, pattern := path, "*"
|
||||
if !dir {
|
||||
directory, pattern = filepath.Dir(path), filepath.Base(path)
|
||||
}
|
||||
return []PluginSourceConfig{{ID: "file", Type: "file", Config: map[string]any{"directory": directory, "pattern": pattern, "from": v["from"]}}}, nil
|
||||
}
|
||||
|
||||
// tlsModes is what each preset tls mode sets, and the preset keys it needs
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -62,3 +63,29 @@ func TestPresetKeysAreChecked(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A preset path is what isDir answers, a directory named with brackets too;
|
||||
// without isDir, off the target host, it is a directory when it ends in '/',
|
||||
// a pattern when it holds a glob, and otherwise ErrPathKind asks
|
||||
func TestPresetPathKind(t *testing.T) {
|
||||
yes := func(string) (bool, error) { return true, nil }
|
||||
for _, c := range []struct {
|
||||
path string
|
||||
isDir IsDir
|
||||
want string
|
||||
}{{"/srv/app/", nil, "/srv/app/ *"}, {"/srv/app/*.log", nil, "/srv/app *.log"}, {"/srv/app[1]", yes, "/srv/app[1] *"}, {"/srv/app", nil, ""}} {
|
||||
p, err := ExpandPreset("tail", map[string]string{"path": c.path}, c.isDir)
|
||||
if c.want == "" {
|
||||
if !errors.Is(err, ErrPathKind) {
|
||||
t.Errorf("%s: %v, want ErrPathKind", c.path, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m := p.PluginSources[0].Config; m["directory"].(string)+" "+m["pattern"].(string) != c.want {
|
||||
t.Errorf("%s: %v, want %s", c.path, m, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
+152
-58
@@ -16,6 +16,10 @@ import (
|
||||
// keys it takes
|
||||
type Side int
|
||||
|
||||
var sideNames = []string{"local", "listener", "chain_listener", "dialer"}
|
||||
|
||||
func (s Side) MarshalText() ([]byte, error) { return []byte(sideNames[s]), nil }
|
||||
|
||||
const (
|
||||
Local Side = iota // no network
|
||||
Listener // the tcp and http sinks
|
||||
@@ -25,12 +29,13 @@ const (
|
||||
|
||||
// Plugin is a catalogue row: a source or sink type and its options table
|
||||
type Plugin struct {
|
||||
Role, Type string // "source" or "sink", and the type its config names
|
||||
Side Side
|
||||
HTTP bool // speaks HTTP: takes bearer tokens and request limits
|
||||
Single bool // one instance per process
|
||||
Summary string
|
||||
options reflect.Type
|
||||
Role string `json:"-"` // "source" or "sink"
|
||||
Type string `json:"type"`
|
||||
Side Side `json:"side"`
|
||||
HTTP bool `json:"http"` // speaks HTTP: takes bearer tokens and request limits
|
||||
Single bool `json:"single"` // one instance per process
|
||||
Summary string `json:"summary"`
|
||||
options reflect.Type
|
||||
}
|
||||
|
||||
var plugins = []Plugin{
|
||||
@@ -49,6 +54,12 @@ var plugins = []Plugin{
|
||||
{"sink", "http_chain", Dialer, true, false, "Post entry batches to an http_chain source", reflect.TypeFor[HTTPChainSinkOptions]()},
|
||||
}
|
||||
|
||||
// Plugins lists the catalogue
|
||||
func Plugins() []Plugin { return slices.Clone(plugins) }
|
||||
|
||||
// Keys describes the plugin's options
|
||||
func (p Plugin) Keys() []Key { return keys(p.options) }
|
||||
|
||||
// LookupPlugin finds the catalogue row of a source or sink type
|
||||
func LookupPlugin(role, typ string) (Plugin, bool) {
|
||||
i := slices.IndexFunc(plugins, func(p Plugin) bool { return p.Role == role && p.Type == typ })
|
||||
@@ -73,13 +84,8 @@ func (e *KeyError) Unwrap() error { return e.Err }
|
||||
// it does, and lw: lists its rules: required, enum=a|b, min=N, max=N,
|
||||
// zero=MEANING (zero is a value, not the default), hint=KIND, listener or dialer.
|
||||
type option struct {
|
||||
key, def, help string
|
||||
required bool
|
||||
enum []string
|
||||
min, max *float64
|
||||
zero string
|
||||
hint, side string
|
||||
index int
|
||||
Key
|
||||
index int
|
||||
}
|
||||
|
||||
// options reads the keys of an options table; a malformed tag panics, as the
|
||||
@@ -92,31 +98,31 @@ func options(t reflect.Type) []option {
|
||||
if key == "" || key == "-" {
|
||||
continue
|
||||
}
|
||||
o := option{key: key, def: f.Tag.Get("default"), help: f.Tag.Get("help"), index: i}
|
||||
o := option{Key{Name: key, Kind: kinds[f.Type.Kind()], Default: f.Tag.Get("default"), Help: f.Tag.Get("help")}, i}
|
||||
for rule := range strings.SplitSeq(f.Tag.Get("lw"), ",") {
|
||||
name, value, _ := strings.Cut(rule, "=")
|
||||
switch name {
|
||||
case "":
|
||||
case "required":
|
||||
o.required = true
|
||||
o.Required = true
|
||||
case "listener", "dialer":
|
||||
o.side = name
|
||||
o.Side = name
|
||||
case "enum":
|
||||
o.enum = strings.Split(value, "|")
|
||||
o.Enum = strings.Split(value, "|")
|
||||
case "min", "max":
|
||||
n, err := strconv.ParseFloat(value, 64)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("%s.%s: lw %s=%q", t.Name(), key, name, value))
|
||||
}
|
||||
if name == "min" {
|
||||
o.min = &n
|
||||
o.Min = &n
|
||||
} else {
|
||||
o.max = &n
|
||||
o.Max = &n
|
||||
}
|
||||
case "zero":
|
||||
o.zero = value
|
||||
o.Zero = value
|
||||
case "hint":
|
||||
o.hint = value
|
||||
o.Hint = value
|
||||
default:
|
||||
panic(fmt.Sprintf("%s.%s: unknown lw rule %q", t.Name(), key, rule))
|
||||
}
|
||||
@@ -126,6 +132,81 @@ func options(t reflect.Type) []option {
|
||||
return opts
|
||||
}
|
||||
|
||||
// Key describes one option, for lw --schema and the composer: its tags, and
|
||||
// its kind, a table's being "table" (tls, auth or acl)
|
||||
type Key struct {
|
||||
Name string `json:"key"`
|
||||
Kind string `json:"kind"`
|
||||
Default string `json:"default,omitempty"`
|
||||
Help string `json:"help,omitempty"`
|
||||
Required bool `json:"required,omitempty"`
|
||||
Enum []string `json:"enum,omitempty"`
|
||||
Min *float64 `json:"min,omitempty"`
|
||||
Max *float64 `json:"max,omitempty"`
|
||||
Zero string `json:"zero,omitempty"`
|
||||
Hint string `json:"hint,omitempty"`
|
||||
Side string `json:"side,omitempty"`
|
||||
}
|
||||
|
||||
var kinds = map[reflect.Kind]string{reflect.String: "string", reflect.Int64: "integer",
|
||||
reflect.Float64: "number", reflect.Bool: "bool", reflect.Slice: "list", reflect.Pointer: "table"}
|
||||
|
||||
// Stage is a flow stage: its key in the flow table, whether it is a list (the
|
||||
// filters), and its options
|
||||
type Stage struct {
|
||||
Key string `json:"key"`
|
||||
List bool `json:"list,omitempty"`
|
||||
Keys []Key `json:"keys"`
|
||||
}
|
||||
|
||||
// FlowStages lists the flow table's stages, in the order entries pass them
|
||||
func FlowStages() []Stage {
|
||||
var out []Stage
|
||||
t := reflect.TypeFor[FlowConfig]()
|
||||
for _, o := range options(t) {
|
||||
ft := t.Field(o.index).Type
|
||||
out = append(out, Stage{o.Name, ft.Kind() == reflect.Slice, keys(ft.Elem())})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// KeysOf describes the options of a table: tls, auth or acl
|
||||
func KeysOf[T any]() []Key { return keys(reflect.TypeFor[T]()) }
|
||||
|
||||
func keys(t reflect.Type) []Key {
|
||||
var out []Key
|
||||
for _, o := range options(t) {
|
||||
if o.Kind == "" {
|
||||
panic(fmt.Sprintf("%s.%s: no kind", t.Name(), o.Name))
|
||||
}
|
||||
out = append(out, o.Key)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// SettingKeys describes the settings by their dotted paths, with the defaults
|
||||
// lw has when it reads a file
|
||||
func SettingKeys() []Key { return settingKeys(reflect.ValueOf(defaults()).Elem(), "") }
|
||||
|
||||
func settingKeys(v reflect.Value, prefix string) []Key {
|
||||
var out []Key
|
||||
for _, o := range options(v.Type()) {
|
||||
switch f := v.Field(o.index); f.Kind() {
|
||||
case reflect.Pointer:
|
||||
out = append(out, settingKeys(f.Elem(), prefix+o.Name+".")...)
|
||||
case reflect.Slice: // the pipelines, which their flags set
|
||||
default:
|
||||
k := o.Key
|
||||
k.Name = prefix + k.Name
|
||||
if !f.IsZero() {
|
||||
k.Default = fmt.Sprint(f.Interface())
|
||||
}
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Decode reads a plugin's config map into its options: unknown keys, kinds,
|
||||
// the defaults, the tag rules, the tls, auth and acl rules of its side, then the
|
||||
// options' own Check. It reads no file.
|
||||
@@ -145,7 +226,7 @@ func decode(role, typ string, m map[string]any) (any, error) {
|
||||
if err := checkKeys(m, p.options, ""); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m = clone(m)
|
||||
m = Clone(m)
|
||||
if err := coerce(p.options, m, ""); err != nil {
|
||||
return nil, err // a file's wrong kind, named by its key
|
||||
}
|
||||
@@ -168,9 +249,8 @@ func decode(role, typ string, m map[string]any) (any, error) {
|
||||
// Settle gives a decoded table its defaults, applies the tag rules, then its
|
||||
// own Check: Decode's last steps, for the flow stages a file decodes typed
|
||||
func Settle(v any) error {
|
||||
rv := reflect.ValueOf(v).Elem()
|
||||
settle(rv)
|
||||
if err := checkTags(rv, ""); err != nil {
|
||||
Fill(v)
|
||||
if err := checkTags(reflect.ValueOf(v).Elem(), ""); err != nil {
|
||||
return err
|
||||
}
|
||||
if c, ok := v.(interface{ Check() error }); ok {
|
||||
@@ -179,15 +259,18 @@ func Settle(v any) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Fill gives a table its defaults and checks nothing: Settle's first step
|
||||
func Fill(v any) { settle(reflect.ValueOf(v).Elem()) }
|
||||
|
||||
// preset sets every default before decoding, so a switch that defaults on
|
||||
// stays on unless set off; it does so in the tables m sets
|
||||
func preset(v reflect.Value, m map[string]any) {
|
||||
for _, o := range options(v.Type()) {
|
||||
f := v.Field(o.index)
|
||||
if sub, ok := m[o.key].(map[string]any); ok && f.Kind() == reflect.Pointer && f.Type().Elem().Kind() == reflect.Struct {
|
||||
if sub, ok := m[o.Name].(map[string]any); ok && f.Kind() == reflect.Pointer && f.Type().Elem().Kind() == reflect.Struct {
|
||||
f.Set(reflect.New(f.Type().Elem()))
|
||||
preset(f.Elem(), sub)
|
||||
} else if o.def != "" {
|
||||
} else if o.Default != "" {
|
||||
setDefault(f, o)
|
||||
}
|
||||
}
|
||||
@@ -203,7 +286,7 @@ func settle(v reflect.Value) {
|
||||
if !f.IsNil() {
|
||||
settle(f.Elem())
|
||||
}
|
||||
case o.def != "" && o.zero == "" && f.Kind() != reflect.Bool && f.IsZero():
|
||||
case o.Default != "" && o.Zero == "" && f.Kind() != reflect.Bool && f.IsZero():
|
||||
setDefault(f, o)
|
||||
}
|
||||
}
|
||||
@@ -213,24 +296,24 @@ func setDefault(f reflect.Value, o option) {
|
||||
var err error
|
||||
switch f.Kind() {
|
||||
case reflect.String:
|
||||
f.SetString(o.def)
|
||||
f.SetString(o.Default)
|
||||
case reflect.Int64:
|
||||
var n int64
|
||||
n, err = strconv.ParseInt(o.def, 10, 64)
|
||||
n, err = strconv.ParseInt(o.Default, 10, 64)
|
||||
f.SetInt(n)
|
||||
case reflect.Float64:
|
||||
var n float64
|
||||
n, err = strconv.ParseFloat(o.def, 64)
|
||||
n, err = strconv.ParseFloat(o.Default, 64)
|
||||
f.SetFloat(n)
|
||||
case reflect.Bool:
|
||||
var b bool
|
||||
b, err = strconv.ParseBool(o.def)
|
||||
b, err = strconv.ParseBool(o.Default)
|
||||
f.SetBool(b)
|
||||
default:
|
||||
err = errors.New("no default for this kind")
|
||||
}
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("%s: default %q: %v", o.key, o.def, err))
|
||||
panic(fmt.Sprintf("%s: default %q: %v", o.Name, o.Default, err))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -243,12 +326,12 @@ func checkTags(v reflect.Value, prefix string) error {
|
||||
f := v.Field(o.index)
|
||||
var err error
|
||||
switch {
|
||||
case o.required && (f.IsZero() || f.Kind() == reflect.String && strings.TrimSpace(f.String()) == ""):
|
||||
case o.Required && (f.IsZero() || f.Kind() == reflect.String && strings.TrimSpace(f.String()) == ""):
|
||||
err = errors.New("must be set")
|
||||
case f.Kind() == reflect.String && len(o.enum) > 0 && (f.String() != "" || o.zero == "" && o.def == "") && !slices.Contains(o.enum, f.String()):
|
||||
err = fmt.Errorf("must be one of %s, got %q", strings.Join(o.enum, ", "), f.String())
|
||||
case f.Kind() == reflect.String && len(o.Enum) > 0 && (f.String() != "" || o.Zero == "" && o.Default == "") && !slices.Contains(o.Enum, f.String()):
|
||||
err = fmt.Errorf("must be one of %s, got %q", strings.Join(o.Enum, ", "), f.String())
|
||||
case f.CanFloat() || f.CanInt():
|
||||
if f.IsZero() && !o.required {
|
||||
if f.IsZero() && !o.Required {
|
||||
break
|
||||
}
|
||||
n := float64(0)
|
||||
@@ -260,16 +343,16 @@ func checkTags(v reflect.Value, prefix string) error {
|
||||
switch {
|
||||
case math.IsNaN(n) || math.IsInf(n, 0):
|
||||
err = errors.New("must be a finite number")
|
||||
case o.min != nil && o.max != nil && !(n >= *o.min && n <= *o.max):
|
||||
err = fmt.Errorf("must be from %s to %s, got %s", num(*o.min), num(*o.max), num(n))
|
||||
case o.min != nil && !(n >= *o.min):
|
||||
err = fmt.Errorf("must be at least %s, got %s", num(*o.min), num(n))
|
||||
case o.max != nil && !(n <= *o.max):
|
||||
err = fmt.Errorf("must be at most %s, got %s", num(*o.max), num(n))
|
||||
case o.Min != nil && o.Max != nil && !(n >= *o.Min && n <= *o.Max):
|
||||
err = fmt.Errorf("must be from %s to %s, got %s", num(*o.Min), num(*o.Max), num(n))
|
||||
case o.Min != nil && !(n >= *o.Min):
|
||||
err = fmt.Errorf("must be at least %s, got %s", num(*o.Min), num(n))
|
||||
case o.Max != nil && !(n <= *o.Max):
|
||||
err = fmt.Errorf("must be at most %s, got %s", num(*o.Max), num(n))
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return &KeyError{prefix + o.key, err}
|
||||
return &KeyError{prefix + o.Name, err}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
@@ -281,8 +364,8 @@ func num(n float64) string { return strconv.FormatFloat(n, 'f', -1, 64) }
|
||||
func sideKeys(t reflect.Type, side string) []string {
|
||||
var keys []string
|
||||
for _, o := range options(t) {
|
||||
if o.side == side {
|
||||
keys = append(keys, o.key)
|
||||
if o.Side == side {
|
||||
keys = append(keys, o.Name)
|
||||
}
|
||||
}
|
||||
return keys
|
||||
@@ -301,7 +384,7 @@ func Coerce(role, typ string, m map[string]any) error {
|
||||
|
||||
func coerce(t reflect.Type, m map[string]any, prefix string) error {
|
||||
for _, o := range options(t) {
|
||||
value, ok := m[o.key]
|
||||
value, ok := m[o.Name]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
@@ -310,25 +393,28 @@ func coerce(t reflect.Type, m map[string]any, prefix string) error {
|
||||
ft = ft.Elem()
|
||||
}
|
||||
if sub, ok := value.(map[string]any); ok && ft.Kind() == reflect.Struct {
|
||||
if err := coerce(ft, sub, prefix+o.key+"."); err != nil {
|
||||
if err := coerce(ft, sub, prefix+o.Name+"."); err != nil {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
}
|
||||
v, err := coerceValue(ft.Kind(), value)
|
||||
if err != nil {
|
||||
return &KeyError{prefix + o.key, err}
|
||||
return &KeyError{prefix + o.Name, err}
|
||||
}
|
||||
m[o.key] = v
|
||||
m[o.Name] = v
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// coerceValue converts text, and a JSON float for an integer; its errors
|
||||
// never hold the value, which may be a misplaced secret
|
||||
// coerceValue converts text, a JSON float for an integer and a TOML integer
|
||||
// for a number; its errors never hold the value, which may be a misplaced secret
|
||||
func coerceValue(kind reflect.Kind, value any) (any, error) {
|
||||
s, text := value.(string)
|
||||
n, integer := value.(int64)
|
||||
switch {
|
||||
case kind == reflect.Float64 && integer:
|
||||
return float64(n), nil
|
||||
case kind == reflect.Slice && text:
|
||||
return []any{s}, nil
|
||||
case kind == reflect.Int64 && text:
|
||||
@@ -357,14 +443,22 @@ func coerceValue(kind reflect.Kind, value any) (any, error) {
|
||||
return value, nil
|
||||
}
|
||||
|
||||
// clone copies m and the tables in it, so decoding leaves the caller's map be
|
||||
func clone(m map[string]any) map[string]any {
|
||||
// Clone copies an options map with its tables and lists, so decoding or an
|
||||
// edit leaves the caller's map be
|
||||
func Clone(m map[string]any) map[string]any {
|
||||
c := make(map[string]any, len(m))
|
||||
for k, v := range m {
|
||||
if sub, ok := v.(map[string]any); ok {
|
||||
v = clone(sub)
|
||||
}
|
||||
c[k] = v
|
||||
c[k] = cloneValue(v)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func cloneValue(v any) any {
|
||||
switch v := v.(type) {
|
||||
case map[string]any:
|
||||
return Clone(v)
|
||||
case []any:
|
||||
return slices.Clone(v)
|
||||
}
|
||||
return v
|
||||
}
|
||||
@@ -13,8 +13,8 @@ func TestEveryRowDecodesFromItsRequiredKeys(t *testing.T) {
|
||||
for _, p := range plugins {
|
||||
m := map[string]any{}
|
||||
for _, o := range options(p.options) {
|
||||
if o.required {
|
||||
m[o.key] = "1"
|
||||
if o.Required {
|
||||
m[o.Name] = "1"
|
||||
}
|
||||
}
|
||||
if err := Coerce(p.Role, p.Type, m); err != nil {
|
||||
|
||||
+193
-26
@@ -4,7 +4,9 @@ import (
|
||||
"cmp"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"os"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -17,11 +19,12 @@ type specKind struct {
|
||||
typed bool // SPEC starts with TYPE
|
||||
many bool // repeatable within a pipeline; the environment adds _1.._N
|
||||
first bool // starts its pipeline, and names it by TYPE unless --pipeline did
|
||||
apply func(p *PipelineConfig, typ string, opts map[string]any) error
|
||||
apply func(p *PipelineConfig, typ string, opts map[string]any, isDir IsDir) error
|
||||
write func(p *PipelineConfig) []string // its specs that build p's part; nil for preset
|
||||
}
|
||||
|
||||
var specKinds = []specKind{
|
||||
{"preset", true, false, true, func(p *PipelineConfig, typ string, opts map[string]any) error {
|
||||
{"preset", true, false, true, func(p *PipelineConfig, typ string, opts map[string]any, isDir IsDir) error {
|
||||
values := map[string][]string{}
|
||||
for k, v := range opts {
|
||||
switch v := v.(type) {
|
||||
@@ -35,9 +38,9 @@ var specKinds = []specKind{
|
||||
return fmt.Errorf("preset keys do not nest: %q", k)
|
||||
}
|
||||
}
|
||||
return applyPreset(p, typ, values)
|
||||
}},
|
||||
{"source", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any) error {
|
||||
return applyPreset(p, typ, values, isDir)
|
||||
}, nil},
|
||||
{"source", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any, _ IsDir) error {
|
||||
ids := make([]string, len(p.PluginSources))
|
||||
for i, s := range p.PluginSources {
|
||||
ids[i] = s.ID
|
||||
@@ -45,8 +48,13 @@ var specKinds = []specKind{
|
||||
id, err := instanceID(typ, opts, ids)
|
||||
p.PluginSources = append(p.PluginSources, PluginSourceConfig{ID: id, Type: typ, Config: opts})
|
||||
return cmp.Or(err, Coerce("source", typ, opts))
|
||||
}, func(p *PipelineConfig) (specs []string) {
|
||||
for _, s := range p.PluginSources {
|
||||
specs = append(specs, formatSpec(s.Type, map[string]any{"id": s.ID}, s.Config))
|
||||
}
|
||||
return specs
|
||||
}},
|
||||
{"sink", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any) error {
|
||||
{"sink", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any, _ IsDir) error {
|
||||
ids := make([]string, len(p.PluginSinks))
|
||||
for i, s := range p.PluginSinks {
|
||||
ids[i] = s.ID
|
||||
@@ -54,27 +62,61 @@ var specKinds = []specKind{
|
||||
id, err := instanceID(typ, opts, ids)
|
||||
p.PluginSinks = append(p.PluginSinks, PluginSinkConfig{ID: id, Type: typ, Config: opts})
|
||||
return cmp.Or(err, Coerce("sink", typ, opts))
|
||||
}, func(p *PipelineConfig) (specs []string) {
|
||||
for _, s := range p.PluginSinks {
|
||||
specs = append(specs, formatSpec(s.Type, map[string]any{"id": s.ID}, s.Config))
|
||||
}
|
||||
return specs
|
||||
}},
|
||||
{"filter", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any) error {
|
||||
{"filter", true, true, false, func(p *PipelineConfig, typ string, opts map[string]any, _ IsDir) error {
|
||||
opts["type"] = typ
|
||||
f := FilterConfig{}
|
||||
err := Scan(opts, &f)
|
||||
p.Flow.Filters = append(p.Flow.Filters, f)
|
||||
return err
|
||||
}, func(p *PipelineConfig) (specs []string) {
|
||||
for _, f := range p.Flow.Filters {
|
||||
Fill(&f)
|
||||
specs = append(specs, typedSpec(&f))
|
||||
}
|
||||
return specs
|
||||
}},
|
||||
{"format", true, false, false, func(p *PipelineConfig, typ string, opts map[string]any) error {
|
||||
{"format", true, false, false, func(p *PipelineConfig, typ string, opts map[string]any, _ IsDir) error {
|
||||
opts["type"] = typ
|
||||
p.Flow.Format = &FormatConfig{}
|
||||
return Scan(opts, p.Flow.Format)
|
||||
}, func(p *PipelineConfig) []string {
|
||||
if p.Flow.Format == nil {
|
||||
return nil
|
||||
}
|
||||
format := *p.Flow.Format
|
||||
Fill(&format)
|
||||
return []string{typedSpec(&format)}
|
||||
}},
|
||||
// Naming a stage turns it on: the file defaults (pass, disabled) would not.
|
||||
{"rate-limit", false, false, false, func(p *PipelineConfig, _ string, opts map[string]any) error {
|
||||
{"rate-limit", false, false, false, func(p *PipelineConfig, _ string, opts map[string]any, _ IsDir) error {
|
||||
p.Flow.RateLimit = &RateLimitConfig{Policy: "drop"}
|
||||
return Scan(opts, p.Flow.RateLimit)
|
||||
}, func(p *PipelineConfig) []string {
|
||||
if p.Flow.RateLimit == nil {
|
||||
return nil
|
||||
}
|
||||
rate := *p.Flow.RateLimit
|
||||
Fill(&rate) // the policy, whose default here is drop
|
||||
return []string{formatSpec("", Values(&rate))}
|
||||
}},
|
||||
{"heartbeat", false, false, false, func(p *PipelineConfig, _ string, opts map[string]any) error {
|
||||
{"heartbeat", false, false, false, func(p *PipelineConfig, _ string, opts map[string]any, _ IsDir) error {
|
||||
p.Flow.Heartbeat = &HeartbeatConfig{Enabled: true}
|
||||
return Scan(opts, p.Flow.Heartbeat)
|
||||
}, func(p *PipelineConfig) []string {
|
||||
if p.Flow.Heartbeat == nil {
|
||||
return nil
|
||||
}
|
||||
beat := *p.Flow.Heartbeat
|
||||
Fill(&beat)
|
||||
m := Values(&beat)
|
||||
m["enabled"] = beat.Enabled
|
||||
return []string{formatSpec("", m)}
|
||||
}},
|
||||
}
|
||||
|
||||
@@ -88,15 +130,28 @@ type pipelineSpec struct {
|
||||
// Spec is one pipeline flag: its name without dashes, and its value.
|
||||
type Spec struct{ Flag, Value string }
|
||||
|
||||
// SpecFlags names the pipeline flags, for the command-line parser.
|
||||
func SpecFlags() []string {
|
||||
flags := []string{"pipeline"}
|
||||
// SpecFlag is a pipeline flag, for the parser and lw --schema: its name
|
||||
// without dashes, its environment variable, which takes _1.._N when Many, and
|
||||
// whether SPEC starts with TYPE
|
||||
type SpecFlag struct {
|
||||
Flag string `json:"flag"`
|
||||
Variable string `json:"variable"`
|
||||
Typed bool `json:"typed"`
|
||||
Many bool `json:"many"`
|
||||
}
|
||||
|
||||
func SpecFlags() []SpecFlag {
|
||||
flags := []SpecFlag{{"pipeline", specVariable("pipeline"), false, false}}
|
||||
for _, k := range specKinds {
|
||||
flags = append(flags, k.flag)
|
||||
flags = append(flags, SpecFlag{k.flag, specVariable(k.flag), k.typed, k.many})
|
||||
}
|
||||
return flags
|
||||
}
|
||||
|
||||
func specVariable(flag string) string {
|
||||
return "LOGWISP_" + strings.ToUpper(strings.ReplaceAll(flag, "-", "_"))
|
||||
}
|
||||
|
||||
// cliSpecs looks up the kind of each command-line spec.
|
||||
func cliSpecs(in []Spec) ([]pipelineSpec, error) {
|
||||
var specs []pipelineSpec
|
||||
@@ -121,12 +176,12 @@ func cliSpecs(in []Spec) ([]pipelineSpec, error) {
|
||||
// counts as unset, so container templates may leave optional ones blank.
|
||||
func envPipelineSpecs() []pipelineSpec {
|
||||
var specs []pipelineSpec
|
||||
if name := os.Getenv("LOGWISP_PIPELINE"); name != "" {
|
||||
specs = append(specs, pipelineSpec{"LOGWISP_PIPELINE", nil, name})
|
||||
if name := os.Getenv(specVariable("pipeline")); name != "" {
|
||||
specs = append(specs, pipelineSpec{specVariable("pipeline"), nil, name})
|
||||
}
|
||||
for i := range specKinds {
|
||||
kind := &specKinds[i]
|
||||
base := "LOGWISP_" + strings.ToUpper(strings.ReplaceAll(kind.flag, "-", "_"))
|
||||
base := specVariable(kind.flag)
|
||||
type numbered struct {
|
||||
n uint64
|
||||
name, spec string
|
||||
@@ -150,9 +205,18 @@ func envPipelineSpecs() []pipelineSpec {
|
||||
return specs
|
||||
}
|
||||
|
||||
// SpecPipelines builds the pipelines of command-line specs, as Load does
|
||||
func SpecPipelines(specs []Spec, isDir IsDir) ([]PipelineConfig, error) {
|
||||
s, err := cliSpecs(specs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return buildPipelines(s, isDir)
|
||||
}
|
||||
|
||||
// buildPipelines turns specs, in order, into pipelines; specs before the first
|
||||
// --pipeline go to one named "cli". Every call returns fresh maps.
|
||||
func buildPipelines(specs []pipelineSpec) ([]PipelineConfig, error) {
|
||||
func buildPipelines(specs []pipelineSpec, isDir IsDir) ([]PipelineConfig, error) {
|
||||
var pipelines []PipelineConfig
|
||||
var seen map[*specKind]bool
|
||||
for _, s := range specs {
|
||||
@@ -181,7 +245,7 @@ func buildPipelines(specs []pipelineSpec) ([]PipelineConfig, error) {
|
||||
return nil, fmt.Errorf("%s %s: must start its pipeline", s.name, s.value)
|
||||
}
|
||||
seen[s.kind] = true
|
||||
if err := s.kind.apply(p, typ, opts); err != nil {
|
||||
if err := s.kind.apply(p, typ, opts, isDir); err != nil {
|
||||
return nil, fmt.Errorf("%s %s: %w", s.name, s.value, err)
|
||||
}
|
||||
}
|
||||
@@ -191,6 +255,104 @@ func buildPipelines(specs []pipelineSpec) ([]PipelineConfig, error) {
|
||||
return pipelines, nil
|
||||
}
|
||||
|
||||
// PipelineSpecs writes pipelines as the flags that build them: --pipeline,
|
||||
// ids, flow stage defaults and heartbeat's enabled explicit, as some flags'
|
||||
// defaults differ from a file's
|
||||
func PipelineSpecs(pipelines []PipelineConfig) []Spec {
|
||||
var specs []Spec
|
||||
for _, p := range pipelines {
|
||||
specs = append(specs, Spec{"pipeline", p.Name})
|
||||
p.Flow = cmp.Or(p.Flow, &FlowConfig{})
|
||||
for _, k := range specKinds {
|
||||
if k.write != nil {
|
||||
for _, value := range k.write(&p) {
|
||||
specs = append(specs, Spec{k.flag, value})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return specs
|
||||
}
|
||||
|
||||
// typedSpec writes a stage whose type key is its spec's TYPE
|
||||
func typedSpec(stage any) string {
|
||||
m := Values(stage)
|
||||
typ, _ := m["type"].(string)
|
||||
delete(m, "type")
|
||||
return formatSpec(typ, m)
|
||||
}
|
||||
|
||||
// formatSpec writes [TYPE,]key=value,...: tables as dotted keys, lists as a
|
||||
// repeated key
|
||||
func formatSpec(typ string, tables ...map[string]any) string {
|
||||
var parts []string
|
||||
if typ != "" {
|
||||
parts = append(parts, escape(typ))
|
||||
}
|
||||
var add func(prefix string, m map[string]any)
|
||||
add = func(prefix string, m map[string]any) {
|
||||
for _, k := range slices.Sorted(maps.Keys(m)) {
|
||||
switch v := m[k].(type) {
|
||||
case map[string]any:
|
||||
add(prefix+k+".", v)
|
||||
case []any:
|
||||
for _, e := range v {
|
||||
parts = append(parts, escape(prefix+k)+"="+escape(text(e)))
|
||||
}
|
||||
default:
|
||||
parts = append(parts, escape(prefix+k)+"="+escape(text(v)))
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, m := range tables {
|
||||
add("", m)
|
||||
}
|
||||
return strings.Join(parts, ",")
|
||||
}
|
||||
|
||||
func text(v any) string {
|
||||
if f, ok := v.(float64); ok {
|
||||
return strconv.FormatFloat(f, 'g', -1, 64)
|
||||
}
|
||||
return fmt.Sprint(v)
|
||||
}
|
||||
|
||||
// escape is unescape's inverse: ',' and '=' escaped, and '\' where an escape
|
||||
// would otherwise start, so a regex's \d stays as typed
|
||||
func escape(s string) string {
|
||||
var b strings.Builder
|
||||
followed := s + "," // as a value is, unless it ends the spec
|
||||
for i := range len(s) {
|
||||
if s[i] == ',' || s[i] == '=' || escapes(followed, i) {
|
||||
b.WriteByte('\\')
|
||||
}
|
||||
b.WriteByte(s[i])
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// Values gives a table's keys that are not zero, as a spec sets them, for the
|
||||
// composer to edit and Scan back
|
||||
func Values(v any) map[string]any {
|
||||
rv, m := reflect.ValueOf(v).Elem(), map[string]any{}
|
||||
for _, o := range options(rv.Type()) {
|
||||
switch f := rv.Field(o.index); {
|
||||
case f.IsZero():
|
||||
case f.Kind() == reflect.String:
|
||||
m[o.Name] = f.String()
|
||||
case f.Kind() == reflect.Slice:
|
||||
var list []any
|
||||
for i := range f.Len() {
|
||||
list = append(list, f.Index(i).Interface())
|
||||
}
|
||||
m[o.Name] = list
|
||||
default:
|
||||
m[o.Name] = f.Interface()
|
||||
}
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// useStdio makes a pipeline without a source read stdin, and one without a
|
||||
// sink write stdout, as a filter does.
|
||||
func useStdio(p *PipelineConfig) {
|
||||
@@ -225,7 +387,7 @@ func parseSpec(spec string, typed bool) (typ string, opts map[string]any, err er
|
||||
case typed && key == "type":
|
||||
err = errors.New(`TYPE already sets "type"`)
|
||||
default:
|
||||
err = setOption(opts, key, unescape(value))
|
||||
err = SetOption(opts, key, unescape(value))
|
||||
}
|
||||
if err != nil {
|
||||
return "", nil, fmt.Errorf("%s: %w", strings.TrimPrefix(typ+","+part, ","), err)
|
||||
@@ -262,8 +424,8 @@ func escapes(s string, i int) bool {
|
||||
return s[i] == '\\' && i+1 < len(s) && strings.IndexByte(`,=\`, s[i+1]) >= 0
|
||||
}
|
||||
|
||||
// setOption stores value at a dotted key; a second value turns it into a list.
|
||||
func setOption(opts map[string]any, key, value string) error {
|
||||
// SetOption stores value at a dotted key; a second value turns it into a list.
|
||||
func SetOption(opts map[string]any, key, value string) error {
|
||||
segments := strings.Split(key, ".")
|
||||
if slices.Contains(segments, "") {
|
||||
return fmt.Errorf("empty key in %q", key)
|
||||
@@ -293,8 +455,7 @@ func setOption(opts map[string]any, key, value string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// instanceID takes the id option; an unnamed instance gets the first of TYPE,
|
||||
// TYPE_2, TYPE_3... that no earlier instance of its role holds.
|
||||
// instanceID takes the id option, or FreeID for an unnamed instance
|
||||
func instanceID(typ string, opts map[string]any, taken []string) (string, error) {
|
||||
id, _ := opts["id"].(string)
|
||||
if _, set := opts["id"]; set && id == "" {
|
||||
@@ -304,9 +465,15 @@ func instanceID(typ string, opts map[string]any, taken []string) (string, error)
|
||||
if id != "" {
|
||||
return id, nil
|
||||
}
|
||||
id = typ
|
||||
return FreeID(typ, taken), nil
|
||||
}
|
||||
|
||||
// FreeID is the first of TYPE, TYPE_2, TYPE_3... that no earlier instance of
|
||||
// its role holds
|
||||
func FreeID(typ string, taken []string) string {
|
||||
id := typ
|
||||
for n := 2; slices.Contains(taken, id); n++ {
|
||||
id = typ + "_" + strconv.Itoa(n)
|
||||
}
|
||||
return id, nil
|
||||
return id
|
||||
}
|
||||
@@ -213,3 +213,17 @@ func TestReloadKeepsSpecPipelines(t *testing.T) {
|
||||
t.Fatalf("reload after the file was removed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Written specs build the pipelines a file would: each flow stage with its
|
||||
// defaults, so the rate limit keeps its file policy and the heartbeat its off
|
||||
func TestWrittenSpecsBuildWhatAFileWould(t *testing.T) {
|
||||
p := PipelineConfig{Name: "p", Flow: &FlowConfig{Format: &FormatConfig{}, RateLimit: &RateLimitConfig{Rate: 5}, Heartbeat: &HeartbeatConfig{}},
|
||||
PluginSources: []PluginSourceConfig{{ID: "in", Type: "null", Config: map[string]any{}}},
|
||||
PluginSinks: []PluginSinkConfig{{ID: "out", Type: "null", Config: map[string]any{}}}}
|
||||
got, err := SpecPipelines(PipelineSpecs([]PipelineConfig{p}), nil)
|
||||
p.Flow = &FlowConfig{Format: &FormatConfig{Type: "raw"}, RateLimit: &RateLimitConfig{Rate: 5, Policy: "pass"},
|
||||
Heartbeat: &HeartbeatConfig{IntervalMS: 1000, Format: "txt"}}
|
||||
if err != nil || !reflect.DeepEqual(got, []PipelineConfig{p}) {
|
||||
t.Fatalf("got %+v %v\nwant %+v", got, err, p)
|
||||
}
|
||||
}
|
||||
+18
-16
@@ -29,20 +29,6 @@ func ValidateConfig(cfg *Config) error {
|
||||
if cfg == nil {
|
||||
return fmt.Errorf("config is nil")
|
||||
}
|
||||
|
||||
if len(cfg.Pipelines) == 0 {
|
||||
return fmt.Errorf("no pipelines configured")
|
||||
}
|
||||
|
||||
// Reject duplicate pipeline names (service map is keyed by name)
|
||||
names := make(map[string]struct{}, len(cfg.Pipelines))
|
||||
for i, p := range cfg.Pipelines {
|
||||
if _, dup := names[p.Name]; dup {
|
||||
return fmt.Errorf("pipeline[%d]: duplicate name %q", i, p.Name)
|
||||
}
|
||||
names[p.Name] = struct{}{}
|
||||
}
|
||||
|
||||
if err := checkTags(reflect.ValueOf(cfg).Elem(), ""); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -57,8 +43,24 @@ func ValidateConfig(cfg *Config) error {
|
||||
}
|
||||
}
|
||||
|
||||
return ValidatePipelines(cfg.Pipelines)
|
||||
}
|
||||
|
||||
// ValidatePipelines is ValidateConfig's check of the pipelines
|
||||
func ValidatePipelines(pipelines []PipelineConfig) error {
|
||||
if len(pipelines) == 0 {
|
||||
return fmt.Errorf("no pipelines configured")
|
||||
}
|
||||
// Reject duplicate pipeline names (service map is keyed by name)
|
||||
names := make(map[string]struct{}, len(pipelines))
|
||||
for i, p := range pipelines {
|
||||
if _, dup := names[p.Name]; dup {
|
||||
return fmt.Errorf("pipeline[%d]: duplicate name %q", i, p.Name)
|
||||
}
|
||||
names[p.Name] = struct{}{}
|
||||
}
|
||||
single := map[string]string{} // role and type of a single-instance plugin: its pipeline
|
||||
for i, p := range cfg.Pipelines {
|
||||
for i, p := range pipelines {
|
||||
if err := lconfig.NonEmpty(p.Name); err != nil {
|
||||
return fmt.Errorf("pipeline[%d].name: %w", i, err)
|
||||
}
|
||||
@@ -196,7 +198,7 @@ func (o *TLSOptions) Check(listener bool) error {
|
||||
other = "listener"
|
||||
}
|
||||
for _, opt := range options(v.Type()) {
|
||||
if f := v.Field(opt.index); opt.side == other && !f.IsZero() && (f.Kind() != reflect.Slice || f.Len() > 0) {
|
||||
if f := v.Field(opt.index); opt.Side == other && !f.IsZero() && (f.Kind() != reflect.Slice || f.Len() > 0) {
|
||||
return fmt.Errorf("tls: %s apply to %ss", and(sideKeys(v.Type(), other)), other)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user