From 71d845faae2cf3885b7fd5ef354e55f2ff7960fd Mon Sep 17 00:00:00 2001 From: maxlerebourg Date: Sat, 6 Jun 2026 11:07:18 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=EF=B8=8F=20Do=20not=20block=20middlew?= =?UTF-8?q?are=20startup=20due=20to=20stream=20intialization=20(#281)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * :bento: not block on stream startup * :bento: fix * :bento: fix lint * :bento: fix lint * :bento: fix lint * :bento: update readme * :books: clarify StreamStartupBlock warning in README Make the fail-open implication of StreamStartupBlock=false explicit: banned IPs are allowed through until the first stream sync completes. * :bug: clear isCrowdsecStreamStartup on alreadyUpdated path When cacheTimeoutKey is already set (another instance/process has populated the stream cache within the update window), the early return in handleStreamCache used to leave isCrowdsecStreamStartup at true, causing this instance to keep sending startup=true on subsequent ticks. Flip the flag in the early-return path so startup is correctly tracked across multi-instance deployments sharing a cache. --------- Co-authored-by: mhx --- README.md | 7 +++++++ bouncer.go | 17 +++++++++++------ pkg/configuration/configuration.go | 2 ++ 3 files changed, 20 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 84bb9b0..43bea45 100644 --- a/README.md +++ b/README.md @@ -465,6 +465,12 @@ make run - int64 - default: 0 - Used only in `stream` and `alone` mode, the maximum number of time we can not reach Crowdsec before blocking traffic (set -1 to never block) +- StreamStartupBlock + - bool + - default: true + - Used only in `stream` and `alone` mode, controls whether the initial stream update runs synchronously or asynchronously during plugin initialization + - When `true`, plugin initialization waits for Crowdsec to be ready before serving traffic. + - **Warning**: When `false`, all requests bypass remediation until the first stream sync completes — banned IPs will be allowed through during this window. Only disable when startup availability is more important than blocking at startup. - DefaultDecisionSeconds - int64 - default: 60 @@ -598,6 +604,7 @@ http: LogFilePath: "" updateIntervalSeconds: 60 updateMaxFailure: 0 + streamStartupBlock: true defaultDecisionSeconds: 60 remediationStatusCode: 403 httpTimeoutSeconds: 10 diff --git a/bouncer.go b/bouncer.go index a723554..39037bd 100644 --- a/bouncer.go +++ b/bouncer.go @@ -64,7 +64,7 @@ const ( //nolint:gochecknoglobals var ( - isStartup = true + isCrowdsecStreamStartup = true isCrowdsecStreamHealthy = true updateFailure int64 streamTicker chan bool @@ -123,7 +123,7 @@ type Bouncer struct { // New creates the crowdsec bouncer plugin. // -//nolint:gocyclo +//nolint:nestif,gocyclo,gocognit func New(_ context.Context, next http.Handler, config *configuration.Config, name string) (http.Handler, error) { config.LogLevel = strings.ToUpper(config.LogLevel) log := logger.NewWithFormat(config.LogLevel, config.LogFilePath, config.LogFormat) @@ -291,8 +291,11 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam return nil, err } } - handleStreamTicker(bouncer) - isStartup = false + if config.StreamStartupBlock { + handleStreamTicker(bouncer) + } else { + go handleStreamTicker(bouncer) + } streamTicker = startTicker("stream", config.UpdateIntervalSeconds, log, func() { handleStreamTicker(bouncer) }) @@ -301,7 +304,7 @@ func New(_ context.Context, next http.Handler, config *configuration.Config, nam // Start metrics ticker if not already running if metricsTicker == nil && config.MetricsUpdateIntervalSeconds > 0 { lastMetricsPush = time.Now() // Initialize lastMetricsPush when starting the metrics ticker - handleMetricsTicker(bouncer) + go handleMetricsTicker(bouncer) metricsTicker = startTicker("metrics", config.MetricsUpdateIntervalSeconds, log, func() { handleMetricsTicker(bouncer) }) @@ -624,6 +627,7 @@ func handleStreamCache(bouncer *Bouncer) error { _, err := bouncer.cacheClient.Get(cacheTimeoutKey) if err == nil { bouncer.log.Debug("handleStreamCache:alreadyUpdated") + isCrowdsecStreamStartup = false return nil } if err.Error() != cache.CacheMiss { @@ -634,7 +638,7 @@ func handleStreamCache(bouncer *Bouncer) error { Scheme: bouncer.crowdsecScheme, Host: bouncer.crowdsecHost, Path: bouncer.crowdsecPath + bouncer.crowdsecStreamRoute, - RawQuery: fmt.Sprintf("startup=%t", !isCrowdsecStreamHealthy || isStartup), + RawQuery: fmt.Sprintf("startup=%t", !isCrowdsecStreamHealthy || isCrowdsecStreamStartup), } body, err := crowdsecQuery(bouncer, streamRouteURL.String(), nil) if err != nil { @@ -664,6 +668,7 @@ func handleStreamCache(bouncer *Bouncer) error { bouncer.cacheClient.Delete(decision.Value) } bouncer.log.Debug("handleStreamCache:updated") + isCrowdsecStreamStartup = false return nil } diff --git a/pkg/configuration/configuration.go b/pkg/configuration/configuration.go index acd34ad..8cae113 100644 --- a/pkg/configuration/configuration.go +++ b/pkg/configuration/configuration.go @@ -84,6 +84,7 @@ type Config struct { UpdateIntervalSeconds int64 `json:"updateIntervalSeconds,omitempty"` MetricsUpdateIntervalSeconds int64 `json:"metricsUpdateIntervalSeconds,omitempty"` UpdateMaxFailure int64 `json:"updateMaxFailure,omitempty"` + StreamStartupBlock bool `json:"streamStartupBlock,omitempty"` DefaultDecisionSeconds int64 `json:"defaultDecisionSeconds,omitempty"` RemediationStatusCode int `json:"remediationStatusCode,omitempty"` HTTPTimeoutSeconds int64 `json:"httpTimeoutSeconds,omitempty"` @@ -146,6 +147,7 @@ func New() *Config { UpdateIntervalSeconds: 60, MetricsUpdateIntervalSeconds: 600, UpdateMaxFailure: 0, + StreamStartupBlock: true, DefaultDecisionSeconds: 60, RemediationStatusCode: http.StatusForbidden, HTTPTimeoutSeconds: 10,