Merge pull request #1510 from hrntknr/limit_listener

Add connection limit for thick daemon.

.github/workflows/kind-e2e.yml

@@ -85,6 +85,10 @@ jobs:
         working-directory: ./e2e
         run: ./test-default-route1.sh
 
+      - name: Test connection limit
+        working-directory: ./e2e
+        run: ./test-connection-limit.sh
+
 #      - name: Test DRA integration
 #        working-directory: ./e2e
 #        run: ./test-dra-integration.sh

cmd/multus-daemon/main.go

@@ -31,6 +31,7 @@ import (
 	"syscall"
 	"time"
 
+	"golang.org/x/net/netutil"
 	utilwait "k8s.io/apimachinery/pkg/util/wait"
 
 	"gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/logging"
@@ -189,6 +190,14 @@ func startMultusDaemon(ctx context.Context, daemonConfig *srv.ControllerNetConf,
 		return fmt.Errorf("failed to start the CNI server using socket %s. Reason: %+v", api.SocketPath(daemonConfig.SocketDir), err)
 	}
 
+	if limit := daemonConfig.ConnectionLimit; limit != nil {
+		if *limit <= 0 {
+			return fmt.Errorf("connection limit must be greater than 0, got %d", *limit)
+		}
+		logging.Debugf("connection limit: %d", *limit)
+		l = netutil.LimitListener(l, *limit)
+	}
+
 	server.Start(ctx, l)
 
 	go func() {

e2e/templates/many-pods.yml.j2

@@ -0,0 +1,23 @@
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: simple-centos1
+  labels:
+    app: many
+spec:
+  replicas: 6
+  selector:
+    matchLabels:
+      app: many
+  template:
+    metadata:
+      labels:
+        app: many
+    spec:
+      containers:
+      - name: simple-centos1
+        image: centos:8
+        command: ["/bin/sleep", "10000"]
+        securityContext:
+          privileged: true

e2e/templates/multus-daemonset-thick.yml.j2

@@ -97,7 +97,8 @@ data:
         "cniConfigDir": "/host/etc/cni/net.d",
         "multusConfigFile": "auto",
         "forceCNIVersion": true,
-        "multusAutoconfigDir": "/host/etc/cni/net.d"
+        "multusAutoconfigDir": "/host/etc/cni/net.d",
+        "connectionLimit": 1
     }
 ---
 apiVersion: apps/v1

e2e/test-connection-limit.sh

@@ -0,0 +1,10 @@
+#!/bin/sh
+set -o errexit
+
+export PATH=${PATH}:./bin
+
+kubectl create -f yamls/many-pods.yml
+kubectl wait --for=condition=ready -l app=many --timeout=300s pod
+
+echo "cleanup resources"
+kubectl delete -f yamls/many-pods.yml

pkg/server/types.go

@@ -77,6 +77,7 @@ type ControllerNetConf struct {
 	LogLevel           string              `json:"logLevel"`
 	LogToStderr        bool                `json:"logToStderr,omitempty"`
 	PerNodeCertificate *PerNodeCertificate `json:"perNodeCertificate,omitempty"`
+	ConnectionLimit    *int                `json:"connectionLimit,omitempty"`
 
 	MetricsPort *int  `json:"metricsPort,omitempty"`
 	EnablePprof *bool `json:"enablePprof,omitempty"`

vendor/golang.org/x/net/netutil/listen.go

@@ -0,0 +1,87 @@
+// Copyright 2013 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package netutil provides network utility functions, complementing the more
+// common ones in the net package.
+package netutil // import "golang.org/x/net/netutil"
+
+import (
+	"net"
+	"sync"
+)
+
+// LimitListener returns a Listener that accepts at most n simultaneous
+// connections from the provided Listener.
+func LimitListener(l net.Listener, n int) net.Listener {
+	return &limitListener{
+		Listener: l,
+		sem:      make(chan struct{}, n),
+		done:     make(chan struct{}),
+	}
+}
+
+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
+}
+
+// acquire acquires the limiting semaphore. Returns true if successfully
+// acquired, false if the listener is closed and the semaphore is not
+// acquired.
+func (l *limitListener) acquire() bool {
+	select {
+	case <-l.done:
+		return false
+	case l.sem <- struct{}{}:
+		return true
+	}
+}
+func (l *limitListener) release() { <-l.sem }
+
+func (l *limitListener) Accept() (net.Conn, error) {
+	if !l.acquire() {
+		// If the semaphore isn't acquired because the listener was closed, expect
+		// that this call to accept won't block, but immediately return an error.
+		// If it instead returns a spurious connection (due to a bug in the
+		// Listener, such as https://golang.org/issue/50216), we immediately close
+		// it and try again. Some buggy Listener implementations (like the one in
+		// the aforementioned issue) seem to assume that Accept will be called to
+		// completion, and may otherwise fail to clean up the client end of pending
+		// connections.
+		for {
+			c, err := l.Listener.Accept()
+			if err != nil {
+				return nil, err
+			}
+			c.Close()
+		}
+	}
+
+	c, err := l.Listener.Accept()
+	if err != nil {
+		l.release()
+		return nil, err
+	}
+	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
+}

vendor/modules.txt

@@ -208,6 +208,7 @@ golang.org/x/net/http2/hpack
 golang.org/x/net/idna
 golang.org/x/net/internal/httpcommon
 golang.org/x/net/internal/timeseries
+golang.org/x/net/netutil
 golang.org/x/net/trace
 # golang.org/x/oauth2 v0.34.0
 ## explicit; go 1.24.0