mirror of
https://gitee.com/ulthon/ulthon_admin.git
synced 2026-08-30 20:55:32 +08:00
a43a9a48b9
- XhprofProfileParserService（Base/App）：纯数组解析 jsonl 行， meta 字段以真实 fixture 为准；simple_url 缺失兜底与采集端回调同算法； watch 正则按 callee 物化（ct/wt_sum/wt_max），非法正则编译计 fail 跳过 - XhprofLogReaderService（Base/App）：增量读取，半行缓冲超限整行丢弃 （不照抄 nginx 8MB 强制切行）；EOF 无换行半行不 yield 不推进 offset； 进度不自动落盘，由导入侧在同一事务内显式 savePosition - XhprofLogImport 定时任务（Base/App）：不受 XHPROF_ENABLE 门控， runs-*.jsonl 按文件名序增量导入；run 逐条 insert + detail chunk 20 insertAll + run_watch insertAll 与 position 同事务落盘；导至 EOF 后 只删已导完文件 - 修复 T1 缺陷：request_time decimal(12,3) 装不下 10 位时间戳， Scheme 与 migration 同步改 decimal(13,3) - 新增 parser 单元测试（真实 fixture 断言，9 用例 42 断言） 与 jsonl fixture；phpunit 注册 unit 测试套件
317 lines
11 KiB
PHP
317 lines
11 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace base\tools\controller\timer;
|
||
|
||
use app\admin\model\XhprofRun;
|
||
use app\admin\model\XhprofRunDetail;
|
||
use app\admin\model\XhprofRunWatch;
|
||
use app\admin\model\XhprofWatch;
|
||
use app\common\controller\TimerController;
|
||
use think\facade\Db;
|
||
use think\facade\Log;
|
||
|
||
/**
|
||
* XHProf jsonl 导入定时任务（Base 层）.
|
||
*
|
||
* 流程：
|
||
* 1. 扫描 XHPROF_FILE_DIR 下 runs-*.jsonl 按文件名序（旧日期先导）
|
||
* 2. 依赖倒置：实例化 app 层 Parser / Reader；watch 正则 per 批次编译一次
|
||
* 3. 每文件：增量读取（position 按 node_id+file_path 定位，inode 变化归零）→
|
||
* run 逐条 insert 拿自增 id → detail chunk 20 insertAll →
|
||
* run_watch 物化聚合 insertAll；同一事务内 insertBatch + savePosition（消崩溃重复窗口）
|
||
* 4. 文件导至 EOF 且 position 落盘后 unlink（只删已导完文件）
|
||
*
|
||
* 门控约定：本任务不受 XHPROF_ENABLE 门控——开关只管采集端，
|
||
* 已落盘的残留文件必须能导完（关闭采集后存量数据照常入库）。
|
||
*
|
||
* 目录单一来源：getenv('XHPROF_FILE_DIR')，与采集端 bootstrap 读同一变量，
|
||
* 禁止与 ThinkPHP env() 混用（两处漂移会导致导入读不到文件）。
|
||
*
|
||
* 业务侧如需定制导入逻辑，重写 app/tools/controller/timer/XhprofLogImport.php 对应方法即可拦截。
|
||
*/
|
||
class XhprofLogImportBase extends TimerController
|
||
{
|
||
/**
|
||
* 防刷间隔（秒）.
|
||
* 仅做控制器侧防刷，不影响定时器侧 frequency 节流。
|
||
*/
|
||
protected $frequency = 60;
|
||
|
||
/**
|
||
* 并发上限 cap：单分片（id=0）。jsonl 是节点本地文件，多节点各自导自己的。
|
||
*/
|
||
protected $concurrency = 1;
|
||
|
||
/**
|
||
* run 行缓冲上限（同时是 detail insertAll 的 chunk；单行实测 ~100KB，20 条一批防单条 SQL 过大）.
|
||
*/
|
||
protected const RUN_BATCH = 20;
|
||
|
||
/**
|
||
* 单文件单轮最大读取行数（行大，2000 行 ≈ 200MB 上界已足够单轮消化；未读完下轮继续）.
|
||
*/
|
||
protected const READ_MAX_LINES = 2000;
|
||
|
||
/**
|
||
* parse_fail_samples 最大记录条数.
|
||
*/
|
||
protected const FAIL_SAMPLES_MAX = 10;
|
||
|
||
public function do()
|
||
{
|
||
// 注意：不受 XHPROF_ENABLE 门控（开关只管采集，残留文件必须能导完）
|
||
$dir = $this->fileDir();
|
||
$files = glob($dir . '/runs-*.jsonl');
|
||
if ($files === false || $files === []) {
|
||
return json_encode([
|
||
'files' => 0, 'runs' => 0, 'watches' => 0, 'fails' => 0, 'deleted' => 0,
|
||
], JSON_UNESCAPED_UNICODE);
|
||
}
|
||
sort($files); // 按文件名序（runs-{Ymd} 字典序即日期序），旧日期先导
|
||
|
||
// 依赖倒置：使用 app 层入口类（业务侧可重写拦截）
|
||
$parser = new \app\common\service\XhprofProfileParserService();
|
||
$reader = new \app\common\service\XhprofLogReaderService();
|
||
|
||
// watch 正则 per 批次编译：本节点启用的规则一次编译，导入全程复用
|
||
$watchRows = XhprofWatch::field('id,regex')->where('status', 1)->select()->toArray();
|
||
$compiled = $parser->compileWatchRegexes($watchRows);
|
||
$patterns = $compiled['patterns'];
|
||
|
||
$totalFiles = 0;
|
||
$totalRuns = 0;
|
||
$totalWatches = 0;
|
||
$totalFails = $compiled['fails'];
|
||
$deleted = 0;
|
||
$errors = [];
|
||
|
||
foreach ($files as $file) {
|
||
try {
|
||
$stats = $this->processFile($parser, $reader, $file, $patterns);
|
||
$totalFiles++;
|
||
$totalRuns += $stats['runs'];
|
||
$totalWatches += $stats['watches'];
|
||
$totalFails += $stats['fails'];
|
||
if ($stats['deleted']) {
|
||
$deleted++;
|
||
}
|
||
} catch (\Throwable $e) {
|
||
Log::error('xhprof_log_import: 处理文件异常 ' . $file . ' - ' . $e->getMessage());
|
||
$errors[] = 'error:' . basename($file) . ':' . mb_substr($e->getMessage(), 0, 200);
|
||
}
|
||
}
|
||
|
||
return json_encode([
|
||
'files' => $totalFiles,
|
||
'runs' => $totalRuns,
|
||
'watches' => $totalWatches,
|
||
'fails' => $totalFails,
|
||
'deleted' => $deleted,
|
||
'errors' => $errors,
|
||
], JSON_UNESCAPED_UNICODE);
|
||
}
|
||
|
||
/**
|
||
* jsonl 落盘目录（与采集端 bootstrap getenv 读同一变量，禁止 ThinkPHP env()）.
|
||
*/
|
||
protected function fileDir(): string
|
||
{
|
||
return getenv('XHPROF_FILE_DIR') ?: '/var/www/html/runtime/xhprof';
|
||
}
|
||
|
||
/**
|
||
* 处理单个文件：增量读取、解析、批量入库（事务内 savePosition）、导完删除.
|
||
*
|
||
* @return array{runs: int, watches: int, fails: int, deleted: bool}
|
||
*/
|
||
protected function processFile(
|
||
\app\common\service\XhprofProfileParserService $parser,
|
||
\app\common\service\XhprofLogReaderService $reader,
|
||
string $file,
|
||
array $patterns
|
||
): array {
|
||
$runs = 0;
|
||
$watches = 0;
|
||
$fails = 0;
|
||
$samples = [];
|
||
$buffer = [];
|
||
|
||
// Reader 是 Generator，让其自然走完（maxLines / EOF 自然结束，进度由本类事务内落盘）
|
||
foreach ($reader->read($file, self::READ_MAX_LINES) as $line) {
|
||
$parsed = $parser->parse($line);
|
||
if ($parsed === null) {
|
||
$fails++;
|
||
$this->addFailSample($samples, $line);
|
||
continue;
|
||
}
|
||
if ($parsed['degraded']) {
|
||
// main() 缺失：行仍入库（降级 0），但计 fail 留痕
|
||
$fails++;
|
||
$this->addFailSample($samples, $line);
|
||
}
|
||
|
||
$buffer[] = [
|
||
'run' => $this->buildRunRow($parsed),
|
||
'raw' => $line,
|
||
'profile' => $parsed['profile'],
|
||
];
|
||
|
||
if (count($buffer) >= self::RUN_BATCH) {
|
||
$flushed = $this->flushBatch($parser, $reader, $file, $buffer, $patterns, $fails, $samples);
|
||
$runs += $flushed['runs'];
|
||
$watches += $flushed['watches'];
|
||
$buffer = [];
|
||
}
|
||
}
|
||
|
||
// flush 剩余；buffer 为空时也补一次 position 落盘（推进被失败行消费掉的区间）
|
||
if (!empty($buffer)) {
|
||
$flushed = $this->flushBatch($parser, $reader, $file, $buffer, $patterns, $fails, $samples);
|
||
$runs += $flushed['runs'];
|
||
$watches += $flushed['watches'];
|
||
} else {
|
||
$position = $reader->getCurrentPosition();
|
||
$reader->savePosition(
|
||
$file,
|
||
$position['inode'],
|
||
$position['offset'],
|
||
$position['last_line_hash'],
|
||
$fails,
|
||
$this->encodeSamples($samples)
|
||
);
|
||
}
|
||
|
||
// 超长行丢弃并入 fails（Reader 整行丢弃计数）
|
||
$fails += $reader->getDroppedLines();
|
||
|
||
// 文件导至 EOF 且 position 落盘后才可删除（只删已导完文件，未导完/尾部半行不删）
|
||
$isDeleted = false;
|
||
if ($reader->reachedEof()) {
|
||
clearstatcache(true, $file);
|
||
$position = $reader->getCurrentPosition();
|
||
$size = filesize($file);
|
||
if ($size !== false && $position['offset'] >= $size) {
|
||
$isDeleted = @unlink($file);
|
||
if ($isDeleted) {
|
||
Log::info('xhprof_log_import: 文件已导完删除 ' . $file);
|
||
}
|
||
}
|
||
}
|
||
|
||
Log::info("xhprof_log_import: 文件 {$file} 完成 runs={$runs} watches={$watches} fails={$fails}");
|
||
|
||
return ['runs' => $runs, 'watches' => $watches, 'fails' => $fails, 'deleted' => $isDeleted];
|
||
}
|
||
|
||
/**
|
||
* 构造 run 入库行（字段对齐 ul_xhprof_run，单位 μs/字节）.
|
||
*/
|
||
protected function buildRunRow(array $parsed): array
|
||
{
|
||
return [
|
||
'node_id' => \app\common\service\HostService::getNodeId(),
|
||
'url' => $parsed['url'],
|
||
'simple_url' => $parsed['simple_url'],
|
||
'method' => $parsed['method'],
|
||
'wall_time' => $parsed['wall_time'],
|
||
'cpu_time' => $parsed['cpu_time'],
|
||
'memory_peak' => $parsed['memory_peak'],
|
||
'request_time' => $parsed['request_time'],
|
||
'create_time' => time(),
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 批量入库：同一事务内 run 逐条 insert（拿自增 id）→ detail/run_watch insertAll → savePosition.
|
||
*
|
||
* 事务语义：入库与 position 落盘要么同时生效、要么同时回滚——
|
||
* 崩溃恢复后从 position 处重读，不会重复导入已提交批次。
|
||
*
|
||
* @param array $buffer [['run' => array, 'raw' => string, 'profile' => array], ...]
|
||
*
|
||
* @return array{runs: int, watches: int}
|
||
*/
|
||
protected function flushBatch(
|
||
\app\common\service\XhprofProfileParserService $parser,
|
||
\app\common\service\XhprofLogReaderService $reader,
|
||
string $file,
|
||
array $buffer,
|
||
array $patterns,
|
||
int $fails,
|
||
array $samples
|
||
): array {
|
||
// 捕获当前安全偏移（最后一条已 yield 完整行末尾），事务内与数据同落盘
|
||
$position = $reader->getCurrentPosition();
|
||
$samplesJson = $this->encodeSamples($samples);
|
||
$runs = 0;
|
||
$watches = 0;
|
||
|
||
Db::transaction(function () use ($parser, $reader, $file, $buffer, $patterns, $position, $fails, $samplesJson, &$runs, &$watches) {
|
||
$detailBatch = [];
|
||
$watchBatch = [];
|
||
$now = time();
|
||
|
||
foreach ($buffer as $item) {
|
||
$runId = (int) XhprofRun::insertGetId($item['run']);
|
||
$runs++;
|
||
|
||
$detailBatch[] = [
|
||
'id' => $runId,
|
||
'profile_data' => $item['raw'],
|
||
];
|
||
|
||
foreach ($parser->materializeWatches($item['profile'], $patterns) as $watchId => $agg) {
|
||
$watchBatch[] = [
|
||
'run_id' => $runId,
|
||
'watch_id' => $watchId,
|
||
'ct' => $agg['ct'],
|
||
'wt_sum' => $agg['wt_sum'],
|
||
'wt_max' => $agg['wt_max'],
|
||
'create_time' => $now,
|
||
];
|
||
$watches++;
|
||
}
|
||
}
|
||
|
||
if (!empty($detailBatch)) {
|
||
XhprofRunDetail::insertAll($detailBatch);
|
||
}
|
||
if (!empty($watchBatch)) {
|
||
XhprofRunWatch::insertAll($watchBatch);
|
||
}
|
||
|
||
$reader->savePosition(
|
||
$file,
|
||
$position['inode'],
|
||
$position['offset'],
|
||
$position['last_line_hash'],
|
||
$fails,
|
||
$samplesJson
|
||
);
|
||
});
|
||
|
||
return ['runs' => $runs, 'watches' => $watches];
|
||
}
|
||
|
||
/**
|
||
* 收集解析失败样本（截断至 500 字符，最多 FAIL_SAMPLES_MAX 条）.
|
||
*/
|
||
protected function addFailSample(array &$samples, string $line): void
|
||
{
|
||
if (count($samples) >= self::FAIL_SAMPLES_MAX) {
|
||
return;
|
||
}
|
||
$samples[] = mb_substr($line, 0, 500);
|
||
}
|
||
|
||
/**
|
||
* 失败样本编码为 json（空样本返回 null，与 position 列 nullable 对齐）.
|
||
*/
|
||
protected function encodeSamples(array $samples): ?string
|
||
{
|
||
return empty($samples) ? null : json_encode($samples, JSON_UNESCAPED_UNICODE);
|
||
}
|
||
}
|