[9.0](backport #5372) Restore connection-level limiter (#5378)

* Restore connection-level limiter (#5372)

Restore connection level limiter to prevent OOM incidents. (removed in #4402)
This limiter is used in addition to the request-level throttle so that once
our in-flight requests reaches max_connections a 429 is returned, but if the
total connections the server uses is over max_connections*1.1 the server drops
the connection before the TLS handshake.

(cherry picked from commit 39199ef)

# Conflicts:
#	NOTICE-fips.txt
#	NOTICE.txt
#	go.mod
#	internal/pkg/api/server.go

* Fix merge

---------

Co-authored-by: Michel Laterman <[email protected]>
Co-authored-by: michel-laterman <[email protected]>

Author: mergify[bot]

NOTICE.txt

@@ -5065,6 +5065,43 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 
 
+--------------------------------------------------------------------------------
+Dependency : golang.org/x/net
+Version: v0.41.0
+Licence type (autodetected): BSD-3-Clause
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/golang.org/x/[email protected]/LICENSE:
+
+Copyright 2009 The Go Authors.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google LLC nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
 --------------------------------------------------------------------------------
 Dependency : golang.org/x/sync
 Version: v0.15.0
@@ -21889,43 +21926,6 @@ THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 
---------------------------------------------------------------------------------
-Dependency : golang.org/x/net
-Version: v0.41.0
-Licence type (autodetected): BSD-3-Clause
---------------------------------------------------------------------------------
-
-Contents of probable licence file $GOMODCACHE/golang.org/x/[email protected]/LICENSE:
-
-Copyright 2009 The Go Authors.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are
-met:
-
-   * Redistributions of source code must retain the above copyright
-notice, this list of conditions and the following disclaimer.
-   * Redistributions in binary form must reproduce the above
-copyright notice, this list of conditions and the following disclaimer
-in the documentation and/or other materials provided with the
-distribution.
-   * Neither the name of Google LLC nor the names of its
-contributors may be used to endorse or promote products derived from
-this software without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
-"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
-LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
-A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
-OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
-SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
-LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
-DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
-THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
-OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
-
 --------------------------------------------------------------------------------
 Dependency : golang.org/x/oauth2
 Version: v0.30.0

changelog/fragments/1756409821-Restore-connection-limiter.yaml

@@ -0,0 +1,38 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Restore connection limiter
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: |
+  Restore connection level limiter to prevent OOM incidents.
+  This limiter is used in addition to the request-level throttle so that once
+  our in-flight requests reaches max_connections a 429 is returned, but if the
+  total connections the server uses is over max_connections*1.1 the server drops
+  the connection before the TLS handshake.
+
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: fleet-server
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/fleet-server/pull/5372
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

go.mod

@@ -36,6 +36,7 @@ require (
 	go.elastic.co/apm/v2 v2.7.1
 	go.elastic.co/ecszerolog v0.2.0
 	go.uber.org/zap v1.27.0
+	golang.org/x/net v0.41.0
 	golang.org/x/sync v0.15.0
 	golang.org/x/time v0.11.0
 	google.golang.org/grpc v1.75.0
@@ -92,7 +93,6 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.39.0 // indirect
 	golang.org/x/mod v0.25.0 // indirect
-	golang.org/x/net v0.41.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/text v0.26.0 // indirect
 	golang.org/x/tools v0.33.0 // indirect

internal/pkg/api/server.go

@@ -16,6 +16,7 @@ import (
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
 	"github.com/elastic/fleet-server/v7/internal/pkg/config"
+	"github.com/elastic/fleet-server/v7/internal/pkg/limit"
 	"github.com/elastic/fleet-server/v7/internal/pkg/logger"
 
 	"github.com/rs/zerolog"
@@ -31,6 +32,7 @@ type server struct {
 // NewServer creates a new HTTP api for the passed addr.
 //
 // The server has an http request limit and endpoint specific rate-limits.
+// There is also a connection limit that will drop connections if too many connections are formed.
 // The underlying API structs (such as *CheckinT) may be shared between servers.
 func NewServer(addr string, cfg *config.Server, opts ...APIOpt) *server {
 	a := &apiServer{}
@@ -78,6 +80,13 @@ func (s *server) Run(ctx context.Context) error {
 		}
 	}()
 
+	// Conn Limiter must be before the TLS handshake in the stack;
+	// The server should not eat the cost of the handshake if there
+	// is no capacity to service the connection.
+	// Also, it appears the HTTP2 implementation depends on the tls.Listener
+	// being at the top of the stack.
+	ln = wrapConnLimitter(ctx, ln, s.cfg)
+
 	if s.cfg.TLS != nil && s.cfg.TLS.IsEnabled() {
 		commonTLSCfg, err := tlscommon.LoadTLSServerConfig(s.cfg.TLS, s.logger)
 		if err != nil {
@@ -160,3 +169,21 @@ func errLogger(ctx context.Context) *slog.Logger {
 	stub := &stubLogger{*log}
 	return slog.New(stub, "", 0)
 }
+
+// wrapConnLimitter will drop connections once the connection count is max_connections*1.1
+// This means that once the limit is reached, the server will resturn 429 responses until the connection count reaches the threshold, then the server will drop connections before the TLS handshake.
+func wrapConnLimitter(ctx context.Context, ln net.Listener, cfg *config.Server) net.Listener {
+	hardLimit := int(float64(cfg.Limits.MaxConnections) * 1.1)
+
+	if hardLimit != 0 {
+		zerolog.Ctx(ctx).Info().
+			Int("hardConnLimit", hardLimit).
+			Msg("server hard connection limiter installed")
+
+		ln = limit.Listener(ln, hardLimit, zerolog.Ctx(ctx))
+	} else {
+		zerolog.Ctx(ctx).Info().Msg("server hard connection limiter disabled")
+	}
+
+	return ln
+}

internal/pkg/limit/listener.go

@@ -0,0 +1,98 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package limit
+
+import (
+	"net"
+	"sync"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+
+	"github.com/elastic/fleet-server/v7/internal/pkg/logger"
+)
+
+// Derived from netutil.LimitListener but works slightly differently.
+// Instead of blocking on the semaphore before acception connection,
+// this implementation immediately accepts connections and if cannot
+// acquire the semaphore it forces the connection closed.
+// Ideally, this limiter is run *before* the TLS handshake occurs
+// to prevent DDOS attack that eats all the server's CPU.
+// The downside to this is that it will Close() valid connections
+// indiscriminately.
+
+func Listener(l net.Listener, n int, log *zerolog.Logger) net.Listener {
+	return &limitListener{
+		Listener: l,
+		sem:      make(chan struct{}, n),
+		done:     make(chan struct{}),
+		log:      log,
+	}
+}
+
+type limitListener struct {
+	net.Listener
+	sem       chan struct{}
+	closeOnce sync.Once     // ensures the done chan is only closed once
+	done      chan struct{} // no values sent; closed when Close is called
+	log       *zerolog.Logger
+}
+
+func (l *limitListener) acquire() bool {
+	select {
+	case <-l.done:
+		return false
+	case l.sem <- struct{}{}:
+		return true
+	default:
+		return false
+	}
+}
+func (l *limitListener) release() { <-l.sem }
+
+func (l *limitListener) Accept() (net.Conn, error) {
+
+	// Accept the connection irregardless
+	c, err := l.Listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+
+	// If we cannot acquire the semaphore, close the connection
+	if acquired := l.acquire(); !acquired {
+		zlog := log.Warn()
+
+		var err error
+		if c != nil {
+			err = c.Close()
+			zlog.Str(logger.ECSServerAddress, c.LocalAddr().String())
+			zlog.Str(logger.ECSClientAddress, c.RemoteAddr().String())
+			zlog.Err(err)
+		}
+		zlog.Int("max", cap(l.sem)).Msg("Connection closed due to max limit")
+
+		return c, nil
+	}
+
+	return &limitListenerConn{Conn: c, release: l.release}, nil
+}
+
+func (l *limitListener) Close() error {
+	err := l.Listener.Close()
+	l.closeOnce.Do(func() { close(l.done) })
+	return err
+}
+
+type limitListenerConn struct {
+	net.Conn
+	releaseOnce sync.Once
+	release     func()
+}
+
+func (l *limitListenerConn) Close() error {
+	err := l.Conn.Close()
+	l.releaseOnce.Do(l.release)
+	return err
+}

internal/pkg/limit/listener_test.go

@@ -0,0 +1,71 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package limit
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/nettest"
+
+	testlog "github.com/elastic/fleet-server/v7/internal/pkg/testing/log"
+)
+
+func TestLimitListener(t *testing.T) {
+	logger := testlog.SetLogger(t)
+	ll, err := nettest.NewLocalListener("tcp")
+	require.NoError(t, err)
+	defer ll.Close()
+	l := Listener(ll, 1, &logger)
+
+	ch := make(chan struct{})
+	done := make(chan struct{})
+
+	t.Log("Form 1st connection")
+	go func() {
+		_, err := net.Dial("tcp", l.Addr().String())
+		require.NoError(t, err)
+	}()
+	// should accept a connection
+	conn, err := l.Accept()
+	require.NoError(t, err, "expected to be able to form one connection")
+
+	t.Log("Form 2nd connection")
+	go func() {
+		conn, err := net.Dial("tcp", l.Addr().String())
+		require.NoError(t, err)
+
+		select {
+		case <-ch:
+		case <-time.After(time.Second):
+			require.Fail(t, "expected channel write before timeout")
+		}
+
+		var p []byte
+		n, err := conn.Read(p)
+		assert.NoError(t, err)
+		assert.Equal(t, 0, n)
+
+		err = conn.Close()
+		require.NoError(t, err)
+		done <- struct{}{}
+	}()
+
+	conn2, err := l.Accept()
+	require.NoError(t, err)
+	n, err := conn2.Write([]byte(`hellow world`))
+	ch <- struct{}{}
+	assert.Error(t, err)
+	assert.Equal(t, 0, n)
+
+	err = conn.Close()
+	require.NoError(t, err)
+	err = l.Close()
+	require.NoError(t, err)
+	<-done
+}