diff --git a/cmd/multus-daemon/main.go b/cmd/multus-daemon/main.go index 11d38f05c..78796b2f0 100644 --- a/cmd/multus-daemon/main.go +++ b/cmd/multus-daemon/main.go @@ -29,6 +29,7 @@ import ( "sync" "syscall" + "golang.org/x/net/netutil" utilwait "k8s.io/apimachinery/pkg/util/wait" "gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/logging" @@ -105,7 +106,7 @@ func main() { } } - if err := startMultusDaemon(ctx, daemonConf, ignoreReadinessIndicator); err != nil { + if err := startMultusDaemon(ctx, daemonConf, ignoreReadinessIndicator, daemonConf.ConnectionLimit); err != nil { logging.Panicf("failed start the multus thick-plugin listener: %v", err) os.Exit(3) } @@ -139,7 +140,7 @@ func main() { logging.Verbosef("multus daemon is exited") } -func startMultusDaemon(ctx context.Context, daemonConfig *srv.ControllerNetConf, ignoreReadinessIndicator bool) error { +func startMultusDaemon(ctx context.Context, daemonConfig *srv.ControllerNetConf, ignoreReadinessIndicator bool, connectionLimit *int) error { if user, err := user.Current(); err != nil || user.Uid != "0" { return fmt.Errorf("failed to run multus-daemon with root: %v, now running in uid: %s", err, user.Uid) } @@ -166,6 +167,11 @@ 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 := connectionLimit; limit != nil && *limit > 0 { + logging.Debugf("connection limit: %d", *limit) + l = netutil.LimitListener(l, *limit) + } + server.Start(ctx, l) go func() { diff --git a/pkg/server/types.go b/pkg/server/types.go index 81d4d6819..9fca09a3e 100644 --- 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"` + ConnectionLimit *int `json:"connectionLimit,omitempty"` MetricsPort *int `json:"metricsPort,omitempty"` diff --git a/vendor/golang.org/x/net/netutil/listen.go b/vendor/golang.org/x/net/netutil/listen.go new file mode 100644 index 000000000..f8b779ea2 --- /dev/null +++ b/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 +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 6aaf1c642..87b82dfe1 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -219,6 +219,7 @@ golang.org/x/net/http2 golang.org/x/net/http2/hpack golang.org/x/net/idna golang.org/x/net/internal/timeseries +golang.org/x/net/netutil golang.org/x/net/trace # golang.org/x/oauth2 v0.10.0 ## explicit; go 1.17