[Filebeat] cel input does not stop after SIGINT

[andrewkroh]

When testing the CEL input against an HTTP host that is down, the CEL input does not return after a Ctrl+C / SIGINT.

Version: filebeat version 8.7.0 (arm64), libbeat 8.7.0 [ea9b0cb built 2022-12-06 16:22:31 +0000 UTC]

Steps to Reproduce:

1. Configure CEL to request a URL that is down.
2. Start Filebeat.
3. Let it run long enough for CEL to start.
4. Hit Ctrl+C.
5. Wait for Filebeat to exit (but it won't).

[elasticmachine]

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

[efd6]

I have seen essentially the same behaviour with packetbeat, so I am not sure that this is specific to CEL. The behaviour is that on the first ^C the beat attempts to start the shutdown sequence but the input keeps trying and on subsequent ^C the signal handlers are no longer responsive at all since presumably they have been removed during the shutdown sequence.

[andrewkroh]

When Filebeat gets a SIGINT it will pass this down to the inputs via the v2.Canceler and then wait for them to stop before it exits. So on an interrupt signal this ctx ¹ will be closed. That should cause the input to "gracefully" stop. Within a second or two of that context being cancelled it should stop making requests and Run should return.

Footnotes

1. https://github.com/elastic/beats/blob/f74ba6910618d49a84f130f2b1bfb136ba647bfd/x-pack/filebeat/input/cel/input.go#L111 ↩

[efd6]

filebeat.inputs:
- type: cel
  interval: 10s
  resource.url: https://api.ipify.norg/?format=json
  program: |
    bytes(get(state.url).Body).as(body, {
        "events": [body.decode_json()]
    })

$ ./filebeat -e -d "*"
<snip>
{"log.level":"debug","@timestamp":"2022-12-07T06:42:50.751+1030","log.logger":"input.cel.retryablehttp","log.origin":{"file.name":"go-retryablehttp@v0.6.6/client.go","file.line":602},"message":"retrying request","service.name":"filebeat","id":"D917A10C119B3659","input_source":"https://api.ipify.norg/?format=json","input_url":"https://api.ipify.norg/?format=json","request":"GET https://api.ipify.norg/?format=json","timeout":4000000000,"remaining":3,"ecs.version":"1.6.0"}
^C{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"service","log.origin":{"file.name":"service/service.go","file.line":54},"message":"Received sigterm/sigint, stopping","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2022-12-07T06:42:52.891+1030","log.origin":{"file.name":"beater/filebeat.go","file.line":431},"message":"Stopping filebeat","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2022-12-07T06:42:52.891+1030","log.origin":{"file.name":"beater/crawler.go","file.line":155},"message":"Stopping Crawler","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2022-12-07T06:42:52.891+1030","log.origin":{"file.name":"beater/crawler.go","file.line":165},"message":"Stopping 1 inputs","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2022-12-07T06:42:52.891+1030","log.origin":{"file.name":"cfgfile/reload.go","file.line":227},"message":"Dynamic config reloader stopped","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"crawler","log.origin":{"file.name":"beater/crawler.go","file.line":170},"message":"Stopping input: 15643148903901771353","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"publisher","log.origin":{"file.name":"pipeline/client.go","file.line":154},"message":"client: closing acker","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"publisher","log.origin":{"file.name":"pipeline/client.go","file.line":159},"message":"client: done closing acker","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"publisher","log.origin":{"file.name":"pipeline/client.go","file.line":161},"message":"client: unlink from queue","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"publisher","log.origin":{"file.name":"pipeline/client.go","file.line":183},"message":"client: cancelled 0 events","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"publisher","log.origin":{"file.name":"pipeline/client.go","file.line":163},"message":"client: done unlink","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"publisher","log.origin":{"file.name":"pipeline/client.go","file.line":166},"message":"client: closing processors","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:42:52.891+1030","log.logger":"publisher","log.origin":{"file.name":"pipeline/client.go","file.line":171},"message":"client: done closing processors","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"warn","@timestamp":"2022-12-07T06:42:54.754+1030","log.logger":"transport","log.origin":{"file.name":"transport/tcp.go","file.line":52},"message":"DNS lookup failure \"api.ipify.norg\": lookup api.ipify.norg: no such host","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"error","@timestamp":"2022-12-07T06:42:54.754+1030","log.logger":"input.cel.retryablehttp","log.origin":{"file.name":"go-retryablehttp@v0.6.6/client.go","file.line":553},"message":"request failed","service.name":"filebeat","id":"D917A10C119B3659","input_source":"https://api.ipify.norg/?format=json","input_url":"https://api.ipify.norg/?format=json","error":{"message":"Get \"https://api.ipify.norg/?format=json\": lookup api.ipify.norg: no such host"},"method":"GET","url":"https://api.ipify.norg/?format=json","ecs.version":"1.6.0"}

Similar result with packetbeat.

$ ./packetbeat run -e -d "*"
<snip>
^C{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"service","log.origin":{"file.name":"service/service.go","file.line":54},"message":"Received sigterm/sigint, stopping","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"info","@timestamp":"2022-12-07T06:49:26.487+1030","log.origin":{"file.name":"beater/packetbeat.go","file.line":188},"message":"Packetbeat send stop signal","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":77},"message":"stop flows worker","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":115},"message":"stop periodic loop","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":236},"message":"exec tick, timeout=false, report=true","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":258},"message":"handle flow: {3089 0 255 255 255 12 255 255 255 255 255 20 24 1 0 1}, [56 16 213 131 53 1 136 102 90 69 208 41 3 235 96 60 192 168 179 10 187 1 115 209 1 0 0 0 0 0 0 0]","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":274},"message":"report flow","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":286},"message":"add event: {2022-12-07 06:49:26.487546 +1030 ACDT m=+16.755932707 {} {\"destination\":{\"ip\":\"3.235.96.60\",\"mac\":\"38-10-D5-83-35-01\",\"port\":443},\"event\":{\"action\":\"network_flow\",\"category\":[\"network\"],\"dataset\":\"flow\",\"duration\":7576287236,\"end\":\"2022-12-06T20:19:24.359Z\",\"kind\":\"event\",\"start\":\"2022-12-06T20:19:16.783Z\",\"type\":[\"connection\",\"end\"]},\"flow\":{\"final\":true,\"id\":\"EQwA////DP//////FBgBAAE4ENWDNQGIZlpF0CkD62A8wKizCrsBc9EBAAAAAAAAAA\"},\"network\":{\"bytes\":255,\"community_id\":\"1:gmc/xdBmuTDFsYRex2PNOxdnev0=\",\"packets\":3,\"transport\":\"tcp\",\"type\":\"ipv4\"},\"source\":{\"bytes\":255,\"ip\":\"192.168.179.10\",\"mac\":\"88-66-5A-45-D0-29\",\"packets\":3,\"port\":53619},\"type\":\"flow\"} <nil> false}","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":258},"message":"handle flow: {3089 0 255 255 255 12 255 255 255 255 255 20 24 1 0 1}, [56 16 213 131 53 1 136 102 90 69 208 41 142 250 70 238 192 168 179 10 187 1 65 205 2 0 0 0 0 0 0 0]","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":274},"message":"report flow","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":286},"message":"add event: {2022-12-07 06:49:26.487546 +1030 ACDT m=+16.755932707 {} {\"destination\":{\"bytes\":1701,\"ip\":\"142.250.70.238\",\"mac\":\"38-10-D5-83-35-01\",\"packets\":9,\"port\":443},\"event\":{\"action\":\"network_flow\",\"category\":[\"network\"],\"dataset\":\"flow\",\"duration\":4529850907,\"end\":\"2022-12-06T20:19:21.814Z\",\"kind\":\"event\",\"start\":\"2022-12-06T20:19:17.284Z\",\"type\":[\"connection\",\"end\"]},\"flow\":{\"final\":true,\"id\":\"EQwA////DP//////FBgBAAE4ENWDNQGIZlpF0CmO+kbuwKizCrsBQc0CAAAAAAAAAA\"},\"network\":{\"bytes\":2457,\"community_id\":\"1:M1WttNU1eqFxdRKxLWBm+XA3+cE=\",\"packets\":14,\"transport\":\"tcp\",\"type\":\"ipv4\"},\"source\":{\"bytes\":756,\"ip\":\"192.168.179.10\",\"mac\":\"88-66-5A-45-D0-29\",\"packets\":5,\"port\":52545},\"type\":\"flow\"} <nil> false}","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":258},"message":"handle flow: {3089 0 255 255 255 12 255 255 255 255 255 20 24 1 0 1}, [56 16 213 131 53 1 136 102 90 69 208 41 54 79 215 244 192 168 179 10 187 1 206 205 3 0 0 0 0 0 0 0]","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.487+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":274},"message":"report flow","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":286},"message":"add event: {2022-12-07 06:49:26.487546 +1030 ACDT m=+16.755932707 {} {\"destination\":{\"bytes\":188,\"ip\":\"54.79.215.244\",\"mac\":\"38-10-D5-83-35-01\",\"packets\":2,\"port\":443},\"event\":{\"action\":\"network_flow\",\"category\":[\"network\"],\"dataset\":\"flow\",\"duration\":509386320,\"end\":\"2022-12-06T20:19:17.797Z\",\"kind\":\"event\",\"start\":\"2022-12-06T20:19:17.288Z\",\"type\":[\"connection\",\"end\"]},\"flow\":{\"final\":true,\"id\":\"EQwA////DP//////FBgBAAE4ENWDNQGIZlpF0Ck2T9f0wKizCrsBzs0DAAAAAAAAAA\"},\"network\":{\"bytes\":496,\"community_id\":\"1:ZZiV2VhioVShCU2zMJHxv+yuyGM=\",\"packets\":5,\"transport\":\"tcp\",\"type\":\"ipv4\"},\"source\":{\"bytes\":308,\"ip\":\"192.168.179.10\",\"mac\":\"88-66-5A-45-D0-29\",\"packets\":3,\"port\":52686},\"type\":\"flow\"} <nil> false}","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":258},"message":"handle flow: {3089 0 255 255 255 12 255 255 255 255 255 20 24 1 0 1}, [56 16 213 131 53 1 136 102 90 69 208 41 54 253 178 147 192 168 179 10 187 1 36 205 4 0 0 0 0 0 0 0]","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":274},"message":"report flow","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":286},"message":"add event: {2022-12-07 06:49:26.487546 +1030 ACDT m=+16.755932707 {} {\"destination\":{\"bytes\":254,\"ip\":\"54.253.178.147\",\"mac\":\"38-10-D5-83-35-01\",\"packets\":3,\"port\":443},\"event\":{\"action\":\"network_flow\",\"category\":[\"network\"],\"dataset\":\"flow\",\"duration\":2382358,\"end\":\"2022-12-06T20:19:17.797Z\",\"kind\":\"event\",\"start\":\"2022-12-06T20:19:17.795Z\",\"type\":[\"connection\",\"end\"]},\"flow\":{\"final\":true,\"id\":\"EQwA////DP//////FBgBAAE4ENWDNQGIZlpF0Ck2/bKTwKizCrsBJM0EAAAAAAAAAA\"},\"network\":{\"bytes\":562,\"community_id\":\"1:4tX/02NaxJhBxLL9MvKf5jDloAE=\",\"packets\":6,\"transport\":\"tcp\",\"type\":\"ipv4\"},\"source\":{\"bytes\":308,\"ip\":\"192.168.179.10\",\"mac\":\"88-66-5A-45-D0-29\",\"packets\":3,\"port\":52516},\"type\":\"flow\"} <nil> false}","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":258},"message":"handle flow: {3089 0 255 255 255 12 255 255 255 255 255 20 24 1 0 1}, [56 16 213 131 53 1 136 102 90 69 208 41 34 71 180 220 192 168 179 10 187 1 117 209 5 0 0 0 0 0 0 0]","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":274},"message":"report flow","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":286},"message":"add event: {2022-12-07 06:49:26.487546 +1030 ACDT m=+16.755932707 {} {\"destination\":{\"bytes\":1952,\"ip\":\"34.71.180.220\",\"mac\":\"38-10-D5-83-35-01\",\"packets\":5,\"port\":443},\"event\":{\"action\":\"network_flow\",\"category\":[\"network\"],\"dataset\":\"flow\",\"duration\":7102646536,\"end\":\"2022-12-06T20:19:24.901Z\",\"kind\":\"event\",\"start\":\"2022-12-06T20:19:17.798Z\",\"type\":[\"connection\",\"end\"]},\"flow\":{\"final\":true,\"id\":\"EQwA////DP//////FBgBAAE4ENWDNQGIZlpF0CkiR7TcwKizCrsBddEFAAAAAAAAAA\"},\"network\":{\"bytes\":10400,\"community_id\":\"1:CCggJ4SWvwkv6/dd5M9HNb3s8uA=\",\"packets\":13,\"transport\":\"tcp\",\"type\":\"ipv4\"},\"source\":{\"bytes\":8448,\"ip\":\"192.168.179.10\",\"mac\":\"88-66-5A-45-D0-29\",\"packets\":8,\"port\":53621},\"type\":\"flow\"} <nil> false}","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":258},"message":"handle flow: {3089 0 255 255 255 12 255 255 255 255 255 20 24 1 0 1}, [56 16 213 131 53 1 136 102 90 69 208 41 142 250 70 227 192 168 179 10 187 1 58 205 6 0 0 0 0 0 0 0]","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":274},"message":"report flow","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":286},"message":"add event: {2022-12-07 06:49:26.487546 +1030 ACDT m=+16.755932707 {} {\"destination\":{\"bytes\":651,\"ip\":\"142.250.70.227\",\"mac\":\"38-10-D5-83-35-01\",\"packets\":6,\"port\":443},\"event\":{\"action\":\"network_flow\",\"category\":[\"network\"],\"dataset\":\"flow\",\"duration\":3274420,\"end\":\"2022-12-06T20:19:19.306Z\",\"kind\":\"event\",\"start\":\"2022-12-06T20:19:19.302Z\",\"type\":[\"connection\",\"end\"]},\"flow\":{\"final\":true,\"id\":\"EQwA////DP//////FBgBAAE4ENWDNQGIZlpF0CmO+kbjwKizCrsBOs0GAAAAAAAAAA\"},\"network\":{\"bytes\":985,\"community_id\":\"1:QUd+oqVgLmz/zR7nE6cFLLoL5EM=\",\"packets\":9,\"transport\":\"tcp\",\"type\":\"ipv4\"},\"source\":{\"bytes\":334,\"ip\":\"192.168.179.10\",\"mac\":\"88-66-5A-45-D0-29\",\"packets\":3,\"port\":52538},\"type\":\"flow\"} <nil> false}","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":258},"message":"handle flow: {3089 0 255 255 255 12 255 255 255 255 255 20 24 1 0 1}, [56 16 213 131 53 1 136 102 90 69 208 41 34 71 180 220 192 168 179 10 187 1 237 209 7 0 0 0 0 0 0 0]","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":274},"message":"report flow","service.name":"packetbeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-12-07T06:49:26.488+1030","log.logger":"flows","log.origin":{"file.name":"flows/worker.go","file.line":286},"message":"add event: {2022-12-07 06:49:26.487546 +1030 ACDT m=+16.755932707 {} {\"destination\":{\"bytes\":132,\"ip\":\"34.71.180.220\",\"mac\":\"38-10-D5-83-35-01\",\"packets\":2,\"port\":443},\"event\":{\"action\":\"network_flow\",\"category\":[\"network\"],\"dataset\":\"flow\",\"duration\":502022106,\"end\":\"2022-12-06T20:19:20.810Z\",\"kind\":\"event\",\"start\":\"2022-12-06T20:19:20.308Z\",\"type\":[\"connection\",\"end\"]},\"flow\":{\"final\":true,\"id\":\"EQwA////DP//////FBgBAAE4ENWDNQGIZlpF0CkiR7TcwKizCrsB7dEHAAAAAAAAAA\"},\"network\":{\"bytes\":354,\"community_id\":\"1:7/pd0+tlzjldfStMGz/u+Yf0s9Q=\",\"packets\":5,\"transport\":\"tcp\",\"type\":\"ipv4\"},\"source\":{\"bytes\":222,\"ip\":\"192.168.179.10\",\"mac\":\"88-66-5A-45-D0-29\",\"packets\":3,\"port\":53741},\"type\":\"flow\"} <nil> false}","service.name":"packetbeat","ecs.version":"1.6.0"}

