Ankusa.HttpClient.Pools (ankusa v0.7.0)

Copy Markdown View Source

The outbound HTTP connection pools: one Finch for the node (Ankusa.HttpClient.Finch, started with the application) that every Ankusa.HttpClient request without a transport of its own goes through, and the bookkeeping that keeps its pools bounded.

Req's default Finch gives each origin a pool of 50 connections that is never closed (pool_max_idle_time: :infinity) and is HTTP/1 only. A node delivering to one origin per tenant (Ankusa.Sink.Http's computed URLs) would hold 10⁵ pools and their sockets until it ran out of file descriptors. Here every pool is started on demand, sized, counted and swept by this module:

  • Pool: one per origin and connection profile (the connection options a request needs: connect timeout, TLS, the hostname a pinned address is verified against) and per instance. pool_size HTTP/1 connections (pool_count shards) each, from the instance's egress.pools config or a sink's pool: override; a request that cannot check one out within pool_timeout fails that attempt as a transport error.
  • Sweep: every 30 s a pool with no request in flight and none for pool_max_idle_ms is stopped and its connections closed (the mirror of Ankusa.Sink.Reaper); a connection idle for conn_max_idle_ms is not reused. A row whose pool died on its own is dropped.
  • Ceilings: max_pools live pools per instance and max_pools_per_tenant per tenant (the tenant whose delivery opened the pool). At a ceiling the least recently used idle pool (of that tenant for its own ceiling, so a tenant with a thousand origins only ever churns its own pools) is closed to make room; when every pool in scope has a request in flight the request is refused as {:error, {:unavailable, :pool_limit}}, which the retry policy retries after its backoff.
  • Source delete: close_source/3 stops the idle pools a deleted source's deliveries last used.
  • Metrics: gauge ankusa_http_pools ([:ankusa, :http, :state], :pools), from count/1; counters ankusa_http_pool_closed_total{reason} and ankusa_http_pool_refused_total{scope}.

The race it accepts

A request counts itself in (:ets.update_counter/3 on the pool's row) before it touches the pool, and a sweep deletes a row only while that count is zero, so a pool is never stopped under a request that found it. A request that arrives after the row was deleted asks this process to open the pool again, which is serialised behind the sweep.

A pool is created from pool_opts/1 with its connection options in the pool itself, so every connection it opens uses them: for a pinned destination that is the address Ankusa.HttpClient.Egress validated, with the original name as the TLS hostname (SNI and certificate check). Nothing in a pool resolves a name on its own.

Summary

Types

request()

What run/2 needs to find or open the pool for one request.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

close_source(instance, tenant, source)

Stop the idle pools the deliveries of source source of tenant last used. Called when the source is deleted; a pool with a request in flight is left to the idle sweep. Asynchronous.

count(instance)

How many pools instance has open: a read of the maintained rows, no Finch call.

create_table()

Create the row table. The application calls it, so the rows outlive a restart of this process: the pools it opened are still running.

finch()

The name of the node's Finch.

finch_child_spec()

Child spec of the node's Finch. Its default pool is only what Finch falls back to for a pool nobody started through run/2: the bounds the plan names (8 connections, idle pool 60 s, idle connection 30 s, HTTP/1 and HTTP/2), and TLS verification on, so a stray pool fails closed.

run(request, fun)

Run fun with the Finch options (name, pool_tag, pool_timeout) of the pool request belongs to, opening it if need be, and release it afterwards.

validate_config!(config)

Log what the instance's pool settings cannot do: a pool smaller than the dispatch concurrency that can reach one origin (several sinks posting to one receiver would queue on checkout, silently capping throughput), and a ceiling whose worst case is more connections than the process may open. Returns :ok; the ranges themselves are Ankusa.Config's.

validate_override!(pool)

Raise ArgumentError unless pool is a usable per-sink :pool override: a keyword list of :pool_size (1 to 1024), :pool_count (1 to 64), :pool_timeout (1 to 600000 ms) and :protocols (:http1, :http2).

Types

request()

@type request() :: %{
  :url => String.t(),
  :connect_options => keyword(),
  optional(:instance) => atom() | nil,
  optional(:tenant) => String.t() | nil,
  optional(:source) => term(),
  optional(:pool) => keyword(),
  optional(:pool_timeout) => timeout() | nil
}

What run/2 needs to find or open the pool for one request.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

close_source(instance, tenant, source)

@spec close_source(atom(), String.t() | nil, String.t()) :: :ok

Stop the idle pools the deliveries of source source of tenant last used. Called when the source is deleted; a pool with a request in flight is left to the idle sweep. Asynchronous.

count(instance)

@spec count(atom() | nil) :: non_neg_integer()

How many pools instance has open: a read of the maintained rows, no Finch call.

create_table()

@spec create_table() :: :ok

Create the row table. The application calls it, so the rows outlive a restart of this process: the pools it opened are still running.

finch()

@spec finch() :: atom()

The name of the node's Finch.

finch_child_spec()

@spec finch_child_spec() :: {module(), keyword()}

Child spec of the node's Finch. Its default pool is only what Finch falls back to for a pool nobody started through run/2: the bounds the plan names (8 connections, idle pool 60 s, idle connection 30 s, HTTP/1 and HTTP/2), and TLS verification on, so a stray pool fails closed.

run(request, fun)

@spec run(request(), (keyword() -> result)) ::
  result | {:error, {:unavailable, atom()}}
when result: term()

Run fun with the Finch options (name, pool_tag, pool_timeout) of the pool request belongs to, opening it if need be, and release it afterwards.

Returns {:error, {:unavailable, :pool_limit}} without calling fun when a ceiling is reached and nothing can be closed, and {:error, {:unavailable, :pools_down}} when this process cannot be reached.

validate_config!(config)

@spec validate_config!(Ankusa.Config.t()) :: :ok

Log what the instance's pool settings cannot do: a pool smaller than the dispatch concurrency that can reach one origin (several sinks posting to one receiver would queue on checkout, silently capping throughput), and a ceiling whose worst case is more connections than the process may open. Returns :ok; the ranges themselves are Ankusa.Config's.

validate_override!(pool)

@spec validate_override!(term()) :: :ok

Raise ArgumentError unless pool is a usable per-sink :pool override: a keyword list of :pool_size (1 to 1024), :pool_count (1 to 64), :pool_timeout (1 to 600000 ms) and :protocols (:http1, :http2).