[filebeat][streaming] - Added support for websocket keep_alive heartbeat, updated metrics & docs

[ShourieG]

Type of change

• Enhancement
• Docs

Proposed commit message

1. Added support for websocket keep_alive config option and
implemented keep_alive heartbeat logic in a separate go routine. 
2. Added support for new metrics introduced with heartbeats. 
3. Updated docs accordingly.
4. Added metrics test which was previously missing.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

Disruptive User Impact

Note

1. I do not see an efficient way to test this with our current testing setup (still exploring this, thus PR still kept as draft). I believe we will have to rely on manual testing for this.

2. Go channels were not used because of complexity related to channel lifecycles and instead a cancellable context with a manual cancel() call is used to avoid any channel related issues.

3. Inbuilt test was not possible for this scenario due to current limitations of the test framework and testing philosophy, hence manual testing was performed with a custom websocket server, for which a screen recording is attached below along with respective filebeat config.

Author's Checklist

• [ ]

How to test this PR locally

Related issues

• Closes Keepalive for websocket connections #42277
• Relates [proofpoint_on_demand]: Datastreams do not recover after websocket error integrations#11816

Use cases

Keeps low throughput WebSocket connections from idling out, which would otherwise result in frequent reconnects.

Screenshots

Screen Recording Of Manual Testing

filebeat config used:

filebeat.inputs:
- type: streaming
  url: ws://localhost:443/ws
  redact:
    fields: ~
  program: |
    bytes(state.response).decode_json().as(body,{
      "events": {
        "message":  body.encode_json(),
      }
    })
  keep_alive:
    enable: true
    interval: 5s

Screen Recording

Test WS server is on the left, filebeat streaming input on the right:

ws_keepAlive_manual_test.mov

Test Server Gist

https://gist.github.com/ShourieG/322d38bdfc95cc7314e1729c3b892ae3

Logs

Manual Testing Debug Logs:

{
    "log.level": "debug",
    "@timestamp": "2025-05-13T15:45:31.012+0530",
    "log.logger": "input.streaming",
    "log.origin": {
        "function": "github.com/elastic/beats/v7/x-pack/filebeat/input/streaming.(*keepAlive).heartBeat.func1",
        "file.name": "streaming/websocket.go",
        "file.line": 76
    },
    "message": "received pong message from websocket server",
    "service.name": "filebeat",
    "id": "6E095AFCCE4611A6",
    "input_source": "ws://localhost:443/ws",
    "input_url": "ws://localhost:443/ws",
    "pong_data": "SENT PONG FRAME",
    "ecs.version": "1.6.0"
}

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @ShourieG? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[elasticmachine]

Pinging @elastic/security-service-integrations (Team:Security-Service Integrations)

[ShourieG]

@efd6, addressed all the suggestions.

[ShourieG]

/test

[ShourieG]

@efd6, I've resolved all the comments and put up a screen recording showing manual testing for this new feature along with a server restart. Sorry for the quality, res is 720p because of GitHub upload size limits.

[ShourieG]

/test

[efd6]

Can you put the test WS server code up somewhere so that we can use it for testing in future?

[ShourieG]

Can you put the test WS server code up somewhere so that we can use it for testing in future?

@efd6, added the following Gist in the PR description

[ShourieG]

@theletterf, resolved all the suggestions.

[ShourieG]

@Mergifyio backport 8.19 9.0

[mergify]

backport 8.19 9.0

✅ Backports have been created

Details

• #45110 [8.19](backport #44204) [filebeat][streaming] - Added support for websocket keep_alive heartbeat, updated metrics & docs has been created for branch 8.19 but encountered conflicts
• #45111 [9.0](backport #44204) [filebeat][streaming] - Added support for websocket keep_alive heartbeat, updated metrics & docs has been created for branch 9.0 but encountered conflicts