Limit HTTP server's concurrency using semaphore.

Signed-off-by: Julius Hinze <juliusmh@proton.me>
k8snetworkplumbingwg · Nov 12, 2024 · 685bb6b · 685bb6b
1 parent 781ecda
commit 685bb6b
Show file tree

Hide file tree

Showing 10 changed files with 217 additions and 6 deletions.
diff --git a/docs/thick-plugin.md b/docs/thick-plugin.md
@@ -72,6 +72,7 @@ is provided.
 - `"logLevel"`: the logging level for the multus daemon logs.
 - `"logToStderr"`: enable this to have the daemon multus logs echoed to stderr
 as well. By default, it is disabled.
+- `concurrentExecs`: integer that, if specified, defines the amount of parallel chroot plugin executions (optional).
 
 In addition, you can add any configuration which is in [configuration reference](https://github.com/k8snetworkplumbingwg/multus-cni/blob/master/docs/configuration.md#multus-cni-configuration-reference). Server configuration override multus CNI configuration (e.g. `/etc/cni/net.d/00-multus.conf`)
 
@@ -89,7 +90,8 @@ Below you can see an example of the daemon configuration:
         "cniVersion": "0.3.1",
         "cniConfigDir": "/host/etc/cni/net.d",
         "multusConfigFile": "auto",
-        "multusAutoconfigDir": "/host/etc/cni/net.d"
+        "multusAutoconfigDir": "/host/etc/cni/net.d",
+        "concurrentExecs": 10
     }
 ```
 

diff --git a/go.mod b/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/vishvananda/netlink v1.1.1-0.20210330154013-f5de75959ad5
 	golang.org/x/net v0.23.0
+	golang.org/x/sync v0.4.0
 	golang.org/x/sys v0.18.0
 	google.golang.org/grpc v1.58.3
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0

diff --git a/go.sum b/go.sum
@@ -148,6 +148,8 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.4.0 h1:zxkM55ReGkDlKSM+Fu41A+zmbZuaPVbGMzvvdUPznYQ=
+golang.org/x/sync v0.4.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

diff --git a/pkg/server/server.go b/pkg/server/server.go
@@ -29,6 +29,7 @@ import (
 	"github.com/containernetworking/cni/pkg/skel"
 	cnitypes "github.com/containernetworking/cni/pkg/types"
 	cni100 "github.com/containernetworking/cni/pkg/types/100"
+	"golang.org/x/sync/semaphore"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -46,7 +47,7 @@ import (
 	netdefinformerv1 "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/client/informers/externalversions/k8s.cni.cncf.io/v1"
 
 	kapi "k8s.io/api/core/v1"
-	meta "k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -251,18 +252,33 @@ func NewCNIServer(daemonConfig *ControllerNetConf, serverConfig []byte, ignoreRe
 		logging.Verbosef("server configured with chroot: %s", daemonConfig.ChrootDir)
 	}
 
-	return newCNIServer(daemonConfig.SocketDir, kubeClient, exec, serverConfig, ignoreReadinessIndicator)
+	return newCNIServer(daemonConfig.SocketDir, kubeClient, exec, serverConfig, ignoreReadinessIndicator, daemonConfig.ConcurrentExecs)
 }
 
-func newCNIServer(rundir string, kubeClient *k8s.ClientInfo, exec invoke.Exec, servConfig []byte, ignoreReadinessIndicator bool) (*Server, error) {
+func newCNIServer(rundir string, kubeClient *k8s.ClientInfo, exec invoke.Exec, servConfig []byte, ignoreReadinessIndicator bool, concurrency *int) (*Server, error) {
 	informerFactory, podInformer := newPodInformer(kubeClient.Client, os.Getenv("MULTUS_NODE_NAME"))
 	netdefInformerFactory, netdefInformer := newNetDefInformer(kubeClient.NetClient)
 	kubeClient.SetK8sClientInformers(podInformer, netdefInformer)
 
 	router := http.NewServeMux()
+	handler := http.Handler(router)
+
+	// limit concurrent requests by using a semaphore
+	if concurrency != nil {
+		sem := semaphore.NewWeighted(int64(*concurrency))
+		handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if err := sem.Acquire(r.Context(), 1); err != nil {
+				http.Error(w, fmt.Sprintf("%v", err), http.StatusInternalServerError)
+				return
+			}
+			defer sem.Release(1)
+			router.ServeHTTP(w, r)
+		})
+	}
+
 	s := &Server{
 		Server: http.Server{
-			Handler: router,
+			Handler: handler,
 		},
 		rundir:       rundir,
 		kubeclient:   kubeClient,

diff --git a/pkg/server/thick_cni_test.go b/pkg/server/thick_cni_test.go
@@ -36,6 +36,7 @@ import (
 	"k8s.io/client-go/tools/record"
 
 	netfake "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/client/clientset/versioned/fake"
+
 	k8s "gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/k8sclient"
 	"gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/server/api"
 	testhelpers "gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/testing"
@@ -274,7 +275,7 @@ func createFakePod(k8sClient *k8s.ClientInfo, podName string) error {
 func startCNIServer(ctx context.Context, runDir string, k8sClient *k8s.ClientInfo, servConfig []byte) (*Server, error) {
 	const period = 0
 
-	cniServer, err := newCNIServer(runDir, k8sClient, &fakeExec{}, servConfig, true)
+	cniServer, err := newCNIServer(runDir, k8sClient, &fakeExec{}, servConfig, true, nil)
 	if err != nil {
 		return nil, err
 	}

diff --git a/pkg/server/types.go b/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"`
+	ConcurrentExecs    *int                `json:"concurrentExecs,omitempty"`
 
 	MetricsPort *int `json:"metricsPort,omitempty"`
 

diff --git a/vendor/golang.org/x/sync/LICENSE b/vendor/golang.org/x/sync/LICENSE
diff --git a/vendor/golang.org/x/sync/PATENTS b/vendor/golang.org/x/sync/PATENTS
diff --git a/vendor/golang.org/x/sync/semaphore/semaphore.go b/vendor/golang.org/x/sync/semaphore/semaphore.go
diff --git a/vendor/modules.txt b/vendor/modules.txt
@@ -224,6 +224,9 @@ golang.org/x/net/trace
 ## explicit; go 1.17
 golang.org/x/oauth2
 golang.org/x/oauth2/internal
+# golang.org/x/sync v0.4.0
+## explicit; go 1.17
+golang.org/x/sync/semaphore
 # golang.org/x/sys v0.18.0
 ## explicit; go 1.18
 golang.org/x/sys/plan9