mirror of
https://gitee.com/ulthon/ulthon_admin.git
synced 2026-08-30 12:45:32 +08:00
f1a2e8fd5d
F2 Code quality review 发现：3 个新增 service 未遵守 .agents/rules/ulthon-naming-convention.md 第 19 行规定 （service 模块文件名需带 Service 后缀）。 变更： - 重命名 6 个文件（3 Base + 3 App）： NginxLogParser(Base)→NginxLogParserService(Base) NginxLogReader(Base)→NginxLogReaderService(Base) NginxLogAggregator(Base)→NginxLogAggregatorService(Base) - 同步类名、use、类型 hint、new 引用 - NginxLogReaderServiceBase 异常消息前缀同步加 Service - controller Base 引用更新（NginxLogImportBase / NginxLogStatAggregateBase） 验证： - php -l 全部 8 个文件语法通过 - grep 残留旧名 = 0 结果（24 处引用全部带 Service 后缀）
227 lines
9.3 KiB
PHP
227 lines
9.3 KiB
PHP
<?php
|
||
|
||
namespace base\common\service;
|
||
|
||
use app\admin\model\NginxLogPosition;
|
||
|
||
/**
|
||
* Nginx 日志增量读取 service（Base 层）.
|
||
*
|
||
* 职责：
|
||
* - 按文件 offset 增量读取日志行（Generator，不读整个文件到内存）
|
||
* - 检测日志轮转（inode 变化 OR filesize < offset，覆盖 rename+rebuild 与 copytruncate 两种模式）
|
||
* - 将读取进度持久化到 NginxLogPosition 表（upsert）
|
||
*
|
||
* 不负责：解析（T5 Parser）、聚合（T7 Aggregator）。
|
||
*
|
||
* 依赖倒置：内部 model 调用走 `app\admin\model\NginxLogPosition`（App 入口类），
|
||
* 使用者可在 app/ 重写该 model 拦截行为；本类不直接 `new NginxLogPosition`（静态调用满足多态）。
|
||
*/
|
||
class NginxLogReaderServiceBase
|
||
{
|
||
/** 单次 fread 的 buffer 大小（64KB） */
|
||
protected const BUFFER_SIZE = 65536;
|
||
|
||
/** 单行无换行符时的硬上限（8MB），超过则强制按一行 yield，防止 OOM */
|
||
protected const MAX_LINE_BYTES = 8388608;
|
||
|
||
/**
|
||
* 增量读取日志文件，yield 完整行.
|
||
*
|
||
* 读取协议：
|
||
* 1. fopen + fstat 获取当前 inode 和 filesize
|
||
* 2. 读 position 行（无则视为新建：inode=当前、offset=0）
|
||
* 3. 轮转检测双条件：inode 不一致 OR filesize < position.offset → 重置 offset=0
|
||
* 4. fseek 到 offset
|
||
* 5. 每读 64KB buffer，定位已读段内最后 `\n`，仅 yield 该位置之前的完整行
|
||
* 6. 新 offset = 最后 `\n` + 1（buffer 中剩余半行保留到下次循环）
|
||
* 7. yield maxLines 后提前 return 并 savePosition
|
||
* 8. EOF 时 savePosition，offset = 文件末尾（若尾部有半行未 yield，offset 回退到半行起点）
|
||
*
|
||
* @param string $filePath 日志文件绝对路径
|
||
* @param int $maxLines 单次 yield 的最大行数（达到即提前退出并保存 offset）
|
||
*
|
||
* @return \Generator<string> 每次迭代返回一行（不含换行符）
|
||
*
|
||
* @throws \RuntimeException 文件无法打开或 stat/fseek 失败时抛出（交由上层 try-catch）
|
||
*/
|
||
public function read(string $filePath, int $maxLines = 100000): \Generator
|
||
{
|
||
$fp = @fopen($filePath, 'rb');
|
||
if ($fp === false) {
|
||
throw new \RuntimeException("NginxLogReaderService: 无法打开文件 {$filePath}");
|
||
}
|
||
|
||
try {
|
||
// 步骤 1：clearstatcache + fstat 取当前 inode/size
|
||
clearstatcache(true, $filePath);
|
||
$stat = fstat($fp);
|
||
if ($stat === false) {
|
||
throw new \RuntimeException("NginxLogReaderService: fstat 失败 {$filePath}");
|
||
}
|
||
$currentInode = (int) $stat['ino'];
|
||
$currentSize = (int) $stat['size'];
|
||
|
||
// 步骤 2：读 position（无则默认 offset=0）
|
||
$position = $this->getLastPosition($filePath);
|
||
$inode = $currentInode;
|
||
$offset = 0;
|
||
if ($position !== null) {
|
||
$inode = (int) $position['inode'];
|
||
$offset = (int) $position['offset'];
|
||
|
||
// 步骤 3：轮转检测双条件
|
||
// inode 不一致 → rename+rebuild 模式（Linux inode 有效时主判定）
|
||
// filesize < offset → copytruncate 模式 / Windows 上 inode 恒为 0 时的主判定
|
||
if ($inode !== $currentInode || $currentSize < $offset) {
|
||
$offset = 0;
|
||
$inode = $currentInode;
|
||
}
|
||
}
|
||
|
||
// 步骤 4：fseek 到 offset
|
||
if ($offset > 0 && fseek($fp, $offset) !== 0) {
|
||
throw new \RuntimeException("NginxLogReaderService: fseek 失败 offset={$offset} file={$filePath}");
|
||
}
|
||
|
||
$yielded = 0;
|
||
$lastLineHash = null;
|
||
$buffer = '';
|
||
// $bufferOffset 始终等于 buffer[0] 对应的文件字节偏移
|
||
$bufferOffset = $offset;
|
||
|
||
// 步骤 5-7：循环读取 + yield
|
||
while (true) {
|
||
$chunk = fread($fp, self::BUFFER_SIZE);
|
||
if ($chunk === false) {
|
||
// 读错误：交上层处理，不保存进度（下次重读）
|
||
throw new \RuntimeException("NginxLogReaderService: fread 失败 file={$filePath}");
|
||
}
|
||
if ($chunk === '') {
|
||
// EOF
|
||
break;
|
||
}
|
||
$buffer .= $chunk;
|
||
|
||
// 内层循环：处理 buffer 中所有完整行（含末尾 \n 的）
|
||
$pos = 0;
|
||
while (($nlPos = strpos($buffer, "\n", $pos)) !== false) {
|
||
$line = substr($buffer, $pos, $nlPos - $pos);
|
||
$pos = $nlPos + 1;
|
||
|
||
yield $line;
|
||
$yielded++;
|
||
$lastLineHash = md5($line);
|
||
|
||
// 步骤 7：maxLines 达到即保存并退出
|
||
if ($yielded >= $maxLines) {
|
||
$newOffset = $bufferOffset + $pos;
|
||
$this->savePosition($filePath, $inode, $newOffset, $lastLineHash);
|
||
return;
|
||
}
|
||
}
|
||
|
||
// 步骤 6：buffer 中已消费 $pos 字节，半行（buffer[$pos..]）保留
|
||
$bufferOffset += $pos;
|
||
$buffer = substr($buffer, $pos);
|
||
|
||
// 防御：buffer 无 \n 累积过长（异常畸形行），强制按一行处理避免 OOM
|
||
if (strlen($buffer) >= self::MAX_LINE_BYTES) {
|
||
yield $buffer;
|
||
$yielded++;
|
||
$lastLineHash = md5($buffer);
|
||
$bufferOffset += strlen($buffer);
|
||
$buffer = '';
|
||
|
||
if ($yielded >= $maxLines) {
|
||
$this->savePosition($filePath, $inode, $bufferOffset, $lastLineHash);
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
|
||
// 步骤 8：EOF 处理
|
||
if ($buffer === '') {
|
||
// 文件恰好在最后一个 \n 处结束：offset = 文件末尾（ftell）
|
||
$finalOffset = ftell($fp);
|
||
$this->savePosition($filePath, $inode, (int) $finalOffset, $lastLineHash);
|
||
} else {
|
||
// 尾部半行（无 \n）：可能是日志正在写入，不 yield，offset 回退到半行起点，下次重读
|
||
$this->savePosition($filePath, $inode, $bufferOffset, $lastLineHash);
|
||
}
|
||
} finally {
|
||
if (is_resource($fp)) {
|
||
fclose($fp);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 查询文件的上次读取位置.
|
||
*
|
||
* @param string $filePath 日志文件绝对路径
|
||
*
|
||
* @return array|null 命中时返回 [inode, offset, last_line_hash, parse_fail_count, parse_fail_samples]，无记录返回 null
|
||
*/
|
||
public function getLastPosition(string $filePath): ?array
|
||
{
|
||
$row = NginxLogPosition::where('file_path', $filePath)->find();
|
||
if ($row === null) {
|
||
return null;
|
||
}
|
||
|
||
return [
|
||
'inode' => (int) $row->getData('inode'),
|
||
'offset' => (int) $row->getData('offset'),
|
||
'last_line_hash' => $row->getData('last_line_hash'),
|
||
'parse_fail_count' => (int) $row->getData('parse_fail_count'),
|
||
'parse_fail_samples' => $row->getData('parse_fail_samples'),
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 保存读取位置（upsert）.
|
||
*
|
||
* Reader 内部在 maxLines 达到或 EOF 时调用，传 lastLineHash（最后一行 md5）。
|
||
* failCount 默认 0：每次新读取批次开始时由 Reader 重置；上层若需记录解析失败，
|
||
* 在 parse 后再次调用本方法覆盖 failCount/failSamples 即可（offset 保持一致）。
|
||
*
|
||
* @param string $filePath 日志文件绝对路径（主键）
|
||
* @param int $inode 当前文件 inode
|
||
* @param int $offset 新的字节偏移
|
||
* @param string|null $lastLineHash 最后一行 md5（用于去重/检测）
|
||
* @param int $failCount 解析失败累计次数（默认 0）
|
||
* @param string|null $failSamples 解析失败样本（默认 null）
|
||
*/
|
||
public function savePosition(string $filePath, int $inode, int $offset, ?string $lastLineHash, int $failCount = 0, ?string $failSamples = null): void
|
||
{
|
||
$now = time();
|
||
|
||
$existing = NginxLogPosition::where('file_path', $filePath)->find();
|
||
if ($existing !== null) {
|
||
$existing->save([
|
||
'inode' => $inode,
|
||
'offset' => $offset,
|
||
'last_read_time' => $now,
|
||
'last_line_hash' => $lastLineHash,
|
||
'parse_fail_count' => $failCount,
|
||
'parse_fail_samples' => $failSamples,
|
||
]);
|
||
|
||
return;
|
||
}
|
||
|
||
NginxLogPosition::create([
|
||
'file_path' => $filePath,
|
||
'inode' => $inode,
|
||
'offset' => $offset,
|
||
'last_read_time' => $now,
|
||
'last_line_hash' => $lastLineHash,
|
||
'parse_fail_count' => $failCount,
|
||
'parse_fail_samples' => $failSamples,
|
||
'create_time' => $now,
|
||
'update_time' => $now,
|
||
]);
|
||
}
|
||
}
|