v0.35.2 lwconf, doc and deploy update

internal/config/config.go

@@ -75,18 +75,18 @@ type FlowConfig struct {
 
 // HeartbeatConfig defines settings for periodic keep-alive or status messages
 type HeartbeatConfig struct {
-	Enabled          bool   `toml:"enabled" help:"send heartbeats"`
+	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:"no effect: heartbeats use the flow's format"`
+	Format           string `toml:"format" 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:"entry format"`
-	Flags           int64  `toml:"flags,omitempty" lw:"zero=by type" help:"formatter flags"`
-	TimestampFormat string `toml:"timestamp_format,omitempty" lw:"zero=RFC 3339 with nanoseconds" help:"Go time layout of timestamps"`
+	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"`
+	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"`
 }
 
@@ -102,10 +102,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 the limit lets through"`
-	BurstEntries      float64 `toml:"burst_entries" lw:"min=0,zero=one second's worth" help:"entries let through at once before the rate applies"`
-	Policy            string  `toml:"policy" default:"pass" lw:"enum=pass|drop" help:"drop entries over the rate, or pass them"`
-	MaxEntrySizeBytes int64   `toml:"max_entry_size_bytes" lw:"min=0,zero=unlimited" help:"drop larger entries"`
+	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"`
 }
 
 // FilterType represents the filter's behavior (include or exclude)
