mirror of
https://gitee.com/ulthon/ulthon_admin.git
synced 2026-08-30 12:45:32 +08:00
feat(nginx-log): NginxLogReader 增量读取 service（Base/App 双层，含轮转检测）
This commit is contained in:
augushong
226
extend/base/common/service/NginxLogReaderBase.php
Normal file
226
extend/base/common/service/NginxLogReaderBase.php
Normal file
@@ -0,0 +1,226 @@
|
||||
<?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 NginxLogReaderBase
|
||||
{
|
||||
/** 单次 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("NginxLogReader: 无法打开文件 {$filePath}");
|
||||
}
|
||||
|
||||
try {
|
||||
// 步骤 1：clearstatcache + fstat 取当前 inode/size
|
||||
clearstatcache(true, $filePath);
|
||||
$stat = fstat($fp);
|
||||
if ($stat === false) {
|
||||
throw new \RuntimeException("NginxLogReader: 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("NginxLogReader: 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("NginxLogReader: 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,
|
||||
]);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user