Полный production-рефакторинг билдера, drift-reconcile и документации по up5

This commit is contained in:
2026-05-10 16:42:09 +05:00
parent b3dad5974c
commit b8d2c4592b
12 changed files with 571 additions and 266 deletions
+3 -4
View File
@@ -8,15 +8,14 @@ peer_ipv4_cidr = "10.8.1.0/24"
enabled = true
peer_limit = "70mbit"
root_rate = "1000mbit"
reconcile_interval_secs = 60
container_ready_timeout_secs = 30
reconcile_interval_secs = 15
container_ready_timeout_secs = 120
class_start_id = 10
ifb_name = "ifb0"
full_rebuild_on_drift = true
[anti_torrent]
enabled = true
max_connections_per_peer = 200
max_connections_per_peer = 100
peer_ipv4_cidr = "10.8.1.0/24"
apply_in_amnezia_netns = true
+4 -5
View File
@@ -15,15 +15,14 @@ peer_ipv4_cidr = "10.8.1.0/24"
enabled = true
peer_limit = "70mbit"
root_rate = "1000mbit"
reconcile_interval_secs = 60
container_ready_timeout_secs = 30
reconcile_interval_secs = 15
container_ready_timeout_secs = 120
class_start_id = 10
ifb_name = "ifb0"
full_rebuild_on_drift = true
[anti_torrent]
enabled = true
max_connections_per_peer = 200
max_connections_per_peer = 100
peer_ipv4_cidr = "10.8.1.0/24"
apply_in_amnezia_netns = true
@@ -62,7 +61,7 @@ swappiness = 1
`traffic.ifb_name` — имя IFB-интерфейса для ingress shaping. В старой системе это `ifb0`. Этот параметр можно оставить в конфиге, потому что бинарь реально создаёт/удаляет этот интерфейс.
`traffic.full_rebuild_on_drift` — безопасная v1-модель: при drift удалить owned topology и собрать заново.
При drift orchestrator всегда делает controlled rebuild owned topology.
`anti_torrent.enabled` — включает nftables connection-count guard против high-peer-count P2P-нагрузки. Это новое расширение, не часть old bash.
+3 -6
View File
@@ -236,11 +236,8 @@ reconcile_traffic():
return NoChange
if drift.exists:
if config.traffic.full_rebuild_on_drift:
teardown_owned_topology(target, desired)
apply_full_topology(target, desired)
else:
apply_incremental_fix(drift)
teardown_owned_topology(target, desired)
apply_full_topology(target, desired)
verify_after_apply(target, desired)
if firewall_drift.exists:
@@ -248,7 +245,7 @@ reconcile_traffic():
verify_firewall_after_apply(target, firewall_desired)
```
For v1, `full_rebuild_on_drift = true` is the safer default.
For production v1, controlled full rebuild on drift is the only supported path.
Parser hardening requirements:
+2 -12
View File
@@ -31,6 +31,7 @@ nft
Сборку лучше выполнять на отдельной build-машине с тем же целевым ABI.
```bash
./tools/build.sh --prepare-build-env --bootstrap-rust --yes
./tools/build.sh
```
@@ -150,15 +151,4 @@ awg2-harness status --config /etc/awg2-harness/config.toml
## 9. Rollback
Если нужно временно вернуться к old bash:
```bash
systemctl stop awg2-harness.service
systemctl disable awg2-harness.service
# вернуть /opt/amnezia/limit_peers.sh и /opt/amnezia/docker_watcher.sh
# вернуть amnezia-tc.service и amnezia-tc-watcher.service
systemctl daemon-reload
systemctl enable --now amnezia-tc.service amnezia-tc-watcher.service
```
Rollback должен быть проверен до первой production-выкладки.
Production-путь AWG2 Harness не включает возврат к old bash в штатный процесс выкладки.
+3 -4
View File
@@ -183,15 +183,14 @@ peer_ipv4_cidr = "10.8.1.0/24"
enabled = true
peer_limit = "70mbit"
root_rate = "1000mbit"
reconcile_interval_secs = 60
container_ready_timeout_secs = 30
reconcile_interval_secs = 15
container_ready_timeout_secs = 120
class_start_id = 10
ifb_name = "ifb0"
full_rebuild_on_drift = true
[anti_torrent]
enabled = true
max_connections_per_peer = 200
max_connections_per_peer = 100
peer_ipv4_cidr = "10.8.1.0/24"
apply_in_amnezia_netns = true
+2 -1
View File
@@ -1,3 +1,4 @@
[toolchain]
channel = "1.85.0"
channel = "1.95.0"
profile = "minimal"
components = ["rustfmt", "clippy"]
+43 -38
View File
@@ -1,54 +1,59 @@
# Build tooling
# AWG2 Harness builder
[`build.sh`](tools/build.sh) — production build gate для [`traffic_core`](traffic_core/Cargo.toml).
Основной скрипт: [`build.sh`](build.sh).
Целевой crate: [`traffic_core`](../traffic_core/Cargo.toml).
Unit: [`awg2-harness.service`](../awg2-harness.service).
Пример конфига: [`config.example.toml`](../config.example.toml).
## Что делает
## Supported host
Скрипт:
- Debian/Ubuntu (apt-based)
- amd64/x86_64
1. Проверяет OS и архитектуру (Debian/Ubuntu, amd64/x86_64).
2. Проверяет наличие инструментов: `cargo`, `rustc`, `gcc`, `pkg-config`, `file`, `ldd`.
3. Проверяет минимальную версию Rust (`rustc >= 1.85.0`).
4. Запускает quality gate:
- `cargo fmt --check` (если не указан `--skip-fmt`),
- `cargo clippy --locked --all-targets -- -D warnings`,
- `cargo test --locked --release`,
- `cargo build --locked --release`.
5. Проверяет собранный бинарь: `file`, `ldd`, `--help`, `sha256sum`.
## Что НЕ делает
- Не устанавливает Rust/toolchain автоматически.
- Не устанавливает системные зависимости через apt.
- Не создаёт production-конфиг автоматически.
## Использование
Из корня репозитория:
## Default mode
```bash
./tools/build.sh
```
Опции:
Дефолтный запуск **не изменяет систему**. Он:
- `--skip-fmt` — пропустить шаг `cargo fmt --check`.
- `--install` — после успешной сборки установить бинарь и systemd unit.
- проверяет prereqs;
- запускает `fmt/clippy/test/build` в locked-режиме;
- печатает `file`, `ldd`, `sha256` для собранного бинаря.
## Режим `--install`
## Flags
`--install`:
- `--prepare-build-env` — установить недостающие build-пакеты;
- `--prepare-runtime-env` — установить недостающие runtime-пакеты;
- `--bootstrap-rust` — установить rustup и pinned toolchain;
- `--install` — установить бинарь + systemd unit (без enable/start);
- `--enable` — `systemctl enable --now awg2-harness.service`;
- `--yes` — подтверждение системных изменений (apt/rustup);
- `--skip-fmt` — пропустить `cargo fmt --check`;
- `--help` — справка.
1. Копирует бинарь в `/usr/local/bin/awg2-harness`.
2. Проверяет, что `/etc/awg2-harness/config.toml` уже существует.
3. Копирует [`awg2-harness.service`](awg2-harness.service).
4. Выполняет `systemd-analyze verify` (если доступно).
5. Выполняет `systemctl daemon-reload`.
## Typical build machine flow
Важно:
```bash
./tools/build.sh --prepare-build-env --bootstrap-rust --yes
./tools/build.sh
```
- Если скрипт запущен от root, команды установки выполняются напрямую.
- Если не от root — используется `sudo` (и проверяется его наличие).
- Скрипт **не** выполняет `systemctl enable --now` автоматически.
## Typical production host flow
Это intentional: включение сервиса и запуск остаются осознанным эксплуатационным шагом.
```bash
./tools/build.sh --prepare-runtime-env --yes
install -d -m 0755 /etc/awg2-harness
install -m 0644 config.toml /etc/awg2-harness/config.toml
./tools/build.sh
./tools/build.sh --install
./tools/build.sh --enable
```
## What the script intentionally does not do
- не устанавливает Docker автоматически;
- не генерирует production config из example;
- не включает сервис без явного `--enable`;
- не управляет установкой AmneziaWG.
+218 -72
View File
@@ -4,103 +4,249 @@ set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
CRATE_MANIFEST="$ROOT_DIR/traffic_core/Cargo.toml"
TARGET_BIN="$ROOT_DIR/traffic_core/target/release/awg2-harness"
MIN_RUSTC="1.85.0"
RUST_TOOLCHAIN_FILE="$ROOT_DIR/rust-toolchain.toml"
RUST_TOOLCHAIN_VERSION="1.95.0"
AUTO_INSTALL=false
PREPARE_BUILD_ENV=false
PREPARE_RUNTIME_ENV=false
BOOTSTRAP_RUST=false
INSTALL=false
ENABLE=false
YES=false
SKIP_FMT=false
RUN_AS_ROOT=()
BUILD_PACKAGES=(
ca-certificates
curl
build-essential
pkg-config
file
shellcheck
)
RUNTIME_PACKAGES=(
iproute2
wireguard-tools
nftables
kmod
util-linux
procps
systemd
)
usage() {
cat <<'EOF'
Usage:
./tools/build.sh
./tools/build.sh --prepare-build-env --yes
./tools/build.sh --prepare-runtime-env --yes
./tools/build.sh --bootstrap-rust --yes
./tools/build.sh --install
./tools/build.sh --enable
Flags:
--prepare-build-env Install missing build dependencies via apt
--prepare-runtime-env Install missing runtime dependencies via apt
--bootstrap-rust Install rustup/toolchain pinned by this repo
--install Install binary and systemd unit (no enable/start)
--enable Enable and start awg2-harness.service
--yes Confirm system mutations (apt/rustup)
--skip-fmt Skip cargo fmt --check
--help Show this help
EOF
}
for arg in "$@"; do
case "$arg" in
--install) AUTO_INSTALL=true ;;
--prepare-build-env) PREPARE_BUILD_ENV=true ;;
--prepare-runtime-env) PREPARE_RUNTIME_ENV=true ;;
--bootstrap-rust) BOOTSTRAP_RUST=true ;;
--install) INSTALL=true ;;
--enable) ENABLE=true ;;
--yes) YES=true ;;
--skip-fmt) SKIP_FMT=true ;;
*) echo "Unknown arg: $arg"; exit 1 ;;
--help|-h) usage; exit 0 ;;
*) echo "Unknown arg: $arg"; usage; exit 1 ;;
esac
done
require_cmd() {
command -v "$1" >/dev/null 2>&1 || { echo "Missing command: $1"; exit 1; }
command -v "$1" >/dev/null 2>&1 || {
echo "Missing command: $1"
exit 1
}
}
version_ge() {
local have="$1"
local need="$2"
[[ "$(printf '%s\n' "$need" "$have" | sort -V | head -n1)" == "$need" ]]
}
if [[ -f /etc/os-release ]]; then
# shellcheck disable=SC1091
source /etc/os-release
case "${ID:-}" in
debian|ubuntu) ;;
*) echo "Unsupported OS: ${ID:-unknown}. Only Debian/Ubuntu are supported."; exit 1 ;;
esac
fi
ARCH="$(uname -m)"
if [[ "$ARCH" != "x86_64" && "$ARCH" != "amd64" ]]; then
echo "Unsupported arch: $ARCH. Only amd64/x86_64 is supported."
exit 1
fi
if ! command -v cargo >/dev/null 2>&1; then
echo "cargo not found. Install rustup first: https://rustup.rs"
exit 1
fi
require_cmd rustc
RUSTC_VERSION="$(rustc --version | awk '{print $2}')"
if ! version_ge "$RUSTC_VERSION" "$MIN_RUSTC"; then
echo "rustc $RUSTC_VERSION is too old. Require >= $MIN_RUSTC"
exit 1
fi
for cmd in gcc pkg-config; do
require_cmd "$cmd"
done
if [[ "$SKIP_FMT" != true ]]; then
cargo fmt --manifest-path "$CRATE_MANIFEST" --all --check
fi
cargo clippy --locked --manifest-path "$CRATE_MANIFEST" --all-targets -- -D warnings
cargo test --locked --manifest-path "$CRATE_MANIFEST" --release
cargo build --locked --manifest-path "$CRATE_MANIFEST" --release
require_cmd file
require_cmd ldd
file "$TARGET_BIN"
ldd "$TARGET_BIN" || true
"$TARGET_BIN" --help >/dev/null
sha256sum "$TARGET_BIN"
ls -lh "$TARGET_BIN"
if [[ "$AUTO_INSTALL" == true ]]; then
require_cmd install
RUN_AS_ROOT=( )
setup_root_runner() {
if [[ "$(id -u)" -eq 0 ]]; then
RUN_AS_ROOT=( )
RUN_AS_ROOT=()
else
require_cmd sudo
RUN_AS_ROOT=(sudo)
fi
}
"${RUN_AS_ROOT[@]}" install -m 0755 "$TARGET_BIN" /usr/local/bin/awg2-harness
"${RUN_AS_ROOT[@]}" mkdir -p /etc/awg2-harness
if [[ ! -f /etc/awg2-harness/config.toml ]]; then
echo "Missing /etc/awg2-harness/config.toml"
echo "Refusing to install example config as production config."
echo "Create config manually from $ROOT_DIR/config.example.toml and re-run with --install."
ensure_supported_host() {
if [[ -f /etc/os-release ]]; then
# shellcheck disable=SC1091
source /etc/os-release
case "${ID:-}" in
debian|ubuntu) ;;
*) echo "Unsupported OS: ${ID:-unknown}. Only Debian/Ubuntu are supported."; exit 1 ;;
esac
fi
local arch
arch="$(uname -m)"
if [[ "$arch" != "x86_64" && "$arch" != "amd64" ]]; then
echo "Unsupported arch: $arch. Only amd64/x86_64 is supported."
exit 1
fi
}
apt_install_missing() {
local packages=("$@")
local missing=()
require_cmd dpkg-query
require_cmd apt-get
for pkg in "${packages[@]}"; do
if ! dpkg-query -W -f='${Status}' "$pkg" 2>/dev/null | grep -q "install ok installed"; then
missing+=("$pkg")
fi
done
if [[ "${#missing[@]}" -eq 0 ]]; then
echo "All required packages already installed"
return 0
fi
echo "Missing packages: ${missing[*]}"
if [[ "$YES" != true ]]; then
echo "Re-run with --yes to install them"
exit 1
fi
setup_root_runner
"${RUN_AS_ROOT[@]}" apt-get update
"${RUN_AS_ROOT[@]}" apt-get install -y --no-install-recommends "${missing[@]}"
}
bootstrap_rust() {
require_cmd curl
if [[ "$YES" != true ]]; then
echo "This will install rustup and Rust toolchain $RUST_TOOLCHAIN_VERSION"
echo "Re-run with --yes to proceed"
exit 1
fi
if ! command -v rustup >/dev/null 2>&1; then
local rustup_script
rustup_script="$(mktemp)"
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs -o "$rustup_script"
sh "$rustup_script" -y --profile minimal --default-toolchain "$RUST_TOOLCHAIN_VERSION"
rm -f "$rustup_script"
fi
export PATH="$HOME/.cargo/bin:$PATH"
require_cmd rustup
rustup toolchain install "$RUST_TOOLCHAIN_VERSION" --profile minimal --component rustfmt --component clippy
rustup component add rustfmt clippy --toolchain "$RUST_TOOLCHAIN_VERSION"
rustc --version
}
prepare_build_gate() {
require_cmd cargo
require_cmd rustc
require_cmd gcc
require_cmd pkg-config
require_cmd file
require_cmd ldd
if [[ ! -f "$RUST_TOOLCHAIN_FILE" ]]; then
echo "Missing rust toolchain pin file: $RUST_TOOLCHAIN_FILE"
exit 1
fi
if [[ ! -f "$ROOT_DIR/traffic_core/Cargo.lock" ]]; then
echo "Missing lockfile: $ROOT_DIR/traffic_core/Cargo.lock"
exit 1
fi
if [[ "$SKIP_FMT" != true ]]; then
cargo fmt --manifest-path "$CRATE_MANIFEST" --all --check
fi
cargo clippy --locked --manifest-path "$CRATE_MANIFEST" --all-targets -- -D warnings
cargo test --locked --manifest-path "$CRATE_MANIFEST" --release
cargo build --locked --manifest-path "$CRATE_MANIFEST" --release
file "$TARGET_BIN"
ldd "$TARGET_BIN" || true
"$TARGET_BIN" --help >/dev/null
sha256sum "$TARGET_BIN"
ls -lh "$TARGET_BIN"
}
install_artifacts() {
require_cmd install
require_cmd systemctl
if [[ ! -f "$TARGET_BIN" ]]; then
echo "Binary not found: $TARGET_BIN"
echo "Run build first: ./tools/build.sh"
exit 1
fi
if [[ ! -f /etc/awg2-harness/config.toml ]]; then
echo "Missing /etc/awg2-harness/config.toml"
echo "Refusing to install example config as production config"
echo "Create config manually from $ROOT_DIR/config.example.toml and re-run with --install"
exit 1
fi
setup_root_runner
"${RUN_AS_ROOT[@]}" install -m 0755 "$TARGET_BIN" /usr/local/bin/awg2-harness
"${RUN_AS_ROOT[@]}" install -m 0644 "$ROOT_DIR/awg2-harness.service" /etc/systemd/system/awg2-harness.service
if command -v systemd-analyze >/dev/null 2>&1; then
"${RUN_AS_ROOT[@]}" systemd-analyze verify /etc/systemd/system/awg2-harness.service
fi
"${RUN_AS_ROOT[@]}" systemctl daemon-reload
}
enable_service() {
require_cmd systemctl
setup_root_runner
"${RUN_AS_ROOT[@]}" systemctl enable --now awg2-harness.service
}
ensure_supported_host
if [[ "$PREPARE_BUILD_ENV" == true ]]; then
apt_install_missing "${BUILD_PACKAGES[@]}"
fi
if [[ "$PREPARE_RUNTIME_ENV" == true ]]; then
apt_install_missing "${RUNTIME_PACKAGES[@]}"
fi
if [[ "$BOOTSTRAP_RUST" == true ]]; then
bootstrap_rust
fi
prepare_build_gate
if [[ "$INSTALL" == true ]]; then
install_artifacts
fi
if [[ "$ENABLE" == true ]]; then
enable_service
fi
echo "Build completed: $TARGET_BIN"
-1
View File
@@ -28,7 +28,6 @@ pub struct TrafficConfig {
pub container_ready_timeout_secs: u64,
pub class_start_id: u16,
pub ifb_name: String,
pub full_rebuild_on_drift: bool,
}
#[derive(Debug, Clone, Deserialize)]
+89 -45
View File
@@ -65,27 +65,21 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
}
fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result<FirewallActual> {
let table = transport
.run(&["nft", "list", "table", "inet", "awg2_harness"])
.unwrap_or_default();
let table = inspect_cmd(
transport,
&["nft", "list", "table", "inet", "awg2_harness"],
"firewall table",
)?;
let table_exists = !table.trim().is_empty();
let counter_exists = table.contains("counter antitorrent_dropped_v4");
let chain_exists = table.contains("chain forward_antitorrent");
let needle_cidr = format!("ip saddr {}", cfg.anti_torrent.peer_ipv4_cidr);
let needle_max = format!(
"ct count over {}",
cfg.anti_torrent.max_connections_per_peer
let rule_matches = firewall_rule_contract_matches(
&table,
&cfg.anti_torrent.peer_ipv4_cidr,
cfg.anti_torrent.max_connections_per_peer,
);
let rule_matches = table.contains(&needle_cidr)
&& table.contains(&needle_max)
&& table.contains("ct state new")
&& table.contains("meter awg2_peer_connlimit_v4")
&& table.contains("counter name antitorrent_dropped_v4")
&& table.contains("drop")
&& table.contains("hook forward")
&& table.contains("policy accept");
Ok(FirewallActual {
table_exists,
@@ -185,6 +179,27 @@ fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Res
}
}
fn inspect_cmd(transport: &dyn NsTransport, args: &[&str], what: &str) -> anyhow::Result<String> {
transport
.run(args)
.map_err(|e| anyhow::anyhow!("{what} inspect failed: {e}"))
}
fn firewall_rule_contract_matches(table: &str, cidr: &str, max_conn: u32) -> bool {
let needle_cidr = format!("ip saddr {cidr}");
let needle_max = format!("ct count over {max_conn}");
table.contains("table inet awg2_harness")
&& table.contains("counter antitorrent_dropped_v4")
&& table.contains("chain forward_antitorrent")
&& table.contains("type filter hook forward priority filter; policy accept")
&& table.contains(&needle_cidr)
&& table.contains("ct state new")
&& table.contains("meter awg2_peer_connlimit_v4")
&& table.contains(&needle_max)
&& table.contains("counter name antitorrent_dropped_v4")
&& table.contains("drop")
}
pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result<Option<(u64, u64)>> {
if !cfg.anti_torrent.enabled {
return Ok(None);
@@ -201,38 +216,25 @@ pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result<Option<(u64, u64
"antitorrent_dropped_v4",
]);
if let Ok(raw) = json {
if let Some(v) = parse_counter_json(&raw)? {
return Ok(Some(v));
match json {
Ok(raw) => {
if let Some(v) = parse_counter_json(&raw)? {
return Ok(Some(v));
}
return Ok(None);
}
Err(e) => {
let msg = e.to_string();
if msg.contains("No such file")
|| msg.contains("No such")
|| msg.contains("not found")
|| msg.contains("not exist")
{
return Ok(None);
}
return Err(anyhow::anyhow!("nft counter inspection failed: {e}"));
}
}
let table = transport
.run(&["nft", "list", "table", "inet", "awg2_harness"])
.unwrap_or_default();
for line in table.lines() {
let t = line.trim();
if !t.contains("packets") {
continue;
}
let toks: Vec<&str> = t.split_whitespace().collect();
let packets = toks
.iter()
.position(|x| *x == "packets")
.and_then(|i| toks.get(i + 1))
.and_then(|s| s.parse::<u64>().ok());
let bytes = toks
.iter()
.position(|x| *x == "bytes")
.and_then(|i| toks.get(i + 1))
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0);
if let Some(p) = packets {
return Ok(Some((p, bytes)));
}
}
Ok(None)
}
fn parse_counter_json(raw: &str) -> anyhow::Result<Option<(u64, u64)>> {
@@ -262,4 +264,46 @@ mod tests {
let got = parse_counter_json(raw).expect("json parse");
assert_eq!(got, Some((7, 448)));
}
#[test]
fn nft_rule_contract_parser_accepts_expected_rule() {
let table = r#"
table inet awg2_harness {
counter antitorrent_dropped_v4
chain forward_antitorrent {
type filter hook forward priority filter; policy accept;
ip saddr 10.8.1.0/24 ct state new meter awg2_peer_connlimit_v4 { ip saddr ct count over 100 } counter name antitorrent_dropped_v4 drop
}
}
"#;
assert!(firewall_rule_contract_matches(table, "10.8.1.0/24", 100));
}
#[test]
fn missing_counter_name_causes_drift() {
let table = r#"
table inet awg2_harness {
counter antitorrent_dropped_v4
chain forward_antitorrent {
type filter hook forward priority filter; policy accept;
ip saddr 10.8.1.0/24 ct state new meter awg2_peer_connlimit_v4 { ip saddr ct count over 100 } drop
}
}
"#;
assert!(!firewall_rule_contract_matches(table, "10.8.1.0/24", 100));
}
#[test]
fn wrong_max_connection_threshold_causes_drift() {
let table = r#"
table inet awg2_harness {
counter antitorrent_dropped_v4
chain forward_antitorrent {
type filter hook forward priority filter; policy accept;
ip saddr 10.8.1.0/24 ct state new meter awg2_peer_connlimit_v4 { ip saddr ct count over 250 } counter name antitorrent_dropped_v4 drop
}
}
"#;
assert!(!firewall_rule_contract_matches(table, "10.8.1.0/24", 100));
}
}
+150 -78
View File
@@ -17,6 +17,13 @@ struct TrafficDrift {
reason: String,
}
#[derive(Debug, Clone)]
struct FilterInspect {
map: BTreeMap<String, u16>,
duplicate_ips: BTreeSet<String>,
duplicate_classes: BTreeSet<u16>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ClassRateCeil {
rate: String,
@@ -106,13 +113,6 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
println!("traffic drift detected: {}", drift.reason);
if !cfg.traffic.full_rebuild_on_drift {
anyhow::bail!(
"traffic drift detected but full_rebuild_on_drift=false: {}",
drift.reason
);
}
teardown_owned_topology(
transport.as_ref(),
&target.interface_name,
@@ -174,8 +174,10 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
.run(&["tc", "class", "show", "dev", &cfg.traffic.ifb_name])
.unwrap_or_default();
let (actual_map_egress, dup_egress) = extract_filter_ip_class_map(&filters_egress);
let (actual_map_ingress, dup_ingress) = extract_filter_ip_class_map(&filters_ingress);
let inspect_egress = extract_filter_ip_class_map(&filters_egress);
let inspect_ingress = extract_filter_ip_class_map(&filters_ingress);
let actual_map_egress = inspect_egress.map;
let actual_map_ingress = inspect_ingress.map;
let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:");
let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:");
let expected_map = expected_peer_class_map(&peers, cfg.traffic.class_start_id);
@@ -216,10 +218,10 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
{
reasons.push("ingress rate/ceil mismatch".to_string());
}
if dup_egress.contains(&ip) {
if inspect_egress.duplicate_ips.contains(&ip) {
reasons.push("duplicate egress filters".to_string());
}
if dup_ingress.contains(&ip) {
if inspect_ingress.duplicate_ips.contains(&ip) {
reasons.push("duplicate ingress filters".to_string());
}
@@ -252,9 +254,11 @@ fn inspect_and_diff(
peer_limit: &str,
peers: &[wg::Peer],
) -> anyhow::Result<TrafficDrift> {
let root_qdisc = transport
.run(&["tc", "qdisc", "show", "dev", iface])
.unwrap_or_default();
let root_qdisc = inspect_cmd(
transport,
&["tc", "qdisc", "show", "dev", iface],
"egress qdisc",
)?;
if !root_qdisc.contains("htb") {
return Ok(TrafficDrift {
has_drift: true,
@@ -262,9 +266,11 @@ fn inspect_and_diff(
});
}
let ingress_qdisc = transport
.run(&["tc", "qdisc", "show", "dev", iface])
.unwrap_or_default();
let ingress_qdisc = inspect_cmd(
transport,
&["tc", "qdisc", "show", "dev", iface],
"ingress qdisc",
)?;
if !ingress_qdisc.contains("ingress") {
return Ok(TrafficDrift {
has_drift: true,
@@ -272,43 +278,65 @@ fn inspect_and_diff(
});
}
let ifb_link = transport
.run(&["ip", "link", "show", ifb])
.unwrap_or_default();
let ifb_link = inspect_cmd(transport, &["ip", "link", "show", ifb], "ifb link")?;
if ifb_link.trim().is_empty() {
return Ok(TrafficDrift {
has_drift: true,
reason: "missing ifb interface".to_string(),
});
}
if !ifb_link.contains(" UP ") && !ifb_link.contains("state UP") {
return Ok(TrafficDrift {
has_drift: true,
reason: "ifb interface is not UP".to_string(),
});
}
let filters_egress = transport
.run(&["tc", "filter", "show", "dev", iface, "parent", "1:"])
.unwrap_or_default();
let filters_ingress = transport
.run(&["tc", "filter", "show", "dev", ifb, "parent", "2:"])
.unwrap_or_default();
let filters_egress = inspect_cmd(
transport,
&["tc", "filter", "show", "dev", iface, "parent", "1:"],
"egress filters",
)?;
let filters_ingress = inspect_cmd(
transport,
&["tc", "filter", "show", "dev", ifb, "parent", "2:"],
"ingress filters",
)?;
let (actual_map_egress, dup_egress) = extract_filter_ip_class_map(&filters_egress);
let (actual_map_ingress, dup_ingress) = extract_filter_ip_class_map(&filters_ingress);
if !dup_egress.is_empty() {
let inspect_egress = extract_filter_ip_class_map(&filters_egress);
let inspect_ingress = extract_filter_ip_class_map(&filters_ingress);
if !inspect_egress.duplicate_ips.is_empty() {
return Ok(TrafficDrift {
has_drift: true,
reason: format!(
"duplicate egress filters detected for {} peer(s)",
dup_egress.len()
inspect_egress.duplicate_ips.len()
),
});
}
if !dup_ingress.is_empty() {
if !inspect_egress.duplicate_classes.is_empty() {
return Ok(TrafficDrift {
has_drift: true,
reason: "duplicate egress class flowid mapping detected".to_string(),
});
}
if !inspect_ingress.duplicate_ips.is_empty() {
return Ok(TrafficDrift {
has_drift: true,
reason: format!(
"duplicate ingress filters detected for {} peer(s)",
dup_ingress.len()
inspect_ingress.duplicate_ips.len()
),
});
}
if !inspect_ingress.duplicate_classes.is_empty() {
return Ok(TrafficDrift {
has_drift: true,
reason: "duplicate ingress class flowid mapping detected".to_string(),
});
}
let actual_map_egress = inspect_egress.map;
let actual_map_ingress = inspect_ingress.map;
let actual_ips_egress: BTreeSet<String> = actual_map_egress.keys().cloned().collect();
let actual_ips_ingress: BTreeSet<String> = actual_map_ingress.keys().cloned().collect();
let expected_ips: BTreeSet<String> = peers.iter().map(|p| p.ipv4.to_string()).collect();
@@ -358,12 +386,16 @@ fn inspect_and_diff(
}
let expected_limit = normalize_tc_rate(peer_limit);
let classes_egress_full = transport
.run(&["tc", "class", "show", "dev", iface])
.unwrap_or_default();
let classes_ingress_full = transport
.run(&["tc", "class", "show", "dev", ifb])
.unwrap_or_default();
let classes_egress_full = inspect_cmd(
transport,
&["tc", "class", "show", "dev", iface],
"egress classes",
)?;
let classes_ingress_full = inspect_cmd(
transport,
&["tc", "class", "show", "dev", ifb],
"ingress classes",
)?;
let expected_root = normalize_tc_rate(root_rate);
let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:");
let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:");
@@ -414,32 +446,40 @@ fn inspect_and_diff(
}
}
let classes_egress = transport
.run(&["tc", "class", "show", "dev", iface, "parent", "1:1"])
.unwrap_or_default();
let classes_ingress = transport
.run(&["tc", "class", "show", "dev", ifb, "parent", "2:1"])
.unwrap_or_default();
let expected_class_count = peers.len();
if count_leaf_classes(&classes_egress) != expected_class_count {
let classes_egress = inspect_cmd(
transport,
&["tc", "class", "show", "dev", iface, "parent", "1:1"],
"egress leaf classes",
)?;
let classes_ingress = inspect_cmd(
transport,
&["tc", "class", "show", "dev", ifb, "parent", "2:1"],
"ingress leaf classes",
)?;
let expected_set: BTreeSet<u16> = expected_map.values().copied().collect();
if collect_leaf_class_ids(&classes_egress) != expected_set {
return Ok(TrafficDrift {
has_drift: true,
reason: "egress per-peer class count mismatch".to_string(),
reason: "egress per-peer class set mismatch".to_string(),
});
}
if count_leaf_classes(&classes_ingress) != expected_class_count {
if collect_leaf_class_ids(&classes_ingress) != expected_set {
return Ok(TrafficDrift {
has_drift: true,
reason: "ingress per-peer class count mismatch".to_string(),
reason: "ingress per-peer class set mismatch".to_string(),
});
}
let qdisc_egress = transport
.run(&["tc", "qdisc", "show", "dev", iface])
.unwrap_or_default();
let qdisc_ingress = transport
.run(&["tc", "qdisc", "show", "dev", ifb])
.unwrap_or_default();
let qdisc_egress = inspect_cmd(
transport,
&["tc", "qdisc", "show", "dev", iface],
"egress leaf qdisc",
)?;
let qdisc_ingress = inspect_cmd(
transport,
&["tc", "qdisc", "show", "dev", ifb],
"ingress leaf qdisc",
)?;
for class_id in expected_map.values() {
let class_egress = format!("1:{class_id}");
let class_ingress = format!("2:{class_id}");
@@ -463,9 +503,17 @@ fn inspect_and_diff(
})
}
fn extract_filter_ip_class_map(filters: &str) -> (BTreeMap<String, u16>, BTreeSet<String>) {
fn inspect_cmd(transport: &dyn NsTransport, args: &[&str], what: &str) -> anyhow::Result<String> {
transport
.run(args)
.map_err(|e| anyhow::anyhow!("{what} inspect failed: {e}"))
}
fn extract_filter_ip_class_map(filters: &str) -> FilterInspect {
let mut out = BTreeMap::new();
let mut duplicates = BTreeSet::new();
let mut duplicate_ips = BTreeSet::new();
let mut class_to_ip = BTreeMap::new();
let mut duplicate_classes = BTreeSet::new();
for block in split_filter_blocks(filters) {
let mut ip: Option<String> = None;
@@ -483,13 +531,20 @@ fn extract_filter_ip_class_map(filters: &str) -> (BTreeMap<String, u16>, BTreeSe
if let (Some(ip), Some(class_id)) = (ip, class_id) {
if out.contains_key(&ip) {
duplicates.insert(ip.clone());
duplicate_ips.insert(ip.clone());
}
if class_to_ip.insert(class_id, ip.clone()).is_some() {
duplicate_classes.insert(class_id);
}
out.insert(ip, class_id);
}
}
(out, duplicates)
FilterInspect {
map: out,
duplicate_ips,
duplicate_classes,
}
}
fn has_fq_codel_for_parent(raw_qdisc: &str, parent_class: &str) -> bool {
@@ -585,16 +640,24 @@ fn hex_to_ipv4_string(hex: &str) -> Option<String> {
Some(Ipv4Addr::new(b1, b2, b3, b4).to_string())
}
fn count_leaf_classes(raw: &str) -> usize {
raw.lines()
.filter_map(|l| {
let line = l.trim();
let pos = line.find("class htb ")?;
let class = &line[(pos + "class htb ".len())..];
let head = class.split_whitespace().next()?;
if head.ends_with(":1") { None } else { Some(()) }
})
.count()
fn collect_leaf_class_ids(raw: &str) -> BTreeSet<u16> {
let mut out = BTreeSet::new();
for l in raw.lines() {
let line = l.trim();
let Some(pos) = line.find("class htb ") else {
continue;
};
let class = &line[(pos + "class htb ".len())..];
let Some(head) = class.split_whitespace().next() else {
continue;
};
if let Some(class_id) = head.split(':').nth(1).and_then(|v| v.parse::<u16>().ok())
&& class_id != 1
{
out.insert(class_id);
}
}
out
}
fn teardown_owned_topology(
@@ -876,25 +939,34 @@ mod tests {
#[test]
fn parse_tc_u32_hex_ipv4() {
let raw = " match 0a080105/ffffffff at 16\n flowid 1:42";
let (m, dup) = extract_filter_ip_class_map(raw);
assert!(dup.is_empty());
assert_eq!(m.get("10.8.1.5").copied(), Some(42));
let inspect = extract_filter_ip_class_map(raw);
assert!(inspect.duplicate_ips.is_empty());
assert!(inspect.duplicate_classes.is_empty());
assert_eq!(inspect.map.get("10.8.1.5").copied(), Some(42));
}
#[test]
fn parse_block_based_filters() {
let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:11";
let (m, dup) = extract_filter_ip_class_map(raw);
assert!(dup.is_empty());
assert_eq!(m.get("10.8.1.2").copied(), Some(10));
assert_eq!(m.get("10.8.1.3").copied(), Some(11));
let inspect = extract_filter_ip_class_map(raw);
assert!(inspect.duplicate_ips.is_empty());
assert!(inspect.duplicate_classes.is_empty());
assert_eq!(inspect.map.get("10.8.1.2").copied(), Some(10));
assert_eq!(inspect.map.get("10.8.1.3").copied(), Some(11));
}
#[test]
fn detect_duplicate_filter_ip() {
let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080102/ffffffff at 16\n flowid 1:10";
let (_m, dup) = extract_filter_ip_class_map(raw);
assert!(dup.contains("10.8.1.2"));
let inspect = extract_filter_ip_class_map(raw);
assert!(inspect.duplicate_ips.contains("10.8.1.2"));
}
#[test]
fn detect_duplicate_filter_class_flowid() {
let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:10";
let inspect = extract_filter_ip_class_map(raw);
assert!(inspect.duplicate_classes.contains(&10));
}
#[test]
+54
View File
@@ -95,6 +95,34 @@ fn parse_fallback_wg_show(raw: &str, peer_cidr: &str) -> anyhow::Result<Vec<Peer
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use std::sync::Mutex;
struct TestTransport {
replies: Mutex<BTreeMap<String, anyhow::Result<String>>>,
}
impl TestTransport {
fn new(entries: Vec<(&str, anyhow::Result<&str>)>) -> Self {
let mut replies = BTreeMap::new();
for (cmd, out) in entries {
let mapped = out.map(|s| s.to_string());
replies.insert(cmd.to_string(), mapped);
}
Self {
replies: Mutex::new(replies),
}
}
}
impl crate::netns::NsTransport for TestTransport {
fn run(&self, args: &[&str]) -> anyhow::Result<String> {
let key = args.join(" ");
let mut lock = self.replies.lock().expect("lock");
lock.remove(&key)
.unwrap_or_else(|| Err(anyhow::anyhow!("missing stub for: {key}")))
}
}
#[test]
fn parse_primary() {
@@ -126,4 +154,30 @@ mod tests {
assert_eq!(peers[0].ipv4.to_string(), "10.8.1.10");
assert_eq!(peers[1].ipv4.to_string(), "10.8.1.2");
}
#[test]
fn command_failure_is_propagated() {
let transport = TestTransport::new(vec![(
"wg show all allowed-ips",
Err(anyhow::anyhow!("wg broken")),
)]);
let err = discover_peers_with_transport(&transport, "10.8.1.0/24").err();
assert!(err.is_some());
let msg = err.expect("err").to_string();
assert!(msg.contains("wg show all allowed-ips failed"));
}
#[test]
fn empty_primary_output_falls_back_to_wg_show() {
let transport = TestTransport::new(vec![
("wg show all allowed-ips", Ok("")),
("wg show", Ok("allowed ips: 10.8.1.4/32, 10.8.1.5/32")),
]);
let peers = discover_peers_with_transport(&transport, "10.8.1.0/24").expect("peers");
assert_eq!(peers.len(), 2);
assert_eq!(peers[0].ipv4.to_string(), "10.8.1.4");
assert_eq!(peers[1].ipv4.to_string(), "10.8.1.5");
}
}