mirror of
https://gitee.com/ulthon/ulthon_admin.git
synced 2026-08-31 05:05:33 +08:00
366 lines
15 KiB
PHP
366 lines
15 KiB
PHP
<?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.lock，LOCK_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 0666：CLI(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;
|
||
}
|
||
}
|