maenekoandClaude Sonnet 5.5
e9df363dd2
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>
339 lines
16 KiB
TypeScript
339 lines
16 KiB
TypeScript
// Честное деление канала между устройствами (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>;
|