Add audit logging for stream content

[mhl-b]

Add support for HTTP content audit logging for streamed content. After #129302 we can aggregate streamed content within REST layer. When INCLUDE_REQUEST_BODY setting is specified every streamed request will be aggregated in SecurityRestFilter. This flag is effectively disable streaming support.

[elasticsearchmachine]

Pinging @elastic/es-distributed-coordination (Team:Distributed Coordination)

[elasticsearchmachine]

Hi @mhl-b, I've created a changelog YAML for you.

[ywangd]

LGTM

I left a somewhat important comment and appreciate if you could address it. Additionally, could you please remove the following code and maybe replace it by an assertion that the request is no longer streaming.

elasticsearch/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditUtil.java

Lines 30 to 32 in 9be73c5

	if (request.isStreamedContent()) {
	return "Request body had not been received at the time of the audit event";
	}

[ywangd]

This seems problematic. According to this comment, the non-volatile field includeRequestBody is guaranteed to see the latest change because it is always read after the volatile field events. It is true within LoggingAuditTrail but no longer the case if we just return it here. I think we can either also read events field here before return or having SecurityRestFilter register its own settings update consumer for the INCLUDE_REQUEST_BODY setting so that it does not need to ask LoggingAuditTrail.

[mhl-b]

SecurityRestFilter needs to know both features : LoggingAuditTrail and includeRequestBody are enabled. Shouldnt be much of a difference. Kind of hacky, cutting through two layers of abstraction.

[ywangd]

So a read for events before returnning should suffice in this case. Otherwise we may see stale value for includeRequestBody

[mhl-b]

Making includeRequestBody volatile should be cleaner then, if I still need to read a volatile field. 9351f93

[DaveCTurner]

LGTM

[DaveCTurner]

one-liner?

Suggested change

	if (get() instanceof LoggingAuditTrail trail) {
	return trail.includeRequestBody();
	} else {
	return false;
	}
	return get() instanceof LoggingAuditTrail trail && trail.includeRequestBody();