x-pack/filebeat/input/streaming: fix CrowdStrike retry cap and soften transient handling (#51712)

x-pack/filebeat/input/streaming: fix CrowdStrike retry cap and soften transient handling

The CrowdStrike FalconHose discover retry loop capped every run at 10
attempts: the unconfigured maximum was checked as an else-if on the
configured branch, so max_attempts greater than 10 and infinite_retries
were silently limited to 10, terminating the input and requiring a
restart during longer upstream outages. Apply the unconfigured cap only
when no retry policy is set so configured limits and infinite_retries
are honoured.

A 200 response with an empty discover body (io.EOF from Decode) is now
reported as a distinct transient error rather than a generic decode
failure, and DEGRADED is only reported after three consecutive failures
so a single transient blip no longer churns the unit health status. The
discover GET failure no longer updates status inside followSession; the
retry loop owns those transitions.

Add regression tests for the retry cap, infinite_retries, deferred
DEGRADED reporting, and the empty-body discover response.

Assisted-By: Cursor (Claude Opus 4.8)

Author: kcreddy

changelog/fragments/1783081692-fix-crowdstrike-streaming-retry-cap.yaml

@@ -0,0 +1,37 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Fix CrowdStrike streaming input retry cap so max_attempts and infinite_retries are honoured.
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: |
+  Fix the CrowdStrike streaming input retry loop so a configured max_attempts
+  greater than 10 and infinite_retries are no longer silently capped at the
+  unconfigured default of 10. The empty-body case from the discover endpoint is
+  now reported as a distinct transient error, and the input no longer reports
+  DEGRADED on the first transient failure.
+
+# Affected component; a word indicating the component this changeset affects.
+component: filebeat
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+#pr: https://github.com/owner/repo/1234
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

x-pack/filebeat/input/streaming/crowdstrike.go

@@ -248,6 +248,10 @@ func (s *falconHoseStream) FollowStream(ctx context.Context) error {
 	var err error
 	attempt := 0
 	const maxAttemptsUnconfigured = 10
+	// Number of consecutive failures tolerated before reporting DEGRADED,
+	// so a single transient blip (e.g. an empty discover response) does not
+	// churn the unit health status.
+	const degradeAfterAttempts = 3
 	for {
 		state, err = s.followSession(ctx, cli, state)
 		if err != nil {
@@ -262,8 +266,14 @@ func (s *falconHoseStream) FollowStream(ctx context.Context) error {
 
 			attempt++
 
-			if s.cfg.Retry != nil && !s.cfg.Retry.InfiniteRetries && attempt >= s.cfg.Retry.MaxAttempts {
-				return fmt.Errorf("max retry attempts (%d) exceeded: %w", s.cfg.Retry.MaxAttempts, err)
+			// The unconfigured cap must only apply when no retry policy is
+			// set. Keeping it as an else-if on the configured branch caused
+			// infinite_retries and max_attempts > 10 to be silently capped
+			// at 10.
+			if s.cfg.Retry != nil {
+				if !s.cfg.Retry.InfiniteRetries && attempt >= s.cfg.Retry.MaxAttempts {
+					return fmt.Errorf("max retry attempts (%d) exceeded: %w", s.cfg.Retry.MaxAttempts, err)
+				}
 			} else if attempt >= maxAttemptsUnconfigured {
 				return fmt.Errorf("max retry attempts (%d unconfigured) exceeded: %w", maxAttemptsUnconfigured, err)
 			}
@@ -280,7 +290,9 @@ func (s *falconHoseStream) FollowStream(ctx context.Context) error {
 				waitTime = rle.wait
 			}
 
-			s.status.UpdateStatus(status.Degraded, err.Error())
+			if attempt >= degradeAfterAttempts {
+				s.status.UpdateStatus(status.Degraded, err.Error())
+			}
 			s.log.Warnw("session warning", "error", err, "attempt", attempt, "wait", waitTime.String())
 
 			select {
@@ -310,9 +322,10 @@ func (s *falconHoseStream) followSession(ctx context.Context, cli *http.Client,
 	}
 	resp, err := cli.Do(req)
 	if err != nil {
-		err = fmt.Errorf("failed GET to discover stream: %w", err)
-		s.status.UpdateStatus(status.Degraded, err.Error())
-		return state, err
+		// Status transitions are owned by the FollowStream retry loop so
+		// that a single transient failure does not immediately report
+		// DEGRADED; just return the error here.
+		return state, fmt.Errorf("failed GET to discover stream: %w", err)
 	}
 	defer resp.Body.Close()
 
@@ -354,6 +367,13 @@ func (s *falconHoseStream) followSession(ctx context.Context, cli *http.Client,
 	}
 	err = dec.Decode(&body)
 	if err != nil {
+		// A 200 response with an empty body yields io.EOF from Decode. This
+		// is a transient upstream condition (the discover endpoint
+		// occasionally returns no content), distinct from a malformed body,
+		// so surface it clearly rather than as a generic decode failure.
+		if errors.Is(err, io.EOF) {
+			return state, errors.New("discover stream returned an empty body")
+		}
 		return state, fmt.Errorf("failed to decode discover body: %w", err)
 	}
 	s.log.Debugw("stream discover metadata", logp.Namespace(s.ns), "meta", mapstr.M(body.Meta))

x-pack/filebeat/input/streaming/crowdstrike_test.go

@@ -24,6 +24,7 @@ import (
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	cursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/monitoring"
@@ -297,6 +298,157 @@ func TestCrowdstrikeOAuthHTTPClientRespectsConfiguredTimeout(t *testing.T) {
 	}
 }
 
+func TestFollowStreamRetryCapHonorsMaxAttempts(t *testing.T) {
+	log := logptest.NewTestingLogger(t, t.Name())
+
+	discoverURL, tokenURL, hits := startEmptyDiscover(t)
+	const maxAttempts = 15 // Deliberately > the unconfigured cap of 10.
+	cfg := emptyDiscoverConfig(t, discoverURL, tokenURL, &retry{MaxAttempts: maxAttempts, WaitMin: time.Millisecond, WaitMax: time.Millisecond})
+	s := newEmptyDiscoverFollower(t, cfg, nil, log)
+
+	err := s.FollowStream(context.Background())
+	if err == nil {
+		t.Fatal("FollowStream() error = nil; want max-attempts error")
+	}
+	// Regression: a configured MaxAttempts greater than 10 must be honoured,
+	// not silently capped at the unconfigured default of 10.
+	want := fmt.Sprintf("max retry attempts (%d) exceeded", maxAttempts)
+	if !strings.Contains(err.Error(), want) {
+		t.Errorf("FollowStream() error = %v; want substring %q", err, want)
+	}
+	if got := hits.Load(); got != int64(maxAttempts) {
+		t.Errorf("discover attempts = %d; want %d", got, maxAttempts)
+	}
+}
+
+func TestFollowStreamInfiniteRetriesDoesNotCap(t *testing.T) {
+	log := logptest.NewTestingLogger(t, t.Name())
+
+	discoverURL, tokenURL, hits := startEmptyDiscover(t)
+	// MaxAttempts is intentionally tiny: InfiniteRetries must override it and
+	// the unconfigured cap of 10 entirely.
+	cfg := emptyDiscoverConfig(t, discoverURL, tokenURL, &retry{InfiniteRetries: true, MaxAttempts: 1, WaitMin: time.Millisecond, WaitMax: time.Millisecond})
+	s := newEmptyDiscoverFollower(t, cfg, nil, log)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- s.FollowStream(ctx) }()
+
+	// Wait until well past the unconfigured cap to prove the input keeps
+	// retrying, then cancel.
+	deadline := time.After(10 * time.Second)
+	for hits.Load() <= 12 {
+		select {
+		case <-deadline:
+			t.Fatalf("discover attempts = %d before timeout; want > 12", hits.Load())
+		case err := <-done:
+			t.Fatalf("FollowStream returned early after %d attempts: %v", hits.Load(), err)
+		case <-time.After(5 * time.Millisecond):
+		}
+	}
+	cancel()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("FollowStream() error = %v; want nil after context cancel", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for FollowStream to return after cancel")
+	}
+}
+
+func TestFollowStreamDefersDegraded(t *testing.T) {
+	log := logptest.NewTestingLogger(t, t.Name())
+
+	discoverURL, tokenURL, _ := startEmptyDiscover(t)
+	// With MaxAttempts=5 and the degrade threshold of 3, the input reports
+	// DEGRADED only on attempts 3 and 4, then terminates on attempt 5 without
+	// reporting DEGRADED on the earlier transient failures.
+	cfg := emptyDiscoverConfig(t, discoverURL, tokenURL, &retry{MaxAttempts: 5, WaitMin: time.Millisecond, WaitMax: time.Millisecond})
+	rec := &degradedRecorder{}
+	s := newEmptyDiscoverFollower(t, cfg, rec, log)
+
+	if err := s.FollowStream(context.Background()); err == nil {
+		t.Fatal("FollowStream() error = nil; want max-attempts error")
+	}
+	if got := rec.count(); got != 2 {
+		t.Errorf("DEGRADED reports = %d; want 2 (attempts 3 and 4 only)", got)
+	}
+}
+
+// startEmptyDiscover starts a token endpoint and a discover endpoint that
+// returns 200 with an empty body (the transient EOF condition observed from
+// CrowdStrike), returning the discover URL, token URL, and a counter of
+// discover requests.
+func startEmptyDiscover(t *testing.T) (discoverURL, tokenURL string, hits *atomic.Int64) {
+	t.Helper()
+	tokenSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"access_token":"token","token_type":"bearer","expires_in":3600}`)
+	}))
+	t.Cleanup(tokenSrv.Close)
+
+	hits = new(atomic.Int64)
+	discoverSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits.Add(1)
+		w.Header().Set("Content-Type", "application/json")
+	}))
+	t.Cleanup(discoverSrv.Close)
+	return discoverSrv.URL, tokenSrv.URL, hits
+}
+
+func emptyDiscoverConfig(t *testing.T, discoverURL, tokenURL string, r *retry) config {
+	t.Helper()
+	u, err := url.Parse(discoverURL)
+	if err != nil {
+		t.Fatalf("failed to parse discover URL: %v", err)
+	}
+	return config{
+		Type: "crowdstrike",
+		URL:  &urlConfig{u},
+		Auth: authConfig{
+			OAuth2: oAuth2Config{
+				ClientID:     "id",
+				ClientSecret: "secret",
+				TokenURL:     tokenURL,
+			},
+		},
+		CrowdstrikeAppID: "test",
+		Retry:            r,
+		Program: `
+			state.response.decode_json().as(body,{
+				"events": [body],
+			})`,
+	}
+}
+
+func newEmptyDiscoverFollower(t *testing.T, cfg config, stat status.StatusReporter, log *logp.Logger) StreamFollower {
+	t.Helper()
+	env := v2.Context{ID: "crowdstrike_test", MetricsRegistry: monitoring.NewRegistry()}
+	s, err := NewFalconHoseFollower(context.Background(), env, cfg, nil, &testPublisher{log}, stat, log, time.Now)
+	if err != nil {
+		t.Fatalf("failed to construct follower: %v", err)
+	}
+	return s
+}
+
+// degradedRecorder counts the number of times DEGRADED is reported.
+type degradedRecorder struct {
+	n atomic.Int64
+}
+
+func (r *degradedRecorder) UpdateStatus(s status.Status, _ string) {
+	if s == status.Degraded {
+		r.n.Add(1)
+	}
+}
+
+func (r *degradedRecorder) count() int64 {
+	return r.n.Load()
+}
+
 func TestFollowStreamCancelsRefreshOnReconnect(t *testing.T) {
 	const totalSessions = 10
 

x-pack/filebeat/input/streaming/crowdstrike_unit_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 
 	cursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
@@ -64,6 +65,27 @@ func TestFollowSession_FirehoseHTTPError(t *testing.T) {
 	}
 }
 
+func TestFollowSession_EmptyDiscoverBody(t *testing.T) {
+	discoverSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// A 200 OK with an empty body, as observed from the CrowdStrike
+		// discover endpoint; Decode returns io.EOF.
+		w.Header().Set("Content-Type", "application/json")
+	}))
+	defer discoverSrv.Close()
+
+	s := newTestStream(t, discoverSrv.URL, discoverSrv.Client())
+	state, err := s.followSession(context.Background(), discoverSrv.Client(), map[string]any{})
+	if err == nil {
+		t.Fatal("expected error from followSession, got nil")
+	}
+	if want := "discover stream returned an empty body"; !strings.Contains(err.Error(), want) {
+		t.Errorf("followSession() error = %v; want substring %q", err, want)
+	}
+	if state == nil {
+		t.Error("expected non-nil state on non-hard error")
+	}
+}
+
 func TestFollowSession_NonObjectMessage(t *testing.T) {
 	logp.TestingSetup()
 
@@ -153,7 +175,7 @@ func newTestStream(t *testing.T, discoverURL string, firehoseClient *http.Client
 
 func newTestStreamWithPublisher(t *testing.T, discoverURL string, firehoseClient *http.Client, pub cursor.Publisher) *falconHoseStream {
 	t.Helper()
-	log := logp.L()
+	log := logptest.NewTestingLogger(t, t.Name())
 	reg := monitoring.NewRegistry()
 	m := newInputMetrics(reg, log)