From cc39376a5d4c87cc1a034396f8fe6f0fb54dec73 Mon Sep 17 00:00:00 2001
From: Dan Kortschak <90160302+efd6@users.noreply.github.com>
Date: Wed, 22 Nov 2023 19:26:40 +1030
Subject: [PATCH] x-pack/filebeat/input/cel: suppress and log max http request
 retry errors (#37160)

The retryablehttp.Client will return an error based on the ErrorHandler
field when the number of requests exceeds the maximum configuration.
In the case that the field is nil, this is a non-nil error that causes
issues with some APIs and does not allow the CEL code to evaluate the
HTTP response status. So add a retryablehttp.ErrorHandler that will log
the failure, but return the response unaltered and a nil error.
---
 CHANGELOG.next.asciidoc                 |  1 +
 x-pack/filebeat/input/cel/input.go      | 15 +++++++++++-
 x-pack/filebeat/input/cel/input_test.go | 31 +++++++++++++++++++++++++
 3 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 6976bed8833..343ececae7a 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -216,6 +216,7 @@ Setting environmental variable ELASTIC_NETINFO:false in Elastic Agent pod will d
 - Made Azure Blob Storage input GA and updated docs accordingly. {pull}37128[37128]
 - Add request trace logging to http_endpoint input. {issue}36951[36951] {pull}36957[36957]
 - Made GCS input GA and updated docs accordingly. {pull}37127[37127]
+- Suppress and log max HTTP request retry errors in CEL input. {pull}37160[37160]
 
 *Auditbeat*
 
diff --git a/x-pack/filebeat/input/cel/input.go b/x-pack/filebeat/input/cel/input.go
index c12ea0fd2e4..69b4da8ac44 100644
--- a/x-pack/filebeat/input/cel/input.go
+++ b/x-pack/filebeat/input/cel/input.go
@@ -730,14 +730,16 @@ func newClient(ctx context.Context, cfg config, log *logp.Logger) (*http.Client,
 	c.CheckRedirect = checkRedirect(cfg.Resource, log)
 
 	if cfg.Resource.Retry.getMaxAttempts() > 1 {
+		maxAttempts := cfg.Resource.Retry.getMaxAttempts()
 		c = (&retryablehttp.Client{
 			HTTPClient:   c,
 			Logger:       newRetryLog(log),
 			RetryWaitMin: cfg.Resource.Retry.getWaitMin(),
 			RetryWaitMax: cfg.Resource.Retry.getWaitMax(),
-			RetryMax:     cfg.Resource.Retry.getMaxAttempts(),
+			RetryMax:     maxAttempts,
 			CheckRetry:   retryablehttp.DefaultRetryPolicy,
 			Backoff:      retryablehttp.DefaultBackoff,
+			ErrorHandler: retryErrorHandler(maxAttempts, log),
 		}).StandardClient()
 	}
 
@@ -831,6 +833,17 @@ func checkRedirect(cfg *ResourceConfig, log *logp.Logger) func(*http.Request, []
 	}
 }
 
+// retryErrorHandler returns a retryablehttp.ErrorHandler that will log retry resignation
+// but return the last retry attempt's response and a nil error so that the CEL code
+// can evaluate the response status itself. Any error passed to the retryablehttp.ErrorHandler
+// is returned unaltered.
+func retryErrorHandler(max int, log *logp.Logger) retryablehttp.ErrorHandler {
+	return func(resp *http.Response, err error, numTries int) (*http.Response, error) {
+		log.Warnw("giving up retries", "method", resp.Request.Method, "url", resp.Request.URL, "retries", max+1)
+		return resp, err
+	}
+}
+
 func newRateLimiterFromConfig(cfg *ResourceConfig) *rate.Limiter {
 	r := rate.Inf
 	b := 1
diff --git a/x-pack/filebeat/input/cel/input_test.go b/x-pack/filebeat/input/cel/input_test.go
index 1a0a7b44211..bc2e8828be5 100644
--- a/x-pack/filebeat/input/cel/input_test.go
+++ b/x-pack/filebeat/input/cel/input_test.go
@@ -683,6 +683,37 @@ var inputTests = []struct {
 			{"hello": "world"},
 		},
 	},
+	{
+		name:   "retry_failure_no_success",
+		server: newTestServer(httptest.NewServer),
+		config: map[string]interface{}{
+			"interval": 1,
+			"resource": map[string]interface{}{
+				"retry": map[string]interface{}{
+					"max_attempts": 2,
+				},
+			},
+			"program": `
+	get(state.url).as(resp, {
+		"url": state.url,
+		"events": [
+			bytes(resp.Body).decode_json(),
+			{"status": resp.StatusCode},
+		],
+	})
+	`,
+		},
+		handler: func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("content-type", "application/json")
+			w.WriteHeader(http.StatusGatewayTimeout)
+			//nolint:errcheck // No point checking errors in test server.
+			w.Write([]byte(`{"error":"we were too slow"}`))
+		},
+		want: []map[string]interface{}{
+			{"error": "we were too slow"},
+			{"status": float64(504)}, // Float because of JSON.
+		},
+	},
 
 	{
 		name:   "POST_request",