Documentation
¶
Overview ¶
Package pipeline provides the scan orchestration engine for stringer. It resolves collectors, runs them sequentially, validates their output, and aggregates results into a ScanResult.
Index ¶
- Variables
- func BoostColocatedSignals(signals []signal.RawSignal)
- func DeduplicateSignals(signals []signal.RawSignal) []signal.RawSignal
- func FilterSuppressed(signals []signal.RawSignal, state *baseline.BaselineState, prefix string) ([]signal.RawSignal, int)
- func SignalHash(s signal.RawSignal) string
- type Pipeline
- type Progress
- type RunningCollector
- type ValidationError
Constants ¶
This section is empty.
Variables ¶
var HeartbeatInterval = 60 * time.Second
HeartbeatInterval controls how often Run reports collectors that are still running via Progress.OnHeartbeat. It is a package-level variable so tests can shorten it; a value <= 0 disables the heartbeat.
Functions ¶
func BoostColocatedSignals ¶ added in v1.4.0
BoostColocatedSignals applies cross-collector confidence boosts to signals that share a file with certain risk-indicator kinds. The index is built once before iteration so boosted signals cannot create new eligibility (no cascading). A signal is never boosted by its own kind.
func DeduplicateSignals ¶
DeduplicateSignals removes duplicate signals based on content hashing. When duplicates are found, the first occurrence is kept. If a later duplicate has a higher Confidence score, the kept signal's Confidence is updated to the higher value.
func FilterSuppressed ¶ added in v1.5.0
func FilterSuppressed(signals []signal.RawSignal, state *baseline.BaselineState, prefix string) ([]signal.RawSignal, int)
FilterSuppressed removes signals whose IDs appear in the baseline. Expired suppressions are NOT filtered (signal reappears after TTL). Returns the filtered signals and count of suppressed signals.
func SignalHash ¶
SignalHash computes a content-based hash for a signal. The hash key is: Source + Kind + FilePath + Line + Title. It uses SHA-256 truncated to 8 hex characters (4 bytes).
Types ¶
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline orchestrates the execution of collectors and aggregates results.
func New ¶
func New(config signal.ScanConfig) (*Pipeline, error)
New creates a Pipeline from the given ScanConfig. It resolves collectors from the global registry. If config.Collectors is empty, all registered collectors are used (sorted by name for deterministic ordering). Returns an error if a requested collector is not found in the registry.
func NewWithCollectors ¶
func NewWithCollectors(config signal.ScanConfig, collectors []collector.Collector) *Pipeline
NewWithCollectors creates a Pipeline with explicitly provided collectors, bypassing the global registry. This is primarily useful for testing.
func (*Pipeline) Run ¶
Run executes all configured collectors in parallel, validates their output, deduplicates signals, and returns the aggregated ScanResult. Each collector runs in its own goroutine using errgroup with context cancellation. Results are collected with proper synchronization and returned in deterministic order matching the input collector list.
Error handling is controlled per-collector via ErrorMode in CollectorOpts:
- Skip: errors are silently ignored
- Warn: errors are logged, pipeline continues (default)
- Fail: first error aborts the entire scan
Signals are deduplicated via content-based hashing (Source + Kind + FilePath + Line + Title). When duplicates are found, the first occurrence is kept and its confidence is updated if a later duplicate has a higher value.
Invalid signals are logged and skipped.
func (*Pipeline) SetProgress ¶ added in v1.11.0
SetProgress installs progress callbacks that Run invokes as collectors finish and while long-running collectors are still in flight. Nil callbacks are ignored. Must be called before Run.
type Progress ¶ added in v1.11.0
type Progress struct {
// OnCollectorDone is invoked the moment a collector finishes, before the
// remaining collectors complete. The result includes name, signals,
// duration and any error.
OnCollectorDone func(result signal.CollectorResult)
// OnHeartbeat is invoked every HeartbeatInterval while at least one
// collector is still running, listing the unfinished collectors and how
// long each has been running.
OnHeartbeat func(running []RunningCollector)
}
Progress holds optional callbacks that Run invokes to report progress while collectors execute. The pipeline itself carries no logging policy; callers (e.g. the scan and report commands) decide how to surface these events.
Callbacks are invoked serially — never concurrently with each other — and should return quickly, since they hold up result bookkeeping.
type RunningCollector ¶ added in v1.11.0
type RunningCollector struct {
// Name is the collector name.
Name string
// Elapsed is how long the collector has been running so far.
Elapsed time.Duration
}
RunningCollector describes a collector that has not yet finished, as reported by Progress.OnHeartbeat.
type ValidationError ¶
type ValidationError struct {
// Field is the struct field that failed validation.
Field string
// Message describes what went wrong.
Message string
}
ValidationError describes a single validation failure for a RawSignal.
func ValidateSignal ¶
func ValidateSignal(s signal.RawSignal) []ValidationError
ValidateSignal checks a RawSignal for validity and returns all validation errors found. An empty slice means the signal is valid.
func (ValidationError) Error ¶
func (v ValidationError) Error() string
Error implements the error interface.
Source Files