In both cases it's necessary to send SIGKILL to get them to exit.

[efd6]

From the common indifference that they both have, I suspect this is deeper than CEL.

[andrewkroh]

There's probably an issue with Packetbeat as well 😢 .

It seems like the go-retryablehttp is continuing despite the ctx being cancelled. Are the underlying HTTP requests made from the CEL program tied into the evaluator's context (e.g. the one from program.ContextEval(ctx, ...)). That could be one reason why the program continues.

Then after go-retryablehttp gives up and control returns to here ¹ the error is not a context.Canceled. Then I think the for loop hits a continue ² (not fully sure if it hits that one). And the process repeats. Ideally the http requests would execute with the ctx that is canceled when Filebeat signals shutdown, but if not perhaps we add a condition into the for loop ³ (e.g. for ctx.Error() == nil {) so that it eventually stops.

Footnotes

1. https://github.com/elastic/beats/blob/f74ba6910618d49a84f130f2b1bfb136ba647bfd/x-pack/filebeat/input/cel/input.go#L201-L204 ↩

2. https://github.com/elastic/beats/blob/f74ba6910618d49a84f130f2b1bfb136ba647bfd/x-pack/filebeat/input/cel/input.go#L300 ↩

3. https://github.com/elastic/beats/blob/f74ba6910618d49a84f130f2b1bfb136ba647bfd/x-pack/filebeat/input/cel/input.go#L179 ↩

[efd6]

Currently the context is not passed to the underlying HTTP client unless it uses OAuth2, in which case it is handed over that way. It looks like the lib.HTTP invocation will need to have a ctx param. I'll add that today.

[efd6]

There is a quick fix though that doesn't need that and should be put in as well.

diff --git a/x-pack/filebeat/input/cel/input.go b/x-pack/filebeat/input/cel/input.go
index 0cd3609716..034a2b6ee6 100644
--- a/x-pack/filebeat/input/cel/input.go
+++ b/x-pack/filebeat/input/cel/input.go
@@ -187,6 +187,13 @@ func (input) run(env v2.Context, src *source, cursor map[string]interface{}, pub
                                        return ctx.Err()
                                case <-time.After(wait):
                                }
+                       } else {
+                               // Exit if we have been cancelled.
+                               select {
+                               case <-ctx.Done():
+                                       return ctx.Err()
+                               default:
+                               }
                        }
 
                        // Process a set of event requests.