531 lines
19 KiB
JavaScript
531 lines
19 KiB
JavaScript
/**
|
|
* Планировщик пингов и замеров скорости для карты сети.
|
|
* Периодически вызывает пинг и speed-test по связям из network-config,
|
|
* сохраняет результаты в S3 (кеш по рёбрам + агрегат network-map-cache/latest.json).
|
|
* Логи пишет в кольцевой буфер для просмотра в UI.
|
|
*/
|
|
|
|
const { readServersFromS3 } = require('../routes/serversRoutes');
|
|
const { loadNetworkConfig } = require('../routes/mikrotikConfigRoutes');
|
|
const { readS3TextObject, writeS3JsonObject } = require('./s3Service');
|
|
|
|
const UI_SETTINGS_KEY = 'bgp_data/rt_ui_settings.json';
|
|
const NETWORK_MAP_CACHE_KEY = 'network-map-cache/latest.json';
|
|
const NEXT_RUN_AT_KEY = 'networkMapSchedulerNextRunAt';
|
|
|
|
const DEFAULT_INTERVAL_MINUTES = 5;
|
|
const DEFAULT_LOG_MAX_LINES = 500;
|
|
const RESTART_RECOVERY_DELAY_MS = 10 * 60 * 1000;
|
|
|
|
/** Кольцевой буфер логов */
|
|
class LogBuffer {
|
|
constructor(maxLines = 500) {
|
|
this.maxLines = Math.max(100, maxLines);
|
|
this.lines = [];
|
|
this.index = 0;
|
|
}
|
|
|
|
push(line) {
|
|
const ts = new Date().toISOString();
|
|
const entry = `${ts} ${line}`;
|
|
if (this.lines.length < this.maxLines) {
|
|
this.lines.push(entry);
|
|
} else {
|
|
this.lines[this.index] = entry;
|
|
this.index = (this.index + 1) % this.maxLines;
|
|
}
|
|
}
|
|
|
|
getLines() {
|
|
if (this.lines.length < this.maxLines) return [...this.lines];
|
|
return [...this.lines.slice(this.index), ...this.lines.slice(0, this.index)];
|
|
}
|
|
|
|
setMaxLines(n) {
|
|
this.maxLines = Math.max(100, n);
|
|
if (this.lines.length > this.maxLines) {
|
|
this.lines = this.getLines().slice(-this.maxLines);
|
|
this.index = 0;
|
|
}
|
|
}
|
|
}
|
|
|
|
const logBuffer = new LogBuffer(DEFAULT_LOG_MAX_LINES);
|
|
|
|
async function loadUiSettings() {
|
|
try {
|
|
const data = await readS3TextObject(UI_SETTINGS_KEY).catch(() => ({ body: '{}' }));
|
|
const parsed = JSON.parse(data?.body || '{}');
|
|
return parsed && typeof parsed === 'object' ? parsed : {};
|
|
} catch (_) {
|
|
return {};
|
|
}
|
|
}
|
|
|
|
/** Построить список связей как в NetworkMapDashboard (tunnelInterfaces → connections) */
|
|
async function buildConnections() {
|
|
const [serversList, config, uiSettings] = await Promise.all([
|
|
readServersFromS3(),
|
|
loadNetworkConfig(),
|
|
loadUiSettings(),
|
|
]);
|
|
const servers = Array.isArray(serversList) ? serversList : [];
|
|
const tunnelInterfaces = Array.isArray(config.tunnelInterfaces) ? config.tunnelInterfaces : [];
|
|
|
|
const getServer = (serverId) =>
|
|
servers.find((s) => s.id === serverId || s.ip === serverId || s.dns === serverId);
|
|
|
|
const connList = [];
|
|
for (const iface of tunnelInterfaces) {
|
|
if (!iface.serverId || !iface.serverId2) continue;
|
|
const s1 = getServer(iface.serverId);
|
|
const s2 = getServer(iface.serverId2);
|
|
if (!s1 || !s2 || s1.ip === s2.ip) continue;
|
|
const s1Key = s1.id || s1.dns || s1.ip;
|
|
const s2Key = s2.id || s2.dns || s2.ip;
|
|
if (!s1Key || !s2Key) continue;
|
|
const isJumphost1 = s1 && ['jumphost', 'home'].includes(String(s1.type || '').toLowerCase());
|
|
const isJumphost2 = s2 && ['jumphost', 'home'].includes(String(s2.type || '').toLowerCase());
|
|
connList.push({
|
|
from: s1.ip,
|
|
to: s2.ip,
|
|
fromKey: s1Key,
|
|
toKey: s2Key,
|
|
internalFromTo: iface.remoteIp || null,
|
|
internalToFrom: iface.localIp || null,
|
|
speedTestServerId: isJumphost1 ? s1Key : isJumphost2 ? s2Key : null,
|
|
interfaceName: iface.name || null,
|
|
});
|
|
}
|
|
return { connections: connList, servers, uiSettings };
|
|
}
|
|
|
|
function edgePingKey(a, b) {
|
|
return [String(a), String(b)].sort().join(':');
|
|
}
|
|
|
|
function speedKey(key1, key2) {
|
|
return [String(key1), String(key2)].sort().join(':');
|
|
}
|
|
|
|
/** Параллельно с лимитом */
|
|
async function runWithLimit(tasks, limit = 4) {
|
|
const results = [];
|
|
let index = 0;
|
|
async function runNext() {
|
|
const i = index++;
|
|
if (i >= tasks.length) return;
|
|
try {
|
|
results[i] = { value: await tasks[i]() };
|
|
} catch (err) {
|
|
results[i] = { error: err };
|
|
}
|
|
await runNext();
|
|
}
|
|
const workers = Array.from({ length: Math.min(limit, tasks.length) }, () => runNext());
|
|
await Promise.all(workers);
|
|
return results;
|
|
}
|
|
|
|
/** Один тест на ноду в момент времени */
|
|
async function runWithOnePerKey(tasks, getKey) {
|
|
const n = tasks.length;
|
|
const keyToIndices = new Map();
|
|
for (let i = 0; i < n; i++) {
|
|
const k = String(getKey(i));
|
|
if (!keyToIndices.has(k)) keyToIndices.set(k, []);
|
|
keyToIndices.get(k).push(i);
|
|
}
|
|
const results = [];
|
|
for (let i = 0; i < n; i++) results[i] = null;
|
|
const keys = Array.from(keyToIndices.keys());
|
|
let round = 0;
|
|
while (true) {
|
|
const batch = [];
|
|
for (const k of keys) {
|
|
const indices = keyToIndices.get(k);
|
|
if (round < indices.length) batch.push(indices[round]);
|
|
}
|
|
if (batch.length === 0) break;
|
|
await Promise.all(
|
|
batch.map(async (i) => {
|
|
try {
|
|
results[i] = { value: await tasks[i]() };
|
|
} catch (err) {
|
|
results[i] = { error: err };
|
|
}
|
|
})
|
|
);
|
|
round++;
|
|
}
|
|
return results;
|
|
}
|
|
|
|
/**
|
|
* Выполнить одну итерацию: пинг и speed-test по связям, записать кеш и логи.
|
|
* @param {object} opts - { baseUrl, runPing, runSpeedTest, log }
|
|
*/
|
|
async function runJob(opts) {
|
|
const { baseUrl, runPing = true, runSpeedTest = true, log = (msg) => logBuffer.push(msg), trigger = 'scheduled' } = opts;
|
|
const pingMap = {};
|
|
const speedMap = {};
|
|
|
|
log(trigger === 'manual' ? '--- Ручной запуск ---' : '--- Запуск по расписанию ---');
|
|
|
|
try {
|
|
const { connections, servers } = await buildConnections();
|
|
if (connections.length === 0) {
|
|
log('Нет связей в network-config (tunnelInterfaces).');
|
|
await writeCache(pingMap, speedMap);
|
|
return { pingMap, speedMap };
|
|
}
|
|
|
|
log(`Связей: ${connections.length}. Пинг: ${runPing ? 'да' : 'нет'}, Скорость: ${runSpeedTest ? 'да' : 'нет'}`);
|
|
|
|
if (runPing) {
|
|
const pingTasks = [];
|
|
for (const c of connections) {
|
|
const from = String(c.from);
|
|
const to = String(c.to);
|
|
if (!from || !to || from === to) continue;
|
|
const srcFrom = servers.find((s) => s.ip === from);
|
|
const srcTo = servers.find((s) => s.ip === to);
|
|
const canPingFrom = srcFrom && ['jumphost', 'home'].includes(String(srcFrom.type || '').toLowerCase()) && c.internalFromTo;
|
|
const canPingTo = srcTo && ['jumphost', 'home'].includes(String(srcTo.type || '').toLowerCase()) && c.internalToFrom;
|
|
|
|
if (canPingFrom) {
|
|
pingTasks.push(async () => {
|
|
const ekey = edgePingKey(from, to);
|
|
try {
|
|
const res = await fetch(`${baseUrl}/api/mikrotik/ping`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({
|
|
serverId: c.fromKey,
|
|
target: c.internalFromTo,
|
|
gatewayIp: c.internalFromTo,
|
|
count: 3,
|
|
forceRefresh: true,
|
|
}),
|
|
});
|
|
const data = await res.json().catch(() => ({}));
|
|
const ms = typeof data?.avgMs === 'number' ? Math.round(data.avgMs) : null;
|
|
pingMap[ekey] = ms;
|
|
const errMsg = !res.ok ? (data?.message || `HTTP ${res.status}`) : null;
|
|
log(`Пинг ${from} → ${to}: ${ms != null ? ms + ' ms' : errMsg || 'ошибка'}`);
|
|
return ms;
|
|
} catch (e) {
|
|
pingMap[ekey] = null;
|
|
log(`Пинг ${from} → ${to}: ошибка ${e?.message || e}`);
|
|
return null;
|
|
}
|
|
});
|
|
} else if (canPingTo) {
|
|
pingTasks.push(async () => {
|
|
const ekey = edgePingKey(from, to);
|
|
try {
|
|
const res = await fetch(`${baseUrl}/api/mikrotik/ping`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({
|
|
serverId: c.toKey,
|
|
target: c.internalToFrom,
|
|
gatewayIp: c.internalToFrom,
|
|
count: 3,
|
|
forceRefresh: true,
|
|
}),
|
|
});
|
|
const data = await res.json().catch(() => ({}));
|
|
const ms = typeof data?.avgMs === 'number' ? Math.round(data.avgMs) : null;
|
|
pingMap[ekey] = ms;
|
|
const errMsg2 = !res.ok ? (data?.message || `HTTP ${res.status}`) : null;
|
|
log(`Пинг ${to} → ${from}: ${ms != null ? ms + ' ms' : errMsg2 || 'ошибка'}`);
|
|
return ms;
|
|
} catch (e) {
|
|
pingMap[ekey] = null;
|
|
log(`Пинг ${to} → ${from}: ошибка ${e?.message || e}`);
|
|
return null;
|
|
}
|
|
});
|
|
}
|
|
}
|
|
await runWithLimit(pingTasks, 4);
|
|
}
|
|
|
|
if (runSpeedTest) {
|
|
// Таймаут запроса: замер может долго идти на медленных каналах (download + upload по durationSeconds каждый)
|
|
const SPEED_TEST_REQUEST_MS = 300000; // 5 мин
|
|
const withSpeed = connections.filter((c) => c.speedTestServerId && c.interfaceName);
|
|
const speedTasks = withSpeed.map((c) => async () => {
|
|
const key = speedKey(c.fromKey, c.toKey);
|
|
const server = servers.find((s) => (s.id || s.dns || s.ip) === c.speedTestServerId);
|
|
const serverLabel = server ? (server.dns || server.ip || server.id) : c.speedTestServerId;
|
|
const speedLabel = `${serverLabel} / ${c.interfaceName}`;
|
|
const ac = new AbortController();
|
|
const timeoutId = setTimeout(() => ac.abort(), SPEED_TEST_REQUEST_MS);
|
|
try {
|
|
const res = await fetch(`${baseUrl}/api/mikrotik/speed-test`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({
|
|
serverId: c.speedTestServerId,
|
|
interfaceName: c.interfaceName,
|
|
forceRefresh: true,
|
|
}),
|
|
signal: ac.signal,
|
|
});
|
|
clearTimeout(timeoutId);
|
|
const data = await res.json().catch(() => ({}));
|
|
if (data?.ok && (data.tcpDownloadBps != null || data.tcpUploadBps != null)) {
|
|
speedMap[key] = {
|
|
tcpDownloadBps: data.tcpDownloadBps,
|
|
tcpUploadBps: data.tcpUploadBps,
|
|
cached: data.cached === true,
|
|
durationSeconds: data.durationSeconds,
|
|
};
|
|
const down = data.tcpDownloadBps != null ? (data.tcpDownloadBps / 1e6).toFixed(1) : '—';
|
|
const up = data.tcpUploadBps != null ? (data.tcpUploadBps / 1e6).toFixed(1) : '—';
|
|
log(`Скорость ${speedLabel}: ↓${down} ↑${up} Mbps`);
|
|
return speedMap[key];
|
|
}
|
|
speedMap[key] = null;
|
|
const speedErr = !res.ok ? (data?.message || `HTTP ${res.status}`) : 'нет данных';
|
|
log(`Скорость ${speedLabel}: ${speedErr}`);
|
|
return null;
|
|
} catch (e) {
|
|
clearTimeout(timeoutId);
|
|
speedMap[key] = null;
|
|
const msg = e?.message || e;
|
|
const isAbort = String(msg).toLowerCase().includes('abort');
|
|
log(`Скорость ${speedLabel}: ошибка ${isAbort ? 'таймаут или отмена запроса' : msg}`);
|
|
return null;
|
|
}
|
|
});
|
|
// В один момент — только один тест на один сервер назначения (удалённый конец туннеля),
|
|
// иначе несколько замеров к одному хостингу делят канал и дают заниженные результаты
|
|
await runWithOnePerKey(speedTasks, (i) => {
|
|
const c = withSpeed[i];
|
|
return c.speedTestServerId === c.fromKey ? c.toKey : c.fromKey;
|
|
});
|
|
}
|
|
|
|
await writeCache(pingMap, speedMap);
|
|
log('Кеш карты сети обновлён.');
|
|
} catch (err) {
|
|
log(`Ошибка: ${err?.message || err}`);
|
|
}
|
|
|
|
return { pingMap, speedMap };
|
|
}
|
|
|
|
async function writeCache(pingMap, speedMap) {
|
|
try {
|
|
await writeS3JsonObject(NETWORK_MAP_CACHE_KEY, {
|
|
pingMap: pingMap || {},
|
|
speedMap: speedMap || {},
|
|
updatedAt: Date.now(),
|
|
});
|
|
} catch (e) {
|
|
logBuffer.push(`Запись кеша в S3 не удалась: ${e?.message || e}`);
|
|
}
|
|
}
|
|
|
|
let schedulerTimer = null;
|
|
let schedulerStartTimeout = null;
|
|
/** Время последнего запуска по расписанию (для расчёта следующего) */
|
|
let lastScheduledRunAt = null;
|
|
/** Реально запланированное время следующего запуска */
|
|
let nextScheduledRunAt = null;
|
|
/** Интервал активного таймера (мс) */
|
|
let activeIntervalMs = null;
|
|
|
|
async function persistNextRunAt(nextRunAt) {
|
|
try {
|
|
const ui = await loadUiSettings();
|
|
ui[NEXT_RUN_AT_KEY] = typeof nextRunAt === 'number' ? Math.round(nextRunAt) : null;
|
|
await writeS3JsonObject(UI_SETTINGS_KEY, ui);
|
|
} catch (e) {
|
|
logBuffer.push(`Не удалось сохранить nextRunAt: ${e?.message || e}`);
|
|
}
|
|
}
|
|
|
|
function getSettingsFromUiSettings(uiSettings) {
|
|
const persistedNextRunAtRaw = uiSettings?.[NEXT_RUN_AT_KEY];
|
|
const persistedNextRunAt = Number(persistedNextRunAtRaw);
|
|
return {
|
|
enabled: Boolean(uiSettings.networkMapSchedulerEnabled),
|
|
intervalMinutes: Math.max(1, Math.min(1440, parseInt(uiSettings.networkMapSchedulerIntervalMinutes, 10) || DEFAULT_INTERVAL_MINUTES)),
|
|
runPing: uiSettings.networkMapSchedulerRunPing !== false,
|
|
runSpeedTest: uiSettings.networkMapSchedulerRunSpeedTest !== false,
|
|
logMaxLines: Math.max(100, Math.min(2000, parseInt(uiSettings.networkMapSchedulerLogMaxLines, 10) || DEFAULT_LOG_MAX_LINES)),
|
|
persistedNextRunAt: Number.isFinite(persistedNextRunAt) ? persistedNextRunAt : null,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Запустить планировщик (setInterval).
|
|
* @param {object} logger - pino logger
|
|
* @param {number} port - порт сервера для вызова API
|
|
*/
|
|
function initNetworkMapScheduler(logger, port) {
|
|
const log = logger || console;
|
|
const baseUrl = `http://127.0.0.1:${port}`;
|
|
|
|
async function tick(source = 'interval') {
|
|
if (source === 'interval') {
|
|
lastScheduledRunAt = Date.now();
|
|
if (activeIntervalMs) {
|
|
nextScheduledRunAt = lastScheduledRunAt + activeIntervalMs;
|
|
persistNextRunAt(nextScheduledRunAt);
|
|
}
|
|
}
|
|
let uiSettings;
|
|
try {
|
|
const raw = await readS3TextObject(UI_SETTINGS_KEY).catch(() => ({ body: '{}' }));
|
|
uiSettings = JSON.parse(raw?.body || '{}') || {};
|
|
} catch (_) {
|
|
return;
|
|
}
|
|
const settings = getSettingsFromUiSettings(uiSettings);
|
|
if (!settings.enabled) return;
|
|
|
|
logBuffer.setMaxLines(settings.logMaxLines);
|
|
logBuffer.push('--- Запуск по расписанию ---');
|
|
runJob({
|
|
baseUrl,
|
|
runPing: settings.runPing,
|
|
runSpeedTest: settings.runSpeedTest,
|
|
trigger: source === 'startup-delay' ? 'startup-delay' : 'scheduled',
|
|
log: (msg) => {
|
|
logBuffer.push(msg);
|
|
log.info({ component: 'network-map-scheduler' }, msg);
|
|
},
|
|
}).catch((err) => {
|
|
logBuffer.push(`Ошибка: ${err?.message || err}`);
|
|
log.error({ component: 'network-map-scheduler', err: err?.message }, 'Scheduler run failed');
|
|
});
|
|
}
|
|
|
|
async function schedule() {
|
|
let uiSettings;
|
|
try {
|
|
const raw = await readS3TextObject(UI_SETTINGS_KEY).catch(() => ({ body: '{}' }));
|
|
uiSettings = JSON.parse(raw?.body || '{}') || {};
|
|
} catch (_) {
|
|
return;
|
|
}
|
|
const settings = getSettingsFromUiSettings(uiSettings);
|
|
if (!settings.enabled) {
|
|
activeIntervalMs = null;
|
|
nextScheduledRunAt = null;
|
|
persistNextRunAt(null);
|
|
log.info({ component: 'network-map-scheduler' }, 'Scheduler disabled in settings');
|
|
return;
|
|
}
|
|
logBuffer.setMaxLines(settings.logMaxLines);
|
|
const intervalMs = settings.intervalMinutes * 60 * 1000;
|
|
activeIntervalMs = intervalMs;
|
|
const now = Date.now();
|
|
const persisted = settings.persistedNextRunAt;
|
|
let firstDelayMs = intervalMs;
|
|
if (typeof persisted === 'number' && Number.isFinite(persisted)) {
|
|
if (persisted > now) {
|
|
// Плановая дата ещё не наступила — сохраняем исходный план.
|
|
firstDelayMs = Math.max(1000, persisted - now);
|
|
} else {
|
|
// Плановая дата уже прошла (контейнер был перезапущен/простаивал) —
|
|
// первый запуск через 10 минут от старта контейнера.
|
|
firstDelayMs = RESTART_RECOVERY_DELAY_MS;
|
|
}
|
|
}
|
|
nextScheduledRunAt = now + firstDelayMs;
|
|
persistNextRunAt(nextScheduledRunAt);
|
|
log.info(
|
|
{
|
|
component: 'network-map-scheduler',
|
|
intervalMinutes: settings.intervalMinutes,
|
|
firstDelayMs,
|
|
recoveryDelayMs: RESTART_RECOVERY_DELAY_MS,
|
|
},
|
|
'Network map scheduler started (first run in %d sec, then every %d min)',
|
|
Math.round(firstDelayMs / 1000),
|
|
settings.intervalMinutes
|
|
);
|
|
schedulerStartTimeout = setTimeout(() => {
|
|
tick('interval');
|
|
schedulerTimer = setInterval(() => tick('interval'), intervalMs);
|
|
}, firstDelayMs);
|
|
}
|
|
|
|
schedule();
|
|
}
|
|
|
|
function stopNetworkMapScheduler() {
|
|
if (schedulerStartTimeout) {
|
|
clearTimeout(schedulerStartTimeout);
|
|
schedulerStartTimeout = null;
|
|
}
|
|
if (schedulerTimer) {
|
|
clearInterval(schedulerTimer);
|
|
schedulerTimer = null;
|
|
}
|
|
activeIntervalMs = null;
|
|
nextScheduledRunAt = null;
|
|
persistNextRunAt(null);
|
|
}
|
|
|
|
function getLogs() {
|
|
return logBuffer.getLines();
|
|
}
|
|
|
|
function getLogMaxLines() {
|
|
return logBuffer.maxLines;
|
|
}
|
|
|
|
function setLogMaxLines(n) {
|
|
const num = Math.max(100, Math.min(2000, parseInt(n, 10) || 500));
|
|
logBuffer.setMaxLines(num);
|
|
}
|
|
|
|
/** Добавить строку в лог (для вывода ошибок из роутов) */
|
|
function pushLog(msg) {
|
|
logBuffer.push(msg);
|
|
}
|
|
|
|
/**
|
|
* Время следующего запуска по расписанию (timestamp или null если выключено).
|
|
* @param {object} settings - результат getSettingsFromUiSettings()
|
|
* @returns {number|null}
|
|
*/
|
|
function getNextRunAt(settings) {
|
|
if (!settings || !settings.enabled) return null;
|
|
if (typeof nextScheduledRunAt === 'number' && Number.isFinite(nextScheduledRunAt)) {
|
|
return nextScheduledRunAt;
|
|
}
|
|
if (typeof lastScheduledRunAt === 'number' && Number.isFinite(lastScheduledRunAt)) {
|
|
const intervalMs = activeIntervalMs || (settings.intervalMinutes || DEFAULT_INTERVAL_MINUTES) * 60 * 1000;
|
|
return lastScheduledRunAt + intervalMs;
|
|
}
|
|
if (typeof settings?.persistedNextRunAt === 'number' && Number.isFinite(settings.persistedNextRunAt)) {
|
|
const now = Date.now();
|
|
if (settings.persistedNextRunAt > now) return settings.persistedNextRunAt;
|
|
return now + RESTART_RECOVERY_DELAY_MS;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
module.exports = {
|
|
runJob,
|
|
initNetworkMapScheduler,
|
|
stopNetworkMapScheduler,
|
|
getLogs,
|
|
getLogMaxLines,
|
|
setLogMaxLines,
|
|
pushLog,
|
|
getSettingsFromUiSettings,
|
|
getNextRunAt,
|
|
loadUiSettings: loadUiSettings,
|
|
buildConnections,
|
|
LogBuffer,
|
|
};
|