From 82ee57356115dbb34766829f402c9037fb5084a5 Mon Sep 17 00:00:00 2001 From: Sebastien Blot Date: Wed, 9 Sep 2026 12:26:24 +0200 Subject: [PATCH 1/6] postoverflow: run them in a goroutine to not block writers, enforce max in-flight alerts, and warn if queue is becoming too full --- cmd/crowdsec/crowdsec.go | 2 +- cmd/crowdsec/output.go | 216 ++++++++++++++++++++++---- cmd/crowdsec/output_test.go | 109 +++++++++++++ pkg/csconfig/crowdsec_service.go | 71 ++++++++- pkg/csconfig/crowdsec_service_test.go | 133 +++++++++++++++- pkg/metrics/global.go | 20 +++ pkg/metrics/metrics.go | 2 + 7 files changed, 505 insertions(+), 48 deletions(-) create mode 100644 cmd/crowdsec/output_test.go diff --git a/cmd/crowdsec/crowdsec.go b/cmd/crowdsec/crowdsec.go index 2d86ba134eb..f23b0c1c681 100644 --- a/cmd/crowdsec/crowdsec.go +++ b/cmd/crowdsec/crowdsec.go @@ -130,7 +130,7 @@ func startOutputRoutines(ctx context.Context, cConfig *csconfig.Config, parsers log.WithField("idx", idx).Info("Starting output routine") outputsTomb.Go(func() error { defer trace.ReportPanic() - return runOutput(ctx, inEvents, outEvents, bucketStore, *parsers.PovfwCtx, parsers.Povfwnodes, apiClient, sd) + return runOutput(ctx, idx, inEvents, outEvents, bucketStore, parsers, apiClient, sd, cConfig.Crowdsec.PostOverflowQueueSize) }) } } diff --git a/cmd/crowdsec/output.go b/cmd/crowdsec/output.go index dc06a29e97f..d314b139af3 100644 --- a/cmd/crowdsec/output.go +++ b/cmd/crowdsec/output.go @@ -3,18 +3,28 @@ package main import ( "context" "fmt" + "strconv" "sync" "time" log "github.com/sirupsen/logrus" + "github.com/crowdsecurity/go-cs-lib/trace" + "github.com/crowdsecurity/crowdsec/pkg/apiclient" leaky "github.com/crowdsecurity/crowdsec/pkg/leakybucket" + "github.com/crowdsecurity/crowdsec/pkg/metrics" "github.com/crowdsecurity/crowdsec/pkg/models" "github.com/crowdsecurity/crowdsec/pkg/parser" "github.com/crowdsecurity/crowdsec/pkg/pipeline" + + "github.com/prometheus/client_golang/prometheus" ) +// Has to stay under the 3s ShutdownCrowdsecRoutines allows outputsTomb, or the +// final flush never runs. +const postOverflowDrainTimeout = 2 * time.Second + type alertBuffer struct { mu sync.Mutex alerts []pipeline.RuntimeAlert @@ -80,24 +90,159 @@ func PushAlerts(ctx context.Context, alerts []pipeline.RuntimeAlert, client *api return nil } +func handleOverflow( + ctx context.Context, + event pipeline.Event, + input chan pipeline.Event, + postOverflowCTX parser.UnixParserCtx, + postOverflowNodes []parser.Node, + sd *StateDumper, + pendingAlerts *alertBuffer, +) error { + event, err := parser.Parse(postOverflowCTX, event, postOverflowNodes, sd.StageParse) + if err != nil { + return fmt.Errorf("postoverflow failed: %w", err) + } + + ov := event.Overflow + log.Info(*ov.Alert.Message) + + // if the Alert is nil, it's to signal bucket is ready for GC, don't track this + // dump after postoveflow processing to avoid missing whitelist info + // Appended without a lock, as before this ran in a worker: dump mode is + // single-routine in practice, and output_routines > 1 already raced here. + if flags.DumpDir != "" && ov.Alert != nil { + sd.BucketOverflows = append(sd.BucketOverflows, event) + } + + if ov.Whitelisted { + log.Infof("[%s] is whitelisted, skip.", *ov.Alert.Message) + return nil + } + + if ov.Reprocess { + select { + case input <- event: + log.Debug("Reprocessing overflow event") + case <-ctx.Done(): + log.Debug("Reprocessing overflow event: parsing is dead, skipping") + } + } + + if flags.DumpDir != "" { + return nil + } + + pendingAlerts.add(ov) + + return nil +} + +type overflowProcessor func(context.Context, pipeline.Event) error + +// Decouples the pipeline from postoverflow latency: parsing inline in outputLoop +// backpressured the whole engine, down to the appsec in-band responses (#4600). +func postOverflowWorker(ctx context.Context, queue chan pipeline.Event, process overflowProcessor) error { + for event := range queue { + // the alerts we'd produce past this point have no one left to flush them + select { + case <-ctx.Done(): + return nil + default: + } + + if err := process(ctx, event); err != nil { + return err + } + } + + return nil +} + +// Warns on crossing 75% full, then on falling back under 25%: once we drop, it's +// too late to react. The gap between the two keeps a queue hovering at one +// threshold from logging on every tick. +func warnQueuePressure(depth int, size int, warned bool) bool { + switch { + case depth*4 >= size*3 && !warned: + log.Warnf("postoverflow queue is %d/%d full, a postoverflow parser is slow (dns?): overflows will be dropped if it fills up", depth, size) + return true + case depth*4 < size && warned: + log.Info("postoverflow queue is draining") + return false + } + + return warned +} + func runOutput( ctx context.Context, + idx int, input chan pipeline.Event, overflow chan pipeline.Event, bucketStore *leaky.BucketStore, - postOverflowCTX parser.UnixParserCtx, - postOverflowNodes []parser.Node, + parsers *parser.Parsers, client *apiclient.ApiClient, sd *StateDumper, + queueSize int, ) error { - var pendingAlerts alertBuffer + pendingAlerts := &alertBuffer{} + + process := func(ctx context.Context, event pipeline.Event) error { + return handleOverflow(ctx, event, input, *parsers.PovfwCtx, parsers.Povfwnodes, sd, pendingAlerts) + } + return outputLoop(ctx, idx, overflow, bucketStore, process, client, pendingAlerts, queueSize, outputsTomb.Dying()) +} + +func outputLoop( + ctx context.Context, + idx int, + overflow chan pipeline.Event, + bucketStore *leaky.BucketStore, + process overflowProcessor, + client *apiclient.ApiClient, + pendingAlerts *alertBuffer, + queueSize int, + dying <-chan struct{}, +) error { ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() + // Dump mode (cscli explain, -dsn) parses inline: one-shot, and the dump stays ordered. + inlinePostOverflow := flags.DumpDir != "" + + labels := prometheus.Labels{"routine": strconv.Itoa(idx)} + queueDepth := metrics.GlobalPostOverflowQueueDepth.With(labels) + dropCounter := metrics.GlobalPostOverflowDropped.With(labels) + + povfw := make(chan pipeline.Event, queueSize) + workerErr := make(chan error, 1) + + var ( + warnedPressure bool + droppedSinceTick int + ) + + if !inlinePostOverflow { + go func() { + defer trace.ReportPanic() + workerErr <- postOverflowWorker(ctx, povfw, process) + }() + } + for { select { case <-ticker.C: + depth := len(povfw) + queueDepth.Set(float64(depth)) + warnedPressure = warnQueuePressure(depth, queueSize, warnedPressure) + + if droppedSinceTick > 0 { + log.Warnf("postoverflow queue full, dropped %d overflow(s)", droppedSinceTick) + droppedSinceTick = 0 + } + batch := pendingAlerts.takeAll() if len(batch) == 0 { break @@ -115,13 +260,34 @@ func runOutput( } return nil }) - case <-outputsTomb.Dying(): + case err := <-workerErr: + // a postoverflow failure killed the output routine before, keep that + return err + case <-dying: + if !inlinePostOverflow { + close(povfw) + + timer := time.NewTimer(postOverflowDrainTimeout) + + select { + case err := <-workerErr: + if err != nil { + log.Errorf("while draining postoverflow queue: %s", err) + } + case <-timer.C: + log.Warnf("timeout draining the postoverflow queue, %d overflow(s) lost", len(povfw)) + } + + timer.Stop() + } + batch := pendingAlerts.takeAll() if len(batch) > 0 { if err := PushAlerts(ctx, batch, client); err != nil { log.Errorf("while pushing leftovers to api : %s", err) } } + return nil case event := <-overflow: // if alert is empty and mapKey is present, the overflow is just to cleanup bucket @@ -130,40 +296,22 @@ func runOutput( break } - /* process post overflow parser nodes */ - event, err := parser.Parse(postOverflowCTX, event, postOverflowNodes, sd.StageParse) - if err != nil { - return fmt.Errorf("postoverflow failed: %w", err) - } - - ov := event.Overflow - log.Info(*ov.Alert.Message) - - // if the Alert is nil, it's to signal bucket is ready for GC, don't track this - // dump after postoveflow processing to avoid missing whitelist info - if flags.DumpDir != "" && ov.Alert != nil { - sd.BucketOverflows = append(sd.BucketOverflows, event) - } - - if ov.Whitelisted { - log.Infof("[%s] is whitelisted, skip.", *ov.Alert.Message) - continue - } - - if ov.Reprocess { - select { - case input <- event: - log.Debug("Reprocessing overflow event") - case <-ctx.Done(): - log.Debug("Reprocessing overflow event: parsing is dead, skipping") + if inlinePostOverflow { + if err := process(ctx, event); err != nil { + return err } - } - if flags.DumpDir != "" { - continue + break } - pendingAlerts.add(ov) + select { + case povfw <- event: + default: + // Dropping is the fail-safe direction: postoverflow runs the + // whitelists, so an unparsed alert would mean false positive bans. + dropCounter.Inc() + droppedSinceTick++ + } } } } diff --git a/cmd/crowdsec/output_test.go b/cmd/crowdsec/output_test.go new file mode 100644 index 00000000000..77e6bdd01ed --- /dev/null +++ b/cmd/crowdsec/output_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/crowdsecurity/crowdsec/pkg/models" + "github.com/crowdsecurity/crowdsec/pkg/pipeline" +) + +func testOverflow(msg string) pipeline.Event { + return pipeline.Event{ + Overflow: pipeline.RuntimeAlert{ + Alert: &models.Alert{Message: &msg}, + }, + } +} + +// Fails if the output loop is not reading: before #4600 a slow postoverflow +// parser blocked this send, backpressuring the engine up to the appsec responses. +func send(t *testing.T, overflow chan pipeline.Event, evt pipeline.Event) { + t.Helper() + + select { + case overflow <- evt: + case <-time.After(2 * time.Second): + require.FailNow(t, "output loop stopped reading overflows") + } +} + +func TestOutputLoopSlowPostOverflow(t *testing.T) { + const queueSize = 4 + + ctx := t.Context() + + release := make(chan struct{}) + processed := make(chan pipeline.Event, 16) + + process := func(_ context.Context, evt pipeline.Event) error { + <-release + processed <- evt + + return nil + } + + overflow := make(chan pipeline.Event) + pendingAlerts := &alertBuffer{} + + loopErr := make(chan error, 1) + dying := make(chan struct{}) + + go func() { + loopErr <- outputLoop(ctx, 0, overflow, nil, process, nil, pendingAlerts, queueSize, dying) + }() + + total := queueSize * 4 + for range total { + send(t, overflow, testOverflow("overflow")) + } + + close(release) + + require.Eventually(t, func() bool { + return len(processed) >= queueSize + }, 5*time.Second, 10*time.Millisecond) + + // the queue is bounded, so the excess was dropped rather than buffered + time.Sleep(100 * time.Millisecond) + require.Less(t, len(processed), total, "nothing was dropped") + + close(dying) + + select { + case err := <-loopErr: + require.NoError(t, err) + case <-time.After(postOverflowDrainTimeout + 2*time.Second): + require.FailNow(t, "output loop did not stop") + } +} + +func TestWarnQueuePressure(t *testing.T) { + tests := []struct { + name string + depth int + size int + warned bool + expected bool + }{ + {name: "quiet below the high water mark", depth: 5, size: 100, warned: false, expected: false}, + {name: "warns on crossing 75%", depth: 75, size: 100, warned: false, expected: true}, + {name: "stays warned while still high", depth: 90, size: 100, warned: true, expected: true}, + { + // a queue hovering just under the high water mark would otherwise + // re-arm and warn again on every tick + name: "stays warned between the two marks", depth: 50, size: 100, warned: true, expected: true, + }, + {name: "re-arms under 25%", depth: 24, size: 100, warned: true, expected: false}, + {name: "re-arms when empty", depth: 0, size: 100, warned: true, expected: false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.expected, warnQueuePressure(tc.depth, tc.size, tc.warned)) + }) + } +} diff --git a/pkg/csconfig/crowdsec_service.go b/pkg/csconfig/crowdsec_service.go index dcdfbdb4c3f..8ccbad466d6 100644 --- a/pkg/csconfig/crowdsec_service.go +++ b/pkg/csconfig/crowdsec_service.go @@ -23,6 +23,7 @@ type CrowdsecServiceCfg struct { ParserRoutinesCount int `yaml:"parser_routines"` BucketsRoutinesCount int `yaml:"buckets_routines"` OutputRoutinesCount int `yaml:"output_routines"` + Pipeline *PipelineCfg `yaml:"pipeline,omitempty"` SimulationConfig SimulationConfig `yaml:"-"` BucketStateFile string `yaml:"state_input_file,omitempty"` // if we need to unserialize buckets at start BucketStateDumpDir string `yaml:"state_output_dir,omitempty"` // if we need to unserialize buckets on shutdown @@ -31,6 +32,59 @@ type CrowdsecServiceCfg struct { SimulationFilePath string `yaml:"-"` ContextToSend map[string][]string `yaml:"-"` + + PostOverflowQueueSize int `yaml:"-"` // resolved from Pipeline +} + +type PipelineStageCfg struct { + Routines *int `yaml:"routines,omitempty"` +} + +// The output stage is the only one with a queue in front of it, for postoverflow. +type PipelineOutputCfg struct { + Routines *int `yaml:"routines,omitempty"` + QueueSize *int `yaml:"queue_size,omitempty"` +} + +// Supersedes the flat *_routines keys, which remain the fallback when unset here. +type PipelineCfg struct { + Parser *PipelineStageCfg `yaml:"parser,omitempty"` + Buckets *PipelineStageCfg `yaml:"buckets,omitempty"` + Output *PipelineOutputCfg `yaml:"output,omitempty"` +} + +// ~30x the worst case burst in #4600 (arrival rate x the 3s dnscache bound), and +// small enough that a stuck pipeline hits the drop counter in minutes. +const defaultPostOverflowQueueSize = 256 + +// The nested value wins; warn when the legacy key disagrees so it doesn't look effective. +func resolveRoutines(stage string, cfg *PipelineStageCfg, legacy int) int { + if cfg == nil || cfg.Routines == nil { + if legacy <= 0 { + return 1 + } + + return legacy + } + + n := *cfg.Routines + if n <= 0 { + n = 1 + } + + if legacy > 1 && legacy != n { + log.Warnf("pipeline.%s.routines (%d) overrides %s_routines (%d)", stage, n, stage, legacy) + } + + return n +} + +func resolveQueueSize(cfg *PipelineOutputCfg, def int) int { + if cfg == nil || cfg.QueueSize == nil || *cfg.QueueSize <= 0 { + return def + } + + return *cfg.QueueSize } // Cache config for DNS lookups (legit bots, rdns PO) @@ -143,17 +197,20 @@ func (c *Config) LoadCrowdsec() error { return fmt.Errorf("load error (simulation): %w", err) } - if c.Crowdsec.ParserRoutinesCount <= 0 { - c.Crowdsec.ParserRoutinesCount = 1 + pipelineCfg := c.Crowdsec.Pipeline + if pipelineCfg == nil { + pipelineCfg = &PipelineCfg{} } - if c.Crowdsec.BucketsRoutinesCount <= 0 { - c.Crowdsec.BucketsRoutinesCount = 1 + c.Crowdsec.ParserRoutinesCount = resolveRoutines("parser", pipelineCfg.Parser, c.Crowdsec.ParserRoutinesCount) + c.Crowdsec.BucketsRoutinesCount = resolveRoutines("buckets", pipelineCfg.Buckets, c.Crowdsec.BucketsRoutinesCount) + outputRoutines := &PipelineStageCfg{} + if pipelineCfg.Output != nil { + outputRoutines.Routines = pipelineCfg.Output.Routines } - if c.Crowdsec.OutputRoutinesCount <= 0 { - c.Crowdsec.OutputRoutinesCount = 1 - } + c.Crowdsec.OutputRoutinesCount = resolveRoutines("output", outputRoutines, c.Crowdsec.OutputRoutinesCount) + c.Crowdsec.PostOverflowQueueSize = resolveQueueSize(pipelineCfg.Output, defaultPostOverflowQueueSize) if err = c.LoadAPIClient(); err != nil { return fmt.Errorf("loading api client: %w", err) diff --git a/pkg/csconfig/crowdsec_service_test.go b/pkg/csconfig/crowdsec_service_test.go index 7c6c1fe5cf2..4a3f3e43901 100644 --- a/pkg/csconfig/crowdsec_service_test.go +++ b/pkg/csconfig/crowdsec_service_test.go @@ -64,6 +64,7 @@ func TestLoadCrowdsec(t *testing.T) { BucketsRoutinesCount: 1, ParserRoutinesCount: 1, OutputRoutinesCount: 1, + PostOverflowQueueSize: defaultPostOverflowQueueSize, ConsoleContextValueLength: 2500, AcquisitionFiles: []string{acquisFullPath}, SimulationFilePath: "./testdata/simulation.yaml", @@ -101,6 +102,7 @@ func TestLoadCrowdsec(t *testing.T) { BucketsRoutinesCount: 1, ParserRoutinesCount: 1, OutputRoutinesCount: 1, + PostOverflowQueueSize: defaultPostOverflowQueueSize, ConsoleContextValueLength: 0, AcquisitionFiles: []string{acquisFullPath, acquisInDirFullPath}, // context is loaded in pkg/alertcontext @@ -136,6 +138,7 @@ func TestLoadCrowdsec(t *testing.T) { BucketsRoutinesCount: 1, ParserRoutinesCount: 1, OutputRoutinesCount: 1, + PostOverflowQueueSize: defaultPostOverflowQueueSize, ConsoleContextValueLength: 10, AcquisitionFiles: []string{}, SimulationFilePath: "", @@ -164,12 +167,13 @@ func TestLoadCrowdsec(t *testing.T) { }, }, expected: &CrowdsecServiceCfg{ - Enable: new(true), - AcquisitionFilePath: notExistFullPath, - AcquisitionFiles: []string{}, - ParserRoutinesCount: 1, - OutputRoutinesCount: 1, - BucketsRoutinesCount: 1, + Enable: new(true), + AcquisitionFilePath: notExistFullPath, + AcquisitionFiles: []string{}, + ParserRoutinesCount: 1, + OutputRoutinesCount: 1, + PostOverflowQueueSize: defaultPostOverflowQueueSize, + BucketsRoutinesCount: 1, }, }, { @@ -224,3 +228,120 @@ dns_cache: require.NoError(t, yaml.Unmarshal([]byte("acquisition_path: ./testdata/acquis.yaml"), &bare)) assert.Nil(t, bare.DNSCache) } + +func TestPipelineCfg(t *testing.T) { + tests := []struct { + name string + yamlConfig string + expectedParser int + expectedBuckets int + expectedOutput int + expectedQueue int + }{ + { + name: "nothing set", + yamlConfig: "", + expectedParser: 1, + expectedBuckets: 1, + expectedOutput: 1, + expectedQueue: 256, + }, + { + name: "legacy keys only", + yamlConfig: ` +parser_routines: 4 +buckets_routines: 2 +output_routines: 3 +`, + expectedParser: 4, + expectedBuckets: 2, + expectedOutput: 3, + expectedQueue: 256, + }, + { + name: "nested keys only", + yamlConfig: ` +pipeline: + parser: + routines: 4 + buckets: + routines: 2 + output: + routines: 3 + queue_size: 512 +`, + expectedParser: 4, + expectedBuckets: 2, + expectedOutput: 3, + expectedQueue: 512, + }, + { + // config.yaml ships parser_routines, so this is what an upgrade looks like + name: "nested overrides legacy", + yamlConfig: ` +parser_routines: 1 +pipeline: + parser: + routines: 8 +`, + expectedParser: 8, + expectedBuckets: 1, + expectedOutput: 1, + expectedQueue: 256, + }, + { + name: "a negative legacy value falls back to the default", + yamlConfig: "parser_routines: -1", + expectedParser: 1, + expectedBuckets: 1, + expectedOutput: 1, + expectedQueue: 256, + }, + { + name: "zero and negative values fall back to the defaults", + yamlConfig: ` +pipeline: + parser: + routines: 0 + output: + routines: -1 + queue_size: 0 +`, + expectedParser: 1, + expectedBuckets: 1, + expectedOutput: 1, + expectedQueue: 256, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + crowdsecCfg := CrowdsecServiceCfg{} + require.NoError(t, yaml.Unmarshal([]byte(tc.yamlConfig), &crowdsecCfg)) + + crowdsecCfg.AcquisitionFilePath = "./testdata/acquis.yaml" + crowdsecCfg.SimulationFilePath = "./testdata/simulation.yaml" + + cfg := &Config{ + ConfigPaths: &ConfigurationPaths{ + ConfigDir: "./testdata", + DataDir: "./data", + HubDir: "./hub", + }, + API: &APICfg{ + Client: &LocalApiClientCfg{ + CredentialsFilePath: "./testdata/lapi-secrets.yaml", + }, + }, + Crowdsec: &crowdsecCfg, + } + + require.NoError(t, cfg.LoadCrowdsec()) + + require.Equal(t, tc.expectedParser, cfg.Crowdsec.ParserRoutinesCount) + require.Equal(t, tc.expectedBuckets, cfg.Crowdsec.BucketsRoutinesCount) + require.Equal(t, tc.expectedOutput, cfg.Crowdsec.OutputRoutinesCount) + require.Equal(t, tc.expectedQueue, cfg.Crowdsec.PostOverflowQueueSize) + }) + } +} diff --git a/pkg/metrics/global.go b/pkg/metrics/global.go index 5d9143db4c6..97adbe4d872 100644 --- a/pkg/metrics/global.go +++ b/pkg/metrics/global.go @@ -114,3 +114,23 @@ var GlobalPourHistogram = prometheus.NewHistogramVec( }, []string{"type", "source"}, ) + +const GlobalPostOverflowQueueDepthMetricName = "cs_postoverflow_queue_depth" + +var GlobalPostOverflowQueueDepth = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: GlobalPostOverflowQueueDepthMetricName, + Help: "Overflows waiting for postoverflow parsing.", + }, + []string{"routine"}, +) + +const GlobalPostOverflowDroppedMetricName = "cs_postoverflow_dropped_total" + +var GlobalPostOverflowDropped = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: GlobalPostOverflowDroppedMetricName, + Help: "Total overflows dropped because the postoverflow queue was full.", + }, + []string{"routine"}, +) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 766dd349aac..5921108f763 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -39,6 +39,7 @@ func RegisterMetrics(metricsLevel MetricsLevelConfig) error { LapiRouteHits, BucketsCurrentCount, CacheMetrics, RegexpCacheMetrics, NodesWlHitsOk, NodesWlHits, + GlobalPostOverflowQueueDepth, GlobalPostOverflowDropped, PapiOrdersReceived, PapiInvalidOrdersReceived, PapiLastPullTimestamp, PapiPollErrors) case MetricsLevelFull: prometheus.MustRegister(GlobalParserHits, GlobalParserHitsOk, GlobalParserHitsKo, @@ -47,6 +48,7 @@ func RegisterMetrics(metricsLevel MetricsLevelConfig) error { LapiRouteHits, LapiMachineHits, LapiBouncerHits, LapiNilDecisions, LapiNonNilDecisions, LapiResponseTime, BucketsPour, BucketsUnderflow, BucketsCanceled, BucketsInstantiation, BucketsOverflow, BucketsCurrentCount, GlobalActiveDecisions, GlobalAlerts, GlobalMachinesLastHeartbeatTimestamp, NodesWlHitsOk, NodesWlHits, + GlobalPostOverflowQueueDepth, GlobalPostOverflowDropped, CacheMetrics, RegexpCacheMetrics, PapiOrdersReceived, PapiInvalidOrdersReceived, PapiLastPullTimestamp, PapiPollErrors) default: From 9e36d7c4e1f5a787c4617ae39e9609d7effbc436 Mon Sep 17 00:00:00 2001 From: Sebastien Blot Date: Wed, 9 Sep 2026 15:58:15 +0200 Subject: [PATCH 2/6] add dropped overflows to usage metrics --- cmd/crowdsec/lpmetrics.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/cmd/crowdsec/lpmetrics.go b/cmd/crowdsec/lpmetrics.go index 1359296560c..244b8a5d3ac 100644 --- a/cmd/crowdsec/lpmetrics.go +++ b/cmd/crowdsec/lpmetrics.go @@ -298,6 +298,12 @@ func (m *MetricsProvider) getAppsecBlockedMetrics() []*models.MetricsDetailItem }, nil, "appsec_blocked", "request") } +func (m *MetricsProvider) getPostOverflowDroppedMetrics() []*models.MetricsDetailItem { + return m.gatherPromMetrics([]string{metrics.GlobalPostOverflowDroppedMetricName}, labelsMapping{ + "routine": "routine", + }, nil, "postoverflow_dropped", "overflow") +} + func (m *MetricsProvider) metricsPayload() *models.AllMetrics { os := &models.OSversion{ Name: new(m.static.osName), @@ -373,6 +379,11 @@ func (m *MetricsProvider) metricsPayload() *models.AllMetrics { met.Metrics[0].Items = append(met.Metrics[0].Items, appsecBlockedMetrics...) } + postOverflowDroppedMetrics := m.getPostOverflowDroppedMetrics() + if len(postOverflowDroppedMetrics) > 0 { + met.Metrics[0].Items = append(met.Metrics[0].Items, postOverflowDroppedMetrics...) + } + return &models.AllMetrics{ LogProcessors: []*models.LogProcessorsMetrics{met}, } From 0bf3725ff1aab9ecd0485cebea464cf9e4d522c3 Mon Sep 17 00:00:00 2001 From: Sebastien Blot Date: Wed, 9 Sep 2026 15:59:09 +0200 Subject: [PATCH 3/6] be more consistent about s --- pkg/csconfig/crowdsec_service.go | 16 ++++++++-------- pkg/csconfig/crowdsec_service_test.go | 2 +- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/pkg/csconfig/crowdsec_service.go b/pkg/csconfig/crowdsec_service.go index 8ccbad466d6..e9e01f04a51 100644 --- a/pkg/csconfig/crowdsec_service.go +++ b/pkg/csconfig/crowdsec_service.go @@ -48,9 +48,9 @@ type PipelineOutputCfg struct { // Supersedes the flat *_routines keys, which remain the fallback when unset here. type PipelineCfg struct { - Parser *PipelineStageCfg `yaml:"parser,omitempty"` - Buckets *PipelineStageCfg `yaml:"buckets,omitempty"` - Output *PipelineOutputCfg `yaml:"output,omitempty"` + Parser *PipelineStageCfg `yaml:"parser,omitempty"` + Bucket *PipelineStageCfg `yaml:"bucket,omitempty"` + Output *PipelineOutputCfg `yaml:"output,omitempty"` } // ~30x the worst case burst in #4600 (arrival rate x the 3s dnscache bound), and @@ -58,7 +58,7 @@ type PipelineCfg struct { const defaultPostOverflowQueueSize = 256 // The nested value wins; warn when the legacy key disagrees so it doesn't look effective. -func resolveRoutines(stage string, cfg *PipelineStageCfg, legacy int) int { +func resolveRoutines(stage string, legacyKey string, cfg *PipelineStageCfg, legacy int) int { if cfg == nil || cfg.Routines == nil { if legacy <= 0 { return 1 @@ -73,7 +73,7 @@ func resolveRoutines(stage string, cfg *PipelineStageCfg, legacy int) int { } if legacy > 1 && legacy != n { - log.Warnf("pipeline.%s.routines (%d) overrides %s_routines (%d)", stage, n, stage, legacy) + log.Warnf("pipeline.%s.routines (%d) overrides %s (%d)", stage, n, legacyKey, legacy) } return n @@ -202,14 +202,14 @@ func (c *Config) LoadCrowdsec() error { pipelineCfg = &PipelineCfg{} } - c.Crowdsec.ParserRoutinesCount = resolveRoutines("parser", pipelineCfg.Parser, c.Crowdsec.ParserRoutinesCount) - c.Crowdsec.BucketsRoutinesCount = resolveRoutines("buckets", pipelineCfg.Buckets, c.Crowdsec.BucketsRoutinesCount) + c.Crowdsec.ParserRoutinesCount = resolveRoutines("parser", "parser_routines", pipelineCfg.Parser, c.Crowdsec.ParserRoutinesCount) + c.Crowdsec.BucketsRoutinesCount = resolveRoutines("bucket", "buckets_routines", pipelineCfg.Bucket, c.Crowdsec.BucketsRoutinesCount) outputRoutines := &PipelineStageCfg{} if pipelineCfg.Output != nil { outputRoutines.Routines = pipelineCfg.Output.Routines } - c.Crowdsec.OutputRoutinesCount = resolveRoutines("output", outputRoutines, c.Crowdsec.OutputRoutinesCount) + c.Crowdsec.OutputRoutinesCount = resolveRoutines("output", "output_routines", outputRoutines, c.Crowdsec.OutputRoutinesCount) c.Crowdsec.PostOverflowQueueSize = resolveQueueSize(pipelineCfg.Output, defaultPostOverflowQueueSize) if err = c.LoadAPIClient(); err != nil { diff --git a/pkg/csconfig/crowdsec_service_test.go b/pkg/csconfig/crowdsec_service_test.go index 4a3f3e43901..4ab83652e49 100644 --- a/pkg/csconfig/crowdsec_service_test.go +++ b/pkg/csconfig/crowdsec_service_test.go @@ -264,7 +264,7 @@ output_routines: 3 pipeline: parser: routines: 4 - buckets: + bucket: routines: 2 output: routines: 3 From 990c562a4c0511dffed848a1cf5f76e94bf1e82a Mon Sep 17 00:00:00 2001 From: Sebastien Blot Date: Wed, 9 Sep 2026 16:04:11 +0200 Subject: [PATCH 4/6] limit spam of dropped overflow logging --- cmd/crowdsec/output.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/cmd/crowdsec/output.go b/cmd/crowdsec/output.go index d314b139af3..8ed1a6dc6b1 100644 --- a/cmd/crowdsec/output.go +++ b/cmd/crowdsec/output.go @@ -25,6 +25,9 @@ import ( // final flush never runs. const postOverflowDrainTimeout = 2 * time.Second +// Drops come in bursts, and the counter carries the exact number anyway. +const postOverflowDropWarnInterval = time.Minute + type alertBuffer struct { mu sync.Mutex alerts []pipeline.RuntimeAlert @@ -221,7 +224,8 @@ func outputLoop( var ( warnedPressure bool - droppedSinceTick int + droppedSinceWarn int + lastDropWarn time.Time ) if !inlinePostOverflow { @@ -238,9 +242,10 @@ func outputLoop( queueDepth.Set(float64(depth)) warnedPressure = warnQueuePressure(depth, queueSize, warnedPressure) - if droppedSinceTick > 0 { - log.Warnf("postoverflow queue full, dropped %d overflow(s)", droppedSinceTick) - droppedSinceTick = 0 + if droppedSinceWarn > 0 && time.Since(lastDropWarn) >= postOverflowDropWarnInterval { + log.Warnf("postoverflow queue full, dropped %d overflow(s) since the last warning", droppedSinceWarn) + droppedSinceWarn = 0 + lastDropWarn = time.Now() } batch := pendingAlerts.takeAll() @@ -310,7 +315,7 @@ func outputLoop( // Dropping is the fail-safe direction: postoverflow runs the // whitelists, so an unparsed alert would mean false positive bans. dropCounter.Inc() - droppedSinceTick++ + droppedSinceWarn++ } } } From f1c320a171a261d94bef39468533c6ff732c48d9 Mon Sep 17 00:00:00 2001 From: Sebastien Blot Date: Wed, 9 Sep 2026 16:04:35 +0200 Subject: [PATCH 5/6] unify logging of postoverflow errors --- cmd/crowdsec/output.go | 60 ++++++++++++++++++------------------- cmd/crowdsec/output_test.go | 4 +-- 2 files changed, 30 insertions(+), 34 deletions(-) diff --git a/cmd/crowdsec/output.go b/cmd/crowdsec/output.go index 8ed1a6dc6b1..862a8e63748 100644 --- a/cmd/crowdsec/output.go +++ b/cmd/crowdsec/output.go @@ -101,12 +101,24 @@ func handleOverflow( postOverflowNodes []parser.Node, sd *StateDumper, pendingAlerts *alertBuffer, -) error { - event, err := parser.Parse(postOverflowCTX, event, postOverflowNodes, sd.StageParse) +) { + parsed, err := parser.Parse(postOverflowCTX, event, postOverflowNodes, sd.StageParse) if err != nil { - return fmt.Errorf("postoverflow failed: %w", err) + scenario := "" + if event.Overflow.Alert != nil && event.Overflow.Alert.Scenario != nil { + scenario = *event.Overflow.Alert.Scenario + } + + log.WithFields(log.Fields{ + "scenario": scenario, + "bucket_id": event.Overflow.BucketId, + "sources": event.Overflow.GetSources(), + }).Errorf("postoverflow failed: %s", err) + + return } + event = parsed ov := event.Overflow log.Info(*ov.Alert.Message) @@ -120,7 +132,7 @@ func handleOverflow( if ov.Whitelisted { log.Infof("[%s] is whitelisted, skip.", *ov.Alert.Message) - return nil + return } if ov.Reprocess { @@ -133,33 +145,27 @@ func handleOverflow( } if flags.DumpDir != "" { - return nil + return } pendingAlerts.add(ov) - - return nil } -type overflowProcessor func(context.Context, pipeline.Event) error +type overflowProcessor func(context.Context, pipeline.Event) // Decouples the pipeline from postoverflow latency: parsing inline in outputLoop // backpressured the whole engine, down to the appsec in-band responses (#4600). -func postOverflowWorker(ctx context.Context, queue chan pipeline.Event, process overflowProcessor) error { +func postOverflowWorker(ctx context.Context, queue chan pipeline.Event, process overflowProcessor) { for event := range queue { // the alerts we'd produce past this point have no one left to flush them select { case <-ctx.Done(): - return nil + return default: } - if err := process(ctx, event); err != nil { - return err - } + process(ctx, event) } - - return nil } // Warns on crossing 75% full, then on falling back under 25%: once we drop, it's @@ -171,7 +177,7 @@ func warnQueuePressure(depth int, size int, warned bool) bool { log.Warnf("postoverflow queue is %d/%d full, a postoverflow parser is slow (dns?): overflows will be dropped if it fills up", depth, size) return true case depth*4 < size && warned: - log.Info("postoverflow queue is draining") + log.Infof("postoverflow queue is draining, back under 25%% (%d/%d)", depth, size) return false } @@ -191,8 +197,8 @@ func runOutput( ) error { pendingAlerts := &alertBuffer{} - process := func(ctx context.Context, event pipeline.Event) error { - return handleOverflow(ctx, event, input, *parsers.PovfwCtx, parsers.Povfwnodes, sd, pendingAlerts) + process := func(ctx context.Context, event pipeline.Event) { + handleOverflow(ctx, event, input, *parsers.PovfwCtx, parsers.Povfwnodes, sd, pendingAlerts) } return outputLoop(ctx, idx, overflow, bucketStore, process, client, pendingAlerts, queueSize, outputsTomb.Dying()) @@ -220,7 +226,7 @@ func outputLoop( dropCounter := metrics.GlobalPostOverflowDropped.With(labels) povfw := make(chan pipeline.Event, queueSize) - workerErr := make(chan error, 1) + workerDone := make(chan struct{}) var ( warnedPressure bool @@ -231,7 +237,8 @@ func outputLoop( if !inlinePostOverflow { go func() { defer trace.ReportPanic() - workerErr <- postOverflowWorker(ctx, povfw, process) + defer close(workerDone) + postOverflowWorker(ctx, povfw, process) }() } @@ -265,9 +272,6 @@ func outputLoop( } return nil }) - case err := <-workerErr: - // a postoverflow failure killed the output routine before, keep that - return err case <-dying: if !inlinePostOverflow { close(povfw) @@ -275,10 +279,7 @@ func outputLoop( timer := time.NewTimer(postOverflowDrainTimeout) select { - case err := <-workerErr: - if err != nil { - log.Errorf("while draining postoverflow queue: %s", err) - } + case <-workerDone: case <-timer.C: log.Warnf("timeout draining the postoverflow queue, %d overflow(s) lost", len(povfw)) } @@ -302,10 +303,7 @@ func outputLoop( } if inlinePostOverflow { - if err := process(ctx, event); err != nil { - return err - } - + process(ctx, event) break } diff --git a/cmd/crowdsec/output_test.go b/cmd/crowdsec/output_test.go index 77e6bdd01ed..4817b32fa83 100644 --- a/cmd/crowdsec/output_test.go +++ b/cmd/crowdsec/output_test.go @@ -39,11 +39,9 @@ func TestOutputLoopSlowPostOverflow(t *testing.T) { release := make(chan struct{}) processed := make(chan pipeline.Event, 16) - process := func(_ context.Context, evt pipeline.Event) error { + process := func(_ context.Context, evt pipeline.Event) { <-release processed <- evt - - return nil } overflow := make(chan pipeline.Event) From 9956a783903fd88c5cee3b53613432e2efbd8ebe Mon Sep 17 00:00:00 2001 From: Sebastien Blot Date: Wed, 9 Sep 2026 16:15:44 +0200 Subject: [PATCH 6/6] up --- cmd/crowdsec/output.go | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/cmd/crowdsec/output.go b/cmd/crowdsec/output.go index 862a8e63748..23d463e2300 100644 --- a/cmd/crowdsec/output.go +++ b/cmd/crowdsec/output.go @@ -104,16 +104,19 @@ func handleOverflow( ) { parsed, err := parser.Parse(postOverflowCTX, event, postOverflowNodes, sd.StageParse) if err != nil { - scenario := "" + // the parser already logged which node failed, so identify the alert we + // are about to lose instead + fields := log.Fields{"sources": event.Overflow.GetSources()} + if event.Overflow.Alert != nil && event.Overflow.Alert.Scenario != nil { - scenario = *event.Overflow.Alert.Scenario + fields["scenario"] = *event.Overflow.Alert.Scenario + } + + if event.Overflow.BucketId != "" { + fields["bucket_id"] = event.Overflow.BucketId } - log.WithFields(log.Fields{ - "scenario": scenario, - "bucket_id": event.Overflow.BucketId, - "sources": event.Overflow.GetSources(), - }).Errorf("postoverflow failed: %s", err) + log.WithFields(fields).Errorf("postoverflow failed, discarding alert: %s", err) return }