Add max_connections to TCP input (elastic#12691)

New configuration option is added to the TCP input of Filebeat to limit the number of incoming connections. By default it is set to 0 which means, it's unlimited.

```
# Max number of concurrent connections, or 0 for no limit. Default: 0
#max_connections: 0
```

Vendored
* `golang.org/x/net/netutil`

filebeat/_meta/common.reference.inputs.yml

@@ -263,6 +263,9 @@ filebeat.inputs:
  # Maximum size in bytes of the message received over TCP
  #max_message_size: 20MiB
 
+ # Max number of concurrent connections, or 0 for no limit. Default: 0
+ #max_connections: 0
+
  # The number of seconds of inactivity before a remote connection is closed.
  #timeout: 300s
 

filebeat/filebeat.reference.yml

@@ -640,6 +640,9 @@ filebeat.inputs:
  # Maximum size in bytes of the message received over TCP
  #max_message_size: 20MiB
 
+ # Max number of concurrent connections, or 0 for no limit. Default: 0
+ #max_connections: 0
+
  # The number of seconds of inactivity before a remote connection is closed.
  #timeout: 300s
 

filebeat/inputsource/tcp/config.go

@@ -35,6 +35,7 @@ type Config struct {
  Host string `config:"host"`
  Timeout time.Duration `config:"timeout" validate:"nonzero,positive"`
  MaxMessageSize cfgtype.ByteSize `config:"max_message_size" validate:"nonzero,positive"`
+ MaxConnections int `config:"max_connections"`
  TLS *tlscommon.ServerConfig `config:"ssl"`
 }
 

filebeat/inputsource/tcp/server.go

@@ -25,6 +25,8 @@ import (
  "net"
  "sync"
 
+ "golang.org/x/net/netutil"
+
  "github.com/elastic/beats/filebeat/inputsource"
  "github.com/elastic/beats/libbeat/common/transport/tlscommon"
  "github.com/elastic/beats/libbeat/logp"
@@ -175,12 +177,26 @@ func (s *Server) allClients() []*client {
 }
 
 func (s *Server) createServer() (net.Listener, error) {
+ var l net.Listener
+ var err error
  if s.tlsConfig != nil {
  t := s.tlsConfig.BuildModuleConfig(s.config.Host)
  s.log.Info("Listening over TLS")
- return tls.Listen("tcp", s.config.Host, t)
+ l, err = tls.Listen("tcp", s.config.Host, t)
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ l, err = net.Listen("tcp", s.config.Host)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ if s.config.MaxConnections > 0 {
+ return netutil.LimitListener(l, s.config.MaxConnections), nil
  }
- return net.Listen("tcp", s.config.Host)
+ return l, nil
 }
 
 func (s *Server) clientsCount() int {

vendor/vendor.json

@@ -2361,6 +2361,12 @@
  "version": "release-branch.go1.9",
  "versionExact": "release-branch.go1.9"
  },
+ {
+ "checksumSHA1": "CkkyCjLPBm0OGzfMuTfaDsd45X4=",
+ "path": "golang.org/x/net/netutil",
+ "revision": "3b0461eec859c4b73bb64fdc8285971fd33e3938",
+ "revisionTime": "2019-06-20T19:42:36Z"
+ },
  {
  "checksumSHA1": "QEm/dePZ0lOnyOs+m22KjXfJ/IU=",
  "path": "golang.org/x/net/proxy",

x-pack/filebeat/filebeat.reference.yml

@@ -759,6 +759,9 @@ filebeat.inputs:
  # Maximum size in bytes of the message received over TCP
  #max_message_size: 20MiB
 
+ # Max number of concurrent connections, or 0 for no limit. Default: 0
+ #max_connections: 0
+
  # The number of seconds of inactivity before a remote connection is closed.
  #timeout: 300s