[filebeat][websocket] - Add support for passing cursor value as query parameter in Websocket URL

[brijesh-elastic]

Enhancement: The enhancement should enable the use of a cursor value as a query parameter, along with the support of time library functions, to allow for dynamic updates to the URL.

Use case: This feature will be beneficial for integrations like Proofpoint On Demand, where it will facilitate the retrieval of historical data by leveraging the 'sinceTime' query parameter.

Example:

filebeat.inputs:
- type: websocket
  url: wss://logstream.proofpoint.com/v1/stream?cid=abc&type=message&sinceTime={sinceTime}
  program: |
    bytes(state.response).decode_json().as(body,{
      "events": {
        "message":  body.encode_json(),
      },
      "cursor": {
        "last_timestamp": body.map(e, e.ts).max()
      }
    })

In above, when data collection first runs, the sinceTime should be set to (now-duration(initial_interval)) to collect historical data. Whenever the WebSocket connection is interrupted or closes unexpectedly, the next time it restarts, the sinceTime should be taken from state.cursor.last_timestamp to continue collecting data from where it left off.

The ideal condition for the {sinceTime} is: has(state.cursor) && has(state.cursor.last_timestamp) ? state.cursor.last_timestamp : (now-duration(initial_interval)).

[elasticmachine]

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

[botelastic]

This issue doesn't have a Team:<team> label.

[ShourieG]

Hi @brijesh-elastic, can you please expand upon the description of this enhancement request with the help of an example, it would make this feature request more descriptive for keeping track of.