feat(upgrade): 支持批量升级主机并优化错误处理
- 新增 UPGRADE_BATCH_SIZE 环境变量，允许配置批量升级的主机数量 - 重构 upgradeController.js，将单主机循环改为按批次处理 - 批量触发升级和查询状态，减少 API 调用次数 - 改进错误处理，确保批次中每个主机的失败都能独立记录日志 - 使用 pendingHosts 集合跟踪批次内未完成的主机，避免重复轮询
This commit is contained in:
@@ -27,6 +27,9 @@ const parseUpgradeConfig = (configStr) => {
|
||||
}
|
||||
};
|
||||
|
||||
const upgradeBatchSizeRaw = Number(process.env.UPGRADE_BATCH_SIZE || 1);
|
||||
const upgradeBatchSize = Number.isFinite(upgradeBatchSizeRaw) ? upgradeBatchSizeRaw : 1;
|
||||
|
||||
module.exports = {
|
||||
port: process.env.PORT || 3000,
|
||||
db: {
|
||||
@@ -42,5 +45,6 @@ module.exports = {
|
||||
upgradeWaitSeconds: Number(process.env.UPGRADE_WAIT_SECONDS || 45),
|
||||
upgradePollIntervalSeconds: Number(process.env.UPGRADE_POLL_INTERVAL_SECONDS || 45),
|
||||
upgradePollTimeoutSeconds: Number(process.env.UPGRADE_POLL_TIMEOUT_SECONDS || 300),
|
||||
upgradeBatchSize,
|
||||
upgradeConfig: parseUpgradeConfig(process.env.UPGRADE_CONFIG)
|
||||
};
|
||||
|
||||
@@ -36,22 +36,44 @@ const processGroup = async (group, groupIdx) => {
|
||||
}
|
||||
|
||||
const hostQueue = group.hosts.map(String);
|
||||
const batchSize = Number(config.upgradeBatchSize ?? 1);
|
||||
const effectiveBatchSize = batchSize <= 0 ? hostQueue.length : Math.max(1, batchSize);
|
||||
console.log(`[${stateKey}] Starting queue. Roomtype: ${roomtype_id}. File: ${fileName}. Hosts: ${hostQueue.join(', ')}`);
|
||||
|
||||
const waitSeconds = config.upgradeWaitSeconds || 45;
|
||||
const pollIntervalSeconds = config.upgradePollIntervalSeconds || 45;
|
||||
const pollTimeoutSeconds = config.upgradePollTimeoutSeconds || 300;
|
||||
|
||||
for (const hostId of hostQueue) {
|
||||
for (let i = 0; i < hostQueue.length; i += effectiveBatchSize) {
|
||||
const batchHosts = hostQueue.slice(i, i + effectiveBatchSize);
|
||||
const sessionUuid = uuidv4();
|
||||
const startTime = new Date();
|
||||
|
||||
console.log(`[${stateKey}] Triggering upgrade for host ${hostId}. Roomtype: ${roomtype_id}. File: ${fileName}.`);
|
||||
console.log(`[${stateKey}] Triggering upgrade for hosts ${batchHosts.join(', ')}. Roomtype: ${roomtype_id}. File: ${fileName}.`);
|
||||
|
||||
try {
|
||||
const upgradeRes = await apiClient.triggerUpgrade(roomtype_id, [hostId], fileName);
|
||||
const upgradeRes = await apiClient.triggerUpgrade(roomtype_id, batchHosts, fileName);
|
||||
if (!upgradeRes.IsSuccess) {
|
||||
console.error(`[${stateKey}] Upgrade trigger failed for host ${hostId}: ${upgradeRes.Message}`);
|
||||
console.error(`[${stateKey}] Upgrade trigger failed for hosts ${batchHosts.join(', ')}: ${upgradeRes.Message}`);
|
||||
for (const hostId of batchHosts) {
|
||||
await loggerService.logHostResult({
|
||||
uuid: sessionUuid,
|
||||
start_time: startTime.toISOString(),
|
||||
roomtype_id: roomtype_id,
|
||||
host_str: hostId,
|
||||
filename: fileName,
|
||||
status: '触发失败',
|
||||
end_time: new Date().toISOString(),
|
||||
file_type: '',
|
||||
config_version: '',
|
||||
firmware_version: ''
|
||||
});
|
||||
}
|
||||
continue;
|
||||
}
|
||||
} catch (e) {
|
||||
console.error(`[${stateKey}] Upgrade trigger error for hosts ${batchHosts.join(', ')}:`, e.message);
|
||||
for (const hostId of batchHosts) {
|
||||
await loggerService.logHostResult({
|
||||
uuid: sessionUuid,
|
||||
start_time: startTime.toISOString(),
|
||||
@@ -64,37 +86,25 @@ const processGroup = async (group, groupIdx) => {
|
||||
config_version: '',
|
||||
firmware_version: ''
|
||||
});
|
||||
continue;
|
||||
}
|
||||
} catch (e) {
|
||||
console.error(`[${stateKey}] Upgrade trigger error for host ${hostId}:`, e.message);
|
||||
await loggerService.logHostResult({
|
||||
uuid: sessionUuid,
|
||||
start_time: startTime.toISOString(),
|
||||
roomtype_id: roomtype_id,
|
||||
host_str: hostId,
|
||||
filename: fileName,
|
||||
status: '触发失败',
|
||||
end_time: new Date().toISOString(),
|
||||
file_type: '',
|
||||
config_version: '',
|
||||
firmware_version: ''
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
console.log(`[${stateKey}] Waiting ${waitSeconds}s for host ${hostId}...`);
|
||||
console.log(`[${stateKey}] Waiting ${waitSeconds}s for hosts ${batchHosts.join(', ')}...`);
|
||||
await sleep(waitSeconds * 1000);
|
||||
|
||||
const pollStartTime = Date.now();
|
||||
let completed = false;
|
||||
const pendingHosts = new Set(batchHosts.map(String));
|
||||
|
||||
while (Date.now() - pollStartTime < pollTimeoutSeconds * 1000) {
|
||||
while (Date.now() - pollStartTime < pollTimeoutSeconds * 1000 && pendingHosts.size > 0) {
|
||||
try {
|
||||
const queryRes = await apiClient.queryStatus([hostId]);
|
||||
const queryRes = await apiClient.queryStatus(batchHosts);
|
||||
if (queryRes && Array.isArray(queryRes.Response)) {
|
||||
const hostStatus = queryRes.Response.find((item) => String(item.HostID) === String(hostId));
|
||||
if (hostStatus) {
|
||||
for (const hostStatus of queryRes.Response) {
|
||||
const hostId = String(hostStatus.HostID);
|
||||
if (!pendingHosts.has(hostId)) {
|
||||
continue;
|
||||
}
|
||||
const rawStatus = hostStatus.Upgrade_status || '';
|
||||
const status = rawStatus.trim();
|
||||
if (['升级完成', '超时失败', '升级失败'].includes(status)) {
|
||||
@@ -111,34 +121,35 @@ const processGroup = async (group, groupIdx) => {
|
||||
firmware_version: hostStatus.Version
|
||||
});
|
||||
console.log(`[${stateKey}] Host ${hostId} logged to DB with status: '${status}'.`);
|
||||
completed = true;
|
||||
break;
|
||||
pendingHosts.delete(hostId);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
console.warn(`[${stateKey}] Invalid query response for host ${hostId}:`, queryRes);
|
||||
console.warn(`[${stateKey}] Invalid query response for hosts ${batchHosts.join(', ')}:`, queryRes);
|
||||
}
|
||||
} catch (e) {
|
||||
console.error(`[${stateKey}] Poll error for host ${hostId}:`, e.message);
|
||||
console.error(`[${stateKey}] Poll error for hosts ${batchHosts.join(', ')}:`, e.message);
|
||||
}
|
||||
|
||||
await sleep(pollIntervalSeconds * 1000);
|
||||
}
|
||||
|
||||
if (!completed) {
|
||||
await loggerService.logHostResult({
|
||||
uuid: sessionUuid,
|
||||
start_time: startTime.toISOString(),
|
||||
roomtype_id: roomtype_id,
|
||||
host_str: hostId,
|
||||
filename: fileName,
|
||||
status: '超时失败',
|
||||
end_time: new Date().toISOString(),
|
||||
file_type: 'Unknown',
|
||||
config_version: '',
|
||||
firmware_version: ''
|
||||
});
|
||||
console.log(`[${stateKey}] Host ${hostId} timeout failed.`);
|
||||
if (pendingHosts.size > 0) {
|
||||
for (const hostId of pendingHosts) {
|
||||
await loggerService.logHostResult({
|
||||
uuid: sessionUuid,
|
||||
start_time: startTime.toISOString(),
|
||||
roomtype_id: roomtype_id,
|
||||
host_str: hostId,
|
||||
filename: fileName,
|
||||
status: '超时失败',
|
||||
end_time: new Date().toISOString(),
|
||||
file_type: 'Unknown',
|
||||
config_version: '',
|
||||
firmware_version: ''
|
||||
});
|
||||
console.log(`[${stateKey}] Host ${hostId} timeout failed.`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user