[filebeat][streaming] - Fixed a scenario where CEL would process invalid/empty messages in case of a websocket error

[ShourieG]

Type of change

• Bug

Proposed commit message

Fixed a scenario where CEL would process invalid/empty messages in case of a websocket connection/read error. This is specially problematic after the addition of retry logic where these errors could cause false positives downstream even though the connection is re-established successfully.

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

None

Author's Checklist

• [ ]

How to test this PR locally

Related issues

• Relates [filebeat][streaming] - Added retry functionality to websocket connections #40601

Use cases

Screenshots

Logs

[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

[mergify]

backport-8.x has been added to help with the transition to the new branch 8.x.
If you don't need it please use backport-skip label and remove the backport-8.x label.

[elasticmachine]

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

[botelastic]

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

[efd6]

I think the proposed commit message needs a clearer explanation of what is being fixed; it's not the situation that's being fixed, it's the response to the situation, so describe that and say what the behaviour is and then how that is rectified.

[efd6]

Suggested change

	- Fixed a scenario in the streaming input where CEL would process invalid/empty messages in case of a websocket error. {pull}42036[42036]
	- Fix streaming input handling of invalid or empty websocket messages. {pull}42036[42036]

[efd6]

Why move this up? Is a message valid when err is non-nil?

[ShourieG]

Shouldn't we still count the received bytes irrespective of whether it's a valid message or not, since this is a metric ?

[efd6]

Suggested change

	} else {
	state["response"] = message
	s.log.Debugw("received websocket message", logp.Namespace("websocket"), "msg", string(message))
	err = s.process(ctx, state, s.cursor, s.now().In(time.UTC))
	if err != nil {
	s.metrics.errorsTotal.Inc()
	s.log.Errorw("failed to process and publish data", "error", err)
	return err
	}
	continue
	}
	state["response"] = message
	s.log.Debugw("received websocket message", logp.Namespace("websocket"), "msg", string(message))
	err = s.process(ctx, state, s.cursor, s.now().In(time.UTC))
	if err != nil {
	s.metrics.errorsTotal.Inc()
	s.log.Errorw("failed to process and publish data", "error", err)
	return err
	}

but why does the error case not return if !isRetryableError(err)?

[ShourieG]

@efd6, It does return, the issue is when it is a RetryableError. In this scenario the message is still passed to CEL during the retry attempts. So if the 3rd attempt is successful the 1st 2 invalid messages are still processed by CEL leading to downstream errors in integration pipelines because the event itself might be malformed or not present. Also this is just unnecessary processing that can be avoided.

[ShourieG]

Also I felt that an else block had better readability in this case since we already have a bunch of returns within the "if".