diff --git a/README.md b/README.md index 6868a1e..efe63a0 100644 --- a/README.md +++ b/README.md @@ -277,7 +277,7 @@ Swap: drift: no ``` -Если peers ещё нет, `apply-limits` корректно напишет `no peers found, skip tc apply`. Это не ошибка: traffic topology строится, когда есть хотя бы один peer allowed IP в целевой подсети. +Если peers ещё нет, это здоровое desired state. `apply-limits` удалит оставшуюся AWG2-owned traffic topology, если раньше peers были, и завершится успешно. В штатном цикле daemon это не создаёт повторяющийся `info`-шум. ### 8. Включить systemd services @@ -345,7 +345,7 @@ awg2-harness --config /etc/awg2-harness/config.toml `apply-firewall` делает два reconcile: host INPUT firewall в таблице `inet awg2_host_firewall` и anti-torrent guard в выбранном namespace. Команда не управляет Docker NAT и host `FORWARD`. -`apply-limits` находит Amnezia container, заходит в его network namespace, читает peers через container-side `awg`/`wg`, строит per-peer HTB/fq_codel классы на AWG/WG интерфейсе и IFB ingress path. Если drift не найден, пишет `traffic rules up-to-date, skip apply`. +`apply-limits` находит Amnezia container, заходит в его network namespace, читает peers через container-side `awg`/`wg`, строит per-peer HTB/fq_codel классы на AWG/WG интерфейсе и IFB ingress path. Peer filters создаются classifier-ом `flower` с единым `pref 10`; legacy peer-`фильтры `u32` считаются drift и мигрируют через controlled rebuild. Healthy state не печатает сообщение на каждом цикле daemon. `status` печатает фактическую диагностику: сервис, контейнер, netns, интерфейс, peers, WireGuard runtime, traffic desired/actual, host firewall drift, Docker/Amnezia port path, anti-torrent counters, swap state, conntrack, recent events и planned reboot. @@ -355,7 +355,7 @@ awg2-harness --config /etc/awg2-harness/config.toml ## Что происходит под капотом -Traffic shaping реализован через `tc` в network namespace контейнера Amnezia. На egress используется root qdisc `htb 1:` на AWG/WG интерфейсе, root class `1:1`, per-peer classes `1:` и `fq_codel` под каждым peer class. На ingress ставится `ingress ffff:` на AWG/WG интерфейс, входящие пакеты перенаправляются через `mirred` в `ifb0`, а там строится отдельная HTB topology `2:` с такими же per-peer лимитами. +Traffic shaping реализован через `tc` в network namespace контейнера Amnezia. На egress используется root qdisc `htb 1:` на AWG/WG интерфейсе, root class `1:1`, per-peer classes `1:` и `fq_codel` под каждым peer class. Egress peer filter сопоставляет `dst_ip /32`, ingress peer filter — `src_ip /32`; оба используют `flower pref 10`. На ingress ставится `ingress ffff:` на AWG/WG интерфейс, входящие пакеты перенаправляются через `mirred` в `ifb0`, а там строится отдельная HTB topology `2:` с такими же per-peer лимитами. Только catch-all redirect в IFB остаётся `u32`. Вся `tc` topology после teardown передаётся одним fail-fast batch через `tc -batch -`, затем обязательно проверяется. Peer discovery берётся из WireGuard/AmneziaWG allowed IPs, а не из handshake state. Поэтому offline peer с `allowed ips: 10.8.1.1/32` тоже получает лимит заранее. diff --git a/config.example.toml b/config.example.toml index 05ec516..eae7438 100644 --- a/config.example.toml +++ b/config.example.toml @@ -48,8 +48,10 @@ default_qdisc = "fq" rmem_max = 67108864 wmem_max = 67108864 -netdev_backlog_enabled = true -netdev_max_backlog = 50000 +# По умолчанию harness не меняет receive backlog хоста. 10000 — только +# стартовое opt-in значение для хоста, где рост drops подтверждён метриками. +netdev_backlog_enabled = false +netdev_max_backlog = 10000 udp_rmem_min_enabled = true udp_rmem_min = 8192 diff --git a/docs/ACCEPTANCE_CHECKLIST.md b/docs/ACCEPTANCE_CHECKLIST.md index c129635..5b97108 100644 --- a/docs/ACCEPTANCE_CHECKLIST.md +++ b/docs/ACCEPTANCE_CHECKLIST.md @@ -38,7 +38,7 @@ - [ ] На AWG/WG-интерфейсе есть root class `1:1` с `traffic.root_rate`. - [ ] Для каждого peer есть class `1:` с `traffic.peer_limit`. - [ ] Для каждого peer есть `fq_codel` под `1:`. -- [ ] Для каждого peer есть egress filter `dst peer_ip -> 1:`. +- [ ] Для каждо peer есть egress filter `flower pref 10 dst_ip peer_ip/32 -> 1:`. - [ ] Есть ingress qdisc `ffff:` на AWG/WG-интерфейсе. - [ ] Есть ingress redirect через `mirred ... redirect dev ifb0`. - [ ] `ifb0` существует и находится в UP state. @@ -46,7 +46,8 @@ - [ ] На `ifb0` есть root class `2:1` с `traffic.root_rate`. - [ ] Для каждого peer есть class `2:` с `traffic.peer_limit`. - [ ] Для каждого peer есть `fq_codel` под `2:`. -- [ ] Для каждого peer есть ingress filter `src peer_ip -> 2:`. +- [ ] Для каждого peer есть ingress filter `flower pref 10 src_ip peer_ip/32 -> 2:`. +- [ ] Все peer-фильтры на каждом parent используют один `pref 10`; `u32` остаётся только у catch-all ingress redirect. - [ ] Class ids начинаются с `traffic.class_start_id`, default `10`. - [ ] Peer IP → class id mapping использует стабильный лексикографический контракт: IP сортируются как строки, не numeric IPv4. @@ -63,7 +64,10 @@ - [ ] Если изменить `rate/ceil` peer class, drift детектируется и topology восстанавливается. - [ ] Если добавить stale peer class/filter, drift детектируется и topology восстанавливается. - [ ] После удаления peer устаревшие class/filter исчезают через rebuild. +- [ ] После удаления последнего peer вся AWG2-owned traffic topology удаляется; повторный reconcile не выполняет teardown снова. - [ ] После добавления peer появляются новые class/filter. +- [ ] Legacy peer-фильтры `u32` детектируются как migration drift и заменяются на `flower` через rebuild. +- [ ] Ошибка в любой строке `tc -batch -` останавливает apply; после успеха проходит verify-after-apply. ## E. Самовосстановление в жизненном цикле @@ -80,6 +84,7 @@ - [ ] На старте daemon сбой `sysctl_tuning` или `swap` завершает процесс, а сбой traffic/firewall только логируется — сервис остаётся живым. - [ ] После обрыва потока `docker events` он перезапускается автоматически (проверяется, например, `systemctl restart docker`). - [ ] После восстановления потока `docker events` ошибка подсистемы снимается: в `events.log` появляется `docker_events=recovered`, а `last_action.state` не остаётся с этой ошибкой навсегда. +- [ ] Один startup/periodic/event-цикл использует один discovery `RuntimeTarget` для traffic и anti-torrent, а не повторяет `docker inspect` и interface discovery. - [ ] Успешный reconcile одной подсистемы не гасит незакрытую ошибку другой: при постоянно падающем `traffic` успешный `firewall` не очищает `last_error` и не пишет `firewall=recovered`. - [ ] Параллельный запуск `apply-limits` во время работы daemon не приводит к гонке: второй процесс ждёт блокировку до 10 секунд и затем сообщает `lock busy timeout`. - [ ] После `kill -9` блокировка освобождается ядром автоматически; оставшийся lock-файл не мешает следующему захвату `/run/awg2-harness/traffic.lock`. @@ -93,7 +98,9 @@ - [ ] `status` показывает недостающие и устаревшие правила peers: в `Traffic actual` — строка `reason` (например, `missing egress class`, `egress per-peer class set mismatch`, `duplicate egress class flowid mapping detected`), в `peer_table` — колонка `drift` по каждому peer. - [ ] `status` явно показывает частичное состояние и drift, а не общий признак «всё хорошо». - [ ] `status` показывает tc filter counters по peer или `filter statistics: unavailable`. +- [ ] Flower counters и legacy u32 counters в `status` разбираются одним parser-ом и привязываются к правильному peer IP. - [ ] Логи доступны через `journalctl -u awg2-harness`. +- [ ] Повтор одной и той же reconcile-ошибки не создаёт `error`-запись на каждом цикле; первая ошибка, изменение её текста и recovery видны как отдельные переходы. - [ ] Отсутствие обязательной host-утилиты (`tc`, `ip`, `docker`, `nsenter`) логируется явно с именем утилиты. - [ ] Последний reconcile result виден в status или journal. - [ ] `status` показывает последние события reconcile/firewall из `/run/awg2-harness/events.log`. @@ -143,6 +150,9 @@ - [ ] В обоих managed-файлах первая строка содержит marker `Managed by awg2-harness sysctl_tuning`. - [ ] Файл без marker по managed-пути не перезаписывается и не удаляется: команда завершается ошибкой. - [ ] В managed-файле отсутствует `udp_wmem_min`. +- [ ] В shipped/default config `netdev_backlog_enabled = false`, `netdev_max_backlog = 10000`, а `net.core.netdev_max_backlog` отсутствует в managed-файле. +- [ ] При смене `netdev_backlog_enabled = true` на `false` key удаляется из persistence; оператор учитывает, что текущее runtime-значение до reboot автоматически не сбрасывается. +- [ ] До/после ручного включения backlog сняты `/proc/net/softnet_stat`, CPU и latency под реальной нагрузкой. - [ ] `net.ipv4.tcp_congestion_control` и `net.core.default_qdisc` присутствуют в managed-файле только при активном BBR. - [ ] `swap` пишет `vm.swappiness` в `/etc/sysctl.d/90-awg2-harness-swap.conf`, а не в `99-awg2-harness.conf`. - [ ] При обновлении с прежней версии `/etc/sysctl.d/99-awg2-harness.conf` удаляется, если несёт маркер `# Managed by awg2-harness`. @@ -184,7 +194,7 @@ - [ ] При превышении порога новые flows дропаются, established flows ниже порога не ломаются. - [ ] После restart контейнера nftables rule пересоздаётся в новом netns. - [ ] При `anti_torrent.enabled = false` удаляется/отключается только AWG2-owned nftables state; Docker/Amnezia rules не flush-ятся. -- [ ] `status` показывает `nf_conntrack_count` / `nf_conntrack_max`, если sysctl доступны. +- [ ] `status` показывает `nf_conntrack_count`, `nf_conntrack_max` и `nf_conntrack_utilization`, если sysctl доступны. ## J. Откат diff --git a/docs/ANTI_TORRENT.md b/docs/ANTI_TORRENT.md index b99f0d4..cacdb70 100644 --- a/docs/ANTI_TORRENT.md +++ b/docs/ANTI_TORRENT.md @@ -97,7 +97,7 @@ sysctl net.netfilter.nf_conntrack_max nft list table inet awg2_harness ``` -Если `nf_conntrack_count` приближается к `nf_conntrack_max`, проблема уже системная: надо разбираться с timeout, abuse или capacity. Не надо автоматически увеличивать `nf_conntrack_max` из orchestrator без отдельного решения. +Если `nf_conntrack_count` приближается к `nf_conntrack_max`, проблема уже системная: надо разбираться с timeout, abuse или capacity. `awg2-harness status` показывает оба значения и вычисленный `nf_conntrack_utilization`, чтобы давление было видно без ручного деления. Не надо автоматически увеличивать `nf_conntrack_max` из orchestrator без отдельного решения. ## 7. Критерии приёмки @@ -106,6 +106,7 @@ nft list table inet awg2_harness - Правило ограничивает `ip saddr `. - Лимит применяется per source VPN IP, а не глобально на всю подсеть. - Drop counter растёт при превышении threshold. +- `status` показывает count/max/utilization conntrack, если эти sysctl доступны. - Нормальные сценарии с числом flows ниже threshold не затрагиваются. - При restart контейнера правило пересоздаётся в новом netns. - При `anti_torrent.enabled = false` удаляется или отключается только AWG2-owned nftables state. diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index b40e148..8164ce9 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -58,8 +58,8 @@ tcp_bbr = "auto" # off | auto | require default_qdisc = "fq" # none | fq | fq_codel rmem_max = 67108864 wmem_max = 67108864 -netdev_backlog_enabled = true -netdev_max_backlog = 50000 +netdev_backlog_enabled = false +netdev_max_backlog = 10000 udp_rmem_min_enabled = true udp_rmem_min = 8192 @@ -129,6 +129,10 @@ swappiness = 1 - `tcp_bbr = auto` не валит запуск при недоступном BBR (warning), `require` — валит; - для `sysctl_tuning`: `modprobe` обязателен только для `tcp_bbr = "require"`. +`netdev_backlog_enabled = false` — безопасный production default: harness не управляет `net.core.netdev_max_backlog`, пока оператор не включит это явно после измерений. Значение `netdev_max_backlog = 10000` используется только при `netdev_backlog_enabled = true`. + +При обновлении с конфига, где backlog раньше был включён, переключение флага в `false` удаляет key из managed-файла, но не перезаписывает уже изменённое runtime-значение. До следующей перезагрузки его можно вернуть вручную к baseline дистрибутива. Перед осознанным включением backlog снимите pre/post-метрики `/proc/net/softnet_stat` (в том числе dropped/time_squeeze) и CPU/latency под реальной нагрузкой. + Отдельно для всего пакета: - `traffic`-модуль требует `modprobe` для IFB при `traffic.enabled = true`. diff --git a/docs/IMPLEMENTATION_SPEC.md b/docs/IMPLEMENTATION_SPEC.md index b504d98..082fa97 100644 --- a/docs/IMPLEMENTATION_SPEC.md +++ b/docs/IMPLEMENTATION_SPEC.md @@ -243,8 +243,8 @@ struct PeerTrafficStatus { `discover_target()`: 1. Прочитать упорядоченный `amnezia.container_names`. -2. Найти первый запущенный контейнер с точным совпадением имени. -3. Получить container ID и PID. +2. Для каждого имени по порядку выполнить один `docker inspect`, который сразу возвращает running state, container ID и PID. +3. Выбрать первый запущенный контейнер с точным совпадением имени; отсутствующий или stopped candidate не меняет порядок fallback. 4. Построить `netns_path`. 5. Определить интерфейс в этом namespace. 6. Вернуть `RuntimeTarget`. @@ -313,7 +313,7 @@ tc class add dev parent 1: classid 1:1 htb rate ceil parent 1:1 classid 1: htb rate ceil tc qdisc add dev parent 1: fq_codel - tc filter add dev parent 1: protocol ip u32 match ip dst flowid 1: + tc filter add dev parent 1: protocol ip pref 10 flower dst_ip /32 classid 1: ``` Ingress через IFB: @@ -329,9 +329,11 @@ tc class add dev parent 2: classid 2:1 htb rate ceil parent 2:1 classid 2: htb rate ceil tc qdisc add dev parent 2: fq_codel - tc filter add dev parent 2: protocol ip u32 match ip src flowid 2: + tc filter add dev parent 2: protocol ip pref 10 flower src_ip /32 classid 2: ``` +Все peer-фильтры на каждом parent используют один `pref 10`: для `flower` это даёт один hash table на priority вместо множества priority structures. Маска всегда `/32`, поэтому один priority не смешивает несовместимые masks. Catch-all redirect в IFB остаётся `u32`, поскольку это одно правило, а не per-peer lookup. + ## 9. Алгоритм reconcile Traffic и firewall — два независимых reconcile с отдельными блокировками. Они не объединены в одну транзакцию: сбой одного не должен блокировать другой. @@ -343,10 +345,12 @@ reconcile_traffic(): target = discover_target() peers = discover_peers(config, target) если peers пуст: - залогировать "no peers found, skip tc apply" и выйти + проверить наличие AWG2-owned topology + если она осталась от прежних peers: teardown_owned_topology(target) + иначе выйти с debug-диагностикой drift = inspect_and_diff(target, peers, config) если drift отсутствует: - залогировать "traffic rules up-to-date, skip apply" и выйти + выйти с debug-диагностикой teardown_owned_topology(target) apply_full_topology(target, peers, config) # внутри — verify-after-apply ``` @@ -365,12 +369,15 @@ apply_firewall(): Требования к устойчивости парсеров: - `tc filter show` разбирается блоками (конечный автомат), а не наивным сканированием токенов по несвязанным строкам; +- единый parser понимает production `flower` и legacy `u32`, включая counters; точный healthy-контракт требует `flower`, `pref 10`, правильное direction-field и `/32`, а legacy `u32` peer rules считаются migration drift; +- нераспознанный или malformed detailed filter block не игнорируется, а делает инспекцию drifted; - отображение `peer_ip -> class_id` проверяется для обеих сторон: egress (`1:`) и ingress (`2:`); - проверка обязана падать, если классы и фильтры существуют, но отображение отличается от детерминированного присвоения. Политика apply/verify: -- все шаги apply — явные вызовы команд; +- `modprobe ifb`, создание IFB и его перевод в UP выполняются строго и до `tc` apply; +- после teardown вся `tc` topology подаётся одним fail-fast сценарием в stdin команды `tc -batch -`; `-force` не используется, поэтому первая ошибка останавливает batch и сохраняет номер строки в stderr; - ожидаемые конфликты идемпотентности допускаются выборочно, через отдельную обёртку, и по закрытому списку признаков: `file exists`, `eexist`, `object already exists`, `exclusive flag on`; - ошибки «объект не найден» (`no such`, `not found`, `cannot find`, `does not exist`, `handle of zero`) допускаются только на шагах удаления — в teardown и при отключении владеемых таблиц; - любая другая ошибка apply фатальна; @@ -437,9 +444,9 @@ table inet awg2_harness { - counter `antitorrent_dropped_v4`; - meter/set state, создаваемый правилом. -Status обязан показывать наличие таблицы/chain/правила, настроенный порог и drop counter. Также он показывает давление на conntrack хоста через `net.netfilter.nf_conntrack_count` и `net.netfilter.nf_conntrack_max`, когда они доступны. +Status обязан показывать наличие таблицы/chain/правила, настроенный порог и drop counter. Также он один раз читает `net.netfilter.nf_conntrack_count` и `net.netfilter.nf_conntrack_max` и, когда оба значения доступны, печатает `nf_conntrack_utilization` в процентах. -При `anti_torrent.enabled = false` демон удаляет только владеемую anti-torrent таблицу. Глобальный ruleset nftables не сбрасывается никогда. Если контейнер в этот момент отсутствует, удаление в его namespace просто пропускается — это не ошибка. +При `anti_torrent.enabled = false` демон удаляет только владеемую anti-torrent таблицу. Глобальный ruleset nftables не сбрасывается никогда. Если контейнер в этот момент именно не найден, удаление в его namespace пропускается — это не ошибка. Ошибка Docker API/daemon/permissions не маскируется под «контейнер не найден» и возвращается оператору. ## 12. Инспекция фактического состояния @@ -456,6 +463,8 @@ Status обязан показывать наличие таблицы/chain/п - классы на ``; - фильтры под parent `2:` на ``. +Один вывод `tc qdisc show dev ` используется одновременно для root, ingress и per-peer `fq_codel`; один вывод `tc class show dev ` — для root и per-peer classes. Инспекция не делает отдельный child process на каждый parent. + Дополнительно детектируются дубли: один и тот же peer IP в нескольких фильтрах и один и тот же class id у нескольких IP. Оба случая считаются drift. `status` показывает компактную сводку и таблицу по peers. @@ -469,6 +478,8 @@ Status обязан показывать наличие таблицы/chain/п Таймер обязателен. События могут быть пропущены при рестарте демона, рестарте Docker или транзиентных ошибках. +Внутри одного startup/periodic/event-цикла traffic и anti-torrent получают один и тот же результат discovery `RuntimeTarget`; повторные `docker inspect`, PID/interface discovery и transport construction в этом же цикле не выполняются. Наличие host `nsenter` кэшируется на время жизни процесса. + Проверка готовности не должна хардкодить `awg0`. Она вызывает общий детектор интерфейса с интервалом 2 секунды до истечения `traffic.container_ready_timeout_secs`. Порядок на старте демона и разница в критичности ошибок: @@ -487,7 +498,7 @@ Status обязан показывать наличие таблицы/chain/п Ошибки отслеживаются по подсистемам (`traffic`, `firewall`, `swap`, `sysctl_tuning`, `docker_events`), а не одним общим полем. Общее поле в `last_action.state` показывает ту ошибку, которая ещё не устранена: успешный reconcile одной подсистемы не гасит незакрытую ошибку другой и не объявляет восстановление того, что не падало. Поток `docker events` при успешном переподключении тоже отмечается как восстановленный — иначе один рестарт Docker оставлял бы ошибку в `status` навсегда при исправно работающем новом потоке. -В `events.log` пишутся только **переходы состояния**: первое появление ошибки, изменение текста ошибки и восстановление после неё. Рутинный успех не пишется. Раньше каждый успешный цикл добавлял строку — при интервале 15 секунд это порядка 17 тысяч записей в сутки, каждая с полным чтением, дополнением и перезаписью файла, и на их фоне терялись записи, ради которых журнал и читают. Полная поцикловая картина остаётся в journald через tracing. +В `events.log` и journald на уровне `info`/`error` пишутся только **переходы состояния**: первое появление ошибки, изменение её текста и восстановление. Рутинный healthy-цикл и повтор той же ошибки остаются на `debug`. Это исключае повторяющийся journal-шум на каждом 15-секундном reconcile, но сохраняет детали для отладки при включённом debug-уровне. ## 14. Блокировки @@ -552,7 +563,7 @@ Swap: desired_active, exclusive, fstab_exclusive, fstab_swap_entries, swappiness_current, swappiness_persisted, drift[, drift_reason] Conntrack: - nf_conntrack_count, nf_conntrack_max + nf_conntrack_count, nf_conntrack_max, nf_conntrack_utilization Last actions: содержимое /run/awg2-harness/last_action.state Recent events: diff --git a/docs/INSTALLATION.md b/docs/INSTALLATION.md index 7eb06fc..1684931 100644 --- a/docs/INSTALLATION.md +++ b/docs/INSTALLATION.md @@ -292,7 +292,7 @@ Swap: drift: no ``` -Если peers ещё нет, `apply-limits` напечатает `no peers found, skip tc apply`. Для пустого сервера это здоровое состояние. +Если peers ещё нет, это здоровое desired state. `apply-limits` успешно завершится и удалит оставшуюся AWG2-owned traffic topology, если до этого peers были. При `swap.enabled = true` AWG2 Harness намеренно делает swap эксклюзивным: один управляемый swapfile, одна swap-запись в `/etc/fstab`, заданный `swappiness`. diff --git a/traffic_core/src/config.rs b/traffic_core/src/config.rs index 2c545f8..6053684 100644 --- a/traffic_core/src/config.rs +++ b/traffic_core/src/config.rs @@ -77,8 +77,8 @@ impl Default for SysctlTuningConfig { default_qdisc: "fq".to_string(), rmem_max: 67_108_864, wmem_max: 67_108_864, - netdev_backlog_enabled: true, - netdev_max_backlog: 50_000, + netdev_backlog_enabled: false, + netdev_max_backlog: 10_000, udp_rmem_min_enabled: true, udp_rmem_min: 8192, } @@ -830,6 +830,8 @@ mod tests { fn sysctl_tuning_default_is_valid() { let cfg = base_config(); assert!(cfg.validate_config().is_ok()); + assert!(!cfg.sysctl_tuning.netdev_backlog_enabled); + assert_eq!(cfg.sysctl_tuning.netdev_max_backlog, 10_000); } #[test] diff --git a/traffic_core/src/daemon.rs b/traffic_core/src/daemon.rs index aaef88f..be1a68e 100644 --- a/traffic_core/src/daemon.rs +++ b/traffic_core/src/daemon.rs @@ -7,9 +7,7 @@ use std::{ time::{Duration, Instant}, }; -use crate::{ - config::Config, docker, firewall, iface, netns::build_transport, swap, sysctl_tuning, traffic, -}; +use crate::{config::Config, docker, firewall, swap, sysctl_tuning, traffic}; use tracing::{error, info}; #[derive(Debug, Clone, Default)] @@ -67,14 +65,15 @@ pub fn run(cfg: Config) -> anyhow::Result<()> { append_event("startup_swap ok"); } - if let Err(e) = traffic::reconcile_traffic(&cfg) { + let startup_target = discover_runtime_target_if_needed(&cfg); + if let Err(e) = reconcile_traffic_for_cycle(&cfg, startup_target.as_ref()) { set_err(&state, "traffic", format!("startup traffic reconcile: {e}")); append_event(&format!("startup_reconcile traffic=error error=\"{e}\"")); } else { set_ok(&state, "traffic", "startup traffic reconcile ok"); append_event("startup_reconcile traffic=ok"); } - if let Err(e) = firewall::apply_firewall(&cfg) { + if let Err(e) = reconcile_firewall_for_cycle(&cfg, startup_target.as_ref()) { set_err( &state, "firewall", @@ -119,17 +118,18 @@ pub fn run(cfg: Config) -> anyhow::Result<()> { ); } + let target = discover_runtime_target_if_needed(&cfg); report( &state, "traffic", "periodic traffic reconcile", - traffic::reconcile_traffic(&cfg), + reconcile_traffic_for_cycle(&cfg, target.as_ref()), ); report( &state, "firewall", "periodic firewall reconcile", - firewall::apply_firewall(&cfg), + reconcile_firewall_for_cycle(&cfg, target.as_ref()), ); thread::sleep(Duration::from_secs( @@ -138,6 +138,44 @@ pub fn run(cfg: Config) -> anyhow::Result<()> { } } +fn discover_runtime_target_if_needed( + cfg: &Config, +) -> Option> { + (cfg.traffic.enabled || cfg.anti_torrent.apply_in_amnezia_netns) + .then(|| docker::discover_target(cfg)) +} + +fn reconcile_traffic_for_cycle( + cfg: &Config, + target: Option<&anyhow::Result>, +) -> anyhow::Result<()> { + if !cfg.traffic.enabled { + return Ok(()); + } + match target { + Some(Ok(target)) => traffic::reconcile_traffic_with_target(cfg, target), + Some(Err(e)) => Err(anyhow::anyhow!("runtime target discovery failed: {e:#}")), + None => Err(anyhow::anyhow!( + "runtime target was not discovered for enabled traffic" + )), + } +} + +fn reconcile_firewall_for_cycle( + cfg: &Config, + target: Option<&anyhow::Result>, +) -> anyhow::Result<()> { + if !cfg.anti_torrent.apply_in_amnezia_netns { + return firewall::apply_firewall(cfg); + } + match target { + Some(target) => firewall::apply_firewall_with_target_result(cfg, target.as_ref()), + None => Err(anyhow::anyhow!( + "runtime target was not discovered for namespace firewall" + )), + } +} + /// Records the outcome of one reconcile, writing an event only when something changed. /// /// A routine success used to append a line every cycle: at a 15 second interval that is @@ -155,14 +193,17 @@ fn report( Ok(()) => { let recovered = set_ok(state, subsystem, &format!("{what} ok")); if recovered { + info!("{subsystem} reconcile recovered"); append_event(&format!("{subsystem}=recovered")); } } Err(e) => { - error!(error = %e, "{subsystem} reconcile error"); let changed = set_err(state, subsystem, format!("{what}: {e}")); if changed { + error!(error = %e, "{subsystem} reconcile error"); append_event(&format!("{subsystem}=error error=\"{e}\"")); + } else { + tracing::debug!(error = %e, "{subsystem} reconcile error unchanged"); } } } @@ -217,44 +258,58 @@ fn consume_docker_events( for line in reader.lines() { let name = line?.trim().to_string(); if cfg.amnezia.container_names.iter().any(|n| n == &name) - && wait_for_interface(cfg, &name, cfg.traffic.container_ready_timeout_secs) + && let Some(target) = + wait_for_target(cfg, &name, cfg.traffic.container_ready_timeout_secs) { append_event(&format!("docker_event container={name} action=start")); - if let Err(e) = traffic::reconcile_traffic(cfg) { - set_err(state, "traffic", format!("event traffic reconcile: {e}")); - append_event(&format!("event_reconcile traffic=error error=\"{e}\"")); + if let Err(e) = traffic::reconcile_traffic_with_target(cfg, &target) { + if set_err(state, "traffic", format!("event traffic reconcile: {e}")) { + error!(error = %e, "traffic event reconcile error"); + append_event(&format!("event_reconcile traffic=error error=\"{e}\"")); + } } else { - set_ok(state, "traffic", "event traffic reconcile ok"); - append_event("event_reconcile traffic=ok"); + if set_ok(state, "traffic", "event traffic reconcile ok") { + info!("traffic event reconcile recovered"); + append_event("event_reconcile traffic=recovered"); + } } - if let Err(e) = firewall::apply_firewall(cfg) { - set_err(state, "firewall", format!("event firewall reconcile: {e}")); - append_event(&format!("event_reconcile firewall=error error=\"{e}\"")); + if let Err(e) = firewall::apply_firewall_with_target_result(cfg, Ok(&target)) { + if set_err(state, "firewall", format!("event firewall reconcile: {e}")) { + error!(error = %e, "firewall event reconcile error"); + append_event(&format!("event_reconcile firewall=error error=\"{e}\"")); + } } else { - set_ok(state, "firewall", "event firewall reconcile ok"); - append_event("event_reconcile firewall=ok"); + if set_ok(state, "firewall", "event firewall reconcile ok") { + info!("firewall event reconcile recovered"); + append_event("event_reconcile firewall=recovered"); + } } } } Ok(()) } -fn wait_for_interface(cfg: &Config, container_name: &str, timeout_secs: u64) -> bool { +fn wait_for_target( + cfg: &Config, + container_name: &str, + timeout_secs: u64, +) -> Option { let start = Instant::now(); while start.elapsed() < Duration::from_secs(timeout_secs.max(1)) { - if let Ok(target) = docker::discover_target(cfg) - && target.container_name == container_name - { - let transport = build_transport(target.container_pid, &target.container_name); - if iface::detect_interface_with_transport(cfg, transport.as_ref()).is_ok() { - return true; + if let Ok(target) = docker::discover_target(cfg) { + if target.container_name == container_name { + return Some(target); } + // A higher-priority configured container is already the active target. A start + // event for a lower-priority candidate cannot change desired state, so waiting + // for the full readiness timeout would only stall the event consumer. + return None; } thread::sleep(Duration::from_secs(2)); } - false + None } impl LastActionState { diff --git a/traffic_core/src/docker.rs b/traffic_core/src/docker.rs index 1e398ab..856fff9 100644 --- a/traffic_core/src/docker.rs +++ b/traffic_core/src/docker.rs @@ -60,18 +60,10 @@ impl PortPathDiagnostics { } pub fn discover_target(cfg: &Config) -> anyhow::Result { - let container_name = find_running_container(&cfg.amnezia.container_names)?; - let container_id = run_capture(&["docker", "inspect", "--format", "{{.Id}}", &container_name])? - .trim() - .to_string(); - let pid_str = run_capture(&[ - "docker", - "inspect", - "--format", - "{{.State.Pid}}", - &container_name, - ])?; - let container_pid = pid_str.trim().parse::()?; + let snapshot = discover_container_snapshot_with(&cfg.amnezia.container_names, run_capture)?; + let container_name = snapshot.name; + let container_id = snapshot.id; + let container_pid = snapshot.pid; let netns_path = format!("/proc/{container_pid}/ns/net"); let interface_name = crate::iface::detect_interface(cfg, container_pid, &container_name)?; @@ -84,16 +76,69 @@ pub fn discover_target(cfg: &Config) -> anyhow::Result { }) } -fn find_running_container(names: &[String]) -> anyhow::Result { - let ps = run_capture(&["docker", "ps", "--format", "{{.Names}}"])?; +#[derive(Debug, Clone, PartialEq, Eq)] +struct ContainerSnapshot { + name: String, + id: String, + pid: u32, +} + +fn discover_container_snapshot_with( + names: &[String], + mut run: F, +) -> anyhow::Result +where + F: FnMut(&[&str]) -> anyhow::Result, +{ for name in names { - if ps.lines().any(|x| x.trim() == name) { - return Ok(name.clone()); + let raw = match run(&[ + "docker", + "inspect", + "--format", + "{{.State.Running}}\t{{.Id}}\t{{.State.Pid}}", + name, + ]) { + Ok(raw) => raw, + Err(e) if is_missing_container_error(&e.to_string()) => continue, + Err(e) => return Err(e.context(format!("cannot inspect container candidate {name}"))), + }; + let fields: Vec<&str> = raw.trim().split('\t').collect(); + if fields.len() != 3 { + anyhow::bail!( + "docker inspect returned malformed target snapshot for {name}: expected 3 fields" + ); } + let running = fields[0] + .parse::() + .map_err(|_| anyhow::anyhow!("invalid running state for container {name}"))?; + if !running { + continue; + } + let id = fields[1].trim(); + if id.is_empty() { + anyhow::bail!("docker inspect returned an empty container ID for {name}"); + } + let pid = fields[2] + .trim() + .parse::() + .map_err(|_| anyhow::anyhow!("invalid PID for running container {name}"))?; + if pid == 0 { + anyhow::bail!("docker inspect returned PID 0 for running container {name}"); + } + return Ok(ContainerSnapshot { + name: name.clone(), + id: id.to_string(), + pid, + }); } Err(AppError::NotFound("amnezia container not running".to_string()).into()) } +fn is_missing_container_error(message: &str) -> bool { + let lower = message.to_ascii_lowercase(); + lower.contains("no such object") || lower.contains("no such container") +} + pub fn run_capture(args: &[&str]) -> anyhow::Result { let (program, rest) = args .split_first() @@ -346,6 +391,61 @@ fn normalize_nft_forward_text(raw: &str) -> String { #[cfg(test)] mod tests { use super::*; + use std::cell::RefCell; + + #[test] + fn target_discovery_uses_one_inspect_per_candidate_in_config_order() { + let calls = RefCell::new(Vec::::new()); + let names = vec!["primary".to_string(), "legacy".to_string()]; + let got = discover_container_snapshot_with(&names, |args| { + let name = args.last().expect("candidate").to_string(); + calls.borrow_mut().push(name.clone()); + match name.as_str() { + "primary" => Ok("false\tstopped-id\t0\n".into()), + "legacy" => Ok("true\trunning-id\t4242\n".into()), + _ => unreachable!(), + } + }) + .expect("running fallback"); + + assert_eq!(calls.into_inner(), ["primary", "legacy"]); + assert_eq!( + got, + ContainerSnapshot { + name: "legacy".into(), + id: "running-id".into(), + pid: 4242, + } + ); + } + + #[test] + fn target_discovery_skips_only_explicitly_missing_candidates() { + let names = vec!["missing".to_string(), "running".to_string()]; + let got = discover_container_snapshot_with(&names, |args| match args.last().copied() { + Some("missing") => anyhow::bail!("Error: No such object: missing"), + Some("running") => Ok("true\tid\t77\n".into()), + _ => unreachable!(), + }) + .expect("running candidate"); + assert_eq!(got.name, "running"); + } + + #[test] + fn target_discovery_does_not_mask_docker_daemon_errors() { + let names = vec!["primary".to_string(), "legacy".to_string()]; + let calls = RefCell::new(0usize); + let err = discover_container_snapshot_with(&names, |_args| { + *calls.borrow_mut() += 1; + anyhow::bail!("permission denied connecting to docker socket") + }) + .expect_err("daemon failure must propagate"); + assert_eq!(*calls.borrow(), 1); + assert!( + err.to_string() + .contains("cannot inspect container candidate") + ); + } #[test] fn parse_userland_proxy_from_docker_info_works() { diff --git a/traffic_core/src/firewall/antitorrent.rs b/traffic_core/src/firewall/antitorrent.rs index 67b01df..36c0d1c 100644 --- a/traffic_core/src/firewall/antitorrent.rs +++ b/traffic_core/src/firewall/antitorrent.rs @@ -13,7 +13,7 @@ use serde_json::Value; use crate::{config::Config, docker, error::AppError, netns::NsTransport}; use super::{ - ANTITORRENT_TABLE, firewall_transport, + ANTITORRENT_TABLE, firewall_transport, firewall_transport_for_target, nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_nft_batch, table_exists}, }; @@ -68,6 +68,64 @@ pub(super) fn apply_antitorrent_firewall(cfg: &Config) -> anyhow::Result<()> { Ok(()) } +pub(super) fn apply_antitorrent_firewall_with_target_result( + cfg: &Config, + target: Result<&docker::RuntimeTarget, &anyhow::Error>, +) -> anyhow::Result<()> { + if !cfg.anti_torrent.apply_in_amnezia_netns { + return apply_antitorrent_firewall(cfg); + } + + match target { + Ok(target) => apply_antitorrent_with_transport( + cfg, + firewall_transport_for_target(cfg, target).as_ref(), + ), + Err(e) if !cfg.anti_torrent.enabled && is_container_absent(e) => Ok(()), + Err(e) if !cfg.anti_torrent.enabled => Err(anyhow::anyhow!( + "cannot reach the Amnezia container to remove the anti-torrent table: {e:#}" + )), + Err(e) => Err(anyhow::anyhow!( + "cannot reach the Amnezia container to reconcile anti-torrent: {e:#}" + )), + } +} + +fn apply_antitorrent_with_transport( + cfg: &Config, + transport: &dyn NsTransport, +) -> anyhow::Result<()> { + if !cfg.anti_torrent.enabled { + run_allow_notfound( + transport, + &["nft", "delete", "table", "inet", ANTITORRENT_TABLE], + )?; + return Ok(()); + } + + let actual = inspect_firewall(transport, cfg)?; + if !firewall_has_drift(&actual) { + return Ok(()); + } + let present = table_exists(transport, ANTITORRENT_TABLE)?; + run_nft_batch( + transport, + &build_antitorrent_batch_script( + &cfg.anti_torrent.peer_ipv4_cidr, + cfg.anti_torrent.max_connections_per_peer, + present, + ), + )?; + let verify = inspect_firewall(transport, cfg)?; + if firewall_has_drift(&verify) { + anyhow::bail!( + "anti-torrent firewall verify-after-apply failed: {}", + firewall_drift_reason(&verify) + ); + } + Ok(()) +} + fn apply_antitorrent_disabled(cfg: &Config) -> anyhow::Result<()> { if cfg.anti_torrent.apply_in_amnezia_netns { match docker::discover_target(cfg) { @@ -162,7 +220,17 @@ pub fn antitorrent_drift_reason(cfg: &Config) -> anyhow::Result> return Ok(None); } let transport = firewall_transport(cfg)?; - let actual = inspect_firewall(transport.as_ref(), cfg)?; + antitorrent_drift_reason_with_transport(cfg, transport.as_ref()) +} + +pub(super) fn antitorrent_drift_reason_with_transport( + cfg: &Config, + transport: &dyn NsTransport, +) -> anyhow::Result> { + if !cfg.anti_torrent.enabled { + return Ok(None); + } + let actual = inspect_firewall(transport, cfg)?; if firewall_has_drift(&actual) { return Ok(Some(firewall_drift_reason(&actual))); } @@ -175,6 +243,16 @@ pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result anyhow::Result> { + if !cfg.anti_torrent.enabled { + return Ok(None); + } let json = transport.run(&[ "nft", "-j", @@ -449,4 +527,31 @@ table inet awg2_harness { "#; assert!(firewall_rule_contract_matches(table, "10.8.1.0/24", 100)); } + + #[test] + fn disabled_namespace_cleanup_allows_only_an_absent_container() { + let mut cfg = crate::firewall::tests::test_config(); + cfg.anti_torrent.enabled = false; + cfg.anti_torrent.apply_in_amnezia_netns = true; + let missing: anyhow::Error = + AppError::NotFound("amnezia container not running".into()).into(); + + apply_antitorrent_firewall_with_target_result(&cfg, Err(&missing)) + .expect("there is no namespace to clean when the container is absent"); + } + + #[test] + fn disabled_namespace_cleanup_does_not_mask_docker_failures() { + let mut cfg = crate::firewall::tests::test_config(); + cfg.anti_torrent.enabled = false; + cfg.anti_torrent.apply_in_amnezia_netns = true; + let unavailable = anyhow::anyhow!("permission denied on Docker socket"); + + let err = apply_antitorrent_firewall_with_target_result(&cfg, Err(&unavailable)) + .expect_err("Docker failures must remain visible"); + assert!( + err.to_string() + .contains("cannot reach the Amnezia container") + ); + } } diff --git a/traffic_core/src/firewall/mod.rs b/traffic_core/src/firewall/mod.rs index e3073d6..14980c4 100644 --- a/traffic_core/src/firewall/mod.rs +++ b/traffic_core/src/firewall/mod.rs @@ -23,6 +23,20 @@ use crate::{ pub use antitorrent::{antitorrent_drift_reason, antitorrent_drop_counter}; pub use host::host_firewall_drift_reason; +pub(crate) fn antitorrent_drift_reason_with_transport( + cfg: &Config, + transport: &dyn NsTransport, +) -> anyhow::Result> { + antitorrent::antitorrent_drift_reason_with_transport(cfg, transport) +} + +pub(crate) fn antitorrent_drop_counter_with_transport( + cfg: &Config, + transport: &dyn NsTransport, +) -> anyhow::Result> { + antitorrent::antitorrent_drop_counter_with_transport(cfg, transport) +} + const ANTITORRENT_TABLE: &str = "awg2_harness"; const HOST_FW_TABLE: &str = "awg2_host_firewall"; @@ -35,6 +49,17 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { Ok(()) } +pub(crate) fn apply_firewall_with_target_result( + cfg: &Config, + target: Result<&docker::RuntimeTarget, &anyhow::Error>, +) -> anyhow::Result<()> { + let _lock = + FileLock::acquire_timeout("/run/awg2-harness/firewall.lock", Duration::from_secs(10))?; + + host::apply_host_firewall(cfg)?; + antitorrent::apply_antitorrent_firewall_with_target_result(cfg, target) +} + /// Anti-torrent rules go wherever the peer source addresses are actually visible. fn firewall_transport(cfg: &Config) -> anyhow::Result> { if cfg.anti_torrent.apply_in_amnezia_netns { @@ -48,6 +73,17 @@ fn firewall_transport(cfg: &Config) -> anyhow::Result> { } } +fn firewall_transport_for_target( + cfg: &Config, + target: &docker::RuntimeTarget, +) -> Box { + if cfg.anti_torrent.apply_in_amnezia_netns { + build_transport(target.container_pid, &target.container_name) + } else { + build_host_transport() + } +} + #[cfg(test)] pub(crate) mod tests { use crate::config::{ diff --git a/traffic_core/src/iface.rs b/traffic_core/src/iface.rs index c8054cd..23ea1ab 100644 --- a/traffic_core/src/iface.rs +++ b/traffic_core/src/iface.rs @@ -1,17 +1,10 @@ use crate::{ config::Config, - netns::{DockerExecTransport, NsTransport, NsenterTransport, nsenter_available}, + netns::{NsTransport, build_transport}, }; pub fn detect_interface(cfg: &Config, pid: u32, container: &str) -> anyhow::Result { - let boxed: Box = if nsenter_available() { - Box::new(NsenterTransport { pid }) - } else { - Box::new(DockerExecTransport { - container: container.to_string(), - }) - }; - + let boxed = build_transport(pid, container); detect_interface_with_transport(cfg, boxed.as_ref()) } diff --git a/traffic_core/src/netns.rs b/traffic_core/src/netns.rs index 25ff5e9..ffa9f72 100644 --- a/traffic_core/src/netns.rs +++ b/traffic_core/src/netns.rs @@ -1,6 +1,7 @@ use std::{ io::Write, process::{Command, Stdio}, + sync::OnceLock, }; pub trait NsTransport: Send + Sync { @@ -129,7 +130,8 @@ impl NsTransport for HostTransport { } pub fn nsenter_available() -> bool { - Command::new("nsenter").arg("--help").output().is_ok() + static AVAILABLE: OnceLock = OnceLock::new(); + *AVAILABLE.get_or_init(|| Command::new("nsenter").arg("--help").output().is_ok()) } pub fn build_transport(pid: u32, container: &str) -> Box { diff --git a/traffic_core/src/status.rs b/traffic_core/src/status.rs index 9e9a9b0..8338a52 100644 --- a/traffic_core/src/status.rs +++ b/traffic_core/src/status.rs @@ -56,6 +56,7 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { print_degraded_sections(cfg); return Ok(()); }; + let transport = build_transport(target.container_pid, &target.container_name); let docker_path_diag = docker::collect_port_path_diagnostics( &target.container_name, @@ -118,7 +119,17 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { print_traffic_desired_section(cfg); println!("Traffic actual:"); - match traffic::inspect_drift_summary(cfg) { + let traffic_drift = if cfg.traffic.enabled { + match &peers { + Ok(peers) => { + traffic::inspect_drift_summary_with_state(cfg, &target, transport.as_ref(), peers) + } + Err(e) => Err(anyhow::anyhow!("peer discovery failed: {e}")), + } + } else { + Ok(None) + }; + match traffic_drift { Ok(None) => println!(" drift: no"), Ok(Some(reason)) => { println!(" drift: yes"); @@ -128,7 +139,6 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { } println!(" topology:"); - let transport = build_transport(target.container_pid, &target.container_name); let egress_qdisc = transport .run(&["tc", "qdisc", "show", "dev", &target.interface_name]) .unwrap_or_default(); @@ -141,7 +151,17 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { println!(" ifb_htb_present: {}", ifb_qdisc.contains("htb")); println!(" peer_table:"); - match traffic::peer_status_table(cfg) { + let peer_table = if cfg.traffic.enabled { + match &peers { + Ok(peers) => { + traffic::peer_status_table_with_state(cfg, &target, transport.as_ref(), peers) + } + Err(e) => Err(anyhow::anyhow!("peer discovery failed: {e}")), + } + } else { + Ok(None) + }; + match peer_table { Ok(Some(rows)) if rows.is_empty() => println!(" no peers discovered"), Ok(Some(rows)) => { for r in rows { @@ -168,8 +188,7 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { match collect_peer_filter_hits( &target.interface_name, &cfg.traffic.ifb_name, - target.container_pid, - &target.container_name, + transport.as_ref(), ) { Ok(map) if map.is_empty() => println!(" filter statistics: unavailable"), Ok(map) => { @@ -182,7 +201,7 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { print_host_firewall_section(cfg); print_docker_amnezia_port_path(&docker_path_diag); - print_antitorrent_section(cfg); + print_antitorrent_section_with_transport(cfg, Some(transport.as_ref())); print_swap_status(cfg); print_conntrack_section(); print_last_actions_section(); @@ -267,6 +286,13 @@ fn print_host_firewall_section(cfg: &Config) { } fn print_antitorrent_section(cfg: &Config) { + print_antitorrent_section_with_transport(cfg, None); +} + +fn print_antitorrent_section_with_transport( + cfg: &Config, + container_transport: Option<&dyn crate::netns::NsTransport>, +) { println!("Anti-torrent:"); println!(" enabled: {}", cfg.anti_torrent.enabled); println!( @@ -276,7 +302,15 @@ fn print_antitorrent_section(cfg: &Config) { if !cfg.anti_torrent.enabled { return; } - match firewall::antitorrent_drift_reason(cfg) { + let drift = if cfg.anti_torrent.apply_in_amnezia_netns { + match container_transport { + Some(transport) => firewall::antitorrent_drift_reason_with_transport(cfg, transport), + None => firewall::antitorrent_drift_reason(cfg), + } + } else { + firewall::antitorrent_drift_reason(cfg) + }; + match drift { Ok(None) => println!(" drift: no"), Ok(Some(reason)) => { println!(" drift: yes"); @@ -284,7 +318,15 @@ fn print_antitorrent_section(cfg: &Config) { } Err(e) => println!(" drift: unknown ({e})"), } - match firewall::antitorrent_drop_counter(cfg) { + let counter = if cfg.anti_torrent.apply_in_amnezia_netns { + match container_transport { + Some(transport) => firewall::antitorrent_drop_counter_with_transport(cfg, transport), + None => firewall::antitorrent_drop_counter(cfg), + } + } else { + firewall::antitorrent_drop_counter(cfg) + }; + match counter { Ok(Some((packets, bytes))) => { println!(" dropped_packets: {}", packets); println!(" dropped_bytes: {}", bytes); @@ -296,16 +338,25 @@ fn print_antitorrent_section(cfg: &Config) { fn print_conntrack_section() { println!("Conntrack:"); + let count = read_sysctl_or_na("net.netfilter.nf_conntrack_count"); + let max = read_sysctl_or_na("net.netfilter.nf_conntrack_max"); + println!(" nf_conntrack_count: {count}"); + println!(" nf_conntrack_max: {max}"); println!( - " nf_conntrack_count: {}", - read_sysctl_or_na("net.netfilter.nf_conntrack_count") - ); - println!( - " nf_conntrack_max: {}", - read_sysctl_or_na("net.netfilter.nf_conntrack_max") + " nf_conntrack_utilization: {}", + conntrack_utilization(&count, &max).unwrap_or_else(|| "n/a".to_string()) ); } +fn conntrack_utilization(count: &str, max: &str) -> Option { + let count = count.trim().parse::().ok()?; + let max = max.trim().parse::().ok()?; + if max == 0 { + return None; + } + Some(format!("{:.1}%", count as f64 * 100.0 / max as f64)) +} + fn read_sysctl_or_na(key: &str) -> String { Command::new("sysctl") .args(["-n", key]) @@ -494,14 +545,12 @@ fn format_bytes(v: u64) -> String { fn collect_peer_filter_hits( iface: &str, ifb: &str, - pid: u32, - container_name: &str, + transport: &dyn crate::netns::NsTransport, ) -> anyhow::Result> { - let transport = build_transport(pid, container_name); let egress = transport.run(&["tc", "-s", "filter", "show", "dev", iface, "parent", "1:"])?; let ingress = transport.run(&["tc", "-s", "filter", "show", "dev", ifb, "parent", "2:"])?; - let egress_hits = parse_tc_filter_hits(&egress); - let ingress_hits = parse_tc_filter_hits(&ingress); + let egress_hits = traffic::parse_peer_filter_hits(&egress); + let ingress_hits = traffic::parse_peer_filter_hits(&ingress); let mut out: BTreeMap = BTreeMap::new(); for (ip, c) in egress_hits { @@ -513,74 +562,19 @@ fn collect_peer_filter_hits( Ok(out) } -fn parse_tc_filter_hits(raw: &str) -> BTreeMap { - let mut out = BTreeMap::new(); - let mut current_ip: Option = None; - for line in raw.lines() { - let t = line.trim(); - - if t.contains("match ") { - current_ip = extract_tc_match_ip(t); - } - - if t.starts_with("Sent ") { - if let Some(ip) = current_ip.clone() { - let hits = parse_sent_packets(t).unwrap_or(0); - out.insert(ip, hits); - } - } else if (t.contains(" success ") || t.starts_with("success ")) - && let Some(ip) = current_ip.clone() - { - let hits = parse_success_packets(t).unwrap_or(0); - out.insert(ip, hits); - } - } - out -} - -fn parse_sent_packets(line: &str) -> Option { - let toks: Vec<&str> = line.split_whitespace().collect(); - if toks.len() >= 5 && toks[0] == "Sent" && toks[2] == "bytes" && toks[4] == "pkt" { - return toks[3].parse::().ok(); - } - None -} - -fn parse_success_packets(line: &str) -> Option { - let toks: Vec<&str> = line.split_whitespace().collect(); - let pos = toks.iter().position(|v| *v == "success")?; - toks.get(pos + 1)?.parse::().ok() -} - -fn extract_tc_match_ip(line: &str) -> Option { - let toks: Vec<&str> = line.split_whitespace().collect(); - let pos = toks.iter().position(|v| *v == "match")?; - let match_val = *toks.get(pos + 1)?; - let (hex, mask) = match_val.split_once('/')?; - if !mask.eq_ignore_ascii_case("ffffffff") || hex.len() != 8 { - return None; - } - let b1 = u8::from_str_radix(&hex[0..2], 16).ok()?; - let b2 = u8::from_str_radix(&hex[2..4], 16).ok()?; - let b3 = u8::from_str_radix(&hex[4..6], 16).ok()?; - let b4 = u8::from_str_radix(&hex[6..8], 16).ok()?; - Some(format!("{b1}.{b2}.{b3}.{b4}")) -} - #[cfg(test)] mod tests { - use super::{parse_sent_packets, parse_success_packets, parse_tc_filter_hits}; + use super::conntrack_utilization; + use crate::traffic; #[test] - fn parse_sent_pkt_counter() { - let line = "Sent 12345 bytes 67 pkt (dropped 0, overlimits 0 requeues 0)"; - assert_eq!(parse_sent_packets(line), Some(67)); - } - - #[test] - fn parse_success_counter() { - let line = "match 0a080102/ffffffff at 16 success 42"; - assert_eq!(parse_success_packets(line), Some(42)); + fn conntrack_utilization_handles_valid_and_unavailable_values() { + assert_eq!( + conntrack_utilization("184", "1000").as_deref(), + Some("18.4%") + ); + assert_eq!(conntrack_utilization("n/a", "1000"), None); + assert_eq!(conntrack_utilization("10", "0"), None); } #[test] @@ -595,8 +589,20 @@ filter parent 1: protocol ip pref 1 u32 chain 0 fh 800::801 order 2049 key ht 80 match 0a080103/ffffffff at 16 success 9 "#; - let map = parse_tc_filter_hits(raw); + let map = traffic::parse_peer_filter_hits(raw); assert_eq!(map.get("10.8.1.2"), Some(&7)); assert_eq!(map.get("10.8.1.3"), Some(&9)); } + + #[test] + fn parse_tc_filter_hits_supports_flower() { + let raw = r#" +filter protocol ip pref 10 flower chain 0 +filter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10 + dst_ip 10.8.1.2/32 + Sent 2048 bytes 17 pkt (dropped 0, overlimits 0 requeues 0) +"#; + let map = traffic::parse_peer_filter_hits(raw); + assert_eq!(map.get("10.8.1.2"), Some(&17)); + } } diff --git a/traffic_core/src/sysctl_tuning.rs b/traffic_core/src/sysctl_tuning.rs index 68c7a41..ec27d6e 100644 --- a/traffic_core/src/sysctl_tuning.rs +++ b/traffic_core/src/sysctl_tuning.rs @@ -635,4 +635,16 @@ mod tests { assert_eq!(plan.bbr_available, None); assert_eq!(plan.bbr_active, None); } + + #[test] + fn default_render_plan_does_not_manage_netdev_backlog() { + let mut c = cfg(); + assert!(!c.sysctl_tuning.netdev_backlog_enabled); + assert_eq!(c.sysctl_tuning.netdev_max_backlog, 10_000); + // Keep this assertion independent of host kernel/module availability. + c.sysctl_tuning.tcp_bbr = "off".into(); + let plan = build_render_plan(&c, false).expect("render plan"); + assert!(!plan.desired.contains_key("net.core.netdev_max_backlog")); + assert!(!plan.persisted.contains("net.core.netdev_max_backlog")); + } } diff --git a/traffic_core/src/traffic/apply.rs b/traffic_core/src/traffic/apply.rs index 660ccbb..55eabb7 100644 --- a/traffic_core/src/traffic/apply.rs +++ b/traffic_core/src/traffic/apply.rs @@ -6,7 +6,11 @@ use crate::{config::Config, netns::NsTransport, wg}; -use super::inspect::{expected_peer_class_map, inspect_and_diff}; +use super::{ + PEER_FILTER_PREF, + inspect::{expected_peer_class_map, inspect_and_diff}, + parse_tc::{is_not_found_error, normalize_tc_rate}, +}; pub(super) fn teardown_owned_topology( transport: &dyn NsTransport, @@ -26,169 +30,14 @@ pub(super) fn apply_full_topology( ifb: &str, peers: &[wg::Peer], ) -> anyhow::Result<()> { - let root = cfg.traffic.root_rate.as_str(); - run_allow_exists(transport, &["modprobe", "ifb"])?; - run_allow_exists( + run_step(transport, &["modprobe", "ifb"])?; + run_step( transport, &["ip", "link", "add", "name", ifb, "type", "ifb"], )?; run_step(transport, &["ip", "link", "set", "dev", ifb, "up"])?; - run_allow_exists( - transport, - &[ - "tc", "qdisc", "add", "dev", iface, "root", "handle", "1:", "htb", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "class", "add", "dev", iface, "parent", "1:", "classid", "1:1", "htb", "rate", - root, "ceil", root, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "qdisc", "add", "dev", iface, "handle", "ffff:", "ingress", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "filter", "add", "dev", iface, "parent", "ffff:", "protocol", "all", "u32", - "match", "u32", "0", "0", "action", "mirred", "egress", "redirect", "dev", ifb, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "qdisc", "add", "dev", ifb, "root", "handle", "2:", "htb", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "class", "add", "dev", ifb, "parent", "2:", "classid", "2:1", "htb", "rate", - root, "ceil", root, - ], - )?; - - // The same map the inspector will verify against, so apply and verify cannot disagree - // about which class a peer belongs to. - let class_map = expected_peer_class_map(peers, cfg.traffic.class_start_id)?; - - for (ip, class_id) in &class_map { - let class_egress = format!("1:{class_id}"); - let class_ingress = format!("2:{class_id}"); - let limit = cfg.traffic.peer_limit.as_str(); - - run_allow_exists( - transport, - &[ - "tc", - "class", - "add", - "dev", - iface, - "parent", - "1:1", - "classid", - class_egress.as_str(), - "htb", - "rate", - limit, - "ceil", - limit, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "qdisc", - "add", - "dev", - iface, - "parent", - class_egress.as_str(), - "fq_codel", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "filter", - "add", - "dev", - iface, - "parent", - "1:", - "protocol", - "ip", - "u32", - "match", - "ip", - "dst", - ip.as_str(), - "flowid", - class_egress.as_str(), - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "class", - "add", - "dev", - ifb, - "parent", - "2:1", - "classid", - class_ingress.as_str(), - "htb", - "rate", - limit, - "ceil", - limit, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "qdisc", - "add", - "dev", - ifb, - "parent", - class_ingress.as_str(), - "fq_codel", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "filter", - "add", - "dev", - ifb, - "parent", - "2:", - "protocol", - "ip", - "u32", - "match", - "ip", - "src", - ip.as_str(), - "flowid", - class_ingress.as_str(), - ], - )?; - } + let batch = build_tc_batch_script(cfg, iface, ifb, peers)?; + run_tc_batch(transport, &batch)?; let verify = inspect_and_diff( transport, @@ -206,28 +55,125 @@ pub(super) fn apply_full_topology( Ok(()) } -fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { - let _ = transport.run(args)?; +fn run_tc_batch(transport: &dyn NsTransport, batch: &str) -> anyhow::Result<()> { + transport + .run_with_stdin(&["tc", "-batch", "-"], batch) + .map_err(|e| anyhow::anyhow!("tc batch apply failed: {e}"))?; Ok(()) } -/// Tolerates only the idempotency conflicts that a rebuild legitimately produces. -/// Anything else is a real apply failure and must propagate. -fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { - match transport.run(args) { - Ok(_) => Ok(()), - Err(e) => { - let msg = e.to_string().to_ascii_lowercase(); - if msg.contains("file exists") - || msg.contains("eexist") - || msg.contains("object already exists") - || msg.contains("exclusive flag on") - { - return Ok(()); - } - Err(e) - } +pub(super) fn empty_topology_drift_reason( + transport: &dyn NsTransport, + iface: &str, + ifb: &str, +) -> anyhow::Result> { + let qdiscs = transport + .run(&["tc", "qdisc", "show", "dev", iface]) + .map_err(|e| anyhow::anyhow!("empty topology qdisc inspect failed: {e}"))?; + if qdiscs.lines().any(|line| { + let line = line.trim(); + (line.contains("qdisc htb 1:") && line.contains(" root")) + || line.contains("qdisc ingress ffff:") + }) { + return Ok(Some("owned tc topology exists without peers".to_string())); } + + match transport.run(&["ip", "link", "show", ifb]) { + Ok(out) if !out.trim().is_empty() => { + Ok(Some(format!("owned IFB {ifb} exists without peers"))) + } + Ok(_) => Ok(None), + Err(e) if is_not_found_error(&e.to_string()) => Ok(None), + Err(e) => Err(anyhow::anyhow!("empty topology IFB inspect failed: {e}")), + } +} + +fn build_tc_batch_script( + cfg: &Config, + iface: &str, + ifb: &str, + peers: &[wg::Peer], +) -> anyhow::Result { + let root = normalize_tc_rate(&cfg.traffic.root_rate); + let limit = normalize_tc_rate(&cfg.traffic.peer_limit); + let class_map = expected_peer_class_map(peers, cfg.traffic.class_start_id)?; + let mut batch = String::new(); + + push_command( + &mut batch, + format!("qdisc add dev {iface} root handle 1: htb"), + ); + push_command( + &mut batch, + format!("class add dev {iface} parent 1: classid 1:1 htb rate {root} ceil {root}"), + ); + push_command( + &mut batch, + format!("qdisc add dev {iface} handle ffff: ingress"), + ); + push_command( + &mut batch, + format!( + "filter add dev {iface} parent ffff: protocol all u32 match u32 0 0 action mirred egress redirect dev {ifb}" + ), + ); + push_command( + &mut batch, + format!("qdisc add dev {ifb} root handle 2: htb"), + ); + push_command( + &mut batch, + format!("class add dev {ifb} parent 2: classid 2:1 htb rate {root} ceil {root}"), + ); + + for (ip, class_id) in class_map { + let egress = format!("1:{class_id}"); + let ingress = format!("2:{class_id}"); + push_command( + &mut batch, + format!( + "class add dev {iface} parent 1:1 classid {egress} htb rate {limit} ceil {limit}" + ), + ); + push_command( + &mut batch, + format!("qdisc add dev {iface} parent {egress} fq_codel"), + ); + push_command( + &mut batch, + format!( + "filter add dev {iface} parent 1: protocol ip pref {PEER_FILTER_PREF} flower dst_ip {ip}/32 classid {egress}" + ), + ); + push_command( + &mut batch, + format!( + "class add dev {ifb} parent 2:1 classid {ingress} htb rate {limit} ceil {limit}" + ), + ); + push_command( + &mut batch, + format!("qdisc add dev {ifb} parent {ingress} fq_codel"), + ); + push_command( + &mut batch, + format!( + "filter add dev {ifb} parent 2: protocol ip pref {PEER_FILTER_PREF} flower src_ip {ip}/32 classid {ingress}" + ), + ); + } + + Ok(batch) +} + +fn push_command(batch: &mut String, command: String) { + batch.push_str(&command); + batch.push('\n'); +} + +fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { + let _ = transport.run(args)?; + Ok(()) } /// Deleting an object that is already gone is the expected outcome of teardown, not an @@ -252,3 +198,155 @@ fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Res } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::{net::Ipv4Addr, sync::Mutex}; + + #[derive(Default)] + struct BatchTransport { + call: Mutex, String)>>, + fail: bool, + } + + impl NsTransport for BatchTransport { + fn run(&self, _args: &[&str]) -> anyhow::Result { + anyhow::bail!("unexpected run") + } + + fn run_with_stdin(&self, args: &[&str], input: &str) -> anyhow::Result { + *self.call.lock().expect("call lock") = Some(( + args.iter().map(|v| (*v).to_string()).collect(), + input.to_string(), + )); + if self.fail { + anyhow::bail!("Command failed -:4") + } + Ok(String::new()) + } + } + + struct EmptyTopologyTransport { + qdiscs: &'static str, + ifb: anyhow::Result<&'static str>, + } + + impl NsTransport for EmptyTopologyTransport { + fn run(&self, args: &[&str]) -> anyhow::Result { + match args { + ["tc", "qdisc", "show", "dev", "awg0"] => Ok(self.qdiscs.to_string()), + ["ip", "link", "show", "ifb0"] => self + .ifb + .as_ref() + .map(|out| (*out).to_string()) + .map_err(|e| anyhow::anyhow!(e.to_string())), + _ => anyhow::bail!("unexpected command: {args:?}"), + } + } + + fn run_with_stdin(&self, args: &[&str], _input: &str) -> anyhow::Result { + anyhow::bail!("unexpected stdin command: {args:?}") + } + } + + fn peers(ips: &[&str]) -> Vec { + ips.iter() + .map(|ip| wg::Peer { + ipv4: ip.parse::().expect("peer IP"), + }) + .collect() + } + + #[test] + fn batch_uses_one_shared_flower_pref_and_keeps_redirect_u32() { + let mut cfg = crate::firewall::tests::test_config(); + cfg.traffic.root_rate = " 1 Gbit ".into(); + cfg.traffic.peer_limit = " 50 Mbit ".into(); + let script = build_tc_batch_script(&cfg, "awg0", "ifb0", &peers(&["10.8.1.2", "10.8.1.3"])) + .expect("render batch"); + + assert_eq!(script.lines().count(), 18, "six base and six per peer"); + assert_eq!(script.matches(" pref 10 flower ").count(), 4); + assert_eq!( + script.lines().filter(|line| line.contains(" u32 ")).count(), + 1, + "redirect only" + ); + assert!(script.contains("flower dst_ip 10.8.1.2/32 classid 1:10")); + assert!(script.contains("flower src_ip 10.8.1.2/32 classid 2:10")); + assert!(script.contains("rate 1gbit ceil 1gbit")); + assert!(script.contains("rate 50mbit ceil 50mbit")); + assert!(!script.lines().any(|line| line.starts_with("tc "))); + } + + #[test] + fn batch_runner_uses_stdin_filename_and_never_force() { + let transport = BatchTransport::default(); + run_tc_batch(&transport, "qdisc add dev awg0 root handle 1: htb\n") + .expect("batch succeeds"); + let (args, input) = transport + .call + .lock() + .expect("call lock") + .clone() + .expect("recorded call"); + assert_eq!(args, ["tc", "-batch", "-"]); + assert!(!args.iter().any(|arg| arg == "-force")); + assert!(input.starts_with("qdisc add")); + } + + #[test] + fn batch_failure_preserves_tc_line_diagnostic() { + let transport = BatchTransport { + fail: true, + ..BatchTransport::default() + }; + let err = run_tc_batch(&transport, "bad command\n").expect_err("must fail"); + assert!(err.to_string().contains("Command failed -:4")); + } + + #[test] + fn empty_peer_state_detects_owned_tc_topology_for_cleanup() { + let transport = EmptyTopologyTransport { + qdiscs: "qdisc htb 1: root refcnt 2 r2q 10 default 0x30\nqdisc ingress ffff: parent ffff:fff1", + ifb: Err(anyhow::anyhow!("must not inspect IFB after owned qdisc")), + }; + + let reason = empty_topology_drift_reason(&transport, "awg0", "ifb0") + .expect("inspect empty desired state"); + assert_eq!( + reason.as_deref(), + Some("owned tc topology exists without peers") + ); + } + + #[test] + fn empty_peer_state_is_idempotent_after_owned_topology_is_gone() { + let transport = EmptyTopologyTransport { + qdiscs: "qdisc noqueue 0: root refcnt 2", + ifb: Err(anyhow::anyhow!("Cannot find device \"ifb0\"")), + }; + + assert_eq!( + empty_topology_drift_reason(&transport, "awg0", "ifb0") + .expect("inspect clean empty desired state"), + None + ); + } + + #[test] + fn empty_peer_state_detects_leftover_ifb_for_cleanup() { + let transport = EmptyTopologyTransport { + qdiscs: "qdisc noqueue 0: root refcnt 2", + ifb: Ok("7: ifb0: mtu 1500"), + }; + + let reason = + empty_topology_drift_reason(&transport, "awg0", "ifb0").expect("inspect leftover IFB"); + assert_eq!( + reason.as_deref(), + Some("owned IFB ifb0 exists without peers") + ); + } +} diff --git a/traffic_core/src/traffic/inspect.rs b/traffic_core/src/traffic/inspect.rs index b2daea1..79e8de9 100644 --- a/traffic_core/src/traffic/inspect.rs +++ b/traffic_core/src/traffic/inspect.rs @@ -9,7 +9,7 @@ use std::collections::{BTreeMap, BTreeSet}; use crate::{netns::NsTransport, wg}; use super::{ - TrafficDrift, + FilterClassifier, FilterDirection, FilterInspect, PEER_FILTER_PREF, TrafficDrift, parse_tc::{ collect_leaf_class_ids, extract_filter_ip_class_map, has_fq_codel_for_parent, has_ingress_redirect_to_ifb, is_not_found_error, link_is_up, normalize_tc_rate, @@ -27,24 +27,19 @@ pub(super) fn inspect_and_diff( peer_limit: &str, peers: &[wg::Peer], ) -> anyhow::Result { - let root_qdisc = inspect_cmd( + let iface_qdiscs = inspect_cmd( transport, &["tc", "qdisc", "show", "dev", iface], - "egress qdisc", + "interface qdiscs", )?; - if !root_qdisc.contains("htb") { + if !iface_qdiscs.contains("htb") { return Ok(TrafficDrift { has_drift: true, reason: "missing egress root htb qdisc".to_string(), }); } - let ingress_qdisc = inspect_cmd( - transport, - &["tc", "qdisc", "show", "dev", iface], - "ingress qdisc", - )?; - if !ingress_qdisc.contains("ingress") { + if !iface_qdiscs.contains("ingress") { return Ok(TrafficDrift { has_drift: true, reason: "missing ingress qdisc".to_string(), @@ -99,6 +94,22 @@ pub(super) fn inspect_and_diff( let inspect_egress = extract_filter_ip_class_map(&filters_egress); let inspect_ingress = extract_filter_ip_class_map(&filters_ingress); + if let Some(reason) = + filter_contract_drift_reason(&inspect_egress, FilterDirection::Destination, "egress") + { + return Ok(TrafficDrift { + has_drift: true, + reason, + }); + } + if let Some(reason) = + filter_contract_drift_reason(&inspect_ingress, FilterDirection::Source, "ingress") + { + return Ok(TrafficDrift { + has_drift: true, + reason, + }); + } if !inspect_egress.duplicate_ips.is_empty() { return Ok(TrafficDrift { has_drift: true, @@ -240,36 +251,21 @@ pub(super) fn inspect_and_diff( } } - 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 = expected_map.values().copied().collect(); - if collect_leaf_class_ids(&classes_egress) != expected_set { + if collect_leaf_class_ids(&classes_egress_full) != expected_set { return Ok(TrafficDrift { has_drift: true, reason: "egress per-peer class set mismatch".to_string(), }); } - if collect_leaf_class_ids(&classes_ingress) != expected_set { + if collect_leaf_class_ids(&classes_ingress_full) != expected_set { return Ok(TrafficDrift { has_drift: true, reason: "ingress per-peer class set mismatch".to_string(), }); } - let qdisc_egress = inspect_cmd( - transport, - &["tc", "qdisc", "show", "dev", iface], - "egress leaf qdisc", - )?; - let qdisc_ingress = inspect_cmd( + let ifb_qdiscs = inspect_cmd( transport, &["tc", "qdisc", "show", "dev", ifb], "ingress leaf qdisc", @@ -277,13 +273,13 @@ pub(super) fn inspect_and_diff( for class_id in expected_map.values() { let class_egress = format!("1:{class_id}"); let class_ingress = format!("2:{class_id}"); - if !has_fq_codel_for_parent(&qdisc_egress, &class_egress) { + if !has_fq_codel_for_parent(&iface_qdiscs, &class_egress) { return Ok(TrafficDrift { has_drift: true, reason: format!("missing fq_codel for egress class {class_egress}"), }); } - if !has_fq_codel_for_parent(&qdisc_ingress, &class_ingress) { + if !has_fq_codel_for_parent(&ifb_qdiscs, &class_ingress) { return Ok(TrafficDrift { has_drift: true, reason: format!("missing fq_codel for ingress class {class_ingress}"), @@ -297,6 +293,46 @@ pub(super) fn inspect_and_diff( }) } +fn filter_contract_drift_reason( + inspect: &FilterInspect, + direction: FilterDirection, + label: &str, +) -> Option { + if inspect.unrecognized_rules > 0 { + return Some(format!( + "{label} contains {} unrecognized peer filter rule(s)", + inspect.unrecognized_rules + )); + } + for filter in &inspect.filters { + if filter.classifier != FilterClassifier::Flower { + return Some(format!( + "{label} peer filter for {} uses legacy u32 instead of flower", + filter.ip + )); + } + if filter.pref != Some(PEER_FILTER_PREF) { + return Some(format!( + "{label} flower filter for {} has pref {:?}, expected {}", + filter.ip, filter.pref, PEER_FILTER_PREF + )); + } + if filter.direction != direction { + return Some(format!( + "{label} flower filter for {} matches the wrong IP direction", + filter.ip + )); + } + if filter.prefix != 32 { + return Some(format!( + "{label} flower filter for {} uses /{}, expected /32", + filter.ip, filter.prefix + )); + } + } + None +} + /// The class-id contract: peer IP strings sorted lexicographically, then /// `class_start_id + index`. /// @@ -327,7 +363,47 @@ fn inspect_cmd(transport: &dyn NsTransport, args: &[&str], what: &str) -> anyhow #[cfg(test)] mod tests { use super::*; - use std::net::Ipv4Addr; + use std::{net::Ipv4Addr, sync::Mutex}; + + #[derive(Default)] + struct HealthyTransport { + calls: Mutex>, + } + + impl NsTransport for HealthyTransport { + fn run(&self, args: &[&str]) -> anyhow::Result { + let key = args.join(" "); + self.calls.lock().expect("calls lock").push(key.clone()); + let out = match key.as_str() { + "tc qdisc show dev awg0" => { + "qdisc htb 1: root\nqdisc ingress ffff: parent ffff:fff1\nqdisc fq_codel 8001: parent 1:10\n" + } + "tc filter show dev awg0 parent ffff:" => { + "filter parent ffff: protocol all pref 49152 u32\n action order 1: mirred egress redirect dev ifb0\n" + } + "ip link show ifb0" => "2: ifb0: mtu 1500 state UP\n", + "tc filter show dev awg0 parent 1:" => { + "filter protocol ip pref 10 flower chain 0\nfilter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10\n dst_ip 10.8.1.2/32\n" + } + "tc filter show dev ifb0 parent 2:" => { + "filter protocol ip pref 10 flower chain 0\nfilter protocol ip pref 10 flower chain 0 handle 0x1 classid 2:10\n src_ip 10.8.1.2/32\n" + } + "tc class show dev awg0" => { + "class htb 1:1 root rate 1000Mbit ceil 1000Mbit\nclass htb 1:10 parent 1:1 rate 70Mbit ceil 70Mbit\n" + } + "tc class show dev ifb0" => { + "class htb 2:1 root rate 1000Mbit ceil 1000Mbit\nclass htb 2:10 parent 2:1 rate 70Mbit ceil 70Mbit\n" + } + "tc qdisc show dev ifb0" => "qdisc htb 2: root\nqdisc fq_codel 8002: parent 2:10\n", + _ => anyhow::bail!("unexpected inspect command: {key}"), + }; + Ok(out.to_string()) + } + + fn run_with_stdin(&self, _args: &[&str], _input: &str) -> anyhow::Result { + anyhow::bail!("unexpected stdin command") + } + } fn peers(ips: &[&str]) -> Vec { ips.iter() @@ -364,4 +440,49 @@ mod tests { fn empty_peer_list_yields_empty_map() { assert!(expected_peer_class_map(&[], 10).expect("map").is_empty()); } + + #[test] + fn healthy_inspection_reuses_full_qdisc_and_class_outputs() { + let transport = HealthyTransport::default(); + let drift = inspect_and_diff( + &transport, + "awg0", + "ifb0", + 10, + "1000mbit", + "70mbit", + &peers(&["10.8.1.2"]), + ) + .expect("inspect"); + assert!(!drift.has_drift, "{}", drift.reason); + + let calls = transport.calls.lock().expect("calls lock"); + assert_eq!(calls.len(), 8); + assert_eq!( + calls + .iter() + .filter(|call| *call == "tc qdisc show dev awg0") + .count(), + 1 + ); + assert_eq!( + calls + .iter() + .filter(|call| call.starts_with("tc class show dev")) + .count(), + 2 + ); + assert!(!calls.iter().any(|call| call.contains(" parent 1:1"))); + assert!(!calls.iter().any(|call| call.contains(" parent 2:1"))); + } + + #[test] + fn legacy_u32_peer_filter_requires_migration() { + let inspect = extract_filter_ip_class_map( + "filter parent 1: protocol ip pref 10 u32\n match 0a080102/ffffffff at 16\n flowid 1:10", + ); + let reason = filter_contract_drift_reason(&inspect, FilterDirection::Destination, "egress") + .expect("legacy drift"); + assert!(reason.contains("legacy u32 instead of flower")); + } } diff --git a/traffic_core/src/traffic/mod.rs b/traffic_core/src/traffic/mod.rs index f0e4f15..f1e4f35 100644 --- a/traffic_core/src/traffic/mod.rs +++ b/traffic_core/src/traffic/mod.rs @@ -16,13 +16,18 @@ use std::{ time::Duration, }; -use apply::{apply_full_topology, teardown_owned_topology}; +use apply::{apply_full_topology, empty_topology_drift_reason, teardown_owned_topology}; use inspect::{expected_peer_class_map, inspect_and_diff}; use parse_tc::{ extract_filter_ip_class_map, normalize_tc_rate, parse_class_rate_ceil_map, rate_ceil_matches_limit, }; +/// All peer filters of one direction deliberately share one flower priority and mask. +/// A different priority would create another classifier lookup before the packet reaches +/// the matching filter and would defeat the purpose of replacing the linear u32 list. +pub(super) const PEER_FILTER_PREF: u16 = 10; + #[derive(Debug, Clone)] struct TrafficDrift { has_drift: bool, @@ -31,9 +36,34 @@ struct TrafficDrift { #[derive(Debug, Clone)] struct FilterInspect { + filters: Vec, map: BTreeMap, duplicate_ips: BTreeSet, duplicate_classes: BTreeSet, + unrecognized_rules: usize, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FilterClassifier { + Flower, + U32, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FilterDirection { + Source, + Destination, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ParsedPeerFilter { + ip: String, + prefix: u8, + class_id: u16, + classifier: FilterClassifier, + direction: FilterDirection, + pref: Option, + packets: Option, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -55,27 +85,32 @@ pub struct PeerTrafficStatus { pub drift_reason: Option, } -pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result> { +pub(crate) fn inspect_drift_summary_with_state( + cfg: &Config, + target: &docker::RuntimeTarget, + transport: &dyn crate::netns::NsTransport, + peers: &[wg::Peer], +) -> anyhow::Result> { if !cfg.traffic.enabled { return Ok(None); } - let target = docker::discover_target(cfg)?; - let transport = build_transport(target.container_pid, &target.container_name); - let peers = wg::discover_peers(cfg, &target)?; - if peers.is_empty() { - return Ok(Some("no peers discovered".to_string())); + return empty_topology_drift_reason( + transport, + &target.interface_name, + &cfg.traffic.ifb_name, + ); } let drift = inspect_and_diff( - transport.as_ref(), + transport, &target.interface_name, &cfg.traffic.ifb_name, cfg.traffic.class_start_id, &cfg.traffic.root_rate, &cfg.traffic.peer_limit, - &peers, + peers, )?; if drift.has_drift { @@ -93,11 +128,40 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { let _lock = FileLock::acquire_timeout("/run/awg2-harness/traffic.lock", Duration::from_secs(10))?; let target = docker::discover_target(cfg)?; + reconcile_traffic_locked(cfg, &target) +} + +pub(crate) fn reconcile_traffic_with_target( + cfg: &Config, + target: &docker::RuntimeTarget, +) -> anyhow::Result<()> { + if !cfg.traffic.enabled { + return Ok(()); + } + let _lock = + FileLock::acquire_timeout("/run/awg2-harness/traffic.lock", Duration::from_secs(10))?; + reconcile_traffic_locked(cfg, target) +} + +fn reconcile_traffic_locked(cfg: &Config, target: &docker::RuntimeTarget) -> anyhow::Result<()> { let transport = build_transport(target.container_pid, &target.container_name); - let peers = wg::discover_peers(cfg, &target)?; + let peers = wg::discover_peers(cfg, target)?; if peers.is_empty() { - println!("no peers found, skip tc apply"); + if let Some(reason) = empty_topology_drift_reason( + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + )? { + tracing::info!(target: "traffic", "{reason}; removing owned topology"); + teardown_owned_topology( + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + )?; + } else { + tracing::debug!(target: "traffic", "no peers and no owned topology"); + } return Ok(()); } @@ -112,11 +176,11 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { )?; if !drift.has_drift { - println!("traffic rules up-to-date, skip apply"); + tracing::debug!(target: "traffic", "traffic rules are up-to-date"); return Ok(()); } - println!("traffic drift detected: {}", drift.reason); + tracing::info!(target: "traffic", reason = %drift.reason, "traffic drift detected"); teardown_owned_topology( transport.as_ref(), @@ -134,14 +198,15 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { Ok(()) } -pub fn peer_status_table(cfg: &Config) -> anyhow::Result>> { +pub(crate) fn peer_status_table_with_state( + cfg: &Config, + target: &docker::RuntimeTarget, + transport: &dyn crate::netns::NsTransport, + peers: &[wg::Peer], +) -> anyhow::Result>> { if !cfg.traffic.enabled { return Ok(None); } - - let target = docker::discover_target(cfg)?; - let transport = build_transport(target.container_pid, &target.container_name); - let peers = wg::discover_peers(cfg, &target)?; if peers.is_empty() { return Ok(Some(Vec::new())); } @@ -181,7 +246,7 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result anyhow::Result anyhow::Result bool { + filter.classifier == FilterClassifier::Flower + && filter.pref == Some(PEER_FILTER_PREF) + && filter.direction == direction + && filter.prefix == 32 +} + +pub(crate) fn parse_peer_filter_hits(raw: &str) -> BTreeMap { + parse_tc::parse_peer_filter_hits(raw) +} diff --git a/traffic_core/src/traffic/parse_tc.rs b/traffic_core/src/traffic/parse_tc.rs index f77af71..bb16206 100644 --- a/traffic_core/src/traffic/parse_tc.rs +++ b/traffic_core/src/traffic/parse_tc.rs @@ -8,7 +8,7 @@ use std::{ net::Ipv4Addr, }; -use super::{ClassRateCeil, FilterInspect}; +use super::{ClassRateCeil, FilterClassifier, FilterDirection, FilterInspect, ParsedPeerFilter}; /// Maps peer IP to class id from `tc filter show`, and reports both kinds of duplicate. /// @@ -16,43 +16,196 @@ use super::{ClassRateCeil, FilterInspect}; /// the IP and flowid are taken from within one block. Scanning tokens across unrelated /// lines would happily pair an IP from one filter with a flowid from the next. pub(super) fn extract_filter_ip_class_map(filters: &str) -> FilterInspect { + let mut parsed = Vec::new(); let mut out = BTreeMap::new(); let mut duplicate_ips = BTreeSet::new(); let mut class_to_ip = BTreeMap::new(); let mut duplicate_classes = BTreeSet::new(); + let mut unrecognized_rules = 0; for block in split_filter_blocks(filters) { - let mut ip: Option = None; - let mut class_id: Option = None; - - for line in block.lines() { - let t = line.trim(); - if t.contains("match") && ip.is_none() { - ip = extract_u32_match_hex(t).and_then(|hex| hex_to_ipv4_string(&hex)); + if let Some(filter) = parse_peer_filter_block(&block) { + if out.contains_key(&filter.ip) { + duplicate_ips.insert(filter.ip.clone()); } - if t.contains("flowid") && class_id.is_none() { - class_id = extract_flowid_class_id(t); + if class_to_ip + .insert(filter.class_id, filter.ip.clone()) + .is_some() + { + duplicate_classes.insert(filter.class_id); } - } - - if let (Some(ip), Some(class_id)) = (ip, class_id) { - if out.contains_key(&ip) { - 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.insert(filter.ip.clone(), filter.class_id); + parsed.push(filter); + } else if block_looks_like_rule(&block) { + unrecognized_rules += 1; } } FilterInspect { + filters: parsed, map: out, duplicate_ips, duplicate_classes, + unrecognized_rules, } } +pub(super) fn parse_peer_filter_hits(filters: &str) -> BTreeMap { + extract_filter_ip_class_map(filters) + .filters + .into_iter() + .filter_map(|filter| filter.packets.map(|packets| (filter.ip, packets))) + .collect() +} + +fn parse_peer_filter_block(block: &str) -> Option { + let lower = block.to_ascii_lowercase(); + let classifier = if lower.split_whitespace().any(|token| token == "flower") { + FilterClassifier::Flower + } else if lower.split_whitespace().any(|token| token == "u32") || lower.contains("match ") { + FilterClassifier::U32 + } else { + return None; + }; + let pref = extract_token_u16(block, "pref"); + let class_id = extract_class_id(block)?; + let packets = extract_packet_count(block); + + let (ip, prefix, direction) = match classifier { + FilterClassifier::Flower => extract_flower_ip(block)?, + FilterClassifier::U32 => extract_u32_ip(block)?, + }; + + Some(ParsedPeerFilter { + ip, + prefix, + class_id, + classifier, + direction, + pref, + packets, + }) +} + +fn block_looks_like_rule(block: &str) -> bool { + let lower = block.to_ascii_lowercase(); + lower.contains("classid ") + || lower.contains("flowid ") + || lower.contains("dst_ip ") + || lower.contains("src_ip ") + || lower.contains("match ") + || lower.contains(" action ") + || lower.contains("\nsent ") + || lower.lines().next().is_some_and(|line| { + line.split_whitespace().any(|token| token == "handle") + && line + .split_whitespace() + .any(|token| token == "flower" || token == "u32") + }) +} + +fn extract_flower_ip(block: &str) -> Option<(String, u8, FilterDirection)> { + for line in block.lines() { + let tokens: Vec<&str> = line.split_whitespace().collect(); + for (key, direction) in [ + ("src_ip", FilterDirection::Source), + ("dst_ip", FilterDirection::Destination), + ] { + let Some(pos) = tokens.iter().position(|token| *token == key) else { + continue; + }; + let raw = *tokens.get(pos + 1)?; + let (addr, prefix) = match raw.split_once('/') { + Some((addr, prefix)) => (addr, prefix.parse::().ok()?), + None => (raw, 32), + }; + let ip = addr.parse::().ok()?; + if prefix > 32 { + return None; + } + return Some((ip.to_string(), prefix, direction)); + } + } + None +} + +fn extract_u32_ip(block: &str) -> Option<(String, u8, FilterDirection)> { + for line in block.lines() { + let tokens: Vec<&str> = line.split_whitespace().collect(); + let Some(pos) = tokens.iter().position(|token| *token == "match") else { + continue; + }; + let (hex, mask) = tokens.get(pos + 1)?.split_once('/')?; + if hex.len() != 8 || mask.len() != 8 { + continue; + } + let prefix = mask_hex_to_prefix(mask)?; + let at = tokens + .iter() + .position(|token| *token == "at") + .and_then(|idx| tokens.get(idx + 1)) + .and_then(|raw| raw.parse::().ok())?; + let direction = match at { + 12 => FilterDirection::Source, + 16 => FilterDirection::Destination, + _ => continue, + }; + return Some((hex_to_ipv4_string(hex)?, prefix, direction)); + } + None +} + +fn mask_hex_to_prefix(mask: &str) -> Option { + let mask = u32::from_str_radix(mask, 16).ok()?; + let prefix = mask.leading_ones() as u8; + let expected = if prefix == 0 { + 0 + } else { + u32::MAX << (32 - prefix) + }; + (mask == expected).then_some(prefix) +} + +fn extract_class_id(block: &str) -> Option { + for line in block.lines() { + let tokens: Vec<&str> = line.split_whitespace().collect(); + for key in ["classid", "flowid"] { + let Some(pos) = tokens.iter().position(|token| *token == key) else { + continue; + }; + let class = tokens.get(pos + 1)?; + return class.split(':').nth(1)?.parse::().ok(); + } + } + None +} + +fn extract_token_u16(block: &str, key: &str) -> Option { + for line in block.lines() { + let tokens: Vec<&str> = line.split_whitespace().collect(); + if let Some(pos) = tokens.iter().position(|token| *token == key) { + return tokens.get(pos + 1)?.parse::().ok(); + } + } + None +} + +fn extract_packet_count(block: &str) -> Option { + for line in block.lines() { + let tokens: Vec<&str> = line.split_whitespace().collect(); + if tokens.first() == Some(&"Sent") + && tokens.get(2) == Some(&"bytes") + && tokens.get(4) == Some(&"pkt") + { + return tokens.get(3)?.parse::().ok(); + } + if let Some(pos) = tokens.iter().position(|token| *token == "success") { + return tokens.get(pos + 1)?.parse::().ok(); + } + } + None +} + pub(super) fn has_fq_codel_for_parent(raw_qdisc: &str, parent_class: &str) -> bool { raw_qdisc.lines().map(str::trim).any(|line| { line.contains("fq_codel") && line.contains("parent") && line.contains(parent_class) @@ -210,27 +363,6 @@ fn split_filter_blocks(raw: &str) -> Vec { blocks } -fn extract_flowid_class_id(line: &str) -> Option { - let toks: Vec<&str> = line.split_whitespace().collect(); - let flowid = toks - .iter() - .position(|v| *v == "flowid") - .and_then(|i| toks.get(i + 1))?; - let class_tail = flowid.split(':').nth(1)?; - class_tail.parse::().ok() -} - -fn extract_u32_match_hex(line: &str) -> Option { - let toks: Vec<&str> = line.split_whitespace().collect(); - let pos = toks.iter().position(|v| *v == "match")?; - let match_val = *toks.get(pos + 1)?; - let (hex, mask) = match_val.split_once('/')?; - if mask.eq_ignore_ascii_case("ffffffff") && hex.len() == 8 { - return Some(hex.to_ascii_lowercase()); - } - None -} - fn hex_to_ipv4_string(hex: &str) -> Option { if hex.len() != 8 { return None; @@ -299,6 +431,56 @@ mod tests { assert!(inspect.duplicate_classes.contains(&10)); } + #[test] + fn parse_debian13_flower_filters_and_statistics() { + let raw = r#" +filter protocol ip pref 10 flower chain 0 +filter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10 + dst_ip 10.8.1.2/32 + not_in_hw + Sent 1200 bytes 12 pkt (dropped 0, overlimits 0 requeues 0) +filter protocol ip pref 10 flower chain 0 handle 0x2 classid 1:11 + dst_ip 10.8.1.3 + not_in_hw + Sent 900 bytes 9 pkt (dropped 0, overlimits 0 requeues 0) +"#; + let inspect = extract_filter_ip_class_map(raw); + assert_eq!(inspect.unrecognized_rules, 0); + assert_eq!(inspect.map.get("10.8.1.2"), Some(&10)); + assert_eq!(inspect.map.get("10.8.1.3"), Some(&11)); + assert!(inspect.filters.iter().all(|filter| { + filter.classifier == FilterClassifier::Flower + && filter.direction == FilterDirection::Destination + && filter.pref == Some(10) + && filter.prefix == 32 + })); + assert_eq!(parse_peer_filter_hits(raw).get("10.8.1.2"), Some(&12)); + } + + #[test] + fn parse_flower_source_prefix_and_classid_on_separate_lines() { + let raw = r#" +filter protocol ip pref 10 flower chain 0 handle 0x7 + src_ip 10.8.1.5/24 + classid 2:42 +"#; + let inspect = extract_filter_ip_class_map(raw); + let filter = inspect.filters.first().expect("flower filter"); + assert_eq!(filter.ip, "10.8.1.5"); + assert_eq!(filter.prefix, 24); + assert_eq!(filter.direction, FilterDirection::Source); + assert_eq!(filter.class_id, 42); + } + + #[test] + fn malformed_peer_rule_is_not_silently_ignored() { + let raw = + "filter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10\n action drop"; + let inspect = extract_filter_ip_class_map(raw); + assert_eq!(inspect.unrecognized_rules, 1); + assert!(inspect.map.is_empty()); + } + #[test] fn detect_fq_codel_parent() { let raw = "qdisc fq_codel 8001: dev wg0 parent 1:10 limit 10240p";