openmoat_proxy/server/
capacity.rs

1//! Connections refused at accept for capacity, in the audit log.
2//!
3//! A connection over `max_connections`, or over the share of those not yet
4//! authenticated, is answered 503 before its head is read. Each is a
5//! refusal whether or not it is recorded; recording every one would let a
6//! flood fill the log, so the first refusal of a kind is recorded at once and
7//! the ones after it are counted and recorded as one row per [`INTERVAL`],
8//! when the next connection is accepted or refused after it
9//! ([`crate::coalesce`]).
10
11use std::time::Instant;
12
13use openmoat_core::{Decision, Verdict};
14
15use crate::audit::{Connection, Recorder};
16pub(super) use crate::coalesce::INTERVAL;
17use crate::coalesce::{Coalesce, counted};
18
19/// Rule id for a connection refused at accept because the proxy, or the
20/// connections not yet authenticated, hold all the slots they may.
21pub const RULE_CAPACITY: &str = "proxy-capacity";
22
23/// Why a connection was refused at accept.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub(super) enum Full {
26    /// Every slot is taken.
27    Slots,
28    /// The connections not yet authenticated hold their share.
29    Pending,
30}
31
32impl Full {
33    const ALL: [Self; 2] = [Self::Slots, Self::Pending];
34
35    /// What the 503 says.
36    pub(super) fn reply(self) -> &'static str {
37        match self {
38            Self::Slots => "too many connections",
39            Self::Pending => "too many connections waiting to authenticate",
40        }
41    }
42}
43
44/// The refusals at accept of one [`Proxy::serve`](crate::Proxy::serve).
45#[derive(Debug, Default)]
46pub(super) struct Refusals {
47    counts: Coalesce<2>,
48}
49
50impl Refusals {
51    /// Count one refusal, and record the count once [`INTERVAL`] has passed
52    /// since the last row of its kind.
53    pub(super) fn refused(&self, full: Full, recorder: &dyn Recorder, now: Instant) {
54        if let Some(count) = self.counts.refused(full as usize, now) {
55            write(full, count, recorder);
56        }
57    }
58
59    /// Record the refusals counted since the last row, where [`INTERVAL`] has
60    /// passed: called as connections are accepted, so a burst that ends is
61    /// still recorded.
62    pub(super) fn flush(&self, recorder: &dyn Recorder, now: Instant) {
63        for (kind, count) in self.counts.due(now) {
64            if let Some(full) = Full::ALL.get(kind) {
65                write(*full, count, recorder);
66            }
67        }
68    }
69}
70
71fn write(full: Full, count: u64, recorder: &dyn Recorder) {
72    let decision = Decision::single(
73        Verdict::Deny,
74        RULE_CAPACITY,
75        format!(
76            "{} refused before a request was read: {}; later ones within {} s are counted in \
77             the next row",
78            counted(count, "connection", "connections"),
79            full.reply(),
80            INTERVAL.as_secs()
81        ),
82    );
83    let connection = Connection {
84        session: None,
85        method: None,
86        destination: None,
87        terminated: false,
88        decision: &decision,
89        latency_us: 0,
90    };
91    // The connections are refused either way; a row that cannot be written
92    // is reported where the recorder reports failures.
93    if let Err(error) = recorder.record(&connection) {
94        recorder.unrecorded(&error);
95    }
96}
97
98#[cfg(test)]
99mod tests;
openmoat_proxy/server/capacity.rs

openmoat_proxy/server/
capacity.rs
