176 lines
7.0 KiB
Go
176 lines
7.0 KiB
Go
// Copyright 2026 Alibaba Group
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package consume
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
)
|
|
|
|
// ValidationError represents a flag-level user error. It wraps a clear,
|
|
// actionable message — the cobra command layer surfaces it to the user
|
|
// with exit code 2 (validation error).
|
|
type ValidationError struct{ Msg string }
|
|
|
|
func (e *ValidationError) Error() string { return e.Msg }
|
|
|
|
// validation sentinels (cobra layer uses errors.Is to set the exit code).
|
|
var (
|
|
// ErrForceRequiresForeground is the plan §3.1 contract: --force only
|
|
// makes sense in foreground mode where the bus runs in the current
|
|
// process (then --force skips the single-instance lock so a second
|
|
// foreground bus can co-exist for a brief debug window). Outside of
|
|
// --foreground, --force would silently produce two daemons writing
|
|
// to the same socket — refuse upfront.
|
|
ErrForceRequiresForeground = &ValidationError{
|
|
Msg: "--force is only meaningful with --foreground (in daemon mode it would produce multiple bus instances; cloud events would be randomly split across connections). To restart the bus: preview dws event stop --all --dry-run, confirm with dws event stop --all --yes, then run dws event consume",
|
|
}
|
|
|
|
// ErrJSONFormatRequiresBounded is the plan §3.1 contract: --format
|
|
// json renders each event as a multi-line JSON object suitable for
|
|
// human inspection. With an unbounded stream the output mixes events
|
|
// without delimiters. Force the user to bound the run.
|
|
ErrJSONFormatRequiresBounded = &ValidationError{
|
|
Msg: "--format json requires --max-events or --duration (an unbounded JSON stream is not parseable). Use --format ndjson for unbounded streams.",
|
|
}
|
|
)
|
|
|
|
// ValidateConfig performs all pre-flight validation that does not require
|
|
// disk / network I/O. Returns a *ValidationError for any rule violation;
|
|
// returns nil if the cfg is launchable. The cobra layer calls this BEFORE
|
|
// calling Run so the user gets clear errors at parse time.
|
|
//
|
|
// Rules implemented:
|
|
// 1. WorkDir / IPCEndpoint / ClientID non-empty
|
|
// 2. --force requires --foreground (plan §3.1)
|
|
// 3. --format json requires --max-events OR --duration (bounded)
|
|
// 4. Routes already pre-parsed (any parse error is reported by ParseRoutes)
|
|
// 5. --output-dir conflict with global --output (caller-supplied flag —
|
|
// we expose ValidateNoOutputConflict separately because global -o is
|
|
// a cobra-layer concern)
|
|
//
|
|
// Rules NOT enforced here (deferred to caller / Run):
|
|
// - Credentials presence (auth.ResolveAppCredentialsStrict already
|
|
// reports a typed error)
|
|
// - bus availability (busctl.Discover handles)
|
|
func ValidateConfig(cfg Config) error {
|
|
if strings.TrimSpace(cfg.WorkDir) == "" {
|
|
return &ValidationError{Msg: "consume: WorkDir is required"}
|
|
}
|
|
if strings.TrimSpace(cfg.IPCEndpoint) == "" {
|
|
return &ValidationError{Msg: "consume: IPCEndpoint is required"}
|
|
}
|
|
if strings.TrimSpace(cfg.ClientID) == "" {
|
|
return &ValidationError{Msg: "consume: ClientID is required"}
|
|
}
|
|
if cfg.Force && !cfg.Foreground {
|
|
return ErrForceRequiresForeground
|
|
}
|
|
if cfg.Format == FormatJSON && cfg.MaxEvents <= 0 && cfg.Duration <= 0 {
|
|
return ErrJSONFormatRequiresBounded
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ValidateNoOutputConflict ensures --output-dir / --route (event-stream
|
|
// sinks) are not combined with the dws global hidden -o/--output flag
|
|
// (request-output to file). The cobra layer reads the global output flag
|
|
// from inherited flags and passes its value here; an empty globalOutput
|
|
// means the flag was unset.
|
|
func ValidateNoOutputConflict(cfg Config, globalOutput string) error {
|
|
if globalOutput == "" {
|
|
return nil
|
|
}
|
|
if cfg.OutputDir != "" || len(cfg.Routes) > 0 {
|
|
return &ValidationError{
|
|
Msg: fmt.Sprintf("--output-dir/--route cannot be combined with global -o/--output=%q (event stream sinks are mutually exclusive with single-file output capture)", globalOutput),
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// IsValidationError reports whether err is a flag-level user error.
|
|
// Cobra command handlers use this to map validation errors to exit code 2.
|
|
func IsValidationError(err error) bool {
|
|
var v *ValidationError
|
|
return errors.As(err, &v)
|
|
}
|
|
|
|
// PrintDryRun writes the resolved configuration to w in a single
|
|
// human-readable block. Called by Run when cfg.DryRun is true. Format
|
|
// avoids JSON so users can `dws event consume --dry-run | head` cleanly.
|
|
//
|
|
// RuntimeToken is the only secret-bearing Config field and is deliberately
|
|
// not read or rendered here. Keep this function allowlist-based: never switch
|
|
// it to generic struct serialization.
|
|
func PrintDryRun(w io.Writer, cfg Config) {
|
|
if w == nil {
|
|
return
|
|
}
|
|
fmt.Fprintln(w, "dws event consume — dry run (no bus connection will be made)")
|
|
fmt.Fprintf(w, " client_id : %s\n", cfg.ClientID)
|
|
fmt.Fprintf(w, " workdir : %s\n", cfg.WorkDir)
|
|
fmt.Fprintf(w, " ipc_endpoint : %s\n", cfg.IPCEndpoint)
|
|
if len(cfg.EventTypes) > 0 {
|
|
fmt.Fprintf(w, " event_types : %s\n", strings.Join(cfg.EventTypes, ","))
|
|
} else {
|
|
fmt.Fprintln(w, " event_types : (catch-all)")
|
|
}
|
|
if cfg.Filter != "" {
|
|
fmt.Fprintf(w, " filter : %s\n", cfg.Filter)
|
|
}
|
|
fmt.Fprintf(w, " format : %s\n", cfg.Format)
|
|
fmt.Fprintf(w, " flatten : %v\n", cfg.Flatten)
|
|
if cfg.OutputDir != "" {
|
|
fmt.Fprintf(w, " output_dir : %s\n", cfg.OutputDir)
|
|
}
|
|
for i, r := range cfg.Routes {
|
|
fmt.Fprintf(w, " route[%d] : %s\n", i, r.Raw)
|
|
}
|
|
if cfg.MaxEvents > 0 {
|
|
fmt.Fprintf(w, " max_events : %d\n", cfg.MaxEvents)
|
|
}
|
|
if cfg.Duration > 0 {
|
|
fmt.Fprintf(w, " duration : %s\n", cfg.Duration)
|
|
}
|
|
fmt.Fprintf(w, " compact : %v\n", cfg.Compact)
|
|
fmt.Fprintf(w, " quiet : %v\n", cfg.Quiet)
|
|
fmt.Fprintf(w, " foreground : %v\n", cfg.Foreground)
|
|
fmt.Fprintf(w, " force : %v\n", cfg.Force)
|
|
}
|
|
|
|
// PrintDryRunMany renders the shared consume configuration plus every local
|
|
// logical consumer without opening the bus.
|
|
func PrintDryRunMany(w io.Writer, cfg Config, specs []ConsumerSpec) {
|
|
preview := cfg
|
|
preview.EventTypes = nil
|
|
for _, spec := range specs {
|
|
preview.EventTypes = append(preview.EventTypes, spec.EventTypes...)
|
|
}
|
|
PrintDryRun(w, preview)
|
|
for i, spec := range specs {
|
|
fmt.Fprintf(w, " consumer[%d] : event_key=%s subscribe_id=%s event_types=%s\n",
|
|
i, spec.EventKey, displayDryRunValue(spec.SubscribeID), strings.Join(spec.EventTypes, ","))
|
|
}
|
|
}
|
|
|
|
func displayDryRunValue(value string) string {
|
|
if value = strings.TrimSpace(value); value != "" {
|
|
return value
|
|
}
|
|
return "(pending)"
|
|
}
|