diff --git a/app/tools/controller/timer/DebugLogImport.php b/app/tools/controller/timer/DebugLogImport.php new file mode 100644 index 0000000..18e43cb --- /dev/null +++ b/app/tools/controller/timer/DebugLogImport.php @@ -0,0 +1,20 @@ + 1, 'run_type' => 'auto', ], + [ + 'name' => 'debug_log_import', + 'type' => 'site', + 'target' => '/tools/timer.DebugLogImport/do', + 'frequency' => 60, + 'concurrency' => 1, + 'run_type' => 'all', + ], ]; diff --git a/extend/base/tools/controller/timer/DebugLogImportBase.php b/extend/base/tools/controller/timer/DebugLogImportBase.php new file mode 100644 index 0000000..16f036d --- /dev/null +++ b/extend/base/tools/controller/timer/DebugLogImportBase.php @@ -0,0 +1,365 @@ +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; + } +}