[filebeat][streaming] - Fix for streaming input handling of invalid or empty websocket messages

[ShourieG]

Type of change

• Bug

Proposed commit message

Fix for streaming input handling of invalid or empty websocket messages.

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]

I think this depends on what the understood semantics of the metric are.

[ShourieG]

As this is not a core part of the change for this PR, I've reverted this now, will take a look at it once more when I do some cleanups.

[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".

[efd6]

It does return, the issue is when it is a RetryableError

Sorry, you are correct.

Suggest then

			if err != nil {
				s.metrics.errorsTotal.Inc()
				if !isRetryableError(err) {
					s.log.Errorw("failed to read websocket data", "error", err)
					return err
				}
				s.log.Debugw("websocket connection encountered an error, attempting to reconnect...", "error", err)
				// close the old connection and reconnect
				if err := c.Close(); err != nil {
					s.metrics.errorsTotal.Inc()
					s.log.Errorw("encountered an error while closing the websocket connection", "error", err)
				}
				// since c is already a pointer, we can reassign it to the new connection and the defer func will still handle it
				c, resp, err = connectWebSocket(ctx, s.cfg, url, s.log)
				handleConnectionResponse(resp, s.metrics, s.log)
				if err != nil {
					s.metrics.errorsTotal.Inc()
					s.log.Errorw("failed to reconnect websocket connection", "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
			}

Note also that the debugw call with the message allocates non-conditionally due to the string conversion. This can be avoided by providing a lazy fmt.Stringer.

[ShourieG]

I'll keep the fmt.Stringer change for a later cleanup PR then.

[efd6]

The type is

type bytesStringer []byte

func (b bytesStringer) String() string { return string(b) }

with

s.log.Debugw("received websocket message", logp.Namespace("websocket"), "msg", bytesStringer(message))

[ShourieG]

@efd6, updated the PR