Skip to content

Harvey / job-executor

转到一个项目
切换导航栏
切换导航栏固定状态
Commit aa2bfd71 由 Harvey
选项

no message

1 个父辈 49e7c073
内嵌 并排
正在显示 4 个修改的文件 包含 324 行增加 和 150 行删除
build/xxl_job/executor.js
...@@ -80,8 +80,8 @@ class Executor { ...@@ -80,8 +80,8 @@ class Executor {
res.send(this.killJob(req.body.jobId || -1)); res.send(this.killJob(req.body.jobId || -1));
}); });
router.post(`${baseUri}/log`, async (req, res, next) => { router.post(`${baseUri}/log`, async (req, res, next) => {
const { logDateTim: logDateTime, logId, fromLineNum } = req.body || {}; const { logDateTim, logId, fromLineNum } = req.body || {};
const data = await this.readLog(logDateTime, logId, fromLineNum); const data = await this.readLog(logDateTim, logId, fromLineNum);
res.send(data); res.send(data);
}); });
} }
...@@ -134,23 +134,47 @@ class Executor { ...@@ -134,23 +134,47 @@ class Executor {
* @return {*} - fromLineNum:日志开始行号; toLineNum:日志结束行号; logContent:日志内容 * @return {*} - fromLineNum:日志开始行号; toLineNum:日志结束行号; logContent:日志内容
*/ */
async readLog(logDateTime, logId, fromLineNum) { async readLog(logDateTime, logId, fromLineNum) {
let logContent; //待实现
let toLineNum; return {
try { code: 200, content: {
const lines = await this.jobManager.readJobLog(logDateTime, logId); "fromLineNum": 0,
lines.splice(0, fromLineNum - 1); "toLineNum": 100,
if (last(lines) === '') "logContent": "test",
lines.pop(); "isEnd": true // 日志是否全部加载完
toLineNum = fromLineNum + lines.length - 1; }
lines.unshift(''); };
logContent = lines.join('\n'); //let logContent
} //let toLineNum
catch (err) { // try {
log.err('readLog error: %o', err.message); // const lines = await this.jobManager.readJobLog(logDateTime, logId)
toLineNum = fromLineNum; // lines.splice(0, fromLineNum - 1)
logContent = err.toString(); // if (last(lines) === '') lines.pop()
} // toLineNum = fromLineNum + lines.length - 1
return { code: 200, content: { fromLineNum, toLineNum, logContent } }; // lines.unshift('')
// logContent = lines.join('\n')
// } catch (err) {
// log.err('readLog error: %o', err.message)
// toLineNum = fromLineNum
// logContent = err.toString()
// }
// 请求数据格式如下，放置在 RequestBody 中，JSON格式：
// {
// "logDateTim":0, // 本次调度日志时间
// "logId":0, // 本次调度日志ID
// "fromLineNum":0 // 日志开始行号，滚动加载日志
// }
// 响应数据格式：
// {
// "code":200, // 200 表示正常、其他失败
// "msg": null // 错误提示消息
// "content":{
// "fromLineNum":0, // 本次请求，日志开始行数
// "toLineNum":100, // 本次请求，日志结束行号
// "logContent":"xxx", // 本次请求日志内容
// "isEnd":true // 日志是否全部加载完
// }
// }
// return { code: 200, content: { fromLineNum, toLineNum, logContent } }
} }
/** /**
* 执行器注册：执行器注册时使用，调度中心会实时感知注册成功的执行器并发起任务调度 * 执行器注册：执行器注册时使用，调度中心会实时感知注册成功的执行器并发起任务调度
......
build/xxl_job/logger.js
const debug = require('debug'); "use strict";
const fs = require('fs'); Object.defineProperty(exports, "__esModule", { value: true });
const os = require('os'); exports.createLogger = void 0;
const util = require('util'); const debug_1 = require("debug");
const { always, propOr } = require('./purefuncs'); const fs_1 = require("fs");
const enableExecutorDebugLog = /^(yes|on|true|enable|enabled|1)$/i.test(`${propOr(false, 'XXL_JOB_DEBUG_LOG', process.env)}`); const os_1 = require("os");
const enableLogLevels = propOr('info:*,warn:*,error:*,debug:*,trace:*', 'DEBUG', process.env); const util_1 = require("util");
const writeStreamOptions = { flags: 'a', encoding: 'utf8', autoClose: true, emitClose: true }; // 配置解析
const noop = always(undefined); const parseEnvBool = (envVar) => {
const noopLogger = { info: noop, err: noop, debug: noop, warn: noop, trace: noop }; const value = process.env[envVar] || '';
const dErr = debug('error'); return /^(yes|on|true|enable|enabled|1)$/i.test(value);
const dInfo = debug('info');
const dWarn = debug('warn');
const dDebug = debug('debug');
const dTrace = debug('trace');
// 自定义对象，包装 debug 模拟日志级别
const createLogger = (ns) => {
const logger = {
info: dInfo.extend(ns),
err: dErr.extend(ns),
debug: dDebug.extend(ns),
warn: dWarn.extend(ns),
trace: dTrace.extend(ns),
};
Object.values(logger).forEach((levelLogger) => Object.assign(levelLogger, { enabled: true, useColors: false }));
return logger;
}; };
module.exports = (ns, logFilePath) => { const getEnvValue = (envVar, defaultValue) => {
// 1. 执行器运行日志，输出到 stderr，限制日志级别 return process.env[envVar] !== undefined ? process.env[envVar] : defaultValue;
if (!logFilePath) { };
if (!enableExecutorDebugLog) // 默认配置
return noopLogger; const DEFAULT_LOG_LEVELS = 'info:*,warn:*,error:*,debug:*,trace:*';
const logger = createLogger(ns); const WRITE_STREAM_OPTIONS = {
Object.entries(logger).forEach(([level, levelLogger]) => levelLogger.enabled = enableLogLevels.includes(level)); flags: 'a',
return logger; encoding: 'utf8',
} autoClose: true,
// 2. 任务执行日志，同时输出到 stderr 和 文件，stderr 限制日志级别，输出到文件不限制级别以供调度中心全量查看 emitClose: true
const writeStream = fs.createWriteStream(logFilePath, writeStreamOptions); };
const log2File = (...args) => writeStream.write(`${util.format(...args)}${os.EOL}`); // 空日志函数
const log2Stderr = (...args) => console.error(util.format(...args)); const noop = (..._args) => { };
const log2FileAndStderr = (...args) => { const noopLogger = {
const content = util.format(...args); error: noop,
writeStream.write(`${content}${os.EOL}`); info: noop,
console.error(content); warn: noop,
debug: noop,
trace: noop,
close: noop
};
// 创建基础日志函数
const createBaseLoggers = (namespace) => {
return {
error: (0, debug_1.default)('error').extend(namespace),
info: (0, debug_1.default)('info').extend(namespace),
warn: (0, debug_1.default)('warn').extend(namespace),
debug: (0, debug_1.default)('debug').extend(namespace),
trace: (0, debug_1.default)('trace').extend(namespace)
}; };
const logger = createLogger(ns); };
// 设置输出 // 配置日志级别
Object.entries(logger).forEach(([level, levelLogger]) => { const configureLogLevels = (logger, enabledLevels) => {
levelLogger.log = enableLogLevels.includes(level) ? log2FileAndStderr : log2File; Object.entries(logger).forEach(([level]) => {
logger[level].enabled = enabledLevels.includes(level);
}); });
// 任务执行完成，关闭文件输出流，后续日志只输出到 stderr };
logger.close = () => { // 创建文件日志写入器
Object.entries(logger).forEach(([level, levelLogger]) => { const createFileLogger = (logFilePath, logger, enabledLevels) => {
Object.assign(levelLogger, { enabled: enableLogLevels.includes(level), log: log2Stderr }); const writeStream = fs_1.default.createWriteStream(logFilePath, WRITE_STREAM_OPTIONS);
const logToFile = (...args) => {
writeStream.write(`${util_1.default.format(...args)}${os_1.default.EOL}`);
};
const logToConsoleAndFile = (...args) => {
const message = util_1.default.format(...args);
console.error(message);
writeStream.write(`${message}${os_1.default.EOL}`);
};
// 配置日志输出方式
Object.entries(logger).forEach(([level, logFn]) => {
logFn.log = enabledLevels.includes(level) ? logToConsoleAndFile : logToFile;
}); });
// 返回完整logger对象
return {
...logger,
close: () => {
writeStream.end(); writeStream.end();
// 切换回仅控制台日志
Object.values(logger).forEach((logFn) => {
logFn.log = console.error;
});
}
}; };
return logger;
}; };
// 主导出函数
const createLogger = (options) => {
// 处理参数重载
const normalizedOptions = typeof options === 'string'
? { namespace: options }
: options;
const { namespace, logFilePath, enabledLevels = getEnvValue('DEBUG', DEFAULT_LOG_LEVELS), debugEnabled = parseEnvBool('XXL_JOB_DEBUG_LOG') } = normalizedOptions;
if (!debugEnabled && !logFilePath) {
return noopLogger;
}
const logger = createBaseLoggers(namespace);
// 配置日志级别
configureLogLevels(logger, enabledLevels);
if (!logFilePath) {
return {
...logger,
close: noop
};
}
// 文件日志配置
return createFileLogger(logFilePath, logger, enabledLevels);
};
exports.createLogger = createLogger;
// 默认导出
exports.default = exports.createLogger;
//# sourceMappingURL=logger.js.map //# sourceMappingURL=logger.js.map
\ No newline at end of file \ No newline at end of file
src/xxl_job/executor.ts
...@@ -4,7 +4,6 @@ const logger = require('./logger') ...@@ -4,7 +4,6 @@ const logger = require('./logger')
const log = logger('xxl-job-executor') const log = logger('xxl-job-executor')
import axios from 'axios'; import axios from 'axios';
export class Executor { export class Executor {
private executorKey private executorKey
private scheduleCenterUrl private scheduleCenterUrl
...@@ -97,8 +96,8 @@ export class Executor { ...@@ -97,8 +96,8 @@ export class Executor {
}) })
router.post(`${baseUri}/log`, async (req, res, next) => { router.post(`${baseUri}/log`, async (req, res, next) => {
const { logDateTim: logDateTime, logId, fromLineNum } = req.body || {} const { logDateTim, logId, fromLineNum } = req.body || {}
const data = await this.readLog(logDateTime, logId, fromLineNum) const data = await this.readLog(logDateTim, logId, fromLineNum)
res.send(data) res.send(data)
}) })
} }
...@@ -159,21 +158,52 @@ export class Executor { ...@@ -159,21 +158,52 @@ export class Executor {
* @return {*} - fromLineNum:日志开始行号; toLineNum:日志结束行号; logContent:日志内容 * @return {*} - fromLineNum:日志开始行号; toLineNum:日志结束行号; logContent:日志内容
*/ */
async readLog(logDateTime, logId, fromLineNum) { async readLog(logDateTime, logId, fromLineNum) {
let logContent //待实现
let toLineNum return {
try { code: 200, content: {
const lines = await this.jobManager.readJobLog(logDateTime, logId) "fromLineNum": 0, // 本次请求，日志开始行数
lines.splice(0, fromLineNum - 1) "toLineNum": 100, // 本次请求，日志结束行号
if (last(lines) === '') lines.pop() "logContent": "test", // 本次请求日志内容
toLineNum = fromLineNum + lines.length - 1 "isEnd": true // 日志是否全部加载完
lines.unshift('') }
logContent = lines.join('\n') }
} catch (err) {
log.err('readLog error: %o', err.message) //let logContent
toLineNum = fromLineNum //let toLineNum
logContent = err.toString() // try {
} // const lines = await this.jobManager.readJobLog(logDateTime, logId)
return { code: 200, content: { fromLineNum, toLineNum, logContent } } // lines.splice(0, fromLineNum - 1)
// if (last(lines) === '') lines.pop()
// toLineNum = fromLineNum + lines.length - 1
// lines.unshift('')
// logContent = lines.join('\n')
// } catch (err) {
// log.err('readLog error: %o', err.message)
// toLineNum = fromLineNum
// logContent = err.toString()
// }
// 请求数据格式如下，放置在 RequestBody 中，JSON格式：
// {
// "logDateTim":0, // 本次调度日志时间
// "logId":0, // 本次调度日志ID
// "fromLineNum":0 // 日志开始行号，滚动加载日志
// }
// 响应数据格式：
// {
// "code":200, // 200 表示正常、其他失败
// "msg": null // 错误提示消息
// "content":{
// "fromLineNum":0, // 本次请求，日志开始行数
// "toLineNum":100, // 本次请求，日志结束行号
// "logContent":"xxx", // 本次请求日志内容
// "isEnd":true // 日志是否全部加载完
// }
// }
// return { code: 200, content: { fromLineNum, toLineNum, logContent } }
} }
/** /**
......
src/xxl_job/logger.ts
const debug = require('debug') import debug from 'debug';
const fs = require('fs') import fs from 'fs';
const os = require('os') import os from 'os';
const util = require('util') import util from 'util';
const { always, propOr } = require('./purefuncs')
// 类型定义
const enableExecutorDebugLog = /^(yes|on|true|enable|enabled|1)$/i.test(`${propOr(false, 'XXL_JOB_DEBUG_LOG', process.env)}`) type LogLevel = 'error' | 'info' | 'warn' | 'debug' | 'trace';
const enableLogLevels = propOr('info:*,warn:*,error:*,debug:*,trace:*', 'DEBUG', process.env)
type LoggerMethods = {
const writeStreamOptions = { flags: 'a', encoding: 'utf8', autoClose: true, emitClose: true } [key in LogLevel]: debug.Debugger;
} & {
const noop = always(undefined) close: () => void;
const noopLogger = { info: noop, err: noop, debug: noop, warn: noop, trace: noop } };
const dErr = debug('error') interface LoggerOptions {
const dInfo = debug('info') namespace: string;
const dWarn = debug('warn') logFilePath?: string;
const dDebug = debug('debug') enabledLevels?: string;
const dTrace = debug('trace') debugEnabled?: boolean;
// 自定义对象，包装 debug 模拟日志级别
const createLogger = (ns) => {
const logger = {
info: dInfo.extend(ns),
err: dErr.extend(ns),
debug: dDebug.extend(ns),
warn: dWarn.extend(ns),
trace: dTrace.extend(ns),
}
Object.values(logger).forEach((levelLogger) => Object.assign(levelLogger, { enabled: true, useColors: false }))
return logger
} }
module.exports = (ns, logFilePath) => { // 配置解析
// 1. 执行器运行日志，输出到 stderr，限制日志级别 const parseEnvBool = (envVar: string): boolean => {
if (!logFilePath) { const value = process.env[envVar] || '';
if (!enableExecutorDebugLog) return noopLogger return /^(yes|on|true|enable|enabled|1)$/i.test(value);
const logger = createLogger(ns) };
Object.entries(logger).forEach(([level, levelLogger]) => levelLogger.enabled = enableLogLevels.includes(level))
return logger const getEnvValue = (envVar: string, defaultValue: string): string => {
return process.env[envVar] !== undefined ? process.env[envVar] : defaultValue;
};
// 默认配置
const DEFAULT_LOG_LEVELS = 'info:*,warn:*,error:*,debug:*,trace:*';
const WRITE_STREAM_OPTIONS: fs.WriteStreamOptions = {
flags: 'a',
encoding: 'utf8',
autoClose: true,
emitClose: true
};
// 空日志函数
const noop = (..._args: any[]): void => {};
const noopLogger: LoggerMethods = {
error: noop,
info: noop,
warn: noop,
debug: noop,
trace: noop,
close: noop
};
// 创建基础日志函数
const createBaseLoggers = (namespace: string): Omit<LoggerMethods, 'close'> => {
return {
error: debug('error').extend(namespace),
info: debug('info').extend(namespace),
warn: debug('warn').extend(namespace),
debug: debug('debug').extend(namespace),
trace: debug('trace').extend(namespace)
};
};
// 配置日志级别
const configureLogLevels = (
logger: Omit<LoggerMethods, 'close'>,
enabledLevels: string
): void => {
Object.entries(logger).forEach(([level]) => {
logger[level as LogLevel].enabled = enabledLevels.includes(level);
});
};
// 创建文件日志写入器
const createFileLogger = (
logFilePath: string,
logger: Omit<LoggerMethods, 'close'>,
enabledLevels: string
): LoggerMethods => {
const writeStream = fs.createWriteStream(logFilePath, WRITE_STREAM_OPTIONS);
const logToFile = (...args: any[]): void => {
writeStream.write(`${util.format(...args)}${os.EOL}`);
};
const logToConsoleAndFile = (...args: any[]): void => {
const message = util.format(...args);
console.error(message);
writeStream.write(`${message}${os.EOL}`);
};
// 配置日志输出方式
Object.entries(logger).forEach(([level, logFn]) => {
logFn.log = enabledLevels.includes(level) ? logToConsoleAndFile : logToFile;
});
// 返回完整logger对象
return {
...logger,
close: (): void => {
writeStream.end();
// 切换回仅控制台日志
Object.values(logger).forEach((logFn) => {
logFn.log = console.error;
});
} }
};
};
// 2. 任务执行日志，同时输出到 stderr 和 文件，stderr 限制日志级别，输出到文件不限制级别以供调度中心全量查看 // 主导出函数
const writeStream = fs.createWriteStream(logFilePath, writeStreamOptions) export const createLogger = (
const log2File = (...args) => writeStream.write(`${util.format(...args)}${os.EOL}`) options: LoggerOptions | string
const log2Stderr = (...args) => console.error(util.format(...args)) ): LoggerMethods => {
const log2FileAndStderr = (...args) => { // 处理参数重载
const content = util.format(...args) const normalizedOptions = typeof options === 'string'
writeStream.write(`${content}${os.EOL}`) ? { namespace: options }
console.error(content) : options;
const {
namespace,
logFilePath,
enabledLevels = getEnvValue('DEBUG', DEFAULT_LOG_LEVELS),
debugEnabled = parseEnvBool('XXL_JOB_DEBUG_LOG')
} = normalizedOptions;
if (!debugEnabled && !logFilePath) {
return noopLogger;
} }
const logger = createLogger(ns)
const logger = createBaseLoggers(namespace);
// 设置输出
Object.entries(logger).forEach(([level, levelLogger]) => { // 配置日志级别
levelLogger.log = enableLogLevels.includes(level) ? log2FileAndStderr : log2File configureLogLevels(logger, enabledLevels);
})
if (!logFilePath) {
// 任务执行完成，关闭文件输出流，后续日志只输出到 stderr return {
logger.close = () => { ...logger,
Object.entries(logger).forEach(([level, levelLogger]) => { close: noop
Object.assign(levelLogger, { enabled: enableLogLevels.includes(level), log: log2Stderr }) };
})
writeStream.end()
} }
return logger // 文件日志配置
} return createFileLogger(logFilePath, logger, enabledLevels);
};
// 默认导出
export default createLogger;
\ No newline at end of file \ No newline at end of file
支持 Markdown 格式
你添加了 0 人 到此讨论。请谨慎行事。
Finish editing this message first!