system/socket: propagate backpressure from output #32192

adriansr · 2022-07-04T19:19:19Z

What does this PR do?

This PR modifies the socket dataset so that it is affected by output backpressure.

Originally, it operated as two separate goroutines. One would read kernel tracing events as fast as possible and maintain a view of the open sockets/flows on the system. Once a flow is deemed terminated, it would be stored on a linked list (done) for it to be published to the output.

A separate goroutine consumes that linked list and publishes the flows.

Under heavy network load, it can generate flows faster than the output is able to accept. Due to the linked list being unbounded, it would store more and more flows, causing Auditbeat's memory usage to grow over time.

This PR simplifies flow publication in such a way that if the output blocks, the main reader goroutine will block too, preventing the consumption of new kernel events and the generation of additional flows until the output is able to accept more documents.

Also it contains a necessary refactor: Moving the linked list implementation to a helper package. This adds a bit of noise.

Why is it important?

To prevent high-memory usage of Auditbeat under heavy network I/O.

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.

Author's Checklist

[ ]

How to test this PR locally

With the below code it's easy to increase the memory usage of Auditbeat dramatically. With this patch, it will stabilize.

Flow flooder script

package main

import (
	"fmt"
	"math/rand"
	"net"
	"os"
)

const (
	NSOCKS     = 8000
	ADDR       = "127.0.0.1"
	ADDRLISTEN = ADDR + ":0"
)

func main() {
	addr, err := net.ResolveUDPAddr("udp4", ADDRLISTEN)
	if err != nil {
		panic(fmt.Errorf("ResolveUDPAddr(%s): %w", ADDRLISTEN, err))
	}
	var sock [NSOCKS]*net.UDPConn
	for idx := range sock {
		c, err := net.ListenUDP("udp4", addr)
		if err != nil {
			panic(fmt.Errorf("ListenUDP(%s): %w", addr, err))
		}
		sock[idx] = c
		laddr := c.LocalAddr().(*net.UDPAddr)
		fmt.Fprintf(os.Stderr, "sock[%d] is at %d\n", idx, laddr.Port)
	}
	data := []byte("Some random data")
	for {
		for _, src := range sock {
			dst := sock[rand.Intn(NSOCKS)]
			if _, err = src.WriteTo(data, dst.LocalAddr()); err != nil {
				panic(fmt.Errorf("WriteTo(%v -> %v): %w", src.LocalAddr(), dst.LocalAddr(), err))
			}
		}
	}
}

Related issues

Closes Auditbeat: Propagate output backpressure in socket dataset #32191

Use cases

Screenshots

Logs

botelastic · 2022-07-04T19:23:04Z

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

elasticmachine · 2022-07-04T19:33:30Z

💚 Build Succeeded

the below badges are clickable and redirect to their specific view in the CI or DOCS

Expand to view the summary

Build stats

Start Time: 2022-07-22T14:58:51.626+0000
Duration: 50 min 44 sec

Test stats 🧪

Test	Results
Failed	0
Passed	300
Skipped	49
Total	349

💚 Flaky test report

Tests succeeded.

🤖 GitHub comments

To re-run your PR in the CI, just comment with:

/test : Re-trigger the build.
/package : Generate the packages and run the E2E tests.
/beats-tester : Run the installation tests with beats-tester.
run elasticsearch-ci/docs : Re-trigger the docs validation. (use unformatted text in the comment!)

efd6

This is my initial look, I will take another look later.

I'd just like to confirm/fix my understanding of the approach here. ISTM that this is decoupling the publication of events from the list of captured events from holding the list of events once the timeouts have happened. To prevent the publication from getting bound by publication rate limiting it. Is this correct?

I think it would be helpful to have a description of the approach in the PR description with a view to including it in the commit message when the squash happens.

x-pack/auditbeat/module/system/socket/state.go

efd6 · 2022-07-06T03:18:15Z

x-pack/auditbeat/module/system/socket/state.go

