feat(log): 新增 debug_log_import 定时导入任务(时间窗口+事务对齐+当天文件保护)

This commit is contained in:
augushong
2026-08-16 08:45:19 +08:00
parent c0cae93562
commit 57873080df
3 changed files with 393 additions and 0 deletions

View File

@@ -0,0 +1,20 @@
<?php
declare(strict_types=1);
namespace app\tools\controller\timer;
use base\tools\controller\timer\DebugLogImportBase;
/**
* debug 日志导入定时任务App 入口类).
*
* 框架路由 /tools/timer.DebugLogImport/do 与 /tools/timer.DebugLogImport/execute
* 均通过本类入口实例化(依赖倒置:定时器框架/路由系统直接调本类,不调 Base
*
* 业务侧如需定制导入行为(如改批大小、加自定义过滤、改入库目标表等),
* 重写本类对应方法即可拦截,无需改动 extend/base/。
*/
class DebugLogImport extends DebugLogImportBase
{
}

View File

@@ -61,4 +61,12 @@ return [
'concurrency' => 1,
'run_type' => 'auto',
],
[
'name' => 'debug_log_import',
'type' => 'site',
'target' => '/tools/timer.DebugLogImport/do',
'frequency' => 60,
'concurrency' => 1,
'run_type' => 'all',
],
];

View File

