Documentation
¶
Overview ¶
Package p2plimits builds the libp2p resource manager a mesh component runs with: limits sized from its connection budget, a per-source-address connection cap and rate in place of libp2p's default of 8, applied to the address a connection comes from and not to a relay it comes through, and a wrapper that counts and logs every inbound connection the manager refuses. libp2p refuses a connection before any byte of TLS and, for its per-IP and rate-limit paths, without a trace event; a dialer sees the connection close or a QUIC CONNECTION_REFUSED, and nothing else records it.
Index ¶
- Constants
- Variables
- func DefaultConnsPerSourceIP(highWaterMark int) int
- func NewResourceManager(limits rcmgr.ConcreteLimitConfig, opts ...rcmgr.Option) (network.ResourceManager, error)
- func RefusalReason(err error) string
- func Streams(n int) rcmgr.ResourceLimits
- type Observed
- func (m *Observed) GetConnLimit() int
- func (m *Observed) LastLoggedAt(source string) (time.Time, bool)
- func (m *Observed) OpenConnection(dir network.Direction, usefd bool, endpoint multiaddr.Multiaddr) (network.ConnManagementScope, error)
- func (m *Observed) SetClock(now func() time.Time)
- func (m *Observed) SourceConns(addr netip.Addr) int
- func (m *Observed) SourcesLogged() int
- type StreamPressure
- type StreamRefusals
- func (s *StreamRefusals) AllowConn(network.Direction, bool)
- func (s *StreamRefusals) AllowMemory(int)
- func (s *StreamRefusals) AllowPeer(peer.ID)
- func (s *StreamRefusals) AllowProtocol(protocol.ID)
- func (s *StreamRefusals) AllowService(string)
- func (s *StreamRefusals) AllowStream(peer.ID, network.Direction)
- func (s *StreamRefusals) BlockConn(network.Direction, bool)
- func (s *StreamRefusals) BlockMemory(size int)
- func (s *StreamRefusals) BlockPeer(peer.ID)
- func (s *StreamRefusals) BlockProtocol(proto protocol.ID)
- func (s *StreamRefusals) BlockProtocolPeer(proto protocol.ID, _ peer.ID)
- func (s *StreamRefusals) BlockService(svc string)
- func (s *StreamRefusals) BlockServicePeer(svc string, _ peer.ID)
- func (s *StreamRefusals) BlockStream(_ peer.ID, _ network.Direction)
- func (s *StreamRefusals) SetClock(now func() time.Time)
Constants ¶
const ( RefusedPerIP = "per_ip_limit" RefusedRateLimit = "rate_limit" RefusedScope = "scope_limit" RefusedOther = "other" )
Why the resource manager refused an inbound connection; the label values of a refusal counter.
const ( StreamScopeProtocol = "protocol" StreamScopeProtocolPeer = "protocol_peer" StreamScopeService = "service" StreamScopeServicePeer = "service_peer" StreamScopePeer = "peer" // StreamScopeConn is the system or transient scope, where the manager // does not yet know the protocol. StreamScopeConn = "conn" )
The scopes a stream refusal is counted under. libp2p's own gauge (libp2p_rcmgr_blocked_resources) carries the scope but not the protocol, and a hub needs the protocol to tell a flood from a limit set too low.
ConnsPerSourceIPShare is the fraction of the inbound connection budget one source address may hold when no cap is configured.
const QUICStreamsPerConnection = 256
QUICStreamsPerConnection is how many streams the far end may hold open on one QUIC connection, as go-libp2p configures quic-go; a stream opened past it waits for one to close, with no error at either end. yamux over TCP has no limit of its own; the resource manager's are the only ones. Every relayed connection to a peer is one stream on the relay's connection to it, so a service called through a router has at most this many relayed callers through it over QUIC.
const RefusalLogInterval = time.Minute
RefusalLogInterval bounds how often one source IP's refusals are logged; the counter carries the rate.
const StreamPressureWarnAt = QUICStreamsPerConnection * 9 / 10
StreamPressureWarnAt is the stream count at which a QUIC connection is reported: nine tenths of what it admits.
const StreamWindow = 256 << 10
StreamWindow is the receive window yamux reserves for a stream when it opens (go-yamux's initialStreamWindow); a stream that carries a fast transfer to a slow reader may grow it, up to the muxer's maximum.
const TransportQUIC = "quic-v1"
TransportQUIC is the transport name libp2p gives a QUIC connection; WebTransport runs over QUIC with the same stream limit.
Variables ¶
var ( ErrPerSourceConns = errors.New("connections per ip limit exceeded") ErrPerSourceRate = errors.New("rate limit exceeded") )
What Observed answers a direct inbound connection with past the per-source cap or rate; the texts are libp2p's own for the same refusals.
var Reasons = []string{RefusedPerIP, RefusedRateLimit, RefusedScope, RefusedOther}
Reasons lists every refusal label, for a counter to pre-register.
Functions ¶
func DefaultConnsPerSourceIP ¶
DefaultConnsPerSourceIP is the per-source-address cap derived from a connection manager high watermark. Members share addresses (pod SNAT, Cloud NAT, offices, a load generator), and the handshake, not the address, is the trust boundary.
func NewResourceManager ¶
func NewResourceManager(limits rcmgr.ConcreteLimitConfig, opts ...rcmgr.Option) (network.ResourceManager, error)
NewResourceManager is libp2p's resource manager with the given limits and its own per-address limiters switched off: Observe applies the per-source cap and rate, which lets it tell a connection that comes from an address apart from one that comes through a relay. opts are passed on, for a MetricsReporter among others.
func RefusalReason ¶
RefusalReason maps the resource manager's errors to the counter's labels. The per-IP and rate-limit paths return plain strings, not sentinel errors.
func Streams ¶
func Streams(n int) rcmgr.ResourceLimits
Streams is a stream budget of n each way, with the memory the streams reserve: yamux reserves one receive window per open stream through the resource manager, so a budget of streams is a budget of memory, and libp2p's default for the latter is scaled from host memory, which on the testnet's routers was 1.1 GiB and full at 1500 members over TCP. The memory is a reservation, released with the stream, and allocated only as data arrives and waits to be read.
Types ¶
type Observed ¶
type Observed struct {
network.ResourceManager
// contains filtered or unexported fields
}
Observed applies the per-source cap and rate to the inbound connections that come from an address, and counts and logs what it or the wrapped manager refuses. A connection that comes through a relay carries the relay's address, not its source's: libp2p charged it to the relay, so a service called by many members through a router saw them all as one address and refused most. Such a connection is bounded by the manager's connection budget and its per-peer scope, and at the relay by the relay's own per-peer and per-address budgets, so no per-source rule is applied to it here. Loopback is exempt, as in libp2p's default, so a local mesh never meets the cap. Observed keeps the wrapped manager's connection limit visible to libp2p's connmgr sanity check.
func New ¶
func New(limits rcmgr.ConcreteLimitConfig, perIP int, refused *prometheus.CounterVec, warnf func(format string, args ...any), opts ...rcmgr.Option) (*Observed, error)
New is NewResourceManager wrapped in Observe: the manager a component's host runs with. A cap under one would refuse every peer, so it is an error here rather than a silent listener.
func Observe ¶
func Observe(inner network.ResourceManager, perIP int, refused *prometheus.CounterVec, warnf func(format string, args ...any)) *Observed
Observe wraps inner with a per-source cap of perIP. refused is counted by reason (see Reasons) and warnf written once per source address per RefusalLogInterval; a nil warnf only counts.
func (*Observed) GetConnLimit ¶
GetConnLimit is what libp2p compares the connection manager's high water mark against at host construction.
func (*Observed) LastLoggedAt ¶
LastLoggedAt is when source was last logged, and whether it is remembered.
func (*Observed) OpenConnection ¶
func (*Observed) SourceConns ¶
SourceConns is how many inbound connections from addr are open.
func (*Observed) SourcesLogged ¶
SourcesLogged is how many source addresses the log limiter remembers.
type StreamPressure ¶
type StreamPressure struct {
// contains filtered or unexported fields
}
StreamPressure reports the connection of a host that carries the most streams, by transport, and warns once per peer per RefusalLogInterval when a QUIC connection is within reach of its stream limit: the next stream between the two waits, which a relay turns into a hop that hangs until its connect timeout and a caller into a dial that times out, neither of which says why.
func ObserveStreamPressure ¶
func ObserveStreamPressure(conns func() []network.Conn, busiest *prometheus.GaugeVec, warnf func(format string, args ...any)) *StreamPressure
ObserveStreamPressure returns a sampler over conns. busiest carries the label transport; a nil warnf only samples.
func (*StreamPressure) Run ¶
func (p *StreamPressure) Run(ctx context.Context, interval time.Duration)
Run samples every interval until ctx ends.
func (*StreamPressure) Sample ¶
func (p *StreamPressure) Sample()
Sample reads every connection once.
func (*StreamPressure) SetClock ¶
func (p *StreamPressure) SetClock(now func() time.Time)
SetClock replaces the clock the log limiter reads; for tests.
type StreamRefusals ¶
type StreamRefusals struct {
// contains filtered or unexported fields
}
StreamRefusals counts the streams the resource manager refused, by scope and protocol (or service), and the memory reservations it refused, and logs each once per RefusalLogInterval. It is libp2p's MetricsReporter; the Allow side is a no-op.
func CountStreamRefusals ¶
func CountStreamRefusals(refused *prometheus.CounterVec, memoryRefused prometheus.Counter, warnf func(format string, args ...any)) *StreamRefusals
CountStreamRefusals returns a reporter for rcmgr.WithMetrics. refused carries the labels scope and protocol; memoryRefused counts the memory reservations refused, which the stream muxer makes one of per stream it opens and the relay one of per connection it relays; a nil warnf only counts.
func (*StreamRefusals) AllowConn ¶
func (s *StreamRefusals) AllowConn(network.Direction, bool)
Connections and memory are observed elsewhere (Observed and libp2p's own gauges); allowances are not of interest.
func (*StreamRefusals) AllowMemory ¶
func (s *StreamRefusals) AllowMemory(int)
func (*StreamRefusals) AllowPeer ¶
func (s *StreamRefusals) AllowPeer(peer.ID)
func (*StreamRefusals) AllowProtocol ¶
func (s *StreamRefusals) AllowProtocol(protocol.ID)
func (*StreamRefusals) AllowService ¶
func (s *StreamRefusals) AllowService(string)
func (*StreamRefusals) AllowStream ¶
func (s *StreamRefusals) AllowStream(peer.ID, network.Direction)
func (*StreamRefusals) BlockMemory ¶
func (s *StreamRefusals) BlockMemory(size int)
BlockMemory is a memory reservation the manager refused: the muxer's window for a new stream, or the relay's buffers for a connection it would relay. The budget it ran into is libp2p_rcmgr_limit{resource= "memory"} on the scope libp2p_rcmgr_memory shows full.
func (*StreamRefusals) BlockPeer ¶
func (s *StreamRefusals) BlockPeer(peer.ID)
func (*StreamRefusals) BlockProtocol ¶
func (s *StreamRefusals) BlockProtocol(proto protocol.ID)
func (*StreamRefusals) BlockProtocolPeer ¶
func (s *StreamRefusals) BlockProtocolPeer(proto protocol.ID, _ peer.ID)
func (*StreamRefusals) BlockService ¶
func (s *StreamRefusals) BlockService(svc string)
func (*StreamRefusals) BlockServicePeer ¶
func (s *StreamRefusals) BlockServicePeer(svc string, _ peer.ID)
func (*StreamRefusals) BlockStream ¶
func (s *StreamRefusals) BlockStream(_ peer.ID, _ network.Direction)
func (*StreamRefusals) SetClock ¶
func (s *StreamRefusals) SetClock(now func() time.Time)
SetClock replaces the clock the log interval is measured on.
Source Files