+func (s *state) ExpireFlows() {
+	start := s.clock()
+	toReport := s.expireFlows()
+	if sent := s.reportFlows(&toReport); sent > 0 {


sent cannot be negative, so sent != 0?

x-pack/auditbeat/module/system/socket/state.go

efd6 · 2022-07-06T03:23:20Z

x-pack/auditbeat/module/system/socket/state.go

+	return toReport
+}
+
+func (l *linkedList) append(b linkedList) {


Just for clarity, the caller to l.append never uses b again, true? Otherwise this is not safe since their b will continue to be a tail of its previous head. This is always what is happening here since b is always a newly constructed value, but it is probably worth documenting that this is the case.

Makes sense. I've updated the function to take a pointer for the second argument and zero it before returning.

efd6 · 2022-07-06T03:31:22Z

x-pack/auditbeat/module/system/socket/state.go

+	var errs multierror.Errors
+	rootPut := func(key string, value interface{}) {
+		if _, err := root.Put(key, value); err != nil {
+			errs = append(errs, err)
+		}
+	}


efd6 · 2022-07-06T03:47:09Z

x-pack/auditbeat/module/system/socket/state.go

+	}
+}
+
+func (s *state) expireFlows() (toReport linkedList) {
 	s.Lock()
 	defer s.Unlock()
 	deadline := s.clock().Add(-s.inactiveTimeout)


This is a minor point, but I'm wondering if s.clock() should be pulled out into now and have that used in the places here.

efd6 · 2022-07-06T03:51:53Z

x-pack/auditbeat/module/system/socket/state.go

+	var toReport linkedList
+	// Send flows to the output as a deferred function to avoid
+	// holding on s mutex when there's backpressure from the output.
+	defer s.reportFlows(&toReport)


efd6 · 2022-07-06T05:16:21Z

x-pack/auditbeat/module/system/socket/state.go

+	}
+}
+
+func (s *state) expireFlows() (toReport linkedList) {


I'm wondering at this point, when it is now a matter of collecting items, if it wouldn't be simpler to have a []linkedElement for reportFlows to work on; the allocations would likely be reduced (particularly if the size hints were used to preallocate) and the code would be simpler. This would involve changes in onSockTerminated and onFlowTerminated as well.

I'm not sure I follow. Removing elements from linkedLists and creating a new linked list from those existing elements is a zero-alloc operation. I'll add a test to ensure this is always the case.

I think checking the escape analysis would be worthwhile. The head that is being obtained for adding to toReport is not the same memory that is in the source since we are not using the pointer, so it must be allocated. Since it is being used after the function returns, it cannot be on the stack, so I think the allocation needs to be on the heap. The tail of that head will all be already allocated and so they are certainly zero-alloc.

You are right; I've looked at the escape analysis and it doesn't escape.

I've gone over this more carefully because I was unable to find a fault in my logic (the -m report is not in source order so I missed it previously and accepted the output I saw) and I can see that the b parameter to append does leak.

./state.go:901:29: leaking param content: b

efd6 · 2022-07-06T05:19:42Z

x-pack/auditbeat/module/system/socket/state_test.go

+	return true
+}
+
+func (ts *testingState) Error(_ error) bool {


efd6 · 2022-07-06T05:20:50Z

x-pack/auditbeat/module/system/socket/state_test.go

+}
+
+func (ts *testingState) feedEvents(evs []event) {
+	for idx, ev := range evs {


Add t.Helper() above this?

adriansr · 2022-07-06T10:00:42Z

@efd6 please re-review. I've updated the description with an explanation.

efd6 · 2022-07-06T11:22:18Z

OK, I have the logic now and it looks OK to me. Thanks for the clarification above. I think it would be helpful to also say that the back pressure is exerted through *state.Mutex. I'll take another look tomorrow.

efd6

Assuming my interpretation of the back pressure being exerted via the *state mutex is correct, this LGTM.

efd6 · 2022-07-07T01:46:17Z

x-pack/auditbeat/module/system/socket/state.go

+			s.reportFlow(f)
+			count++


If *state.reportFlow returns a conditional 1 for success and 0 for failure then the count would more accurately reflect what was reported (depending on whether you are interested in the attempts or the succeeds).

makes sense, done.

efd6 · 2022-07-07T01:47:24Z

x-pack/auditbeat/module/system/socket/state.go

+	var toReport linkedList
+	// Send flows to the output as a deferred function to avoid
+	// holding on s mutex when there's backpressure from the output.
+	defer s.reportFlows(&toReport)


Is it worth logging the count here?

I didn't want to log in the main thread as this will report a flow every time a socket is closed.

The logging in ExpireFlows is there to have a way to tell when backpressure is happening without causing too much impact.

x-pack/auditbeat/module/system/socket/helper/linkedlist.go

adriansr · 2022-07-13T10:15:19Z

We have users testing this change. Waiting for feedback before moving out of the draft state.

Co-authored-by: Dan Kortschak <90160302+efd6@users.noreply.github.com>

elasticmachine · 2022-07-22T10:18:52Z

Pinging @elastic/security-external-integrations (Team:Security-External Integrations)

This PR modifies the socket dataset so that it is affected by output backpressure. Originally, it operated as two separate goroutines. One would read kernel tracing events as fast as possible and maintain a view of the open sockets/flows on the system. Once a flow is deemed terminated, it would be stored on a linked list (done) for it to be published to the output. A separate goroutine consumes that linked list and publishes the flows. Under heavy network load, it can generate flows faster than the output is able to accept. Due to the linked list being unbounded, it would store more and more flows, causing Auditbeat's memory usage to grow over time. This PR simplifies flow publication in such a way that if the output blocks, the main reader goroutine will block too, preventing the consumption of new kernel events and the generation of additional flows until the output is able to accept more documents. Also it contains a necessary refactor: Moving the linked list implementation to a helper package. This adds a bit of noise. (cherry picked from commit e362ca7) # Conflicts: # x-pack/auditbeat/module/system/socket/state.go # x-pack/auditbeat/module/system/socket/state_test.go

This PR modifies the socket dataset so that it is affected by output backpressure. Originally, it operated as two separate goroutines. One would read kernel tracing events as fast as possible and maintain a view of the open sockets/flows on the system. Once a flow is deemed terminated, it would be stored on a linked list (done) for it to be published to the output. A separate goroutine consumes that linked list and publishes the flows. Under heavy network load, it can generate flows faster than the output is able to accept. Due to the linked list being unbounded, it would store more and more flows, causing Auditbeat's memory usage to grow over time. This PR simplifies flow publication in such a way that if the output blocks, the main reader goroutine will block too, preventing the consumption of new kernel events and the generation of additional flows until the output is able to accept more documents. Also it contains a necessary refactor: Moving the linked list implementation to a helper package. This adds a bit of noise. (cherry picked from commit e362ca7)

…put (#32480) This PR modifies the socket dataset so that it is affected by output backpressure. Originally, it operated as two separate goroutines. One would read kernel tracing events as fast as possible and maintain a view of the open sockets/flows on the system. Once a flow is deemed terminated, it would be stored on a linked list (done) for it to be published to the output. A separate goroutine consumes that linked list and publishes the flows. Under heavy network load, it can generate flows faster than the output is able to accept. Due to the linked list being unbounded, it would store more and more flows, causing Auditbeat's memory usage to grow over time. This PR simplifies flow publication in such a way that if the output blocks, the main reader goroutine will block too, preventing the consumption of new kernel events and the generation of additional flows until the output is able to accept more documents. Also it contains a necessary refactor: Moving the linked list implementation to a helper package. This adds a bit of noise. (cherry picked from commit e362ca7) Co-authored-by: Adrian Serrano <adrisr83@gmail.com>

This PR modifies the socket dataset so that it is affected by output backpressure. Originally, it operated as two separate goroutines. One would read kernel tracing events as fast as possible and maintain a view of the open sockets/flows on the system. Once a flow is deemed terminated, it would be stored on a linked list (done) for it to be published to the output. A separate goroutine consumes that linked list and publishes the flows. Under heavy network load, it can generate flows faster than the output is able to accept. Due to the linked list being unbounded, it would store more and more flows, causing Auditbeat's memory usage to grow over time. This PR simplifies flow publication in such a way that if the output blocks, the main reader goroutine will block too, preventing the consumption of new kernel events and the generation of additional flows until the output is able to accept more documents. Also it contains a necessary refactor: Moving the linked list implementation to a helper package. This adds a bit of noise. (cherry picked from commit e362ca7)

This PR modifies the socket dataset so that it is affected by output backpressure. Originally, it operated as two separate goroutines. One would read kernel tracing events as fast as possible and maintain a view of the open sockets/flows on the system. Once a flow is deemed terminated, it would be stored on a linked list (done) for it to be published to the output. A separate goroutine consumes that linked list and publishes the flows. Under heavy network load, it can generate flows faster than the output is able to accept. Due to the linked list being unbounded, it would store more and more flows, causing Auditbeat's memory usage to grow over time. This PR simplifies flow publication in such a way that if the output blocks, the main reader goroutine will block too, preventing the consumption of new kernel events and the generation of additional flows until the output is able to accept more documents. Also it contains a necessary refactor: Moving the linked list implementation to a helper package. This adds a bit of noise. (cherry picked from commit e362ca7) Co-authored-by: Adrian Serrano <adrisr83@gmail.com>

This PR modifies the socket dataset so that it is affected by output backpressure. Originally, it operated as two separate goroutines. One would read kernel tracing events as fast as possible and maintain a view of the open sockets/flows on the system. Once a flow is deemed terminated, it would be stored on a linked list (done) for it to be published to the output. A separate goroutine consumes that linked list and publishes the flows. Under heavy network load, it can generate flows faster than the output is able to accept. Due to the linked list being unbounded, it would store more and more flows, causing Auditbeat's memory usage to grow over time. This PR simplifies flow publication in such a way that if the output blocks, the main reader goroutine will block too, preventing the consumption of new kernel events and the generation of additional flows until the output is able to accept more documents. Also it contains a necessary refactor: Moving the linked list implementation to a helper package. This adds a bit of noise.

botelastic bot added the needs_team Indicates that the issue/PR needs a Team:* label label Jul 4, 2022

mergify bot assigned adriansr Jul 4, 2022

adriansr added the Team:Security-External Integrations label Jul 4, 2022

botelastic bot removed the needs_team Indicates that the issue/PR needs a Team:* label label Jul 4, 2022

adriansr added enhancement in progress Pull request is currently in progress. needs_team Indicates that the issue/PR needs a Team:* label labels Jul 4, 2022

botelastic bot removed the needs_team Indicates that the issue/PR needs a Team:* label label Jul 4, 2022

adriansr requested a review from efd6 July 5, 2022 20:02

efd6 reviewed Jul 6, 2022

View reviewed changes

adriansr requested a review from efd6 July 6, 2022 09:59

efd6 approved these changes Jul 7, 2022

View reviewed changes

efd6 approved these changes Jul 12, 2022

View reviewed changes

x-pack/auditbeat/module/system/socket/helper/linkedlist.go Show resolved Hide resolved

x-pack/auditbeat/module/system/socket/helper/linkedlist.go Show resolved Hide resolved

x-pack/auditbeat/module/system/socket/helper/linkedlist.go Outdated Show resolved Hide resolved

adriansr and others added 12 commits July 22, 2022 12:08

Simple backpressure test

a475fab

Fix unit test

abe98c4

Fix golangci-lint issues

b4e8c8f

Correct tests

e46c520

Fix linter error

1aa9e56

Update x-pack/auditbeat/module/system/socket/state.go

f941f0e

Co-authored-by: Dan Kortschak <90160302+efd6@users.noreply.github.com>

Update x-pack/auditbeat/module/system/socket/state.go

3522904

Co-authored-by: Dan Kortschak <90160302+efd6@users.noreply.github.com>

Nit: cleanup imports ordering

2868462

Correct style

d6be13b

Make linkedList.append invalidate its argument

c992b3c

expireFlows: Use a single clock value

a160a3d

Add zero-alloc test

7b708e6

adriansr and others added 7 commits July 22, 2022 12:08

Address comments on test file

9619859

Refactor LinkedList to helper package

4989b3b

Update x-pack/auditbeat/module/system/socket/helper/linkedlist.go

bc980b8

Co-authored-by: Dan Kortschak <90160302+efd6@users.noreply.github.com>

Improve reportFlows accounting

990b76c

Correct license on new files

9e0c257

Improve LinkedList.Remove docs

cbde64a

Changelog entry

ef5c302

adriansr force-pushed the ab_socket_backpressure branch from 8845b36 to ef5c302 Compare July 22, 2022 10:16

adriansr marked this pull request as ready for review July 22, 2022 10:18

adriansr requested a review from a team as a code owner July 22, 2022 10:18

adriansr added backport-v8.3.0 Automated backport with mergify backport-7.17 Automated backport to the 7.17 branch with mergify bug and removed enhancement labels Jul 22, 2022

Feature ---> Bugfix

d7382f9

adriansr removed the in progress Pull request is currently in progress. label Jul 25, 2022

adriansr merged commit e362ca7 into elastic:main Jul 25, 2022

mergify bot mentioned this pull request Jul 25, 2022

[7.17](backport #32192) system/socket: propagate backpressure from output #32479

Merged

mergify bot mentioned this pull request Jul 25, 2022

[8.3](backport #32192) system/socket: propagate backpressure from output #32480

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

system/socket: propagate backpressure from output #32192

system/socket: propagate backpressure from output #32192

adriansr commented Jul 4, 2022 •

edited

Loading

botelastic bot commented Jul 4, 2022

elasticmachine commented Jul 4, 2022 •

edited by jenkins-beats-ci bot

Loading

Build stats

Test stats 🧪

efd6 left a comment

efd6 Jul 6, 2022

adriansr Jul 6, 2022

efd6 Jul 6, 2022

adriansr Jul 6, 2022

efd6 Jul 6, 2022

efd6 Jul 6, 2022

adriansr Jul 12, 2022

efd6 Jul 6, 2022

efd6 Jul 6, 2022

adriansr Jul 6, 2022

efd6 Jul 6, 2022

efd6 Jul 6, 2022

efd6 Jul 7, 2022

efd6 Jul 6, 2022

adriansr Jul 6, 2022

efd6 Jul 6, 2022

adriansr Jul 6, 2022

adriansr commented Jul 6, 2022

efd6 commented Jul 6, 2022 •

edited

Loading

efd6 left a comment

efd6 Jul 7, 2022

adriansr Jul 12, 2022

efd6 Jul 7, 2022

adriansr Jul 11, 2022

adriansr commented Jul 13, 2022

elasticmachine commented Jul 22, 2022

system/socket: propagate backpressure from output #32192

system/socket: propagate backpressure from output #32192

Conversation

adriansr commented Jul 4, 2022 • edited Loading

What does this PR do?

Why is it important?

Checklist

Author's Checklist

How to test this PR locally

Related issues

Use cases

Screenshots

Logs

botelastic bot commented Jul 4, 2022

elasticmachine commented Jul 4, 2022 • edited by jenkins-beats-ci bot Loading

💚 Build Succeeded

Build stats

Test stats 🧪

💚 Flaky test report

🤖 GitHub comments

efd6 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adriansr commented Jul 6, 2022

efd6 commented Jul 6, 2022 • edited Loading

efd6 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adriansr commented Jul 13, 2022

elasticmachine commented Jul 22, 2022

adriansr commented Jul 4, 2022 •

edited

Loading

elasticmachine commented Jul 4, 2022 •

edited by jenkins-beats-ci bot

Loading

efd6 commented Jul 6, 2022 •

edited

Loading