@@ -174,31 +174,31 @@ type ConsoleSourceOptions struct {
 // TCPChainSourceOptions defines settings for a stdlib TCP listener ingesting
 // NDJSON entries from upstream logwisp tcp_chain sinks
 type TCPChainSourceOptions struct {
-	TLS            *TLSOptions  `toml:"tls"`
-	Host           string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
+	TLS            *TLSOptions  `toml:"tls" help:"TLS for the connections it accepts"`
+	Host           string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on, one family per listener: 0.0.0.0 every IPv4 interface, :: every IPv6 interface, 127.0.0.1 or ::1 this machine only"`
 	Port           int64        `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
 	BufferSize     int64        `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
 	MaxConnections int64        `toml:"max_connections" lw:"min=0,zero=unlimited" help:"connections at once"`
 	ReadTimeoutMS  int64        `toml:"read_timeout_ms" lw:"min=0,zero=none" help:"milliseconds a connection may idle"`
 	HelloTimeoutMS int64        `toml:"hello_timeout_ms" default:"10000" lw:"min=1" help:"milliseconds for a peer's hello"`
 	TrustNode      bool         `toml:"trust_node" default:"true" help:"keep the node label a peer declares; off, the label is its address"`
-	Auth           *AuthOptions `toml:"auth"`
-	ACL            *ACLOptions  `toml:"acl"`
+	Auth           *AuthOptions `toml:"auth" help:"how peers authenticate: mtls by client certificate, scram by user and password"`
+	ACL            *ACLOptions  `toml:"acl" help:"client addresses admitted or refused, and per-client limits"`
 }
 
 // HTTPChainSourceOptions defines settings for a stdlib HTTP listener ingesting
 // NDJSON batches from upstream logwisp http_chain sinks
 type HTTPChainSourceOptions struct {
-	TLS           *TLSOptions  `toml:"tls"`
-	Host          string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
+	TLS           *TLSOptions  `toml:"tls" help:"TLS for the connections it accepts"`
+	Host          string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on, one family per listener: 0.0.0.0 every IPv4 interface, :: every IPv6 interface, 127.0.0.1 or ::1 this machine only"`
 	Port          int64        `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
 	IngestPath    string       `toml:"ingest_path" default:"/ingest" lw:"hint=path" help:"path batches are posted to"`
 	BufferSize    int64        `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
 	MaxBodyBytes  int64        `toml:"max_body_bytes" default:"8388608" lw:"min=1" help:"largest batch accepted"`
 	ReadTimeoutMS int64        `toml:"read_timeout_ms" default:"30000" lw:"min=1" help:"milliseconds to read a request"`
 	TrustNode     bool         `toml:"trust_node" default:"true" help:"keep the node label a peer declares; off, the label is its address"`
-	Auth          *AuthOptions `toml:"auth"`
-	ACL           *ACLOptions  `toml:"acl"`
+	Auth          *AuthOptions `toml:"auth" help:"how peers authenticate: mtls by client certificate, scram by user and password"`
+	ACL           *ACLOptions  `toml:"acl" help:"client addresses admitted or refused, and per-client limits"`
 }
 
 // --- Sinks ---
@@ -228,7 +228,7 @@ type FileSinkOptions struct {
 	Name            string  `toml:"name" lw:"required" help:"base name of the files"`
 	MaxSizeMB       int64   `toml:"max_size_mb" default:"100" lw:"min=1" help:"size at which a file rotates"`
 	MaxTotalSizeMB  int64   `toml:"max_total_size_mb" default:"1000" lw:"min=1" help:"size of all files, beyond which the oldest go"`
-	MinDiskFreeMB   int64   `toml:"min_disk_free_mb" lw:"zero=none" help:"free space kept on the disk; negative means 100"`
+	MinDiskFreeMB   int64   `toml:"min_disk_free_mb" lw:"zero=none" help:"megabytes kept free on the disk, the oldest .log files deleted for it; a negative value means 100"`
 	RetentionHours  float64 `toml:"retention_hours" default:"168" lw:"min=0,zero=kept" help:"hours a .log file is kept; 0 keeps them"`
 	BufferSize      int64   `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued"`
 	FlushIntervalMs int64   `toml:"flush_interval_ms" default:"100" lw:"min=1" help:"milliseconds between flushes"`
@@ -236,8 +236,8 @@ type FileSinkOptions struct {
 
 // TCPSinkOptions defines settings for a TCP server sink
 type TCPSinkOptions struct {
-	TLS               *TLSOptions  `toml:"tls"`
-	Host              string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
+	TLS               *TLSOptions  `toml:"tls" help:"TLS for the connections it accepts"`
+	Host              string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on, one family per listener: 0.0.0.0 every IPv4 interface, :: every IPv6 interface, 127.0.0.1 or ::1 this machine only"`
 	Port              int64        `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
 	BufferSize        int64        `toml:"buffer_size" default:"1000" lw:"min=1" help:"entries queued for all clients"`
 	ClientBufferSize  int64        `toml:"client_buffer_size" default:"256" lw:"min=1" help:"entries queued per client"`
@@ -245,14 +245,14 @@ type TCPSinkOptions struct {
 	KeepAlive         bool         `toml:"keep_alive" default:"true" help:"TCP keep-alive"`
 	KeepAlivePeriodMS int64        `toml:"keep_alive_period_ms" default:"30000" lw:"min=1" help:"milliseconds idle before a keep-alive probe"`
 	MaxConnections    int64        `toml:"max_connections" lw:"min=0,zero=unlimited" help:"clients at once"`
-	Auth              *AuthOptions `toml:"auth"`
-	ACL               *ACLOptions  `toml:"acl"`
+	Auth              *AuthOptions `toml:"auth" help:"how peers authenticate: mtls by client certificate, scram by user and password"`
+	ACL               *ACLOptions  `toml:"acl" help:"client addresses admitted or refused, and per-client limits"`
 }
 
 // HTTPSinkOptions defines settings for an HTTP SSE server sink
 type HTTPSinkOptions struct {
-	TLS              *TLSOptions  `toml:"tls"`
-	Host             string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on; :: for IPv6"`
+	TLS              *TLSOptions  `toml:"tls" help:"TLS for the connections it accepts"`
+	Host             string       `toml:"host" default:"0.0.0.0" lw:"hint=host" help:"address to listen on, one family per listener: 0.0.0.0 every IPv4 interface, :: every IPv6 interface, 127.0.0.1 or ::1 this machine only"`
 	Port             int64        `toml:"port" lw:"required,min=1,max=65535" help:"port to listen on"`
 	StreamPath       string       `toml:"stream_path" default:"/stream" lw:"hint=path" help:"path of the event stream"`
 	StatusPath       string       `toml:"status_path" default:"/status" lw:"hint=path" help:"path of the status document"`
@@ -263,14 +263,14 @@ type HTTPSinkOptions struct {
 	ReplayLines      int64        `toml:"replay_lines" lw:"min=0,zero=none" help:"recent entries a new stream gets first"`
 	LoginPage        bool         `toml:"login_page" help:"scram behind auth.trusted_proxies only: a login page at /auth/login"`
 	ViewerPage       bool         `toml:"viewer_page" help:"scram behind auth.trusted_proxies only: the viewer after a login; without scram it is always at /"`
-	Auth             *AuthOptions `toml:"auth"`
-	ACL              *ACLOptions  `toml:"acl"`
+	Auth             *AuthOptions `toml:"auth" help:"how peers authenticate: mtls by client certificate, scram by user and password"`
+	ACL              *ACLOptions  `toml:"acl" help:"client addresses admitted or refused, and per-client limits"`
 }
 
 // TCPChainSinkOptions defines settings for a stdlib TCP client forwarding
 // entries to a downstream logwisp tcp_chain source
 type TCPChainSinkOptions struct {
-	TLS               *TLSOptions  `toml:"tls"`
+	TLS               *TLSOptions  `toml:"tls" help:"TLS to the source it connects to"`
 	Node              string       `toml:"node" lw:"zero=the host name" help:"origin label of the entries"`
 	Host              string       `toml:"host" lw:"required,hint=host" help:"host of the tcp_chain source"`
 	Port              int64        `toml:"port" lw:"required,min=1,max=65535" help:"port of the tcp_chain source"`
@@ -281,13 +281,13 @@ type TCPChainSinkOptions struct {
 	BackoffMaxMS      int64        `toml:"backoff_max_ms" default:"30000" lw:"min=1" help:"longest wait before reconnecting; below the first, the default"`
 	KeepAlivePeriodMS int64        `toml:"keep_alive_period_ms" default:"30000" lw:"min=1" help:"milliseconds idle before a keep-alive probe"`
 	KeepAlive         bool         `toml:"keep_alive" default:"true" help:"TCP keep-alive"`
-	Auth              *AuthOptions `toml:"auth"`
+	Auth              *AuthOptions `toml:"auth" help:"how it authenticates with the source: scram logs in, mtls checks the source's identity"`
 }
 
 // HTTPChainSinkOptions defines settings for a stdlib HTTP client posting
 // NDJSON batches to a downstream logwisp http_chain source
 type HTTPChainSinkOptions struct {
-	TLS              *TLSOptions  `toml:"tls"`
+	TLS              *TLSOptions  `toml:"tls" help:"TLS to the source it connects to"`
 	Node             string       `toml:"node" lw:"zero=the host name" help:"origin label of the entries"`
 	Host             string       `toml:"host" lw:"required,hint=host" help:"host of the http_chain source"`
 	Port             int64        `toml:"port" lw:"required,min=1,max=65535" help:"port of the http_chain source"`
@@ -299,7 +299,7 @@ type HTTPChainSinkOptions struct {
 	RequestTimeoutMS int64        `toml:"request_timeout_ms" default:"10000" lw:"min=1" help:"milliseconds for a request: dial, write and response"`
 	BackoffMinMS     int64        `toml:"backoff_min_ms" default:"500" lw:"min=1" help:"first wait before retrying"`
 	BackoffMaxMS     int64        `toml:"backoff_max_ms" default:"30000" lw:"min=1" help:"longest wait before retrying; below the first, the default"`
-	Auth             *AuthOptions `toml:"auth"`
+	Auth             *AuthOptions `toml:"auth" help:"how it authenticates with the source: scram logs in, mtls checks the source's identity"`
 }
 
 // --- Auth Options ---
@@ -312,14 +312,14 @@ type HTTPChainSinkOptions struct {
 type AuthOptions struct {
 	Type            string   `toml:"type" default:"none" lw:"enum=none|mtls|scram" help:"how peers authenticate"`
 	Identity        string   `toml:"identity" lw:"enum=cn|san_dns|san_uri|san_email,zero=cn under mtls; no binding under scram" help:"certificate field naming the peer; under scram it must equal the user"`
-	Allow           []string `toml:"allow" help:"mtls: identities admitted; with allow_patterns empty, any the CA issued"`
+	Allow           []string `toml:"allow" help:"mtls: identities admitted; with neither this nor allow_patterns set, any the CA issued"`
 	AllowPatterns   []string `toml:"allow_patterns" lw:"hint=regex" help:"mtls: RE2 patterns of identities admitted; anchor them"`
-	NodeBinding     string   `toml:"node_binding" lw:"listener,enum=none|assert|force,zero=force" help:"chain sources: how a peer's identity binds its node label; overrides trust_node"`
-	CredentialsFile string   `toml:"credentials_file" lw:"listener,hint=file" help:"scram: verifiers, from lw auth add-user"`
+	NodeBinding     string   `toml:"node_binding" lw:"chain_listener,enum=none|assert|force,zero=force" help:"chain sources: force labels every entry with the peer's identity, assert refuses a peer declaring another node, none leaves it to trust_node"`
+	CredentialsFile string   `toml:"credentials_file" lw:"listener,hint=file" help:"scram: file of the users who may log in, made by lw auth add-user"`
 	// exp has whole seconds, a sub-second expires_in reads as unknown, and 10 s
 	// leaves room for renewal ahead of expiry
 	TokenLifetimeMS int64    `toml:"token_lifetime_ms" lw:"listener,http,min=10000,max=86400000,zero=15 minutes" help:"scram on HTTP: bearer token lifetime"`
-	TrustedProxies  []string `toml:"trusted_proxies" lw:"listener,http,hint=cidr" help:"scram on the http sink: proxies ending the browsers' TLS; logins are then unbound"`
+	TrustedProxies  []string `toml:"trusted_proxies" lw:"listener,http,hint=cidr" help:"scram on the http sink: addresses or CIDRs of the proxies that end browsers' TLS, through which browsers then log in"`
 	Username        string   `toml:"username" help:"scram: the user a dialer logs in as, or a listener's one user"`
 	PasswordFile    string   `toml:"password_file" lw:"hint=file" help:"scram: file holding the password; /dev/fd/N or a pipe is read once, /dev/tty asks"`
 }
@@ -329,8 +329,8 @@ type AuthOptions struct {
 // the L4 proxies whose PROXY header names the client the rules then see; the
 // per-client limits count that client. ACLOptions.Check holds the rules.
 type ACLOptions struct {
-	Allow                      []string `toml:"allow" lw:"hint=cidr" help:"addresses admitted; empty, all deny does not list"`
-	Deny                       []string `toml:"deny" lw:"hint=cidr" help:"addresses refused"`
+	Allow                      []string `toml:"allow" lw:"hint=cidr" help:"addresses or CIDRs that may connect; unset, any that deny does not refuse"`
+	Deny                       []string `toml:"deny" lw:"hint=cidr" help:"addresses or CIDRs refused, even when allow lists them"`
 	ProxyProtocol              string   `toml:"proxy_protocol" default:"off" lw:"enum=off|optional|required" help:"read a PROXY header from proxy_from"`
 	ProxyFrom                  []string `toml:"proxy_from" lw:"hint=cidr" help:"proxies that may send a PROXY header"`
 	MaxConnectionsPerClient    int64    `toml:"max_connections_per_client" lw:"min=0,zero=unlimited" help:"connections a client may hold"`
@@ -346,12 +346,12 @@ type ACLOptions struct {
 // may present cert_file and key_file.
 type TLSOptions struct {
 	Enabled            bool     `toml:"enabled" help:"use TLS"`
-	CertFile           string   `toml:"cert_file" lw:"hint=file" help:"certificate presented; a dialer's for mTLS"`
+	CertFile           string   `toml:"cert_file" lw:"hint=file" help:"certificate file this end presents; a dialer's, for a listener with client_auth"`
 	KeyFile            string   `toml:"key_file" lw:"hint=file" help:"key of cert_file"`
 	SelfSigned         bool     `toml:"self_signed" lw:"listener" help:"a self-signed certificate on a key made at startup"`
-	IssuerCertFile     string   `toml:"issuer_cert_file" lw:"listener,hint=file" help:"CA certificate (lw tls ca) that issues the listener's at startup"`
+	IssuerCertFile     string   `toml:"issuer_cert_file" lw:"listener,hint=file" help:"CA certificate file (lw tls ca) that signs a certificate made at startup"`
 	IssuerKeyFile      string   `toml:"issuer_key_file" lw:"listener,hint=file" help:"key of issuer_cert_file"`
-	Hosts              []string `toml:"hosts" lw:"listener" help:"names and addresses a made certificate carries beyond host, this machine's and loopback"`
+	Hosts              []string `toml:"hosts" lw:"listener" help:"more names and addresses for a certificate made at startup, which always carries host, this machine's name and localhost"`
 	ClientAuth         bool     `toml:"client_auth" lw:"listener" help:"require and verify client certificates"`
 	ClientCAFile       string   `toml:"client_ca_file" lw:"listener,hint=file" help:"CA bundle verifying client certificates"`
 	CAFile             string   `toml:"ca_file" lw:"dialer,hint=file,zero=the system roots" help:"CA bundle verifying the server"`

internal/config/preset.go

@@ -63,10 +63,19 @@ func mirror[T any](p PresetParam, key string) PresetParam {
 // IsDir tells whether a preset's path is a directory, whose files it follows,
 // or a file pattern; a path ending in '/' is a directory without asking. Off
 // the target host it is the user's Answers, or nil: a path with a glob is then
-// a pattern, and any other fails with ErrPathKind.
+// a pattern, and any other fails at key path with a PathKindError.
 type IsDir func(path string) (bool, error)
 
-var ErrPathKind error = &KeyError{"path", errors.New("a directory or a file pattern?")}
+// ErrPathKind is what every PathKindError unwraps to
+var ErrPathKind = errors.New("a directory or a file?")
+
+// PathKindError asks whether Path, which a preset names, is a directory
+type PathKindError struct{ Path string }
+
+func (e *PathKindError) Error() string {
+	return fmt.Sprintf("%q: %v end a directory with / to follow its files", e.Path, ErrPathKind)
+}
+func (e *PathKindError) Unwrap() error { return ErrPathKind }
 
 // HostIsDir asks this host, a leading ~ being the home directory: a path that
 // does not exist is a pattern
@@ -85,16 +94,16 @@ func Answers(dirs map[string]bool) IsDir {
 		if strings.ContainsAny(path, `*?[\`) {
 			return false, nil
 		}
-		return false, ErrPathKind
+		return false, &KeyError{"path", &PathKindError{path}}
 	}
 }
 
 // Keys several presets share
 var (
-	pathParam   = PresetParam{Name: "path", Help: "file, directory (its files) or glob to follow; default: standard input"}
+	pathParam   = PresetParam{Name: "path", Help: "a file, a directory (its files) or a glob such as /var/log/*.log, to follow; default: standard input"}
 	fromParam   = mirror[FileSourceOptions](PresetParam{Name: "from", Default: "end", Help: "where reading a file starts"}, "from")
 	formatParam = func(def string) PresetParam {
-		return mirror[FormatConfig](PresetParam{Name: "format", Default: def, Help: "entry format"}, "type")
+		return mirror[FormatConfig](PresetParam{Name: "format", Default: def, Help: "raw writes each message as it came, txt a line with time and level, json one object an entry"}, "type")
 	}
 	tlsParams = []PresetParam{
 		mirror[TLSOptions](PresetParam{Name: "cert", Help: "certificate file, for tls=files"}, "cert_file"),
@@ -107,13 +116,13 @@ var (
 		mirror[ACLOptions](PresetParam{Name: "allow", List: true, Help: "addresses or CIDRs that may connect, of listen's family (either behind proxy=); default: all; ',' between them"}, "allow"),
 		mirror[ACLOptions](PresetParam{Name: "deny", List: true, Help: "addresses or CIDRs refused, allow or not; ',' between them"}, "deny"),
 	}
-	transportParam = PresetParam{Name: "transport", Default: "tcp", Help: "tcp_chain or http_chain", Enum: transports()}
+	transportParam = PresetParam{Name: "transport", Default: "tcp", Help: "tcp sends over one lasting connection (tcp_chain), http posts batches (http_chain)", Enum: transports()}
 	usersParam     = func(who string) PresetParam {
 		return mirror[AuthOptions](PresetParam{Name: "users", Help: "credentials file (lw auth add-user): " + who + " log in with SCRAM"}, "credentials_file")
 	}
-	userParams = []PresetParam{
-		{Name: "user", Help: "SCRAM user: an edge's login, or a listener's only user in place of users"},
-		mirror[AuthOptions](PresetParam{Name: "password_file", Help: "file holding user's password; /dev/fd/N or a pipe is read once; default: /dev/tty, which asks"}, "password_file"),
+	userParams = func(help string) []PresetParam {
+		return []PresetParam{{Name: "user", Help: help},
+			mirror[AuthOptions](PresetParam{Name: "password_file", Help: "file holding user's password; /dev/fd/N or a pipe is read once; default: /dev/tty, which asks"}, "password_file")}
 	}
 )
 
@@ -146,7 +155,7 @@ var presets = []Preset{
 			return nil
 		}},
 	{"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")},
+		[]PresetParam{{Name: "path", Required: true, Help: "a file, a directory (its files) or a glob such as /var/log/*.log, to follow"}, fromParam, formatParam("raw")},
 		func(p *PipelineConfig, v map[string]string, isDir IsDir) error {
 			sources, err := pathOrStdin(v, isDir)
 			p.PluginSources, p.PluginSinks = sources, pipeDefault("").PluginSinks
@@ -155,12 +164,12 @@ var presets = []Preset{
 		}},
 	{"serve", "Serve files or standard input live: a browser viewer at /, SSE at /stream",
 		slices.Concat([]PresetParam{pathParam, fromParam, formatParam("json"),
-			{Name: "listen", Default: "127.0.0.1:8080", Help: "HOST:PORT; [IPV6]:PORT"},
+			{Name: "listen", Default: "127.0.0.1:8080", Help: "HOST:PORT to listen on: 127.0.0.1:8080 this machine only, 0.0.0.0:8080 every IPv4 interface, [::]:8080 every IPv6 interface"},
 			serveTLS,
 			usersParam("readers"),
-		}, userParams, []PresetParam{
+		}, userParams("the one SCRAM user readers log in as, in place of users"), []PresetParam{
 			mirror[AuthOptions](PresetParam{Name: "proxy", List: true, Help: "addresses or CIDRs of the TLS-terminating proxies browsers come through; ',' between them"}, "trusted_proxies"),
-			mirror[HTTPSinkOptions](PresetParam{Name: "viewer", Default: "false", Help: "the login page and viewer, behind proxy; without user or users the viewer is always on"}, "viewer_page"),
+			mirror[HTTPSinkOptions](PresetParam{Name: "viewer", Default: "false", Help: "a login page, then the viewer, for browsers behind proxy; without user or users the viewer is on anyway"}, "viewer_page"),
 		}, aclParams, tlsParams),
 		func(p *PipelineConfig, v map[string]string, isDir IsDir) error {
 			sources, err := pathOrStdin(v, isDir)
@@ -177,11 +186,11 @@ var presets = []Preset{
 			case err != nil:
 				return err
 			case auth != nil && v["tls"] == "off" && v["proxy"] == "":
-				return fmt.Errorf("user and users need tls (self, issuer or files), or proxy")
+				return &KeyError{"tls", errors.New("user and users need self, issuer or files, or proxy")}
 			case auth != nil:
 				sink["auth"] = auth
 			case v["proxy"] != "":
-				return fmt.Errorf("proxy needs user or users: browsers log in with SCRAM")
+				return &KeyError{"proxy", errors.New("needs user or users: browsers log in with SCRAM")}
 			}
 			if v["proxy"] != "" {
 				auth["trusted_proxies"] = list(v["proxy"])
@@ -189,12 +198,12 @@ var presets = []Preset{
 			switch v["viewer"] {
 			case "true":
 				if v["proxy"] == "" {
-					return fmt.Errorf("viewer=true needs proxy, behind which browsers log in; without users the viewer is always on")
+					return &KeyError{"viewer", errors.New("true needs proxy, behind which browsers log in; without users the viewer is always on")}
 				}
 				sink["login_page"], sink["viewer_page"] = true, true
 			case "false":
 			default:
-				return fmt.Errorf("viewer %q (valid: true, false)", v["viewer"])
+				return &KeyError{"viewer", fmt.Errorf("must be true or false, got %q", v["viewer"])}
 			}
 			sink["replay_lines"] = int64(1000) // a browser opened late sees the recent entries
 			p.Flow.Format = &FormatConfig{Type: v["format"]}
@@ -203,12 +212,12 @@ var presets = []Preset{
 		}},
 	{"edge", "Forward files or standard input to an aggregator over TLS, authenticated",
 		slices.Concat([]PresetParam{pathParam, fromParam,
-			{Name: "to", Required: true, Help: "the aggregator's HOST:PORT; [IPV6]:PORT"},
+			{Name: "to", Required: true, Help: "the aggregator's HOST:PORT, an IPv6 address in brackets: logs.example.net:9000, [2001:db8::1]:9000"},
 			transportParam,
 			mirror[TLSOptions](PresetParam{Name: "ca", Help: "CA file that verifies the aggregator; default: system roots"}, "ca_file"),
 			{Name: "pin", Help: "sha256//BASE64 of the aggregator's key, which it logs with tls=self"},
 			{Name: "server_name", Help: "name the aggregator's certificate carries; default: to's host"},
-		}, userParams, []PresetParam{
+		}, userParams("SCRAM user this edge logs in to the aggregator as"), []PresetParam{
 			mirror[TLSOptions](PresetParam{Name: "cert", Help: "client certificate file, for an aggregator with client_ca"}, "cert_file"),
 			mirror[TLSOptions](PresetParam{Name: "key", Help: "client key file"}, "key_file"),
 			{Name: "node", Help: "origin label; default: this host's name"},
@@ -241,7 +250,7 @@ var presets = []Preset{
 			case v["cert"] != "" && v["key"] != "":
 				auth = map[string]any{"type": "mtls"}
 			default:
-				return fmt.Errorf("edge needs user, or cert and key: it never sends unauthenticated")
+				return &KeyError{"user", errors.New("required unless cert and key are set; an edge never sends unauthenticated")}
 			}
 			sink := map[string]any{"host": host, "port": port, "tls": tls, "auth": auth}
 			if v["node"] != "" {
@@ -252,11 +261,11 @@ var presets = []Preset{
 		}},
 	{"aggregator", "Receive from edges over TLS, authenticated, to files or standard output",
 		slices.Concat([]PresetParam{
-			{Name: "listen", Default: "0.0.0.0:9000", Help: "HOST:PORT; [IPV6]:PORT; [::] for IPv6 only"},
+			{Name: "listen", Default: "0.0.0.0:9000", Help: "HOST:PORT to listen on: 0.0.0.0:9000 every IPv4 interface, [::]:9000 every IPv6 interface, 127.0.0.1:9000 this machine only"},
 			transportParam,
 			aggregatorTLS,
 			usersParam("edges"),
-		}, userParams, []PresetParam{
+		}, userParams("the one SCRAM user edges log in as, in place of users"), []PresetParam{
 			mirror[TLSOptions](PresetParam{Name: "client_ca", Help: "CA file that verifies edge certificates (mTLS)"}, "client_ca_file"),
 			mirror[FileSinkOptions](PresetParam{Name: "out", Help: "directory for the received entries; default: standard output"}, "directory"),
 			formatParam("json"),
@@ -283,7 +292,7 @@ var presets = []Preset{
 			case v["client_ca"] != "":
 				src["auth"] = map[string]any{"type": "mtls"}
 			default:
-				return fmt.Errorf("aggregator needs user, users or client_ca: it never receives unauthenticated")
+				return &KeyError{"users", errors.New("required unless user or client_ca is set; an aggregator never receives unauthenticated")}
 			}
 			p.PluginSources = append(p.PluginSources, PluginSourceConfig{ID: "edges", Type: typ, Config: src})
 			p.PluginSinks = pipeDefault("").PluginSinks
@@ -331,13 +340,13 @@ func applyPreset(p *PipelineConfig, name string, values map[string][]string, isD
 		case j < 0:
 			return fmt.Errorf("preset %s has no key %q (valid: %s)", name, k, strings.Join(keys, ", "))
 		case len(each) > 1 && !r.Params[j].List:
-			return fmt.Errorf("preset %s key %q takes one value", name, k)
+			return &KeyError{k, errors.New("takes one value")}
 		}
 		v[k] = cmp.Or(strings.Join(each, ","), v[k])
 	}
 	for _, k := range r.Params {
 		if k.Required && v[k.Name] == "" {
-			return fmt.Errorf("preset %s needs %s", name, k.Name)
+			return &KeyError{k.Name, errors.New("must be set")}
 		}
 	}
 	p.Name = cmp.Or(p.Name, name)
@@ -404,20 +413,20 @@ func listener(v map[string]string, key PresetParam) (map[string]any, error) {
 	mode := cmp.Or(v["tls"], key.Default)
 	i := slices.IndexFunc(tlsModes, func(m tlsMode) bool { return m.name == mode })
 	if i < 0 || !slices.Contains(key.Enum, mode) {
-		return nil, fmt.Errorf("tls %q (valid: %s)", mode, strings.Join(key.Enum, ", "))
+		return nil, &KeyError{"tls", fmt.Errorf("must be one of %s, got %q", strings.Join(key.Enum, ", "), mode)}
 	}
 	m := tlsModes[i]
 	for _, other := range tlsModes {
 		for k := range other.needs {
 			if _, needed := m.needs[k]; needed && v[k] == "" {
-				return nil, fmt.Errorf("tls=%s needs %s", mode, k)
+				return nil, &KeyError{k, fmt.Errorf("must be set for tls=%s", mode)}
 			} else if !needed && v[k] != "" {
-				return nil, fmt.Errorf("%s does not apply to tls=%s", k, mode)
+				return nil, &KeyError{k, fmt.Errorf("does not apply to tls=%s", mode)}
 			}
 		}
 	}
 	if v["hosts"] != "" && !m.made {
-		return nil, fmt.Errorf("hosts does not apply to tls=%s", mode)
+		return nil, &KeyError{"hosts", fmt.Errorf("does not apply to tls=%s", mode)}
 	}
 	if mode == "off" {
 		return opts, nil
@@ -439,11 +448,11 @@ func listener(v map[string]string, key PresetParam) (map[string]any, error) {
 func scramAuth(v map[string]string) (map[string]any, error) {
 	switch {
 	case v["user"] != "" && v["users"] != "":
-		return nil, fmt.Errorf("user or users, not both")
+		return nil, &KeyError{"user", errors.New("set user or users, not both")}
 	case v["user"] != "":
 		return map[string]any{"type": "scram", "username": v["user"], "password_file": cmp.Or(v["password_file"], "/dev/tty")}, nil
 	case v["password_file"] != "":
-		return nil, fmt.Errorf("password_file needs user")
+		return nil, &KeyError{"password_file", errors.New("needs user")}
 	case v["users"] != "":
 		return map[string]any{"type": "scram", "credentials_file": v["users"]}, nil
 	}
@@ -454,7 +463,7 @@ func splitAddr(key, addr string) (string, int64, error) {
 	host, p, err := net.SplitHostPort(addr)
 	port, perr := strconv.ParseUint(p, 10, 16)
 	if err != nil || perr != nil || port == 0 {
-		return "", 0, fmt.Errorf("%s %q: want HOST:PORT, an IPv6 address in brackets", key, addr)
+		return "", 0, &KeyError{key, fmt.Errorf("must be HOST:PORT, an IPv6 address in brackets as in [::1]:9000, got %q", addr)}
 	}
 	return host, int64(port), nil
 }
@@ -462,7 +471,7 @@ func splitAddr(key, addr string) (string, int64, error) {
 // chainType is a transport's chain plugin type, both a source and a sink
 func chainType(transport string) (string, error) {
 	if !slices.Contains(transports(), transport) {
-		return "", fmt.Errorf("transport %q (valid: %s)", transport, strings.Join(transports(), ", "))
+		return "", &KeyError{"transport", fmt.Errorf("must be one of %s, got %q", strings.Join(transports(), ", "), transport)}
 	}
 	return transport + "_chain", nil
 }

internal/config/preset_test.go

@@ -39,34 +39,39 @@ func TestPresetStartsAndNamesItsPipeline(t *testing.T) {
 }
 
 // Keys are checked against the preset's own, required ones enforced, and
-// the edge and aggregator refuse to run without authentication.
+// the edge and aggregator refuse to run without authentication; a refusal of
+// one key is a KeyError at it, the others name the preset.
 func TestPresetKeysAreChecked(t *testing.T) {
 	isolateConfig(t)
 	for spec, want := range map[string]string{
 		"nope":                             `no preset "nope" (valid: pipe, tail, serve, edge, aggregator)`,
 		"tail:path=x,colour=1":             `preset tail has no key "colour" (valid: path, from, format)`,
-		"tail":                             "preset tail needs path",
+		"tail":                             "path: must be set",
 		"tail:path=x,format.type=json":     "preset keys do not nest",
-		"tail:path=x,path=y":               `key "path" takes one value`,
-		"serve:listen=8080":                `listen "8080": want HOST:PORT`,
-		"serve:tls=self,cert=c":            "cert does not apply to tls=self",
-		"serve:tls=files,cert=c":           "tls=files needs key",
-		"serve:viewer=yes":                 `viewer "yes"`,
-		"serve:proxy=127.0.0.1":            "proxy needs user or users",
-		"serve:viewer=true":                "viewer=true needs proxy",
-		"serve:user=u":                     "user and users need tls (self, issuer or files), or proxy",
-		"serve:tls=self,user=u,users=f":    "user or users, not both",
-		"edge:to=agg:9000":                 "it never sends unauthenticated",
-		"edge:to=agg:9000,password_file=p": "password_file needs user",
-		"edge:to=agg:9000,user=u,password_file=p,transport=udp": `transport "udp"`,
-		"aggregator:users=u,tls=off":                            `tls "off" (valid: `,
-		"aggregator:user=u,users=f":                             "user or users, not both",
-		"aggregator":                                            "it never receives unauthenticated",
+		"tail:path=x,path=y":               "path: takes one value",
+		"serve:listen=8080":                `listen: must be HOST:PORT, an IPv6 address in brackets as in [::1]:9000, got "8080"`,
+		"serve:tls=self,cert=c":            "cert: does not apply to tls=self",
+		"serve:tls=files,cert=c":           "key: must be set for tls=files",
+		"serve:viewer=yes":                 `viewer: must be true or false, got "yes"`,
+		"serve:proxy=127.0.0.1":            "proxy: needs user or users",
+		"serve:viewer=true":                "viewer: true needs proxy",
+		"serve:user=u":                     "tls: user and users need self, issuer or files, or proxy",
+		"serve:tls=self,user=u,users=f":    "user: set user or users, not both",
+		"edge:to=agg:9000":                 "user: required unless cert and key are set; an edge never sends unauthenticated",
+		"edge:to=agg:9000,password_file=p": "password_file: needs user",
+		"edge:to=agg:9000,user=u,password_file=p,transport=udp": `transport: must be one of tcp, http, got "udp"`,
+		"aggregator:users=u,tls=off":                            `tls: must be one of self, issuer, files, got "off"`,
+		"aggregator:user=u,users=f":                             "user: set user or users, not both",
+		"aggregator":                                            "users: required unless user or client_ca is set; an aggregator never receives unauthenticated",
 	} {
 		_, err := Load(Args{Specs: specs("preset", spec)})
 		if err == nil || !strings.Contains(err.Error(), want) {
 			t.Errorf("%s: err = %v, want %s", spec, err, want)
 		}
+		ke, keyed := errors.AsType[*KeyError](err)
+		if keyed == strings.Contains(want, "preset") || keyed && !strings.HasPrefix(want, ke.Key+": ") {
+			t.Errorf("%s: %v placed at %v", spec, err, ke)
+		}
 	}
 }
 
@@ -127,7 +132,7 @@ func TestPresetUserWithoutPasswordFileAsks(t *testing.T) {
 // A preset path ending in '/' is a directory, there yet or not; any other is
 // what isDir answers, a directory named with brackets too: this host's,
 // reading ~ as home, or off it the user's answer, a single file too; without
-// one it is a pattern when it holds a glob, and otherwise ErrPathKind asks
+// one it is a pattern when it holds a glob, and otherwise PathKindError asks
 func TestPresetPathKind(t *testing.T) {
 	home := t.TempDir()
 	t.Setenv("HOME", home)
@@ -145,8 +150,8 @@ func TestPresetPathKind(t *testing.T) {
 		{"~/logs", HostIsDir, "~/logs *"}, {home + "/later/", HostIsDir, home + "/later/ *"}} {
 		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)
+			if pk, ok := errors.AsType[*PathKindError](err); !errors.Is(err, ErrPathKind) || !ok || pk.Path != c.path {
+				t.Errorf("%s: %v, want ErrPathKind asking about it", c.path, err)
 			}
 			continue
 		}

internal/config/schema.go

@@ -35,7 +35,7 @@ type Plugin struct {
 	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
+	Single  bool   `json:"single"` // one instance per process, as standard input is one
 	Summary string `json:"summary"`
 	options reflect.Type
 }
@@ -45,15 +45,15 @@ var plugins = []Plugin{
 	{"source", "file", Local, false, false, "Follow the files of a directory that match a pattern", reflect.TypeFor[FileSourceOptions]()},
 	{"source", "random", Local, false, false, "Generated entries, to check a deployment before real logs", reflect.TypeFor[RandomSourceOptions]()},
 	{"source", "null", Local, false, false, "Nothing; holds a source's place until one is chosen", reflect.TypeFor[NullSourceOptions]()},
-	{"source", "tcp_chain", ChainListener, false, false, "Entries from tcp_chain sinks of other instances", reflect.TypeFor[TCPChainSourceOptions]()},
-	{"source", "http_chain", ChainListener, true, false, "Entry batches from http_chain sinks of other instances", reflect.TypeFor[HTTPChainSourceOptions]()},
+	{"source", "tcp_chain", ChainListener, false, false, "Receive entries from other nodes' tcp_chain sinks, each over one lasting TCP connection", reflect.TypeFor[TCPChainSourceOptions]()},
+	{"source", "http_chain", ChainListener, true, false, "Receive entries from other nodes' http_chain sinks, as batches each posts over HTTP", reflect.TypeFor[HTTPChainSourceOptions]()},
 	{"sink", "console", Local, false, false, "Standard output or standard error", reflect.TypeFor[ConsoleSinkOptions]()},
 	{"sink", "file", Local, false, false, "Rotating files", reflect.TypeFor[FileSinkOptions]()},
 	{"sink", "null", Local, false, false, "Discards; holds a sink's place until one is chosen", reflect.TypeFor[NullSinkOptions]()},
 	{"sink", "tcp", Listener, false, false, "Serve entries to TCP clients, one per line", reflect.TypeFor[TCPSinkOptions]()},
 	{"sink", "http", Listener, true, false, "Serve entries as server-sent events, with a browser viewer", reflect.TypeFor[HTTPSinkOptions]()},
-	{"sink", "tcp_chain", Dialer, false, false, "Forward entries to a tcp_chain source", reflect.TypeFor[TCPChainSinkOptions]()},
-	{"sink", "http_chain", Dialer, true, false, "Post entry batches to an http_chain source", reflect.TypeFor[HTTPChainSinkOptions]()},
+	{"sink", "tcp_chain", Dialer, false, false, "Forward entries to another node's tcp_chain source over one lasting TCP connection", reflect.TypeFor[TCPChainSinkOptions]()},
+	{"sink", "http_chain", Dialer, true, false, "Forward entries to another node's http_chain source in batches, an HTTP POST each", reflect.TypeFor[HTTPChainSinkOptions]()},
 }
 
 // Plugins lists the catalogue
@@ -64,12 +64,9 @@ func Plugins() []Plugin { return slices.Clone(plugins) }
 func (p Plugin) Keys() []Key { return p.applying(keys(p.options)) }
 
 func (p Plugin) applying(all []Key) []Key {
-	side, out := "listener", []Key{}
-	if p.Side == Dialer {
-		side = "dialer"
-	}
+	out := []Key{}
 	for _, k := range all {
-		if (k.Side == "" || k.Side == side) && (!k.HTTP || p.HTTP) {
+		if (k.Side == "" || k.Side == sideNames[p.Side] || k.Side == "listener" && p.Side == ChainListener) && (!k.HTTP || p.HTTP) {
 			if k.Keys != nil {
 				k.Keys = p.applying(k.Keys)
 			}
@@ -99,10 +96,10 @@ func (e *KeyError) Error() string { return e.Key + ": " + e.Err.Error() }
 func (e *KeyError) Unwrap() error { return e.Err }
 
 // option is one key of an options table, from its field's tags: toml names
-// it, default: holds the default as typed on a command line, help: says what
-// 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, and http (only plugins that speak HTTP take it).
+// it, default: holds the default as a command line types it, help: says what
+// it does, lw: lists its rules: required, enum=a|b, min=N, max=N, zero=MEANING
+// (a value, not the default), hint=KIND, a side (listener, a chain source's
+// too; chain_listener; dialer), and http (only plugins that speak HTTP take it).
 type option struct {
 	Key
 	index int
@@ -125,7 +122,7 @@ func options(t reflect.Type) []option {
 			case "":
 			case "required":
 				o.Required = true
-			case "listener", "dialer":
+			case "listener", "chain_listener", "dialer":
 				o.Side = name
 			case "http":
 				o.HTTP = true

internal/config/schema_test.go

@@ -173,7 +173,8 @@ func TestLoadReportsEveryProblemALine(t *testing.T) {
 }
 
 // A plugin lists the table keys that apply to it: those of its side, a chain
-// source's being a listener's, and the HTTP ones only when it speaks HTTP
+// source's being a listener's and its own, and the HTTP ones only when it
+// speaks HTTP
 func TestTableKeysFollowThePlugin(t *testing.T) {
 	has := func(role, typ, path string) bool {
 		p, _ := LookupPlugin(role, typ)
@@ -190,7 +191,7 @@ func TestTableKeysFollowThePlugin(t *testing.T) {
 	for path, want := range map[string]bool{
 		"sink tcp_chain tls.pin_sha256": true, "sink tcp_chain tls.self_signed": false,
 		"source tcp_chain tls.self_signed": true, "source tcp_chain tls.pin_sha256": false,
-		"source tcp_chain auth.node_binding": true, "sink tcp_chain auth.credentials_file": false,
+		"source tcp_chain auth.node_binding": true, "sink http auth.node_binding": false, "sink tcp_chain auth.credentials_file": false,
 		"sink tcp acl.requests_per_second_per_client": false, "sink tcp auth.token_lifetime_ms": false,
 		"sink http acl.requests_per_second_per_client": true, "sink http auth.trusted_proxies": true,
 	} {

internal/config/validate.go

@@ -59,10 +59,10 @@ func ValidatePipelines(pipelines []PipelineConfig) error {
 		}
 		names[p.Name] = true
 		if len(p.PluginSources) == 0 {
-			at(path+".plugin_sources", errors.New("no sources defined"))
+			at(path+".plugin_sources", errors.New("add a source; a pipeline needs at least one"))
 		}
 		if len(p.PluginSinks) == 0 {
-			at(path+".plugin_sinks", errors.New("no sinks defined"))
+			at(path+".plugin_sinks", errors.New("add a sink; a pipeline needs at least one"))
 		}
 		if err := checkFlow(p.Flow, path+".flow"); err != nil {
 			errs = append(errs, err)
@@ -77,7 +77,7 @@ func ValidatePipelines(pipelines []PipelineConfig) error {
 			case !ok:
 				at(path+".type", fmt.Errorf("unknown %s type %q", role, typ))
 			case plugin.Single && single[role+typ] != "":
-				at(path, fmt.Errorf("one %s %s may run, and pipeline %q has it", typ, role, single[role+typ]))
+				at(path, fmt.Errorf("one %s %s may run, and pipeline %q has it, as lw has one standard input; give this pipeline another %s", typ, role, single[role+typ], role))
 			default:
 				ids[list+id] = true
 				if plugin.Single {
@@ -414,15 +414,17 @@ func ParsePrefix(entry, network string) (netip.Prefix, error) {
 	if err != nil {
 		addr, aerr := netip.ParseAddr(entry)
 		if aerr != nil || addr.Zone() != "" {
-			return netip.Prefix{}, fmt.Errorf("neither an address nor a CIDR (and no zone)")
+			return netip.Prefix{}, errors.New("not an IP address or CIDR; host names and IPv6 zones are not accepted")
 		}
 		prefix = netip.PrefixFrom(addr, addr.BitLen())
 	}
 	switch a := prefix.Addr(); {
 	case a.Is4In6():
 		return netip.Prefix{}, fmt.Errorf("IPv4-mapped; write it as IPv4")
-	case network == "tcp4" && !a.Is4(), network == "tcp6" && a.Is4():
-		return netip.Prefix{}, fmt.Errorf("not of the listener's family (%s)", network)
+	case network == "tcp4" && !a.Is4():
+		return netip.Prefix{}, errors.New("IPv6, and the listener's host is IPv4; write an IPv4 address or CIDR")
+	case network == "tcp6" && a.Is4():
+		return netip.Prefix{}, errors.New("IPv4, and the listener's host is IPv6; write an IPv6 address or CIDR")
 	}
 	return prefix, nil
 }
@@ -520,11 +522,15 @@ func (c *FilterConfig) Check() error {
 }
 
 // Scan decodes a flow stage's command-line map into target, rejecting keys
-// target does not declare: a misspelled key must fail, not be dropped.
+// target does not declare: a misspelled key must fail, not be dropped. Its
+// values take their kinds as a plugin's do, a wrong one refused at its key.
 func Scan(configMap map[string]any, target any) error {
 	if err := checkKeys(configMap, reflect.TypeOf(target), ""); err != nil {
 		return err
 	}
+	if err := coerce(reflect.TypeOf(target).Elem(), configMap, ""); err != nil {
+		return err
+	}
 	return lconfig.ScanMap(configMap, target)
 }