"use strict"; /** * 巡检调度引擎 * * 职责: * 1. 定时扫描启用了巡检的设备 * 2. 根据各设备的巡检周期,判断是否需要执行 * 3. 并发执行多台设备的巡检 * 4. 采集指标 → 判断告警 → 写入结果和告警记录 * * 设计说明: * - 每 30 秒轮询一次(检查哪些设备到巡检时间了) * - 并发数默认 5,可配置 * - 单设备超时默认 30 秒 * - 巡检结果通过回调通知上层(更新 UI 用) */ const deviceStore = require("./deviceStore.cjs"); const alertRuleStore = require("./alertRuleStore.cjs"); const inspectionResultStore = require("./inspectionResultStore.cjs"); const alertRecordStore = require("./alertRecordStore.cjs"); const adapterRegistry = require("./adapterRegistry.cjs"); const { INSPECTION_STATUS } = require("./types.cjs"); const DEFAULT_POLL_INTERVAL_MS = 30 * 1000; // 30 秒轮询一次 const DEFAULT_CONCURRENCY = 5; const DEFAULT_DEVICE_TIMEOUT_MS = 30 * 1000; class InspectionScheduler { /** * @param {object} options * @param {function} options.readFn - 读取存储 (key) => string * @param {function} options.writeFn - 写入存储 (key, value) => void * @param {function} options.getConnectionFn - 获取 SSH 连接 (hostId) => connection 或 null * @param {function} options.execFn - 执行命令 (hostId, command) => Promise * @param {function} [options.onResult] - 单设备巡检完成回调 * @param {number} [options.pollIntervalMs] - 轮询间隔 * @param {number} [options.concurrency] - 并发数 * @param {number} [options.timeoutMs] - 单设备超时 */ constructor(options) { this.readFn = options.readFn; this.writeFn = options.writeFn; this.getConnectionFn = options.getConnectionFn; this.execFn = options.execFn; this.onResult = options.onResult || (() => {}); this.onAlertChanged = options.onAlertChanged || (() => {}); this.pollIntervalMs = options.pollIntervalMs || DEFAULT_POLL_INTERVAL_MS; this.concurrency = options.concurrency || DEFAULT_CONCURRENCY; this.timeoutMs = options.timeoutMs || DEFAULT_DEVICE_TIMEOUT_MS; // hostId → { hostname, label } 映射,用于告警记录显示设备名称 this._hostInfo = options.hostInfo || {}; this._timer = null; this._running = false; this._scanQueue = []; // 待巡检的设备队列 this._activeScans = 0; // 当前正在执行的巡检数 } // ---------- 生命周期 ---------- start() { if (this._running) return; this._running = true; console.log("[InspectionScheduler] 巡检调度器已启动"); // 启动时立即跑一轮(异步,不阻塞 start 返回) setImmediate(() => this._tick()); } stop() { this._running = false; if (this._timer) { clearTimeout(this._timer); this._timer = null; } console.log("[InspectionScheduler] 巡检调度器已停止"); } isRunning() { return this._running; } _scheduleNextTick() { if (!this._running) return; this._timer = setTimeout(() => this._tick(), this.pollIntervalMs); } // ---------- 主循环 ---------- async _tick() { if (!this._running) return; try { // 1. 获取所有启用巡检的设备 const devices = deviceStore.getEnabledDevices(this.readFn); // 2. 筛选出到期需要巡检的设备 const now = Date.now(); const dueDevices = devices.filter((dev) => { const intervalMs = dev.effective_interval * 60 * 1000; const lastScan = dev.last_scan_at || 0; return now - lastScan >= intervalMs; }); if (dueDevices.length > 0) { console.log(`[InspectionScheduler] 到期设备 ${dueDevices.length} 台,加入巡检队列`); // 加入队列 for (const dev of dueDevices) { if (!this._scanQueue.find((d) => d.host_id === dev.host_id)) { this._scanQueue.push(dev); } } // 启动并发执行 this._processQueue(); } } catch (e) { console.error("[InspectionScheduler] 调度异常:", e.message); } // 3. 安排下一次轮询 this._scheduleNextTick(); } _processQueue() { while (this._activeScans < this.concurrency && this._scanQueue.length > 0) { const device = this._scanQueue.shift(); this._activeScans++; this._scanDevice(device).finally(() => { this._activeScans--; // 继续处理队列 this._processQueue(); }); } } // ---------- 单设备巡检 ---------- async _scanDevice(deviceExt) { const startTime = Date.now(); const hostId = deviceExt.host_id; try { // 1. 获取设备类型信息 const catalog = deviceStore.getCatalogById(deviceExt.catalog_id); if (!catalog) { throw new Error(`未找到设备类型: ${deviceExt.catalog_id}`); } // 2. 创建采集器 const adapter = adapterRegistry.createAdapter( catalog.adapter_class, { host: hostId }, { execFn: async (cmd) => { if (!this.execFn) throw new Error("execFn 未配置"); return this.execFn(hostId, cmd); }, timeoutMs: this.timeoutMs, } ); if (!adapter) { throw new Error(`采集器未注册: ${catalog.adapter_class}`); } // 3. 执行采集(带超时) const result = await this._withTimeout( adapter.collect(), this.timeoutMs, "巡检超时" ); const durationMs = Date.now() - startTime; // 4. 保存巡检结果(带上 catalog_id 供后续 RCA 根因分析用) const record = inspectionResultStore.addResult( this.readFn, this.writeFn, { device_id: hostId, catalog_id: deviceExt.catalog_id || catalog.id || null, online: result.online, status: result.status, duration_ms: durationMs, metrics: result.metrics, alerts: [], error_message: result.error, } ); // 5. 更新设备最后巡检状态 deviceStore.updateLastScan( this.readFn, this.writeFn, hostId, result.status, result.error ); // 6. 判断告警 const triggeredAlerts = this._evaluateAlerts(deviceExt, catalog, result.metrics); record.alerts = triggeredAlerts; if (triggeredAlerts.length > 0) this.onAlertChanged(); // 7. 回调通知 this.onResult({ device_id: hostId, result: record, alerts: triggeredAlerts, }); return record; } catch (e) { const durationMs = Date.now() - startTime; // 失败也要记录 const record = inspectionResultStore.addResult( this.readFn, this.writeFn, { device_id: hostId, online: false, status: e.message.includes("超时") ? INSPECTION_STATUS.TIMEOUT : INSPECTION_STATUS.FAILED, duration_ms: durationMs, metrics: {}, alerts: [], error_message: e.message, } ); deviceStore.updateLastScan( this.readFn, this.writeFn, hostId, e.message.includes("超时") ? INSPECTION_STATUS.TIMEOUT : INSPECTION_STATUS.FAILED, e.message ); // 执行错误告警 —— 统一归类为 rule_offline(与设备离线告警合并) // 不再额外调 _evaluateAlerts(online: false) 避免重复创建同规则告警 const failStatus = e.message.includes("超时") ? "timeout" : "failed"; alertRecordStore.createExecutionErrorAlert( this.readFn, this.writeFn, { device_id: hostId, device_name: deviceExt.label || deviceExt.hostname || hostId, catalog_id: deviceExt.catalog_id, }, failStatus, e.message || "巡检执行失败" ); this.onAlertChanged(); this.onResult({ device_id: hostId, result: record, alerts: [], error: e.message, }); return record; } } // ---------- 告警判断 ---------- _evaluateAlerts(deviceExt, catalog, metrics) { const triggered = []; try { // 获取该设备适用的规则 const rules = alertRuleStore.getRulesForDevice(this.readFn, deviceExt); const hostInfo = this._hostInfo[deviceExt.host_id] || {}; const deviceName = hostInfo.label || hostInfo.hostname || deviceExt.host_id; console.log(`[InspectionScheduler] 告警评估 device=${deviceName}(${deviceExt.host_id}) online=${metrics.online} 规则数=${rules.length}`); for (const rule of rules) { // 特殊处理:online 指标 if (rule.metric === "online") { const isOnline = metrics.online !== false; // threshold 为 "false" 表示离线时告警 const shouldAlert = rule.threshold === "false" ? !isOnline : isOnline; console.log(`[InspectionScheduler] 离线规则: name=${rule.name} enabled=${rule.enabled} threshold=${rule.threshold} isOnline=${isOnline} shouldAlert=${shouldAlert}`); const alertResult = alertRecordStore.processAlert( this.readFn, this.writeFn, { device_id: deviceExt.host_id, device_name: deviceName }, rule, isOnline ? "在线" : "离线", shouldAlert ); if (alertResult) { console.log(`[InspectionScheduler] 告警触发:`, JSON.stringify(alertResult)); triggered.push(alertResult); } continue; } // 普通数值指标 const value = this._getMetricValue(metrics, rule.metric); if (value === null || value === undefined) continue; const isTriggered = alertRuleStore.evaluateRule(value, rule.operator, rule.threshold); const hostInfo = this._hostInfo[deviceExt.host_id] || {}; const deviceName = hostInfo.label || hostInfo.hostname || deviceExt.host_id; const alertResult = alertRecordStore.processAlert( this.readFn, this.writeFn, { device_id: deviceExt.host_id, device_name: deviceName }, rule, value, isTriggered ); if (alertResult) triggered.push(alertResult); } } catch (e) { console.error("[InspectionScheduler] 告警判断异常:", e.message); } return triggered; } _getMetricValue(metrics, metricKey) { if (!metrics) return null; // 直接取 if (metrics[metricKey] !== undefined) return metrics[metricKey]; // 磁盘最高使用率 if (metricKey === "disk_usage" && Array.isArray(metrics.disk_usage)) { return Math.max(...metrics.disk_usage.map((d) => d.usage_percent || 0)); } return null; } // ---------- 手动巡检 ---------- /** * 手动触发单设备巡检 */ async scanDeviceNow(hostId) { const ext = deviceStore.getExtension(this.readFn, hostId); if (!ext) throw new Error(`设备未配置巡检: ${hostId}`); const interval = deviceStore.getEffectiveInterval(this.readFn, hostId); const device = { ...ext, effective_interval: interval }; // 加入队列头部 this._scanQueue.unshift(device); this._processQueue(); return { queued: true, host_id: hostId }; } /** * 手动触发所有启用巡检的设备 */ async scanAllNow() { const devices = deviceStore.getEnabledDevices(this.readFn); let count = 0; for (const dev of devices) { if (!this._scanQueue.find((d) => d.host_id === dev.host_id)) { this._scanQueue.push(dev); count++; } } this._processQueue(); return { queued: count }; } // ---------- 工具 ---------- _withTimeout(promise, ms, message) { return new Promise((resolve, reject) => { const timer = setTimeout(() => reject(new Error(message || "超时")), ms); promise.then( (v) => { clearTimeout(timer); resolve(v); }, (e) => { clearTimeout(timer); reject(e); } ); }); } getStatus() { return { running: this._running, queue_size: this._scanQueue.length, active_scans: this._activeScans, concurrency: this.concurrency, }; } } module.exports = InspectionScheduler;