maeneko/forgetting
1
0
Fork 0
Code Issues Actions Packages Projects Releases 15 Wiki Activity

Add shaping, Prometheus, monitor tab and peer audit

Browse Source 
Baseline TypeScript version before the Go rewrite: CAKE fair sharing with
automatic deprioritisation of heavy devices, per-key priority, /metrics with
a metrics token, the built-in Prometheus install/remove, the Monitor tab,
grouped sidebar tabs and the dead-peer audit.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
maeneko
2026-10-14 15:27:05 +03:00
co-authored by Claude Sonnet 5.5
parent 7fbff82e76
commit e9df363dd2
33 changed files with 2708 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
awg-ctrl/index.ts
+240 -17
View File
@@ -1,15 +1,16 @@
// Copyright (c) 2026 Ivan Vasilev
// This source code is licensed under the MIT license found in the
// LICENSE file in the root directory of this source tree.
import fs, { readFileSync, writeFileSync, existsSync, mkdirSync } from "fs";
import fs, { readFileSync, writeFileSync, existsSync, mkdirSync, openSync, closeSync } from "fs";
import path from "path";
import os from "os";
import crypto from "crypto";
import { execSync, spawnSync } from "child_process";
import { execSync, spawnSync, execFile, spawn } from "child_process";
import * as zlib from "zlib";
import express, { Request, Response, NextFunction } from "express";
import Database from "better-sqlite3";
import winston from "winston";
import { createShaping } from "./shaping";
const logsDir = path.join(process.cwd(), "logs");
if (!existsSync(logsDir)) mkdirSync(logsDir, { recursive: true });
@@ -575,6 +576,26 @@ function getPeersData(): Record<string, { online: boolean; lastHandshake: number
} catch (e) { logger.warn("getPeersData failed", { error: e }); return {}; }
}
// Шейпинг (CAKE) и сэмплер тяжёлых устройств. Счётчики читаем асинхронно: цикл
// не должен блокировать event loop, как это делают остальные вызовы awg.
const shaping = createShaping({
iface: CONFIG.interface,
getCfg, setCfg, logger,
// CAKE видит только внутренний IP-пакет, на проводе к нему добавляются
// IPv4 (20) + UDP (8) + заголовок WG (32) = 60 байт и случайный префикс S4.
// Формулу сверять iperf3-замером на стенде (SHAPE_OVERHEAD перекрывает её).
overhead: () => 60 + (Number(CONFIG.awgParams.S4) || 0),
peers: () => new Promise((resolve, reject) => {
execFile("awg", ["show", CONFIG.interface, "dump"], { encoding: "utf8", timeout: 10_000 }, (err, out) => {
if (err) { reject(err); return; }
const rows = out.split("\n").slice(1).map(l => l.split("\t"))
.filter(f => f[0] && f[3])
.map(f => ({ ip: f[3].split(",")[0].split("/")[0], rx: Number(f[5]) || 0, tx: Number(f[6]) || 0 }));
resolve(rows);
});
}),
});
function rebuildConf() {
const users = db.prepare("SELECT name, ip, pub_key, psk_key FROM users").all() as UserRow[];
const devices = stmts.peersAll.all();
@@ -680,6 +701,7 @@ function restartAwg() {
// в [Peer]-блоки, которые rebuildConf() пересобирает прямо сейчас.
const gen = reloadAwgParams();
syncPeers();
void shaping.apply(); // tc-правила жили на интерфейсе, который мы только что пересоздали
const versions = readAwgVersions();
CONFIG.awgModule = versions.module;
CONFIG.awgTools = versions.tools;
@@ -803,6 +825,7 @@ function startInterface() {
const r = spawnSync("awg-quick", ["up", CONFIG.interface]);
if (r.status !== 0) throw new Error(`awg-quick up failed: ${r.stderr?.toString()}`);
syncPeers();
void shaping.apply();
return getInterfaceStatus();
}
@@ -849,12 +872,30 @@ function addUser(username: string): UserRow {
};
}
// Снять пира с интерфейса и убедиться, что его там нет. Раньше результат `awg set … remove`
// не проверялся: строку из базы удаляли, а пир в ядре оставался живым, и про него уже никто не
// знал. Теперь порядок такой: сначала ядро, потом база — при отказе строка остаётся, и
// повторная попытка (надгробие панели, следующий sync) найдёт пира.
function kernelPeers(): Set<string> | null {
const r = spawnSync("awg", ["show", CONFIG.interface, "peers"]);
return r.status === 0 ? new Set(r.stdout.toString().split("\n").map(l => l.trim()).filter(Boolean)) : null;
}
function dropFromKernel(pub: string): boolean {
const r = spawnSync("awg", ["set", CONFIG.interface, "peer", pub, "remove"]);
if (r.status === 0) return true;
const left = kernelPeers();
if (left && !left.has(pub)) return true; // ответ странный, но пира уже нет
logger.error("peer remove from kernel failed", { pub, stderr: r.stderr?.toString() });
return false;
}
function removeUser(username: string) {
const user = stmts.get.get(username);
if (!user) throw new Error("Пользователь не найден");
if (!dropFromKernel(user.pub_key)) throw new Error("Не удалось снять пира с интерфейса");
stmts.delete.run(username);
spawnSync("awg", ["set", CONFIG.interface, "peer", user.pub_key, "remove"]);
rebuildConf();
for (const ext of [".conf", ".key"]) {
@@ -920,7 +961,9 @@ function replacePeer(oldPub: string, newPub: string): { ip: string; psk_key: str
stmts.peerRekey.run(oldPub, newPub);
throw new Error(`awg set failed: ${r.stderr?.toString()}`);
}
spawnSync("awg", ["set", CONFIG.interface, "peer", oldPub, "remove"]);
// Старый ключ не снялся — пир остаётся «призраком» в ядре до аудита панели; смену ключа не откатываем:
// новый уже стоит, а клиент к этому моменту получил бы ошибку.
if (!dropFromKernel(oldPub)) logger.error("peer rekey: старый ключ остался в ядре", { owner: old.owner });
rebuildConf();
logger.info("peer rekeyed", { owner: old.owner, ip: old.ip });
return { ip: old.ip, psk_key: old.psk_key };
@@ -941,8 +984,8 @@ function rotatePeerPsk(pub: string): { ip: string; psk_key: string } {
function removePeer(pub: string) {
const peer = stmts.peerGet.get(pub);
if (!peer) throw new HttpError(404, "Пир не найден");
if (!dropFromKernel(pub)) throw new Error("Не удалось снять пира с интерфейса");
stmts.peerDelete.run(pub);
spawnSync("awg", ["set", CONFIG.interface, "peer", pub, "remove"]);
rebuildConf();
logger.info("peer removed", { owner: peer.owner });
}
@@ -964,18 +1007,15 @@ function syncOwnedPeers(panel: string, keep: string[]): {
const prefix = panel + "/";
const removed: PeerRow[] = [];
let claimed = 0;
db.transaction(() => {
for (const p of stmts.peersAll.all()) {
if (p.owner.startsWith(prefix)) {
if (!want.has(p.pub_key)) { stmts.peerDelete.run(p.pub_key); removed.push(p); }
} else if (LEGACY_OWNER.test(p.owner) && want.has(p.pub_key)) {
stmts.peerOwner.run(prefix + p.owner, p.pub_key);
claimed++;
}
for (const p of stmts.peersAll.all()) {
if (p.owner.startsWith(prefix)) {
// Ядро первым: не снялся — строка остаётся, следующий sync попробует снова.
if (!want.has(p.pub_key) && dropFromKernel(p.pub_key)) { stmts.peerDelete.run(p.pub_key); removed.push(p); }
} else if (LEGACY_OWNER.test(p.owner) && want.has(p.pub_key)) {
stmts.peerOwner.run(prefix + p.owner, p.pub_key);
claimed++;
}
})();
for (const p of removed) spawnSync("awg", ["set", CONFIG.interface, "peer", p.pub_key, "remove"]);
}
if (removed.length || claimed) rebuildConf();
if (removed.length) logger.info("peers synced out", { panel, removed: removed.map(p => p.owner) });
@@ -992,7 +1032,9 @@ const app = express();
const SYNC_PATH = "/api/peers/sync";
const jsonSmall = express.json({ limit: "1kb" });
const jsonSync = express.json({ limit: "512kb" });
app.use((req, res, next) => (req.path === SYNC_PATH ? jsonSync : jsonSmall)(req, res, next));
// /api/audit/clean тоже несёт до SYNC_MAX ключей — под 1kb он принимал бы около двадцати.
const BIG_BODY_PATHS = new Set([SYNC_PATH, "/api/audit/clean"]);
app.use((req, res, next) => (BIG_BODY_PATHS.has(req.path) ? jsonSync : jsonSmall)(req, res, next));
function verifyInternalToken(token: string): boolean {
try {
@@ -1075,6 +1117,8 @@ app.get("/api/users/stats", auth, handler((_req, res) => {
lastHandshake: peers[u.pub_key]?.lastHandshake ?? 0,
rx: peers[u.pub_key]?.rx ?? 0,
tx: peers[u.pub_key]?.tx ?? 0,
// «normal» | «background»: фоновый — пока канал занят, а устройство давно качает больше своей доли
priority: shaping.priorityOf(u.ip),
})),
});
}));
@@ -1111,10 +1155,43 @@ app.get("/api/peers", auth, handler((_req, res) => {
lastHandshake: stats[p.pub_key]?.lastHandshake ?? 0,
rx: stats[p.pub_key]?.rx ?? 0,
tx: stats[p.pub_key]?.tx ?? 0,
priority: shaping.priorityOf(p.ip),
})),
});
}));
// Пиры, которые стоят в ядре, но неизвестны awg-ctrl (нет ни в users, ни в peers): снять их
// после сбоя `awg set … remove` некому. Только чтение — панель (core) сама решает, что делать.
app.get("/api/audit", auth, handler((_req, res) => {
const r = spawnSync("awg", ["show", CONFIG.interface, "dump"]);
if (r.status !== 0) throw new HttpError(503, "awg show не ответил");
const known = new Set<string>([
...(db.prepare("SELECT pub_key FROM users").all() as { pub_key: string }[]).map(u => u.pub_key),
...stmts.peersAll.all().map(p => p.pub_key),
]);
const now = Math.floor(Date.now() / 1000);
const ghosts = r.stdout.toString().split("\n").slice(1).map(l => l.split("\t")).filter(f => f[0] && !known.has(f[0])).map(f => {
const lastHandshake = Number(f[4]) || 0;
return { pub_key: f[0], lastHandshake, online: lastHandshake > 0 && now - lastHandshake < 180, rx: Number(f[5]) || 0, tx: Number(f[6]) || 0 };
});
res.json({ ghosts });
}));
// Снять «призраков» из ядра. Ключи присылает панель, но здесь каждый перепроверяется: пир, который
// успел появиться в users/peers, не трогаем. Строк в базе у «призраков» нет — снимается только ядро.
app.post("/api/audit/clean", auth, handler((req, res) => {
const { pub_keys } = (req.body ?? {}) as { pub_keys?: unknown };
if (!Array.isArray(pub_keys) || pub_keys.length > SYNC_MAX) throw new HttpError(400, "Неверный список ключей");
const removed: string[] = [], skipped: string[] = [];
for (const k of pub_keys.map(checkPubKey)) {
const known = stmts.peerGet.get(k) || (db.prepare("SELECT 1 FROM users WHERE pub_key = ?").get(k));
if (known || k === getServerPublicKey() || !kernelPeers()?.has(k)) { skipped.push(k); continue; }
(dropFromKernel(k) ? removed : skipped).push(k);
}
if (removed.length) logger.warn("ghost peers removed from kernel", { count: removed.length, removed });
res.json({ removed, skipped });
}));
// Должен стоять ДО PUT /api/peers/:pub — иначе тот примет «sync» за ключ.
app.put(SYNC_PATH, auth, handler((req, res) => {
const { owner, keep } = (req.body ?? {}) as { owner?: unknown; keep?: unknown };
@@ -1175,6 +1252,148 @@ app.post("/awg/upgrade", auth, handler((_req, res) => {
}
}));
// Деление канала между устройствами. Путь под /api/ — иначе awg-agent на ноде его не пропустит
// (и без PATCH: агент отвечает на него 400).
app.get("/api/shaping", auth, handler(async (_req, res) => {
res.json(await shaping.status());
}));
app.put("/api/shaping", auth, handler(async (req, res) => {
const b = (req.body ?? {}) as Record<string, unknown>;
const mbit = (v: unknown, what: string): number => {
const n = Number(v);
if (!Number.isInteger(n) || n < 0 || n > 100_000) throw new HttpError(400, `${what}: целое число Мбит/с от 1 до 100000`);
return n;
};
if (typeof b.enabled !== "boolean" || typeof b.auto !== "boolean")
throw new HttpError(400, "enabled и auto должны быть true/false");
const downMbit = mbit(b.downMbit ?? 0, "Загрузка");
const upMbit = mbit(b.upMbit ?? 0, "Отдача");
if (b.enabled && (downMbit < 1 || upMbit < 1))
throw new HttpError(400, "Для включения укажите ширину канала в обе стороны");
try {
await shaping.update({ enabled: b.enabled, auto: b.auto, downMbit, upMbit });
} catch (e) {
// tc/nft/модуль не нашлись или ядро отказало: оператору нужна причина, а не 500.
throw new HttpError(500, e instanceof Error ? e.message : "Не удалось применить шейпинг");
}
res.json(await shaping.status());
}));
// Метрики для Prometheus. awg-ctrl отдаёт JSON, а текстовый формат собирает awg-ui: на ноде
// агент пропускает только /api/*, и один формат на ядре и нодах проще сверять. Кэш на 5 с —
// несколько скрейперов не должны гонять `awg show` чаще.
const METRICS_TTL_MS = 5_000;
let metricsCache: { at: number; data: Promise<unknown> } | null = null;
function linkCounter(name: string): number {
try { return Number(readFileSync(`/sys/class/net/${CONFIG.interface}/statistics/${name}`, "utf8")) || 0; }
catch { return 0; }
}
// Нагрузка хоста из /proc. Счётчики CPU кумулятивные (секунды по режимам, как у node_exporter): долю считает
// Prometheus через rate(), здесь мгновенного «процента» нет. Память — MemAvailable, а не MemFree: кэш ядра свободен.
const CPU_MODES = ["user", "nice", "system", "idle", "iowait", "irq", "softirq", "steal"] as const;
function readHost() {
try {
const first = readFileSync("/proc/stat", "utf8").split("\n")[0].trim().split(/\s+/).slice(1).map(Number);
const cpu: Record<string, number> = {};
CPU_MODES.forEach((m, i) => { cpu[m] = (first[i] || 0) / 100; }); // USER_HZ = 100 на всех обычных ядрах
const mem = readFileSync("/proc/meminfo", "utf8");
const kb = (k: string) => Number(mem.match(new RegExp(`^${k}:\\s+(\\d+) kB`, "m"))?.[1] ?? 0) * 1024;
return {
cpu, cores: os.cpus().length, load: os.loadavg(),
mem: { total: kb("MemTotal"), available: kb("MemAvailable") },
};
} catch { return null; }
}
async function collectMetrics() {
const st = getInterfaceStatus();
const stats = st.up ? getPeersData() : {};
const list: { kind: "user" | "device"; name: string; ip: string; online: boolean; rx: number; tx: number; priority: string }[] = [
...(db.prepare("SELECT name, ip, pub_key FROM users WHERE vpn_key != ''").all() as UserRow[]).map(u => ({
kind: "user" as const, name: u.name, ip: u.ip,
online: stats[u.pub_key]?.online ?? false, rx: stats[u.pub_key]?.rx ?? 0, tx: stats[u.pub_key]?.tx ?? 0,
priority: shaping.priorityOf(u.ip),
})),
...stmts.peersAll.all().map(p => ({
kind: "device" as const, name: "", ip: p.ip,
online: stats[p.pub_key]?.online ?? false, rx: stats[p.pub_key]?.rx ?? 0, tx: stats[p.pub_key]?.tx ?? 0,
priority: shaping.priorityOf(p.ip),
})),
];
const all = Object.values(stats);
return {
up: st.up,
gen: CONFIG.gen,
peers: { total: st.peers, online: all.filter(p => p.online).length },
// С точки зрения сервера: rx — принято от клиентов (их отдача), tx — отправлено клиентам (их загрузка).
rx_bytes: all.reduce((a, p) => a + p.rx, 0),
tx_bytes: all.reduce((a, p) => a + p.tx, 0),
link: {
rx_bytes: linkCounter("rx_bytes"), tx_bytes: linkCounter("tx_bytes"),
rx_packets: linkCounter("rx_packets"), tx_packets: linkCounter("tx_packets"),
rx_dropped: linkCounter("rx_dropped"), tx_dropped: linkCounter("tx_dropped"),
rx_errors: linkCounter("rx_errors"), tx_errors: linkCounter("tx_errors"),
},
shaping: await shaping.status(),
host: readHost(),
list,
};
}
app.get("/api/metrics", auth, handler(async (_req, res) => {
if (!metricsCache || Date.now() - metricsCache.at > METRICS_TTL_MS) {
const data = collectMetrics();
metricsCache = { at: Date.now(), data };
data.catch(() => { metricsCache = null; }); // ошибку не кэшируем
}
res.json(await metricsCache.data);
}));
// Установка и удаление встроенного мониторинга (Prometheus) по кнопке в панели. Действие одно из двух и без
// параметров: awg-ctrl запускает ту же команду, что оператор набрал бы руками (`awg-ctrl monitor install|remove`,
// root), так что код один. Панель сама пакетов не ставит и не удаляет — она только просит об этом здесь.
const MONITOR_CLI = process.env.AWG_CLI ?? "/usr/local/bin/awg-ctrl";
const MONITOR_LOG = process.env.MONITOR_LOG ?? path.resolve(process.cwd(), "..", "logs", "monitor-job.log");
type MonitorAction = "install" | "remove";
let monitorJob: { action: MonitorAction; startedAt: number; exit: number | null } | null = null;
const logTail = (): string => {
try {
const t = readFileSync(MONITOR_LOG, "utf8").slice(-4000);
// apt рисует прогресс возвратом каретки и цветами — в панели это мусор
return t.replace(/\x1b\[[0-9;]*[A-Za-z]/g, "").replace(/\r/g, "\n");
} catch { return ""; }
};
function startMonitorJob(action: MonitorAction, res: Response) {
if (monitorJob && monitorJob.exit === null) { res.status(409).json({ error: "Операция уже идёт" }); return; }
if (!existsSync(MONITOR_CLI)) throw new HttpError(501, `CLI не найден (${MONITOR_CLI}): выполните на сервере awg-ctrl monitor ${action}`);
mkdirSync(path.dirname(MONITOR_LOG), { recursive: true });
const fd = openSync(MONITOR_LOG, "w");
const job = { action, startedAt: Date.now(), exit: null as number | null };
monitorJob = job;
const done = (code: number) => { job.exit = code; try { closeSync(fd); } catch { /* уже закрыт */ } };
const child = spawn(MONITOR_CLI, ["monitor", action], { stdio: ["ignore", fd, fd] });
child.on("error", e => { logger.error("monitor: не запустился", { action, error: e.message }); done(127); });
child.on("exit", code => { logger.info("monitor: завершён", { action, code }); done(code ?? 1); });
res.status(202).json({ running: true, action });
}
app.post("/api/monitor/install", auth, handler((_req, res) => { startMonitorJob("install", res); }));
app.post("/api/monitor/remove", auth, handler((_req, res) => { startMonitorJob("remove", res); }));
app.get("/api/monitor/job", auth, handler((_req, res) => {
res.json({
action: monitorJob?.action ?? null,
running: !!monitorJob && monitorJob.exit === null,
exit: monitorJob?.exit ?? null,
log: monitorJob ? logTail() : "",
});
}));
app.get("/awg/status", auth, handler((_req, res) => {
res.json(getInterfaceStatus());
}));
@@ -1188,3 +1407,7 @@ app.listen(SERVER.port, "127.0.0.1", () => {
});
ensureInterfaceUp();
syncPeers();
// Обязательно и на старте: при загрузке интерфейс поднимает systemd (ExecStartPre),
// ни один из путей awg-quick up внутри awg-ctrl тогда не срабатывает.
void shaping.apply();
shaping.start();
awg-ctrl/shaping.ts
+338
View File
@@ -0,0 +1,338 @@
// Честное деление канала между устройствами (CAKE) + автоподстройка по RFC 6057.
//
// Схема:
// загрузка (к клиенту) — CAKE root на awg1, dual-dsthost, diffserv3
// отдача (от клиента) — ingress awg1 → ifb-awg, CAKE dual-srchost, besteffort
// тяжёлые при перегрузке — DSCP CS1 (bulk-tin CAKE) через отдельную nft-таблицу
//
// Ничего не пишет в awg1.conf (PostUp не трогаем): всё применяется из процесса и
// идемпотентно. Правила tc пропадают вместе с интерфейсом, поэтому apply() зовётся
// на старте, после restartAwg()/startInterface() и из страховочной проверки в цикле.
//
// Сэмплер только КЛАССИФИЦИРУЕТ. Он никого не удаляет и не отключает — ни по
// времени, ни по неактивности (правило проекта: удаление только по явному списку
// от панели).
import { execFile, spawn } from "child_process";
import type { Logger } from "winston";
export interface ShapingSettings {
enabled: boolean;
downMbit: number;
upMbit: number;
auto: boolean;
}
export interface PeerCounters {
ip: string;
rx: number; // принято от клиента (его отдача)
tx: number; // отправлено клиенту (его загрузка)
}
export interface ShapingDeps {
iface: string;
getCfg: (key: string, fallback: string) => string;
setCfg: (key: string, value: string) => void;
logger: Pick<Logger, "info" | "warn" | "error">;
// Байт на пакет, которые туннель добавляет на проводе (CAKE видит только внутренний IP-пакет).
overhead: () => number;
peers: () => Promise<PeerCounters[]>;
}
export interface QdiscStats { packets: number; bytes: number; drops: number; backlog: number }
export interface ShapingStatus {
settings: ShapingSettings;
active: boolean; // CAKE реально стоит на интерфейсе
congested: boolean;
heavy: string[]; // IP устройств в bulk-классе
devices: number; // активных устройств в окне
rateDown: number; // Мбит/с за последний тик
rateUp: number;
down?: QdiscStats;
up?: QdiscStats;
}
// Окна и пороги. Переопределение через env — только для тестов на стенде.
const num = (v: string | undefined, d: number) => (v && Number.isFinite(Number(v)) && Number(v) > 0 ? Number(v) : d);
const SAMPLE_S = num(process.env.SHAPE_SAMPLE_S, 10);
const WINDOW_S = num(process.env.SHAPE_WINDOW_S, 600);
const CONGEST_ON = 0.85; // доля ширины, выше которой канал считается перегруженным
const CONGEST_OFF = 0.70; // и ниже которой перегрузка снимается (гистерезис)
const ACTIVE_MBIT = 0.1; // меньше — устройство не считается активным
const HEAVY_MULT = 1.5; // тяжёлое: ≥ 1.5 честных долей…
const HEAVY_MIN_FRAC = 0.05; // …и не меньше 5% ширины канала
const CHECK_EVERY = 3; // страховочная проверка qdisc — раз в N тиков
const IFB = "ifb-awg";
const NFT_TABLE = "awg_shape";
function exec(cmd: string, args: string[]): Promise<string> {
return new Promise((resolve, reject) => {
execFile(cmd, args, { encoding: "utf8", timeout: 15_000 }, (err, stdout, stderr) => {
if (err) reject(new Error(`${cmd} ${args.join(" ")}: ${(stderr || err.message).toString().trim()}`));
else resolve(stdout);
});
});
}
function execStdin(cmd: string, args: string[], input: string): Promise<void> {
return new Promise((resolve, reject) => {
const p = spawn(cmd, args, { stdio: ["pipe", "ignore", "pipe"] });
let err = "";
p.stderr.on("data", d => { err += d; });
p.stdin.on("error", () => undefined); // процесс умер раньше, чем дочитал stdin — причина придёт в close
p.on("error", reject);
p.on("close", code => code === 0 ? resolve() : reject(new Error(`${cmd}: ${err.trim() || `exit ${code}`}`)));
p.stdin.end(input);
});
}
interface Tick { dt: number; bytes: number }
export function createShaping(deps: ShapingDeps) {
const { iface, logger } = deps;
// ───────────── настройки ─────────────
function settings(): ShapingSettings {
return {
enabled: deps.getCfg("shapeEnabled", "0") === "1",
downMbit: Number(deps.getCfg("shapeDownMbit", "0")) || 0,
upMbit: Number(deps.getCfg("shapeUpMbit", "0")) || 0,
auto: deps.getCfg("shapeAuto", "1") === "1",
};
}
function save(s: ShapingSettings) {
deps.setCfg("shapeEnabled", s.enabled ? "1" : "0");
deps.setCfg("shapeDownMbit", String(s.downMbit));
deps.setCfg("shapeUpMbit", String(s.upMbit));
deps.setCfg("shapeAuto", s.auto ? "1" : "0");
}
// ───────────── применение ─────────────
// Все изменения tc/nft идут строго по одному: два параллельных apply() дали бы
// полусобранные qdisc.
let chain: Promise<unknown> = Promise.resolve();
const serial = <T>(fn: () => Promise<T>): Promise<T> => {
const r = chain.then(fn, fn);
chain = r.catch(() => undefined);
return r;
};
let lastError = "";
const noteError = (e: unknown) => {
const m = e instanceof Error ? e.message : String(e);
if (m !== lastError) logger.warn("shaping: " + m); // одинаковую ошибку в лог не спамим
lastError = m;
};
const cakeOverhead = () => {
const o = Number(process.env.SHAPE_OVERHEAD);
return Number.isFinite(o) && o >= 0 && process.env.SHAPE_OVERHEAD ? o : deps.overhead();
};
async function nftApply(heavy: string[]) {
const elements = heavy.length ? `\n elements = { ${heavy.join(", ")} }` : "";
// add+delete в одной транзакции: таблица пересоздаётся атомарно, без окна без правил.
await execStdin("nft", ["-f", "-"], `
add table inet ${NFT_TABLE}
delete table inet ${NFT_TABLE}
table inet ${NFT_TABLE} {
set heavy {
type ipv4_addr${elements}
}
chain postrouting {
type filter hook postrouting priority -150; policy accept;
oifname "${iface}" ip daddr @heavy ip dscp set cs1
}
}
`);
}
async function hasCake(dev: string): Promise<boolean> {
try { return /\bcake\b/.test(await exec("tc", ["qdisc", "show", "dev", dev, "root"])); }
catch { return false; }
}
async function hasIngressRedirect(): Promise<boolean> {
try { return /mirred/.test(await exec("tc", ["filter", "show", "dev", iface, "parent", "ffff:"])); }
catch { return false; }
}
async function applyNow() {
const s = settings();
if (!s.enabled || s.downMbit <= 0 || s.upMbit <= 0) { await removeNow(); return; }
const oh = cakeOverhead();
// Загрузка (к клиенту). replace = change, если qdisc уже стоит: скорость меняется на лету.
await exec("tc", ["qdisc", "replace", "dev", iface, "root", "cake",
"bandwidth", `${s.downMbit}mbit`, "diffserv3", "dual-dsthost", "overhead", String(oh)]);
// Отдача (от клиента): зеркалим входящий трафик awg1 на ifb и делим там.
try { await exec("ip", ["link", "show", IFB]); }
catch { await exec("ip", ["link", "add", IFB, "type", "ifb"]); }
await exec("ip", ["link", "set", IFB, "up"]);
await exec("tc", ["qdisc", "replace", "dev", IFB, "root", "cake",
"bandwidth", `${s.upMbit}mbit`, "besteffort", "dual-srchost", "overhead", String(oh)]);
if (!(await hasIngressRedirect())) {
await exec("tc", ["qdisc", "del", "dev", iface, "ingress"]).catch(() => undefined);
await exec("tc", ["qdisc", "add", "dev", iface, "handle", "ffff:", "ingress"]);
await exec("tc", ["filter", "add", "dev", iface, "parent", "ffff:", "protocol", "all", "prio", "1",
"matchall", "action", "mirred", "egress", "redirect", "dev", IFB]);
}
await nftApply([...heavySet]);
lastError = "";
logger.info("shaping applied", { down: s.downMbit, up: s.upMbit, overhead: oh, heavy: heavySet.size });
}
async function removeNow() {
await exec("tc", ["qdisc", "del", "dev", iface, "root"]).catch(() => undefined);
await exec("tc", ["qdisc", "del", "dev", iface, "ingress"]).catch(() => undefined);
await exec("ip", ["link", "del", IFB]).catch(() => undefined);
await exec("nft", ["delete", "table", "inet", NFT_TABLE]).catch(() => undefined);
heavySet.clear(); history.clear(); totals.length = 0; congested = false;
}
// Не бросает: отсутствие tc/nft/модуля не должно валить awg-ctrl (в отличие от ensureInterfaceUp).
function apply(): Promise<void> {
return serial(applyNow).catch(noteError);
}
async function update(s: ShapingSettings): Promise<void> {
save(s);
await serial(applyNow); // пользователю нужна честная ошибка, а не молчаливый warn
lastError = "";
}
// ───────────── сэмплер (RFC 6057) ─────────────
const prev = new Map<string, { rx: number; tx: number }>();
const history = new Map<string, Tick[]>(); // загрузка по IP
const totals: Tick[] = []; // суммарная загрузка (всё, что ушло клиентам)
const heavySet = new Set<string>();
let congested = false;
let lastRateDown = 0, lastRateUp = 0;
let lastTickAt = 0;
let tickNo = 0;
const windowTicks = Math.max(3, Math.round(WINDOW_S / SAMPLE_S));
const push = (arr: Tick[], t: Tick) => { arr.push(t); while (arr.length > windowTicks) arr.shift(); };
const mbit = (arr: Tick[]) => {
const dt = arr.reduce((a, t) => a + t.dt, 0);
return dt > 0 ? arr.reduce((a, t) => a + t.bytes, 0) * 8 / dt / 1e6 : 0;
};
async function tick() {
const s = settings();
if (!s.enabled) return;
const now = Date.now();
const dt = lastTickAt ? (now - lastTickAt) / 1000 : SAMPLE_S;
lastTickAt = now;
tickNo++;
// Страховка: интерфейс могли поднять руками (awg-quick down/up) — qdisc тогда исчез.
if (tickNo % CHECK_EVERY === 0 && !(await hasCake(iface))) {
logger.warn("shaping: qdisc на интерфейсе пропал, применяю заново");
await apply();
}
let peers: PeerCounters[];
try { peers = await deps.peers(); } catch (e) { noteError(e); return; }
let totalDown = 0, totalUp = 0;
const seen = new Set<string>();
for (const p of peers) {
seen.add(p.ip);
const before = prev.get(p.ip);
// Счётчики ядра сбрасываются при down/up интерфейса и замене ключа пира → отрицательная дельта = 0.
const dTx = before && p.tx >= before.tx ? p.tx - before.tx : 0;
const dRx = before && p.rx >= before.rx ? p.rx - before.rx : 0;
prev.set(p.ip, { rx: p.rx, tx: p.tx });
totalDown += dTx; totalUp += dRx;
let h = history.get(p.ip);
if (!h) history.set(p.ip, h = []);
push(h, { dt, bytes: dTx });
}
for (const ip of [...prev.keys()]) if (!seen.has(ip)) { prev.delete(ip); history.delete(ip); heavySet.delete(ip); }
push(totals, { dt, bytes: totalDown });
lastRateDown = totalDown * 8 / dt / 1e6;
lastRateUp = totalUp * 8 / dt / 1e6;
if (!s.auto) {
if (heavySet.size || congested) { heavySet.clear(); congested = false; await serial(() => nftApply([])).catch(noteError); }
return;
}
if (totals.length < windowTicks) return; // пока окно не набралось, решений не принимаем
const avg = mbit(totals);
const wasCongested = congested;
if (!congested && avg >= CONGEST_ON * s.downMbit) congested = true;
if ( congested && avg < CONGEST_OFF * s.downMbit) congested = false;
const nextHeavy = new Set<string>();
let active = 0;
if (congested) {
const rates = [...history.entries()].filter(([, h]) => h.length >= windowTicks).map(([ip, h]) => [ip, mbit(h)] as const);
active = rates.filter(([, r]) => r >= ACTIVE_MBIT).length;
const share = s.downMbit / Math.max(1, active);
for (const [ip, r] of rates) {
const wasHeavy = heavySet.has(ip);
// Входим на 1.5 доли, выходим ниже 1.0 доли — без дребезга на границе.
if (r >= Math.max(HEAVY_MULT * share, HEAVY_MIN_FRAC * s.downMbit) || (wasHeavy && r >= share)) nextHeavy.add(ip);
}
}
const changed = nextHeavy.size !== heavySet.size || [...nextHeavy].some(ip => !heavySet.has(ip));
if (changed || wasCongested !== congested) {
heavySet.clear(); nextHeavy.forEach(ip => heavySet.add(ip));
logger.info("shaping: класс тяжёлых обновлён", { congested, avgMbit: Math.round(avg), heavy: [...heavySet] });
await serial(() => nftApply([...heavySet])).catch(noteError);
}
}
let timer: NodeJS.Timeout | null = null;
function start() {
if (timer) return;
let busy = false;
timer = setInterval(() => {
if (busy) return; // предыдущий тик ещё идёт — пропускаем, а не копим очередь
busy = true;
tick().catch(noteError).finally(() => { busy = false; });
}, SAMPLE_S * 1000);
timer.unref();
}
// ───────────── состояние ─────────────
async function qdiscStats(dev: string): Promise<QdiscStats | undefined> {
try {
const arr = JSON.parse(await exec("tc", ["-s", "-j", "qdisc", "show", "dev", dev, "root"])) as Record<string, unknown>[];
const q = arr.find(x => x.kind === "cake");
if (!q) return undefined;
return {
packets: Number(q.packets) || 0, bytes: Number(q.bytes) || 0,
drops: Number(q.drops) || 0, backlog: Number(q.backlog) || 0,
};
} catch { return undefined; }
}
async function status(): Promise<ShapingStatus> {
const s = settings();
const [active, down, up] = s.enabled
? await Promise.all([hasCake(iface), qdiscStats(iface), qdiscStats(IFB)])
: [false, undefined, undefined];
return {
settings: s, active, congested, heavy: [...heavySet],
devices: [...history.values()].filter(h => mbit(h) >= ACTIVE_MBIT).length,
rateDown: Math.round(lastRateDown * 10) / 10,
rateUp: Math.round(lastRateUp * 10) / 10,
down, up,
};
}
// Текущий приоритет устройства по его внутреннему IP: «фоновый» — пока он в bulk-классе.
const priorityOf = (ip: string): "normal" | "background" => (heavySet.has(ip) ? "background" : "normal");
return { settings, update, apply, start, status, priorityOf };
}
export type Shaping = ReturnType<typeof createShaping>;