v0.38.0 lwconf updates for deployment
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
lwconf: sparse stages, every problem of a part, pastes read by form
This commit is contained in:
1 parent
15ea410a0a
commit
6a168212de
17 files changed
+376
-180
No files matched your search
+20
-17
@@ -69,28 +69,31 @@ type PipelineConfig struct {
|
||||
// --- Flow ---
|
||||
|
||||
// FlowConfig holds the stages between sources and sinks, in the order entries
|
||||
// pass them; the heartbeat runs beside them
|
||||
// pass them; the heartbeat runs beside them. Help is each stage's summary.
|
||||
type FlowConfig struct {
|
||||
RateLimit *RateLimitConfig `toml:"rate_limit"`
|
||||
Filters []FilterConfig `toml:"filters,omitempty"`
|
||||
Format *FormatConfig `toml:"format"`
|
||||
Heartbeat *HeartbeatConfig `toml:"heartbeat"`
|
||||
RateLimit *RateLimitConfig `toml:"rate_limit" help:"Drop entries beyond a rate, and those over a size"`
|
||||
Filters []FilterConfig `toml:"filters,omitempty" help:"Pass or drop entries by RE2 patterns, in order"`
|
||||
Format *FormatConfig `toml:"format" help:"How entries are written: raw, txt or json"`
|
||||
Heartbeat *HeartbeatConfig `toml:"heartbeat" help:"A keep-alive entry on a timer, beside the entries"`
|
||||
}
|
||||
|
||||
// --- Flow stages ---
|
||||
//
|
||||
// A stage key's zero is its unset, the default or what zero means there, so
|
||||
// a stage is written with the keys set alone.
|
||||
|
||||
// HeartbeatConfig defines settings for periodic keep-alive or status messages
|
||||
type HeartbeatConfig struct {
|
||||
Enabled bool `toml:"enabled" help:"send heartbeats; false keeps these settings and sends none"`
|
||||
IntervalMS int64 `toml:"interval_ms" default:"1000" lw:"min=100" help:"milliseconds between heartbeats"`
|
||||
IncludeTimestamp bool `toml:"include_timestamp" help:"stamp each heartbeat with its time"`
|
||||
IncludeStats bool `toml:"include_stats" help:"add type, beat_count and, from the second beat, the measured interval_ms"`
|
||||
Format string `toml:"format" default:"txt" lw:"enum=txt|json|raw" help:"kept for older files; heartbeats are written in the flow's format"`
|
||||
Enabled bool `toml:"enabled,omitempty" help:"send heartbeats; false keeps these settings and sends none"`
|
||||
IntervalMS int64 `toml:"interval_ms,omitempty" default:"1000" lw:"min=100" help:"milliseconds between heartbeats"`
|
||||
IncludeTimestamp bool `toml:"include_timestamp,omitempty" help:"stamp each heartbeat with its time"`
|
||||
IncludeStats bool `toml:"include_stats,omitempty" help:"add type, beat_count and, from the second beat, the measured interval_ms"`
|
||||
Format string `toml:"format,omitempty" default:"txt" lw:"enum=txt|json|raw" help:"kept for older files; heartbeats are written in the flow's format"`
|
||||
}
|
||||
|
||||
// FormatConfig is how a pipeline writes its entries
|
||||
type FormatConfig struct {
|
||||
Type string `toml:"type" default:"raw" lw:"enum=raw|txt|json" help:"raw writes each message as it came, txt a line with time and level, json one object an entry"`
|
||||
Type string `toml:"type,omitempty" default:"raw" lw:"enum=raw|txt|json" help:"raw writes each message as it came, txt a line with time and level, json one object an entry"`
|
||||
Flags int64 `toml:"flags,omitempty" lw:"zero=by type" help:"for experts, a sum of bits over the type's choice: 1 writes messages verbatim, 16 drops the time, 32 the level"`
|
||||
TimestampFormat string `toml:"timestamp_format,omitempty" lw:"zero=RFC 3339 with nanoseconds" help:"timestamp layout: the reference time 2006-01-02 15:04:05 written as timestamps should look, such as 02 Jan 15:04"`
|
||||
SanitizerPolicy string `toml:"sanitizer_policy,omitempty" lw:"enum=raw|json|txt|shell,zero=by type" help:"how control characters are escaped"`
|
||||
@@ -108,10 +111,10 @@ const (
|
||||
|
||||
// RateLimitConfig defines the configuration for pipeline-level rate limiting
|
||||
type RateLimitConfig struct {
|
||||
EntriesPerSecond float64 `toml:"entries_per_second" lw:"min=0,zero=off" help:"entries a second let through; unset, the stage limits nothing, sizes included"`
|
||||
BurstEntries float64 `toml:"burst_entries" lw:"min=0,zero=one second's worth" help:"entries that may pass in a burst before entries_per_second holds them back"`
|
||||
Policy string `toml:"policy" default:"pass" lw:"enum=pass|drop" help:"drop enforces entries_per_second and max_entry_size_bytes; pass lets every entry through"`
|
||||
MaxEntrySizeBytes int64 `toml:"max_entry_size_bytes" lw:"min=0,zero=unlimited" help:"drop larger entries; needs entries_per_second and policy drop"`
|
||||
EntriesPerSecond float64 `toml:"entries_per_second,omitempty" lw:"min=0,zero=off" help:"entries a second let through; unset, the stage limits nothing, sizes included"`
|
||||
BurstEntries float64 `toml:"burst_entries,omitempty" lw:"min=0,zero=one second's worth" help:"entries that may pass in a burst before entries_per_second holds them back"`
|
||||
Policy string `toml:"policy,omitempty" default:"pass" lw:"enum=pass|drop" help:"drop enforces entries_per_second and max_entry_size_bytes; pass lets every entry through"`
|
||||
MaxEntrySizeBytes int64 `toml:"max_entry_size_bytes,omitempty" lw:"min=0,zero=unlimited" help:"drop larger entries; needs entries_per_second and policy drop"`
|
||||
}
|
||||
|
||||
// FilterType represents the filter's behavior (include or exclude)
|
||||
@@ -136,8 +139,8 @@ const (
|
||||
|
||||
// FilterConfig represents the configuration for a single filter
|
||||
type FilterConfig struct {
|
||||
Type FilterType `toml:"type" default:"include" lw:"enum=include|exclude" help:"pass only matching entries, or drop them"`
|
||||
Logic FilterLogic `toml:"logic" default:"or" lw:"enum=or|and" help:"match any pattern, or all"`
|
||||
Type FilterType `toml:"type,omitempty" default:"include" lw:"enum=include|exclude" help:"pass only matching entries, or drop them"`
|
||||
Logic FilterLogic `toml:"logic,omitempty" default:"or" lw:"enum=or|and" help:"match any pattern, or all"`
|
||||
Patterns []string `toml:"patterns,omitempty" lw:"required,hint=regex" help:"RE2 patterns"`
|
||||
}
|
||||
|
||||
|
||||
+36
-36
@@ -172,14 +172,15 @@ type Key struct {
|
||||
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), its options, and the values that turn it on, which a file's
|
||||
// defaults leave off
|
||||
// Stage is a flow stage: its key in the flow table, what it does, whether it
|
||||
// is a list (the filters), its options, and the values that turn it on, which
|
||||
// a file's defaults leave off
|
||||
type Stage struct {
|
||||
Key string `json:"key"`
|
||||
List bool `json:"list,omitempty"`
|
||||
Keys []Key `json:"keys"`
|
||||
On map[string]any `json:"on,omitempty"`
|
||||
Key string `json:"key"`
|
||||
Summary string `json:"summary"`
|
||||
List bool `json:"list,omitempty"`
|
||||
Keys []Key `json:"keys"`
|
||||
On map[string]any `json:"on,omitempty"`
|
||||
}
|
||||
|
||||
// FlowStages lists the flow table's stages, in the order entries pass them
|
||||
@@ -188,7 +189,7 @@ func FlowStages() []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()), maps.Clone(stageOn[o.Name])})
|
||||
out = append(out, Stage{o.Name, o.Help, ft.Kind() == reflect.Slice, keys(ft.Elem()), maps.Clone(stageOn[o.Name])})
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -232,8 +233,8 @@ func settingKeys(v reflect.Value, prefix string) []Key {
|
||||
|
||||
// Decode reads a plugin's config map into its options: unknown keys, kinds,
|
||||
// the defaults, the tag rules and the options' own Check, then the host, tls,
|
||||
// acl and auth rules of its side; a leading ~ in a path is the home directory.
|
||||
// It reads no file.
|
||||
// acl and auth rules of its side, its error joining every problem found; a
|
||||
// leading ~ in a path is the home directory. It reads no file.
|
||||
func Decode[T any](role, typ string, m map[string]any) (*T, error) {
|
||||
v, err := decode(role, typ, m)
|
||||
if err != nil {
|
||||
@@ -259,30 +260,27 @@ func decode(role, typ string, m map[string]any) (any, error) {
|
||||
if err := lconfig.ScanMap(m, v.Interface()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := Settle(v.Interface()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
errs := []error{Settle(v.Interface())}
|
||||
expandPaths(v.Elem())
|
||||
if n, ok := v.Interface().(networked); ok {
|
||||
if err := p.checkNetwork(n); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
errs = append(errs, p.checkNetwork(n))
|
||||
}
|
||||
if err := errors.Join(errs...); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return v.Interface(), nil
|
||||
}
|
||||
|
||||
// Settle gives a decoded table its defaults, applies the tag rules, then its
|
||||
// own Check, as Decode does before its side's rules, for the flow stages a
|
||||
// file decodes typed
|
||||
// Settle gives a decoded table its defaults, applies the tag rules and its own
|
||||
// Check, as Decode does before its side's rules, for the flow stages a file
|
||||
// decodes typed; its error joins every problem
|
||||
func Settle(v any) error {
|
||||
Fill(v)
|
||||
if err := checkTags(reflect.ValueOf(v).Elem(), ""); err != nil {
|
||||
return err
|
||||
}
|
||||
err := checkTags(reflect.ValueOf(v).Elem(), "")
|
||||
if c, ok := v.(interface{ Check() error }); ok {
|
||||
return c.Check()
|
||||
err = errors.Join(err, c.Check())
|
||||
}
|
||||
return nil
|
||||
return err
|
||||
}
|
||||
|
||||
// Fill gives a table its defaults and checks nothing: Settle's first step
|
||||
@@ -369,11 +367,12 @@ func setDefault(f reflect.Value, o option) {
|
||||
}
|
||||
}
|
||||
|
||||
// checkTags applies the tag rules to a table's own keys, naming each by
|
||||
// prefix and key. A required list needs an entry. A zero number is the
|
||||
// default or its zero meaning, and meets the range unless required; an empty
|
||||
// enum is one only by its zero or default.
|
||||
// checkTags applies the tag rules to a table's own keys, joining a problem a
|
||||
// key, each named by prefix and key. A required list needs an entry. A zero
|
||||
// number is the default or its zero meaning, and meets the range unless
|
||||
// required; an empty enum is one only by its zero or default.
|
||||
func checkTags(v reflect.Value, prefix string) error {
|
||||
var errs []error
|
||||
for _, o := range options(v.Type()) {
|
||||
f := v.Field(o.index)
|
||||
var err error
|
||||
@@ -404,10 +403,10 @@ func checkTags(v reflect.Value, prefix string) error {
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return &KeyError{prefix + o.Name, err}
|
||||
errs = append(errs, &KeyError{prefix + o.Name, err})
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
func num(n float64) string { return strconv.FormatFloat(n, 'f', -1, 64) }
|
||||
@@ -425,7 +424,8 @@ func sideKeys(t reflect.Type, side string) []string {
|
||||
|
||||
// Coerce gives command-line and environment values, which are strings, and
|
||||
// JSON's floats the kinds a plugin's options declare, so a dump prints them
|
||||
// typed. Keys and types it does not know are Decode's to refuse.
|
||||
// typed; its error joins a key each it cannot. Keys and types it does not
|
||||
// know are Decode's to refuse.
|
||||
func Coerce(role, typ string, m map[string]any) error {
|
||||
p, ok := LookupPlugin(role, typ)
|
||||
if !ok {
|
||||
@@ -435,6 +435,7 @@ func Coerce(role, typ string, m map[string]any) error {
|
||||
}
|
||||
|
||||
func coerce(t reflect.Type, m map[string]any, prefix string) error {
|
||||
var errs []error
|
||||
for _, o := range options(t) {
|
||||
value, ok := m[o.Name]
|
||||
if !ok {
|
||||
@@ -445,18 +446,17 @@ 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.Name+"."); err != nil {
|
||||
return err
|
||||
}
|
||||
errs = append(errs, coerce(ft, sub, prefix+o.Name+"."))
|
||||
continue
|
||||
}
|
||||
v, err := coerceValue(ft.Kind(), value)
|
||||
if err != nil {
|
||||
return &KeyError{prefix + o.Name, err}
|
||||
errs = append(errs, &KeyError{prefix + o.Name, err})
|
||||
continue
|
||||
}
|
||||
m[o.Name] = v
|
||||
}
|
||||
return nil
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// coerceValue converts text, a JSON float for an integer and a TOML integer
|
||||
|
||||
@@ -93,7 +93,7 @@ func TestCoerceGivesTheDeclaredKinds(t *testing.T) {
|
||||
|
||||
// ValidateConfig decodes every plugin and flow stage without opening a file,
|
||||
// and names the path of what it refuses: a rule between a table's keys by
|
||||
// the table's
|
||||
// the table's, a list entry by its index
|
||||
func TestValidateConfigNamesThePath(t *testing.T) {
|
||||
valid := func() *Config {
|
||||
return &Config{Color: "auto", Pipelines: []PipelineConfig{{
|
||||
@@ -114,6 +114,9 @@ func TestValidateConfigNamesThePath(t *testing.T) {
|
||||
"pipelines[0].plugin_sinks[out].config.tls: client_auth requires client_ca_file": func(c *Config) {
|
||||
c.Pipelines[0].PluginSinks[0].Config["tls"].(map[string]any)["client_auth"] = true
|
||||
},
|
||||
`pipelines[0].plugin_sinks[out].config.acl.allow[1]: "10.0.0.0.1": not an IP address or CIDR; host names and IPv6 zones are not accepted`: func(c *Config) {
|
||||
c.Pipelines[0].PluginSinks[0].Config["acl"] = map[string]any{"allow": []any{"10.0.0.0/8", "10.0.0.0.1"}}
|
||||
},
|
||||
"pipelines[0].plugin_sinks[out].config.acl: requests_per_second_per_client applies only to HTTP listeners": func(c *Config) {
|
||||
c.Pipelines[0].PluginSinks[0].Type = "tcp"
|
||||
c.Pipelines[0].PluginSinks[0].Config["acl"] = map[string]any{"requests_per_second_per_client": 2.5}
|
||||
@@ -152,19 +155,25 @@ func TestValidateConfigNamesThePath(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A configuration that does not load names every problem, a line each, each
|
||||
// part's first: the settings, each flow stage, each plugin, each pipeline
|
||||
// A configuration that does not load names every problem, a line each: the
|
||||
// settings, each flow stage, each plugin and each pipeline, every key's of a
|
||||
// part, and the first rule between a table's keys
|
||||
func TestLoadReportsEveryProblemALine(t *testing.T) {
|
||||
isolateConfig(t)
|
||||
_, err := Load(Args{Overrides: []string{"--color=bogus"}, Specs: specs("pipeline", "a", "source", "file",
|
||||
"sink", "tcp:port=70000", "rate-limit", "policy=bogus", "filter", "include:patterns=(",
|
||||
_, err := Load(Args{Overrides: []string{"--color=bogus"}, Specs: specs("pipeline", "a", "source", "file:from=middle",
|
||||
"sink", "tcp:port=70000,buffer_size=-1,tls.enabled=true,acl.allow=x", "rate-limit", "policy=bogus", "filter", "include:patterns=(,logic=nor",
|
||||
"pipeline", "a", "sink", "null")})
|
||||
want := strings.Join([]string{"invalid configuration:",
|
||||
`color: must be one of auto, always, never, got "bogus"`,
|
||||
`pipelines[0].flow.rate_limit.policy: must be one of pass, drop, got "bogus"`,
|
||||
`pipelines[0].flow.filters[0].logic: must be one of or, and, got "nor"`,
|
||||
"pipelines[0].flow.filters[0].patterns[0]: \"(\": error parsing regexp: missing closing ): `(`",
|
||||
"pipelines[0].plugin_sources[file].config.directory: must be set",
|
||||
`pipelines[0].plugin_sources[file].config.from: must be one of start, end, got "middle"`,
|
||||
"pipelines[0].plugin_sinks[tcp].config.port: must be from 1 to 65535, got 70000",
|
||||
"pipelines[0].plugin_sinks[tcp].config.buffer_size: must be at least 1, got -1",
|
||||
`pipelines[0].plugin_sinks[tcp].config.acl.allow[0]: "x": not an IP address or CIDR; host names and IPv6 zones are not accepted`,
|
||||
"pipelines[0].plugin_sinks[tcp].config.tls: listeners need cert_file and key_file, self_signed, or issuer_cert_file and issuer_key_file",
|
||||
`pipelines[1].name: duplicate name "a"`,
|
||||
}, "\n")
|
||||
if err == nil || err.Error() != want {
|
||||
|
||||
+54
-15
@@ -48,8 +48,9 @@ var specKinds = []specKind{
|
||||
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) {
|
||||
var ids []string
|
||||
for _, s := range p.PluginSources {
|
||||
specs = append(specs, formatSpec(s.Type, map[string]any{"id": s.ID}, s.Config))
|
||||
specs, ids = append(specs, formatSpec(s.Type, givenID(s.ID, s.Type, ids), s.Config)), append(ids, s.ID)
|
||||
}
|
||||
return specs
|
||||
}},
|
||||
@@ -62,8 +63,9 @@ var specKinds = []specKind{
|
||||
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) {
|
||||
var ids []string
|
||||
for _, s := range p.PluginSinks {
|
||||
specs = append(specs, formatSpec(s.Type, map[string]any{"id": s.ID}, s.Config))
|
||||
specs, ids = append(specs, formatSpec(s.Type, givenID(s.ID, s.Type, ids), s.Config)), append(ids, s.ID)
|
||||
}
|
||||
return specs
|
||||
}},
|
||||
@@ -75,7 +77,6 @@ var specKinds = []specKind{
|
||||
return err
|
||||
}, func(p *PipelineConfig) (specs []string) {
|
||||
for _, f := range p.Flow.Filters {
|
||||
Fill(&f)
|
||||
specs = append(specs, typedSpec(&f))
|
||||
}
|
||||
return specs
|
||||
@@ -89,7 +90,6 @@ var specKinds = []specKind{
|
||||
return nil
|
||||
}
|
||||
format := *p.Flow.Format
|
||||
Fill(&format)
|
||||
return []string{typedSpec(&format)}
|
||||
}},
|
||||
{"rate-limit", false, false, false, func(p *PipelineConfig, _ string, opts map[string]any, _ IsDir) error {
|
||||
@@ -100,8 +100,10 @@ var specKinds = []specKind{
|
||||
return nil
|
||||
}
|
||||
rate := *p.Flow.RateLimit
|
||||
Fill(&rate) // the policy, whose default here is drop
|
||||
return []string{formatSpec("", Values(&rate))}
|
||||
m := Values(&rate)
|
||||
Fill(&rate)
|
||||
m["policy"] = rate.Policy // a spec's default is drop, a file's pass
|
||||
return []string{formatSpec("", m)}
|
||||
}},
|
||||
{"heartbeat", false, false, false, func(p *PipelineConfig, _ string, opts map[string]any, _ IsDir) error {
|
||||
p.Flow.Heartbeat = &HeartbeatConfig{}
|
||||
@@ -111,7 +113,6 @@ var specKinds = []specKind{
|
||||
return nil
|
||||
}
|
||||
beat := *p.Flow.Heartbeat
|
||||
Fill(&beat)
|
||||
m := Values(&beat)
|
||||
m["enabled"] = beat.Enabled
|
||||
return []string{formatSpec("", m)}
|
||||
@@ -137,6 +138,18 @@ func StageOn(stage string, opts map[string]any) map[string]any {
|
||||
return on
|
||||
}
|
||||
|
||||
// FillOn gives the keys a spec of the stage turns on their file defaults
|
||||
// where unset, as that spec writes them: a stage from a file and from its
|
||||
// spec are then one
|
||||
func FillOn(stage string, v any) {
|
||||
rv := reflect.ValueOf(v).Elem()
|
||||
for _, o := range options(rv.Type()) {
|
||||
if _, on := stageOn[stage][o.Name]; on && o.Default != "" && rv.Field(o.index).IsZero() {
|
||||
setDefault(rv.Field(o.index), o)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Spec is one pipeline flag: its name without dashes, and its value.
|
||||
type Spec struct{ Flag, Value string }
|
||||
|
||||
@@ -219,13 +232,22 @@ func envPipelineSpecs(environ []string) []pipelineSpec {
|
||||
}
|
||||
|
||||
// EnvPipelines builds the pipeline of LOGWISP_ variables, NAME=value each, as
|
||||
// Load does without pipeline flags
|
||||
func EnvPipelines(environ []string, isDir IsDir) ([]PipelineConfig, error) {
|
||||
// Load does without pipeline flags; unread are the other LOGWISP_ variables
|
||||
// set, sorted: settings, or names no flag has
|
||||
func EnvPipelines(environ []string, isDir IsDir) (pipelines []PipelineConfig, unread []string, err error) {
|
||||
specs := envPipelineSpecs(environ)
|
||||
if len(specs) == 0 {
|
||||
return nil, errors.New("no LOGWISP_ pipeline variables")
|
||||
for _, kv := range environ {
|
||||
name, value, _ := strings.Cut(kv, "=")
|
||||
if strings.HasPrefix(name, "LOGWISP_") && value != "" && !slices.ContainsFunc(specs, func(s pipelineSpec) bool { return s.name == name }) {
|
||||
unread = append(unread, name)
|
||||
}
|
||||
}
|
||||
return buildPipelines(specs, isDir)
|
||||
slices.Sort(unread)
|
||||
if len(specs) == 0 {
|
||||
return nil, unread, errors.New("no LOGWISP_ pipeline variables")
|
||||
}
|
||||
pipelines, err = buildPipelines(specs, isDir)
|
||||
return pipelines, unread, err
|
||||
}
|
||||
|
||||
// SpecPipelines builds the pipelines of command-line specs, as Load does
|
||||
@@ -280,8 +302,9 @@ func buildPipelines(specs []pipelineSpec, isDir IsDir) ([]PipelineConfig, error)
|
||||
}
|
||||
|
||||
// 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
|
||||
// the ids and keys set, and a typed stage's type, the rate limit's policy and
|
||||
// the heartbeat's enabled explicit, as those flags' defaults differ from a
|
||||
// file's
|
||||
func PipelineSpecs(pipelines []PipelineConfig) []Spec {
|
||||
var specs []Spec
|
||||
for _, p := range pipelines {
|
||||
@@ -298,14 +321,30 @@ func PipelineSpecs(pipelines []PipelineConfig) []Spec {
|
||||
return specs
|
||||
}
|
||||
|
||||
// typedSpec writes a stage whose type key is its spec's TYPE
|
||||
// typedSpec writes a stage whose type key is its spec's TYPE: the type set,
|
||||
// or its default
|
||||
func typedSpec(stage any) string {
|
||||
m := Values(stage)
|
||||
if _, set := m["type"]; !set {
|
||||
filled := reflect.New(reflect.TypeOf(stage).Elem())
|
||||
filled.Elem().Set(reflect.ValueOf(stage).Elem())
|
||||
Fill(filled.Interface())
|
||||
m["type"] = Values(filled.Interface())["type"]
|
||||
}
|
||||
typ, _ := m["type"].(string)
|
||||
delete(m, "type")
|
||||
return formatSpec(typ, m)
|
||||
}
|
||||
|
||||
// givenID is an instance's id as its spec gives it: none for the one
|
||||
// instanceID would take, after the instances before
|
||||
func givenID(id, typ string, before []string) map[string]any {
|
||||
if id == FreeID(typ, before) {
|
||||
return nil
|
||||
}
|
||||
return map[string]any{"id": id}
|
||||
}
|
||||
|
||||
// formatSpec writes TYPE[:key=value,...] or key=value,...: tables as dotted
|
||||
// keys, lists as a repeated key
|
||||
func formatSpec(typ string, tables ...map[string]any) string {
|
||||
|
||||
@@ -216,16 +216,21 @@ func TestReloadKeepsSpecPipelines(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
// Written specs build the pipelines a file would, with the keys it set: a
|
||||
// typed stage names its type, the rate limit keeps its file policy and the
|
||||
// heartbeat its off, and an id is written only where a spec would not take it
|
||||
func TestWrittenSpecsBuildWhatAFileWould(t *testing.T) {
|
||||
p := PipelineConfig{Name: "p", Flow: &FlowConfig{Format: &FormatConfig{}, RateLimit: &RateLimitConfig{EntriesPerSecond: 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{EntriesPerSecond: 5, Policy: "pass"},
|
||||
Heartbeat: &HeartbeatConfig{IntervalMS: 1000, Format: "txt"}}
|
||||
PluginSinks: []PluginSinkConfig{{ID: "null_2", Type: "null", Config: map[string]any{}}, {ID: "null", Type: "null", Config: map[string]any{}}}}
|
||||
specs := PipelineSpecs([]PipelineConfig{p})
|
||||
got, err := SpecPipelines(specs, nil)
|
||||
p.Flow = &FlowConfig{Format: &FormatConfig{Type: "raw"}, RateLimit: &RateLimitConfig{EntriesPerSecond: 5, Policy: "pass"}, Heartbeat: &HeartbeatConfig{}}
|
||||
if err != nil || !reflect.DeepEqual(got, []PipelineConfig{p}) {
|
||||
t.Fatalf("got %+v %v\nwant %+v", got, err, p)
|
||||
}
|
||||
if want := []Spec{{"pipeline", "p"}, {"source", "null:id=in"}, {"sink", "null:id=null_2"}, {"sink", "null"}, {"format", "raw"},
|
||||
{"rate-limit", "entries_per_second=5,policy=pass"}, {"heartbeat", "enabled=false"}}; !reflect.DeepEqual(specs, want) {
|
||||
t.Fatalf("specs %v, want %v", specs, want)
|
||||
}
|
||||
}
|
||||
+60
-40
@@ -24,9 +24,9 @@ import (
|
||||
|
||||
// ValidateConfig checks a configuration without reading a file it names:
|
||||
// the settings, then each pipeline's flow stages, and its sources and sinks
|
||||
// decoded in full. Its error joins every problem, one a part, each a
|
||||
// *KeyError naming the refused key's path, save "no pipelines configured".
|
||||
// lw --check opens the files.
|
||||
// decoded in full. Its error joins every problem, each a *KeyError naming the
|
||||
// refused key's path, save "no pipelines configured". lw --check opens the
|
||||
// files.
|
||||
func ValidateConfig(cfg *Config) error {
|
||||
if cfg == nil {
|
||||
return fmt.Errorf("config is nil")
|
||||
@@ -99,8 +99,8 @@ func ValidatePipelines(pipelines []PipelineConfig) error {
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// checkFlow settles copies of a pipeline's flow stages, which keep their
|
||||
// defaults for the dump: a problem a stage
|
||||
// checkFlow settles copies of a pipeline's flow stages, which keep the keys
|
||||
// set for the dump
|
||||
func checkFlow(f *FlowConfig, path string) error {
|
||||
if f == nil {
|
||||
return nil
|
||||
@@ -126,8 +126,16 @@ func checkFlow(f *FlowConfig, path string) error {
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// At addresses err to the path of the table it was found in
|
||||
// At addresses err to the path of the table it was found in, each problem of
|
||||
// a joined one
|
||||
func At(path string, err error) error {
|
||||
if joined, ok := err.(interface{ Unwrap() []error }); ok {
|
||||
var errs []error
|
||||
for _, e := range joined.Unwrap() {
|
||||
errs = append(errs, At(path, e))
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
if ke, ok := errors.AsType[*KeyError](err); ok {
|
||||
return &KeyError{path + "." + ke.Key, ke.Err}
|
||||
}
|
||||
@@ -167,19 +175,17 @@ func (o *HTTPChainSinkOptions) network() (string, *TLSOptions, *AuthOptions, *AC
|
||||
return o.Host, o.TLS, o.Auth, nil
|
||||
}
|
||||
|
||||
// checkNetwork applies the host, tls, acl and auth rules of the plugin's side
|
||||
// checkNetwork applies the host, tls, acl and auth rules of the plugin's
|
||||
// side, joining their problems; acl entries are read by the host's family
|
||||
func (p Plugin) checkNetwork(n networked) error {
|
||||
host, t, a, acl := n.network()
|
||||
var errs []error
|
||||
if _, err := core.Network(host); err != nil {
|
||||
return &KeyError{"host", err}
|
||||
errs = append(errs, &KeyError{"host", err})
|
||||
} else {
|
||||
errs = append(errs, table("acl", acl.Check(host, p.HTTP, a != nil && len(a.TrustedProxies) > 0)))
|
||||
}
|
||||
if err := t.Check(p.Side != Dialer); err != nil {
|
||||
return table("tls", err)
|
||||
}
|
||||
if err := acl.Check(host, p.HTTP, a != nil && len(a.TrustedProxies) > 0); err != nil {
|
||||
return table("acl", err)
|
||||
}
|
||||
return table("auth", a.Check(p.Side, p.HTTP, t.State()))
|
||||
return errors.Join(append(errs, table("tls", t.Check(p.Side != Dialer)), table("auth", a.Check(p.Side, p.HTTP, t.State())))...)
|
||||
}
|
||||
|
||||
// Check holds the rules between tls keys on a listener or a dialer; a table
|
||||
@@ -221,8 +227,9 @@ func (o *TLSOptions) Check(listener bool) error {
|
||||
return fmt.Errorf("tls: cert_file and key_file must be set together")
|
||||
}
|
||||
if o.PinSHA256 != "" {
|
||||
_, err := ParsePins(o.PinSHA256)
|
||||
return err
|
||||
if _, err := ParsePins(o.PinSHA256); err != nil {
|
||||
return &KeyError{"tls.pin_sha256", err}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -237,7 +244,7 @@ func ParsePins(s string) ([][]byte, error) {
|
||||
b64, ok := strings.CutPrefix(strings.TrimSpace(p), PinPrefix)
|
||||
h, err := base64.StdEncoding.DecodeString(b64)
|
||||
if !ok || err != nil || len(h) != sha256.Size {
|
||||
return nil, fmt.Errorf("tls: pin_sha256 %q: want %sBASE64 of a SHA-256, ';' between several", p, PinPrefix)
|
||||
return nil, fmt.Errorf("pin %q: want %sBASE64 of a SHA-256, ';' between several", p, PinPrefix)
|
||||
}
|
||||
pins = append(pins, h)
|
||||
}
|
||||
@@ -290,7 +297,7 @@ func (o *AuthOptions) Check(side Side, http bool, t TLSState) error {
|
||||
case o.Type == "mtls":
|
||||
for i, pattern := range o.AllowPatterns {
|
||||
if _, err := regexp.Compile(pattern); err != nil {
|
||||
return fmt.Errorf("auth: allow_patterns[%d] %q: %w", i, pattern, err)
|
||||
return &KeyError{fmt.Sprintf("auth.allow_patterns[%d]", i), fmt.Errorf("%q: %w", pattern, err)}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
@@ -320,9 +327,9 @@ func (o *AuthOptions) Check(side Side, http bool, t TLSState) error {
|
||||
// Binds the certificate to the user: a peer needs its own of both
|
||||
return fmt.Errorf("auth: identity under type %q binds the client certificate to the user and requires tls.client_auth", "scram")
|
||||
}
|
||||
for _, e := range o.TrustedProxies {
|
||||
for i, e := range o.TrustedProxies {
|
||||
if _, err := ParsePrefix(e, "tcp"); err != nil {
|
||||
return fmt.Errorf("auth: trusted_proxies entry %q: %w", e, err)
|
||||
return &KeyError{fmt.Sprintf("auth.trusted_proxies[%d]", i), fmt.Errorf("%q: %w", e, err)}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
@@ -357,7 +364,8 @@ func (o *AuthOptions) scramKey() string {
|
||||
|
||||
// Check holds the acl rules of a listener on host, http for one serving
|
||||
// requests, proxied for one behind auth.trusted_proxies, whose rules take
|
||||
// clients of either family, as does a PROXY header
|
||||
// clients of either family, as does a PROXY header; an entry's problem is
|
||||
// at its key
|
||||
func (o *ACLOptions) Check(host string, http, proxied bool) error {
|
||||
if o == nil {
|
||||
return nil
|
||||
@@ -384,17 +392,18 @@ func (o *ACLOptions) Check(host string, http, proxied bool) error {
|
||||
if header || proxied {
|
||||
clients = "tcp"
|
||||
}
|
||||
var errs []error
|
||||
for _, list := range []struct {
|
||||
key, network string
|
||||
entries []string
|
||||
}{{"proxy_from", network, o.ProxyFrom}, {"allow", clients, o.Allow}, {"deny", clients, o.Deny}} {
|
||||
for _, e := range list.entries {
|
||||
for i, e := range list.entries {
|
||||
if _, err := ParsePrefix(e, list.network); err != nil {
|
||||
return fmt.Errorf("acl: %s entry %q: %w", list.key, e, err)
|
||||
errs = append(errs, &KeyError{fmt.Sprintf("acl.%s[%d]", list.key, i), fmt.Errorf("%q: %w", e, err)})
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// httpKey names the first key of table v tagged http that is set
|
||||
@@ -537,8 +546,8 @@ func Scan(configMap map[string]any, target any) error {
|
||||
return lconfig.ScanMap(configMap, target)
|
||||
}
|
||||
|
||||
// checkKeys walks nested tables against the toml tags of t, refusing unknown
|
||||
// keys. Map-typed fields hold free-form keys and are not descended into.
|
||||
// checkKeys walks nested tables against the toml tags of t, refusing every
|
||||
// unknown key. Map-typed fields hold free-form keys and are not descended into.
|
||||
func checkKeys(m map[string]any, t reflect.Type, prefix string) error {
|
||||
for t.Kind() == reflect.Pointer || t.Kind() == reflect.Slice {
|
||||
t = t.Elem()
|
||||
@@ -552,17 +561,17 @@ func checkKeys(m map[string]any, t reflect.Type, prefix string) error {
|
||||
fields[name] = f.Type
|
||||
}
|
||||
}
|
||||
var errs []error
|
||||
for _, key := range slices.Sorted(maps.Keys(m)) {
|
||||
ft, ok := fields[key]
|
||||
if !ok {
|
||||
return fmt.Errorf("unknown key %q", prefix+key)
|
||||
errs = append(errs, fmt.Errorf("unknown key %q", prefix+key))
|
||||
continue
|
||||
}
|
||||
var tables []map[string]any
|
||||
switch v := m[key].(type) {
|
||||
case map[string]any:
|
||||
if err := checkKeys(v, ft, prefix+key+"."); err != nil {
|
||||
return err
|
||||
}
|
||||
errs = append(errs, checkKeys(v, ft, prefix+key+"."))
|
||||
case []map[string]any:
|
||||
tables = v
|
||||
case []any:
|
||||
@@ -573,12 +582,10 @@ func checkKeys(m map[string]any, t reflect.Type, prefix string) error {
|
||||
}
|
||||
}
|
||||
for i, table := range tables {
|
||||
if err := checkKeys(table, ft, fmt.Sprintf("%s%s[%d].", prefix, key, i)); err != nil {
|
||||
return err
|
||||
}
|
||||
errs = append(errs, checkKeys(table, ft, fmt.Sprintf("%s%s[%d].", prefix, key, i)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// checkFileKeys rejects keys the configuration file declares that Config does
|
||||
@@ -600,14 +607,27 @@ func checkFileKeys(path string) error {
|
||||
}
|
||||
|
||||
// ParseFile reads a configuration file's text, refusing the keys lw refuses
|
||||
// there, and opens nothing it names
|
||||
func ParseFile(data []byte) (*Config, error) {
|
||||
// there, and opens nothing it names; settings are the paths it sets outside
|
||||
// the pipelines, sorted
|
||||
func ParseFile(data []byte) (cfg *Config, settings []string, err error) {
|
||||
root, err := fileTable(data)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
cfg := &Config{}
|
||||
return cfg, Scan(root, cfg)
|
||||
var walk func(m map[string]any, prefix string)
|
||||
walk = func(m map[string]any, prefix string) {
|
||||
for k, v := range m {
|
||||
if sub, ok := v.(map[string]any); ok {
|
||||
walk(sub, prefix+k+".")
|
||||
} else if prefix != "" || k != "pipelines" {
|
||||
settings = append(settings, prefix+k)
|
||||
}
|
||||
}
|
||||
}
|
||||
walk(root, "")
|
||||
slices.Sort(settings)
|
||||
cfg = &Config{}
|
||||
return cfg, settings, Scan(root, cfg)
|
||||
}
|
||||
|
||||
// fileTable is a file's root table without the keys lw ignores there
|
||||
|
||||
Reference in new issue
Block a user