@@ -0,0 +1,365 @@
<?php
declare(strict_types=1);
namespace base\tools\controller\timer;
use app\admin\model\DebugLog;
use app\common\controller\TimerController;
use app\common\service\DebugLogReaderService;
use think\facade\Db;
use think\facade\Log;
/**
* debug 日志 JSONL/CSV 导入定时任务Base 层).
*
* 流程:
* 1. flock 节点互斥({runtimeRoot}debug_log_import.lockLOCK_EX|LOCK_NB
* 抢锁失败直接返回 locked——防同节点双 timer 进程/手动触发与调度重叠,
* 重叠会从同一 position 双读产生重复行ul_debug_log 无唯一键无 DB 层防护)
* 2. 时间窗口循环deadline = start + maxRunTime每轮 glob 扫描四模式:
* {runtimeRoot}log/ 下的 [0-9]*.jsonl 与 [0-9]*.csv以及
* {runtimeRoot}{app}/log/ 下的 [0-9]*.jsonl 与 [0-9]*.csv{app} 为各应用名)
* ——多应用模式下日志分散在主 runtime/log/CLI 引导)与 runtime/{app}/log/
* admin/tools 等应用),必须同时扫主目录与应用子目录;禁止 [0-9] 锚定
* 之外的宽匹配,不碰 *.log
* 3. 每文件按扩展名走 reader->read()jsonl/ reader->readCsv()csv
* fgetcsv 有状态解析引号内换行),缓冲到 BATCH 条
* 4. Db::transaction { DebugLog::insertAll + reader->savePosition }——
* 数据与 position 同事务落盘,崩溃恢复不重复不丢失
* 5. unlink 三重与reachedEof() && offset >= filesize && 文件名日期 < 今天;
* 当天文件永不删FPM 写入方可能持有已删 inode 句柄静默丢数据)
* 6. 本轮所有文件无新数据 → sleep(1) 再重扫
* 7. 连续 IDLE_EXIT_ROUNDS 轮无新数据且已运行 >= MIN_RUN_SECONDS 时提前收敛
* 退出(自然静默场景:首轮消费完后约 10-12s 返回,无需跑满 55s 窗口;
* timer 正常调度同样受益,提前退出无害且省资源)
* 8. 返回 json 统计 {files, rows, fails, deleted, locked}finally 释放 flock
*
* runtime 根目录app()->getRootPath().'runtime/'。
* 【禁止】用 App::getRuntimePath()——本任务经 /tools/timer.* 路由触发时上下文
* 是 tools 应用getRuntimePath() 返回 runtime/tools/,会漏扫其余应用目录。
*
* 调度与防刷:$frequency = null 显式关闭控制器侧防刷TimerControllerBase 仅在
* is_int($frequency) 时 protectVisit调度节奏完全由定时器侧 frequency=60s
* 的 Cache 节流控制;$maxRunTime = 55 有意大于技能建议的 frequency/2——重叠
* 风险由 flock 节点互斥兜底补偿(抢不到锁即让位),换取静默期的吞吐余量。
*
* 部署假设runtime/log 必须是节点本地卷,【禁止】多容器共享 runtime 目录——
* 共享卷 + run_type=all 会导致同批文件被多节点重复导入position 按 node_id
* 隔离DB 层不拦截重复行)。
*
* 依赖倒置reader 经 app\common\service\DebugLogReaderService 入口实例化、
* model 走 app\admin\model\DebugLog。业务侧如需定制导入逻辑重写
* app/tools/controller/timer/DebugLogImport.php 对应方法即可拦截。
*/
class DebugLogImportBase extends TimerController
{
/**
* 防刷间隔null = 关闭控制器侧防刷(调度由定时器侧 frequency 节流).
*/
protected $frequency = null;
/**
* 并发上限 cap单分片id=0。日志文件是节点本地卷多节点各自导自己的。
*/
protected $concurrency = 1;
/**
* 时间窗口55 配 frequency=60s有意大于技能建议的 frequency/2
* 重叠风险由 flock 节点互斥兜底(见类头注释「调度与防刷」).
*/
protected $maxRunTime = 55;
/**
* 单事务入库行数(同时是 insertAll 的批大小).
*/
protected const BATCH = 100;
/**
* 单文件单轮最大读取行数未读完下轮继续debug 行是短文本1000 行单轮足够).
*/
protected const READ_MAX_LINES = 1000;
/**
* 提前收敛:连续无新数据的轮数达到此值…….
*/
protected const IDLE_EXIT_ROUNDS = 10;
/**
* ……且已运行时长达到此值deadline 循环提前退出.
*/
protected const MIN_RUN_SECONDS = 10;
public function do()
{
$stats = ['files' => 0, 'rows' => 0, 'fails' => 0, 'deleted' => 0, 'locked' => false];
// 节点互斥:抢锁失败说明本节点已有实例在跑(调度重叠/手动触发重叠),让位。
// 锁文件 chmod 0666CLI(root) 与 FPM(www-data) 都会经本入口跑 do()
// 先创建者放开权限,避免另一侧 fopen 'w' 因 644 root 属主失败
$lockPath = $this->runtimeRoot() . 'debug_log_import.lock';
$lock = @fopen($lockPath, 'w');
if ($lock === false) {
Log::error('debug_log_import: 打开互斥锁文件失败 ' . $lockPath);
return json_encode($stats, JSON_UNESCAPED_UNICODE);
}
@chmod($lockPath, 0666);
if (!flock($lock, LOCK_EX | LOCK_NB)) {
fclose($lock);
$stats['locked'] = true;
return json_encode($stats, JSON_UNESCAPED_UNICODE);
}
try {
$reader = new DebugLogReaderService();
$startTime = time();
$deadline = $startTime + $this->maxRunTime;
$idleRounds = 0;
$activeFiles = [];
while (time() < $deadline) {
$roundRows = 0;
foreach ($this->globFiles() as $file) {
try {
$result = $this->processFile($reader, $file);
$roundRows += $result['rows'];
$stats['rows'] += $result['rows'];
$stats['fails'] += $result['fails'];
if ($result['deleted']) {
$stats['deleted']++;
}
if ($result['rows'] > 0 || $result['fails'] > 0 || $result['deleted']) {
$activeFiles[$file] = true;
}
} catch (\Throwable $e) {
Log::error('debug_log_import: 处理文件异常 ' . $file . ' - ' . $e->getMessage());
}
}
if ($roundRows > 0) {
// 本轮有消费:立即重扫继续消化积压(无 sleep
$idleRounds = 0;
continue;
}
// 本轮无新数据:短暂休眠后重扫;持续静默且已跑够最短时长则提前收敛
$idleRounds++;
if ($idleRounds >= self::IDLE_EXIT_ROUNDS && (time() - $startTime) >= self::MIN_RUN_SECONDS) {
break;
}
sleep(1);
}
$stats['files'] = count($activeFiles);
return json_encode($stats, JSON_UNESCAPED_UNICODE);
} finally {
flock($lock, LOCK_UN);
fclose($lock);
}
}
/**
* runtime 根目录(全应用共享的 runtime/,非当前应用的 runtime/{app}/.
*
* 必须经 app()->getRootPath() 拼:本任务经 /tools/timer.* 路由触发时上下文
* 是 tools 应用App::getRuntimePath() 返回 runtime/tools/ 会漏扫其余目录。
*/
protected function runtimeRoot(): string
{
return app()->getRootPath() . 'runtime/';
}
/**
* 扫描全部待导入文件:主 runtime/log/ 与 runtime/{app}/log/ 双目录、jsonl/csv 双扩展.
*
* [0-9] 前缀锚定日期命名文件({Ymd}.jsonl / {ymd}.csv排除 *.log 与其他
* 杂项;去重后按文件名序(日期字典序),旧日期先导。
*
* @return string[] 绝对路径列表
*/
protected function globFiles(): array
{
$root = $this->runtimeRoot();
$patterns = [
$root . 'log/[0-9]*.jsonl',
$root . 'log/[0-9]*.csv',
$root . '*/log/[0-9]*.jsonl',
$root . '*/log/[0-9]*.csv',
];
$files = [];
foreach ($patterns as $pattern) {
$matched = glob($pattern);
if ($matched === false) {
continue;
}
foreach ($matched as $file) {
if (is_file($file)) {
$files[$file] = true;
}
}
}
$files = array_keys($files);
sort($files);
return $files;
}
/**
* 处理单个文件:增量读取、缓冲批量入库(事务内 savePosition、导完删除.
*
* @return array{rows: int, fails: int, deleted: bool}
*/
protected function processFile(DebugLogReaderService $reader, string $file): array
{
$rows = 0;
$deleted = false;
$buffer = [];
// failCount 传"position 旧值 + 本轮新增"的累计(继承 T5 消费契约)
$lastPosition = $reader->getLastPosition($file);
$prevFailCount = ($lastPosition !== null) ? (int) $lastPosition['parse_fail_count'] : 0;
$storedOffset = ($lastPosition !== null) ? (int) $lastPosition['offset'] : 0;
$storedInode = ($lastPosition !== null) ? (int) $lastPosition['inode'] : -1;
$ext = strtolower(pathinfo($file, PATHINFO_EXTENSION));
$generator = ($ext === 'csv')
? $reader->readCsv($file, self::READ_MAX_LINES)
: $reader->read($file, self::READ_MAX_LINES);
foreach ($generator as $row) {
$buffer[] = $row;
if (count($buffer) >= self::BATCH) {
$rows += $this->flushBatch($reader, $file, $buffer, $prevFailCount);
$buffer = [];
}
}
// 本轮解析失败(坏 JSON 行 + 字段数不符 + 超长丢弃)
$fails = $reader->getParseFails() + $reader->getDroppedLines();
if (!empty($buffer)) {
$rows += $this->flushBatch($reader, $file, $buffer, $prevFailCount);
} else {
// 无产出也可能推进了 position毒行消费 / inode 轮转):有变化才落盘,
// 静默稳态(今天的文件已读到 EOF 无新行)不做无谓 UPDATE
$position = $reader->getCurrentPosition();
if ($fails > 0 || $position['offset'] !== $storedOffset || $position['inode'] !== $storedInode) {
$reader->savePosition(
$file,
$position['inode'],
$position['offset'],
$position['last_line_hash'],
$prevFailCount + $fails,
$reader->getFailSamples()
);
}
}
// unlink 三重与:干净 EOF && offset 覆盖全文件 && 文件名日期严格早于今天
//当天文件永不删——FPM 写入方可能持有已删 inode 句柄静默丢数据)
if ($reader->reachedEof()) {
clearstatcache(true, $file);
$position = $reader->getCurrentPosition();
$size = filesize($file);
if ($size !== false && $position['offset'] >= $size && $this->isPastDateFile($file)) {
$deleted = @unlink($file);
if ($deleted) {
Log::info('debug_log_import: 文件已导完删除 ' . $file);
}
}
}
// 日志克制:仅在本文件确有活动时记 1 条(静默文件不刷屏;写库日志本身
// 会进导入链路,无活动不写可将自反馈噪声压到有界基线)
if ($rows > 0 || $fails > 0 || $deleted) {
Log::info("debug_log_import: 文件 {$file} 完成 rows={$rows} fails={$fails} deleted=" . ($deleted ? 1 : 0));
}
return ['rows' => $rows, 'fails' => $fails, 'deleted' => $deleted];
}
/**
* 批量入库:同一事务内 DebugLog::insertAll + reader->savePosition.
*
* 事务语义:数据与 position 要么同时生效、要么同时回滚——崩溃恢复后从
* position 处重读,不重复导入已提交批次,也不丢失未提交进度。
*
* @param string $file 日志文件绝对路径
* @param array $buffer 行数组(键与 DebugLogToolkit::FIELDS 一致)
* @param int $prevFailCount position 里旧 parse_fail_count
*/
protected function flushBatch(DebugLogReaderService $reader, string $file, array $buffer, int $prevFailCount): int
{
// 捕获当前安全偏移(最后一条已 yield 完整行末尾),事务内与数据同落盘
$position = $reader->getCurrentPosition();
$failCount = $prevFailCount + $reader->getParseFails() + $reader->getDroppedLines();
$failSamples = $reader->getFailSamples();
Db::transaction(function () use ($reader, $file, $buffer, $position, $failCount, $failSamples) {
DebugLog::insertAll($buffer);
$reader->savePosition(
$file,
$position['inode'],
$position['offset'],
$position['last_line_hash'],
$failCount,
$failSamples
);
});
return count($buffer);
}
/**
* 文件名日期是否严格早于今天(当天与不可判定文件一律不删).
*/
protected function isPastDateFile(string $file): bool
{
$fileDate = $this->resolveFileDate($file);
if ($fileDate === null) {
// 两种格式都解析失败 → 不可判定 → 永不删除(安全侧)
return false;
}
$today = new \DateTime('today');
return $fileDate < $today;
}
/**
* 按扩展名定死格式解析文件名日期(禁止 strtotime——数字串语义不可靠.
*
* .jsonl → 'Ymd'{Ymd}.jsonl.csv → 先 'Ymd' 失败再 'ymd'
* (遗留 CSV {ymd}.csv 两位年规则 00-69→20xx实测 '260602' 走 'Ymd'
* 返 false 不会误判。createFromFormat 返回 false 或带任何
* errors/warnings月份越界翻转、尾部杂字符等均视为解析失败。
*/
protected function resolveFileDate(string $file): ?\DateTimeInterface
{
$ext = strtolower(pathinfo($file, PATHINFO_EXTENSION));
$name = pathinfo($file, PATHINFO_FILENAME);
if ($ext !== 'jsonl' && $ext !== 'csv') {
return null;
}
$formats = ($ext === 'csv') ? ['Ymd', 'ymd'] : ['Ymd'];
foreach ($formats as $format) {
$dt = \DateTime::createFromFormat($format, $name);
$errors = \DateTime::getLastErrors();
$hasErrors = is_array($errors)
&& ($errors['error_count'] > 0 || $errors['warning_count'] > 0);
if ($dt !== false && !$hasErrors) {
return $dt;
}
}
return null;
}
}