mirror of
https://github.com/whyour/qinglong.git
synced 2026-08-13 04:02:56 +08:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
855f591992 | ||
|
|
d30eb2008c | ||
|
|
e1ce0f3fa9 | ||
|
|
82514e65e1 | ||
|
|
31261190f0 |
@@ -206,6 +206,7 @@ export default (app: Router) => {
|
|||||||
}),
|
}),
|
||||||
}),
|
}),
|
||||||
async (req: Request, res: Response, next: NextFunction) => {
|
async (req: Request, res: Response, next: NextFunction) => {
|
||||||
|
const logger: Logger = Container.get('logger');
|
||||||
try {
|
try {
|
||||||
let { filename, content, path } = req.body as {
|
let { filename, content, path } = req.body as {
|
||||||
filename: string;
|
filename: string;
|
||||||
@@ -223,6 +224,7 @@ export default (app: Router) => {
|
|||||||
await writeFileWithLock(filePath, content);
|
await writeFileWithLock(filePath, content);
|
||||||
return res.send({ code: 200 });
|
return res.send({ code: 200 });
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
|
logger.error('🔥 error saving script: %o', e);
|
||||||
return next(e);
|
return next(e);
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -16,6 +16,17 @@ export class HttpServerService {
|
|||||||
metricsService.record('http_service_start', 1, {
|
metricsService.record('http_service_start', 1, {
|
||||||
port: port.toString(),
|
port: port.toString(),
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Set server timeouts to prevent premature connection drops
|
||||||
|
if (this.server) {
|
||||||
|
// Timeout for receiving the entire request (including body) - 5 minutes
|
||||||
|
this.server.requestTimeout = 300000;
|
||||||
|
// Timeout for headers - 2 minutes
|
||||||
|
this.server.headersTimeout = 120000;
|
||||||
|
// Keep-alive timeout - 65 seconds (slightly more than typical load balancer timeout)
|
||||||
|
this.server.keepAliveTimeout = 65000;
|
||||||
|
}
|
||||||
|
|
||||||
resolve(this.server);
|
resolve(this.server);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
+2
-28
@@ -14,7 +14,6 @@ import {
|
|||||||
import config from '../config';
|
import config from '../config';
|
||||||
import { credentials } from '@grpc/grpc-js';
|
import { credentials } from '@grpc/grpc-js';
|
||||||
import { ApiClient } from '../protos/api';
|
import { ApiClient } from '../protos/api';
|
||||||
import { CrontabModel } from '../data/cron';
|
|
||||||
|
|
||||||
class TaskLimit {
|
class TaskLimit {
|
||||||
private dependenyLimit = new PQueue({ concurrency: 1 });
|
private dependenyLimit = new PQueue({ concurrency: 1 });
|
||||||
@@ -132,38 +131,13 @@ class TaskLimit {
|
|||||||
let runs = this.queuedCrons.get(cron.id);
|
let runs = this.queuedCrons.get(cron.id);
|
||||||
const result = runs?.length ? [...runs, fn] : [fn];
|
const result = runs?.length ? [...runs, fn] : [fn];
|
||||||
const repeatTimes = this.repeatCronNotifyMap.get(cron.id) || 0;
|
const repeatTimes = this.repeatCronNotifyMap.get(cron.id) || 0;
|
||||||
|
if (result?.length > 5) {
|
||||||
// Check instance mode from database to determine queue limit
|
|
||||||
let maxQueueSize = 10; // Default for multi-instance mode (increased from 5)
|
|
||||||
let isSingleInstanceMode = false;
|
|
||||||
try {
|
|
||||||
const cronRecord = await CrontabModel.findOne({
|
|
||||||
where: { id: Number(cron.id) },
|
|
||||||
});
|
|
||||||
|
|
||||||
// Default to single instance mode (0) for backward compatibility
|
|
||||||
// allow_multiple_instances is 1 for multi-instance, 0 or null/undefined for single instance
|
|
||||||
isSingleInstanceMode = cronRecord?.allow_multiple_instances !== 1;
|
|
||||||
|
|
||||||
if (isSingleInstanceMode) {
|
|
||||||
// For single instance mode, allow up to 2 queued tasks
|
|
||||||
// This allows the new task to be queued while the old one is being killed
|
|
||||||
maxQueueSize = 2;
|
|
||||||
}
|
|
||||||
} catch (error) {
|
|
||||||
Logger.error(
|
|
||||||
`[schedule][检查实例模式失败] 任务ID: ${cron.id}, 错误: ${error}`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (result?.length > maxQueueSize) {
|
|
||||||
if (repeatTimes < 3) {
|
if (repeatTimes < 3) {
|
||||||
this.repeatCronNotifyMap.set(cron.id, repeatTimes + 1);
|
this.repeatCronNotifyMap.set(cron.id, repeatTimes + 1);
|
||||||
const modeStr = isSingleInstanceMode ? '单实例' : '多实例';
|
|
||||||
this.client.systemNotify(
|
this.client.systemNotify(
|
||||||
{
|
{
|
||||||
title: '任务重复运行',
|
title: '任务重复运行',
|
||||||
content: `任务:${cron.name}(${modeStr}模式),命令:${cron.command},定时:${cron.schedule},处于运行中的超过 ${maxQueueSize} 个,请检查定时设置`,
|
content: `任务:${cron.name},命令:${cron.command},定时:${cron.schedule},处于运行中的超过 5 个,请检查定时设置`,
|
||||||
},
|
},
|
||||||
(err, res) => {
|
(err, res) => {
|
||||||
if (err) {
|
if (err) {
|
||||||
|
|||||||
+3
-27
@@ -15,12 +15,11 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
|
|||||||
});
|
});
|
||||||
|
|
||||||
// Default to single instance mode (0) for backward compatibility
|
// Default to single instance mode (0) for backward compatibility
|
||||||
// allow_multiple_instances is 1 for multi-instance, 0 or null/undefined for single instance
|
const allowSingleInstances =
|
||||||
const isSingleInstanceMode =
|
existingCron?.allow_multiple_instances === 0;
|
||||||
existingCron?.allow_multiple_instances !== 1;
|
|
||||||
|
|
||||||
if (
|
if (
|
||||||
isSingleInstanceMode &&
|
allowSingleInstances &&
|
||||||
existingCron &&
|
existingCron &&
|
||||||
existingCron.pid &&
|
existingCron.pid &&
|
||||||
(existingCron.status === CrontabStatus.running ||
|
(existingCron.status === CrontabStatus.running ||
|
||||||
@@ -50,18 +49,6 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
|
|||||||
);
|
);
|
||||||
const cp = spawn(cmd, { shell: '/bin/bash' });
|
const cp = spawn(cmd, { shell: '/bin/bash' });
|
||||||
|
|
||||||
// Update status to running after spawning the process
|
|
||||||
try {
|
|
||||||
await CrontabModel.update(
|
|
||||||
{ status: CrontabStatus.running, pid: cp.pid },
|
|
||||||
{ where: { id: Number(cron.id) } },
|
|
||||||
);
|
|
||||||
} catch (error) {
|
|
||||||
Logger.error(
|
|
||||||
`[schedule][更新任务状态失败] 任务ID: ${cron.id}, 错误: ${error}`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
cp.stderr.on('data', (data) => {
|
cp.stderr.on('data', (data) => {
|
||||||
Logger.info(
|
Logger.info(
|
||||||
'[schedule][执行任务失败] 命令: %s, 错误信息: %j',
|
'[schedule][执行任务失败] 命令: %s, 错误信息: %j',
|
||||||
@@ -79,17 +66,6 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
|
|||||||
|
|
||||||
cp.on('exit', async (code) => {
|
cp.on('exit', async (code) => {
|
||||||
taskLimit.removeQueuedCron(cron.id);
|
taskLimit.removeQueuedCron(cron.id);
|
||||||
// Update status to idle after task completes
|
|
||||||
try {
|
|
||||||
await CrontabModel.update(
|
|
||||||
{ status: CrontabStatus.idle, pid: undefined },
|
|
||||||
{ where: { id: Number(cron.id) } },
|
|
||||||
);
|
|
||||||
} catch (error) {
|
|
||||||
Logger.error(
|
|
||||||
`[schedule][更新任务状态失败] 任务ID: ${cron.id}, 错误: ${error}`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
Logger.info(
|
Logger.info(
|
||||||
'[schedule][执行任务结束] 参数: %s, 退出码: %j',
|
'[schedule][执行任务结束] 参数: %s, 退出码: %j',
|
||||||
JSON.stringify({
|
JSON.stringify({
|
||||||
|
|||||||
+45
-7
@@ -1,8 +1,9 @@
|
|||||||
import { lock } from 'proper-lockfile';
|
import { lock } from 'proper-lockfile';
|
||||||
import os from 'os';
|
import os from 'os';
|
||||||
import path from 'path';
|
import path from 'path';
|
||||||
import { writeFile, open, chmod } from 'fs/promises';
|
import { writeFile, open, chmod, FileHandle } from 'fs/promises';
|
||||||
import { fileExist } from '../config/util';
|
import { fileExist } from '../config/util';
|
||||||
|
import Logger from '../loaders/logger';
|
||||||
|
|
||||||
function getUniqueLockPath(filePath: string) {
|
function getUniqueLockPath(filePath: string) {
|
||||||
const sanitizedPath = filePath
|
const sanitizedPath = filePath
|
||||||
@@ -19,13 +20,32 @@ export async function writeFileWithLock(
|
|||||||
if (typeof options === 'string') {
|
if (typeof options === 'string') {
|
||||||
options = { encoding: options };
|
options = { encoding: options };
|
||||||
}
|
}
|
||||||
if (!(await fileExist(filePath))) {
|
|
||||||
const fileHandle = await open(filePath, 'w');
|
|
||||||
fileHandle.close();
|
|
||||||
}
|
|
||||||
const lockfilePath = getUniqueLockPath(filePath);
|
|
||||||
|
|
||||||
const release = await lock(filePath, {
|
// Ensure file exists before locking
|
||||||
|
if (!(await fileExist(filePath))) {
|
||||||
|
let fileHandle: FileHandle | undefined;
|
||||||
|
try {
|
||||||
|
fileHandle = await open(filePath, 'w');
|
||||||
|
} catch (error) {
|
||||||
|
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||||
|
throw new Error(`Failed to create file ${filePath}: ${errorMessage}`);
|
||||||
|
} finally {
|
||||||
|
if (fileHandle !== undefined) {
|
||||||
|
try {
|
||||||
|
await fileHandle.close();
|
||||||
|
} catch (closeError) {
|
||||||
|
// Log close error but don't throw to avoid masking the original error
|
||||||
|
Logger.error(`Failed to close file handle for ${filePath}:`, closeError);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const lockfilePath = getUniqueLockPath(filePath);
|
||||||
|
let release: (() => Promise<void>) | undefined;
|
||||||
|
|
||||||
|
try {
|
||||||
|
release = await lock(filePath, {
|
||||||
retries: {
|
retries: {
|
||||||
retries: 10,
|
retries: 10,
|
||||||
factor: 2,
|
factor: 2,
|
||||||
@@ -34,9 +54,27 @@ export async function writeFileWithLock(
|
|||||||
},
|
},
|
||||||
lockfilePath,
|
lockfilePath,
|
||||||
});
|
});
|
||||||
|
} catch (error) {
|
||||||
|
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||||
|
throw new Error(`Failed to acquire lock for ${filePath}: ${errorMessage}`);
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
await writeFile(filePath, content, { encoding: 'utf8', ...options });
|
await writeFile(filePath, content, { encoding: 'utf8', ...options });
|
||||||
if (options?.mode) {
|
if (options?.mode) {
|
||||||
await chmod(filePath, options.mode);
|
await chmod(filePath, options.mode);
|
||||||
}
|
}
|
||||||
|
} catch (error) {
|
||||||
|
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||||
|
throw new Error(`Failed to write to file ${filePath}: ${errorMessage}`);
|
||||||
|
} finally {
|
||||||
|
if (release) {
|
||||||
|
try {
|
||||||
await release();
|
await release();
|
||||||
|
} catch (error) {
|
||||||
|
// Log but don't throw on release failure
|
||||||
|
Logger.error(`Failed to release lock for ${filePath}:`, error);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user