[Filebeat] cel input - non HTTP 200 status_code causes rapid retry

[andrewkroh]

When the CEL program returns a state.status_code that is non HTTP 200 then the input will repeat the request rapidly. This occurs for status codes that are outside the range that are handled by go-retryablehttp like [201, 499] except 429. If I omit the status_code then it basically works as expected by making a request, reporting the error, and then waiting for the next interval. So perhaps when a non-200 status_code is reported then the auto-generated error.message could include information about the status code and it would wait for the next interval?

In 6.231 sec the input made 10893 requests. That's ~1700 request per second. 😨

% grep -h "HTTP request" cel.trace | wc -l
   10893
% head -1 cel.trace| jq '."@timestamp"'
"2022-12-08T10:38:10.445-0500"
% tail -1 cel.trace| jq '."@timestamp"'
"2022-12-08T10:38:16.676-0500"

Version: filebeat version 8.7.0 (arm64), libbeat 8.7.0 [f08eed8 built 2022-12-08 15:32:00 +0000 UTC]

Reproduction

package main

import (
	"log"
	"net/http"
)

func main() {
	handler := func(writer http.ResponseWriter, request *http.Request) {
		http.Error(writer, "Go away", http.StatusForbidden)
	}
	log.Fatal(http.ListenAndServe("localhost:9888", http.HandlerFunc(handler)))
}

---

filebeat.inputs:
  - type: cel
    interval: 1m

    resource:
      url: http://localhost:9888
      retry:
        min_wait: 10s
        max_wait: 120s
      tracer:
        filename: cel.trace

    program: |
      get('http://localhost:9888').as(resp, bytes(resp.Body).as(body, {
          "events": (resp.StatusCode == 200) ? try(body.decode_json()) : {"error": {"message": "unexpected status code " + string(resp.StatusCode)} },
          "status_code": resp.StatusCode,
          "want_more": false,
      }))

output.console.pretty: true

[elasticmachine]

Pinging @elastic/security-external-integrations (Team:Security-External Integrations)

[efd6]

Proposed fix.

diff --git a/x-pack/filebeat/input/cel/input.go b/x-pack/filebeat/input/cel/input.go
index c1a1f1a504..0f77a177b1 100644
--- a/x-pack/filebeat/input/cel/input.go
+++ b/x-pack/filebeat/input/cel/input.go
@@ -529,10 +529,14 @@ func handleResponse(log *logp.Logger, state map[string]interface{}, limiter *rat
                                        waitUntil = t
                                }
                        }
-                       fallthrough
-               default:
-                       delete(state, "events")
                        return false, waitUntil, nil
+               default:
+                       status := http.StatusText(statusCode)
+                       if status == "" {
+                               status = "unknown status code"
+                       }
+                       state["events"] = map[string]interface{}{"error.message": fmt.Sprintf("failed http request with %s: %d", status, statusCode)}
+                       return true, time.Time{}, nil
                }
        }
        return true, waitUntil, nil