Sovite.Listener (sovite v0.6.0)

Copy Markdown View Source

A TCP listener with an acceptor pool and connection limits.

Each accepted connection is served by its own process, started from a Sovite.Listener.Handler module under the listener's connection supervisor. Connections over a limit are refused: the handler's optional reject/3 callback can send a protocol-level refusal first.

children = [
  {Sovite.Listener, port: 2525, handler: MyHandler, handler_opts: []}
]

IPv4-mapped IPv6 peer addresses are reported as IPv4. IPv6 listeners are IPv6-only, so 0.0.0.0 and :: can listen on the same port.

PROXY protocol

Behind HAProxy or a load balancer, the listener can read a PROXY protocol header (version 1 or 2, see Sovite.ProxyProtocol) from each connection to learn the real client. It is read in the connection process, by handshake/2, so slow clients cannot hold up accepting. With a PROXY command over TCP, the connection info's remote_* and local_* addresses are the header's source and destination; peer_ip and peer_port are always the proxy's end of the connection. A LOCAL command (a health check) or an unknown family keeps the peer's addresses. A missing or invalid header, or one that does not arrive in time, closes the connection.

Connection limits, including :max_connections_per_ip, and the :connection telemetry events see the peer: the proxy's address, not the client's, since they apply before the header is read.

Draining and updating

drain/1 stops accepting connections and asks the open ones to finish: each connection process gets a {:sovite_listener, :drain} message, and should close once it is idle. connection_count/1 tells when they are gone. update/2 changes the handler options and the PROXY protocol and per-IP limits for new connections, without closing the listen socket.

Options

  • :port - TCP port, 0 for any free port. Required.
  • :ip - address to listen on. Defaults to {0, 0, 0, 0}.
  • :fd - a listening socket inherited from the parent process, such as one from systemd socket activation (Sovite.Systemd.listen_fds/1), used instead of opening one on :ip and :port. :ip must then be an IPv6 address if the socket is an IPv6 one. Defaults to nil.
  • :handler - a Sovite.Listener.Handler module. Required.
  • :handler_opts - passed to the handler. Defaults to [].
  • :id - label in telemetry metadata and connection info. Defaults to "<ip>:<port>".
  • :name - registered name of the listener supervisor.
  • :acceptors - number of acceptor processes. Defaults to 10.
  • :max_connections - concurrent connections. Defaults to 1000.
  • :max_connections_per_ip - concurrent connections from one remote address, or nil for no limit. Defaults to nil.
  • :proxy_protocol - expect a PROXY protocol header. Defaults to false.
  • :proxy_networks - with :proxy_protocol, the Sovite.Net networks of the proxies. Connections from them must start with a header; others are served as direct clients, with no header read. nil, the default, means every connection must start with one.
  • :proxy_timeout - milliseconds to wait for the header. Defaults to 10 seconds.

Telemetry

  • [:sovite, :listener, :connection, :start] - %{system_time}, %{listener, remote_ip, remote_port}
  • [:sovite, :listener, :connection, :stop] - %{duration}, same metadata
  • [:sovite, :listener, :connection, :rejected] - %{}, %{listener, remote_ip, reason}
  • [:sovite, :listener, :proxy, :error] - %{}, %{listener, remote_ip, reason}: no valid PROXY header, so the connection is closed. remote_ip is the peer's address and reason a Sovite.ProxyProtocol.read/3 error.

Summary

Types

connection_info()

Passed to the handler's start_link/2, and returned updated by handshake/2

reject_reason()

Why a connection was refused.

Functions

connection_count(listener)

Returns the number of open connections.

drain(listener)

Stops accepting connections, and sends {:sovite_listener, :drain} to every open connection's process. The listener keeps running until it is stopped, so its connections are not cut off.

handshake(info, timeout \\ 5000)

Waits until the listener has handed the socket in info over to the calling connection process, then reads the PROXY protocol header if the connection must send one. Call it before using the socket, and use the connection info it returns.

sockname(listener)

Returns the address and port the listener is bound to.

start_link(opts)

Starts the listener. See the module docs for options.

update(listener, opts)

Changes options for the connections accepted from now on: :handler_opts, :max_connections_per_ip, :proxy_protocol, :proxy_networks, and :proxy_timeout. Open connections keep the options they started with.

Types

connection_info()

@type connection_info() :: %{
  listener: String.t(),
  socket: :gen_tcp.socket(),
  remote_ip: :inet.ip_address(),
  remote_port: :inet.port_number(),
  local_ip: :inet.ip_address(),
  local_port: :inet.port_number(),
  peer_ip: :inet.ip_address(),
  peer_port: :inet.port_number(),
  proxy: Sovite.ProxyProtocol.Header.t() | nil
}

Passed to the handler's start_link/2, and returned updated by handshake/2:

  • :remote_ip, :remote_port - the client. With the PROXY protocol, the header's source once handshake/2 returns.
  • :local_ip, :local_port - the server address the client connected to. With the PROXY protocol, the header's destination.
  • :peer_ip, :peer_port - the other end of the TCP connection: the client, or the proxy.
  • :proxy - the PROXY protocol header, or nil when none was read.

reject_reason()

@type reject_reason() :: :max_connections | :max_connections_per_ip

Why a connection was refused.

Functions

connection_count(listener)

@spec connection_count(Supervisor.supervisor()) :: non_neg_integer()

Returns the number of open connections.

drain(listener)

@spec drain(Supervisor.supervisor()) :: :ok

Stops accepting connections, and sends {:sovite_listener, :drain} to every open connection's process. The listener keeps running until it is stopped, so its connections are not cut off.

handshake(info, timeout \\ 5000)

@spec handshake(connection_info(), timeout()) ::
  {:ok, connection_info()} | {:error, :timeout | {:proxy, term()}}

Waits until the listener has handed the socket in info over to the calling connection process, then reads the PROXY protocol header if the connection must send one. Call it before using the socket, and use the connection info it returns.

timeout is for the handover; the header has the listener's :proxy_timeout. Returns {:error, :timeout} without a handover, or {:error, {:proxy, reason}} without a valid header, after closing the socket (see Sovite.ProxyProtocol.read/3 for the reasons).

sockname(listener)

@spec sockname(Supervisor.supervisor()) ::
  {:ok, {:inet.ip_address(), :inet.port_number()}} | {:error, term()}

Returns the address and port the listener is bound to.

start_link(opts)

@spec start_link(keyword()) :: Supervisor.on_start()

Starts the listener. See the module docs for options.

update(listener, opts)

@spec update(Supervisor.supervisor(), keyword()) :: :ok

Changes options for the connections accepted from now on: :handler_opts, :max_connections_per_ip, :proxy_protocol, :proxy_networks, and :proxy_timeout. Open connections keep the options they started with.