mirror of
https://gitee.com/ulthon/ulthon_admin.git
synced 2026-08-30 20:55:32 +08:00
8faa52ce90
- 6 张表加 node_id 字段 + position/stat 改复合唯一键（migration） - 6 个 Scheme 同步 node_id 注解 + 唯一键 - Reader 带 node_id 参数 + 懒加载接管（getLastPosition 回退查 node_id=''） - Aggregator 全局行（node_id=''）+ 按节点循环 insertAggregatesForScope - Stat 所有查询方法加 where node_id 条件 - Dashboard 控制器/视图加节点筛选下拉框 + AJAX 带 node_id - AccessLog 列表加节点列 + 采集进度卡片显示节点信息 - import 定时任务 run_type 改 all（每节点采集自己的日志） - v2.4.0 升级脚本追加 ALTER TABLE（幂等 check-then-alter） - 菜单调整：去掉 Nginx 顶级菜单 + 读取位置管理，改为系统管理下挂两个子菜单 - ulthon-timer 文档补充 run_type=all 多节点部署注意事项
283 lines
10 KiB
PHP
283 lines
10 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace base\tools\controller\timer;
|
||
|
||
use app\admin\model\NginxAccessLog;
|
||
use app\common\controller\TimerController;
|
||
use think\facade\Log;
|
||
|
||
/**
|
||
* nginx 访问日志导入定时任务（Base 层）.
|
||
*
|
||
* 流程：
|
||
* 1. env 总开关校验（严格布尔判断，详见 notepad Task 2 #14）
|
||
* 2. 读取 env 配置的日志文件清单
|
||
* 3. 依赖倒置：实例化 app 层 Parser / Reader
|
||
* 4. 每个文件首行格式自检 + 串行增量读取 + 批量入库
|
||
* 5. savePosition 写 parse_fail_count / parse_fail_samples
|
||
*
|
||
* 业务侧如需定制导入逻辑，重写 app/tools/controller/timer/NginxLogImport.php 对应方法即可拦截。
|
||
*/
|
||
class NginxLogImportBase extends TimerController
|
||
{
|
||
/**
|
||
* 防刷间隔（秒）.
|
||
* 仅做控制器侧防刷，不影响定时器侧 frequency 节流。
|
||
*/
|
||
protected $frequency = 60;
|
||
|
||
/**
|
||
* 并发上限 cap：单分片（id=0）。
|
||
*/
|
||
protected $concurrency = 1;
|
||
|
||
/**
|
||
* 批量 insert chunk 大小。
|
||
*/
|
||
protected const INSERT_CHUNK = 500;
|
||
|
||
/**
|
||
* 单次读取最大行数（与 Reader 默认上限一致）。
|
||
*/
|
||
protected const READ_MAX_LINES = 100000;
|
||
|
||
/**
|
||
* parse_fail_samples 最大记录条数。
|
||
*/
|
||
protected const FAIL_SAMPLES_MAX = 10;
|
||
|
||
public function do()
|
||
{
|
||
// 1. 严格布尔判断（env convert 'true'→true / 'false'→false）
|
||
// 见 notepad Task 2 #14：必须 !== true，不能用 !== 'true' 字符串比较
|
||
if (env('nginx_log.enable') !== true) {
|
||
Log::debug('nginx_log disabled, skip');
|
||
return 'disabled';
|
||
}
|
||
|
||
// 2. 读日志文件清单（分号分隔）
|
||
$logFilesConfig = (string) env('nginx_log.log_files', '');
|
||
$files = array_filter(array_map('trim', explode(';', $logFilesConfig)), function ($f) {
|
||
return $f !== '';
|
||
});
|
||
if (empty($files)) {
|
||
return 'no_files';
|
||
}
|
||
|
||
// 3. 依赖倒置：使用 app 层入口类（业务侧可重写拦截）
|
||
$parser = new \app\common\service\NginxLogParserService();
|
||
$reader = new \app\common\service\NginxLogReaderService();
|
||
|
||
// 静态资源过滤开关
|
||
$excludeStatic = (int) sysconfig('nginx_log', 'exclude_static', 1) === 1;
|
||
|
||
$totalFiles = 0;
|
||
$totalLines = 0;
|
||
$totalFails = 0;
|
||
$errors = [];
|
||
|
||
foreach ($files as $file) {
|
||
try {
|
||
if (!is_file($file)) {
|
||
throw new \RuntimeException("文件不存在: {$file}");
|
||
}
|
||
|
||
// 4. 首行格式自检（避免对非 nginx 日志文件做无意义解析）
|
||
// 取首个非空行喂给 parser->isFormatMatch
|
||
$sampleLine = $this->readFirstNonEmptyLine($file);
|
||
if ($sampleLine === null || !$parser->isFormatMatch($sampleLine)) {
|
||
Log::error("nginx_log_import: 格式不匹配,跳过文件 {$file}");
|
||
$errors[] = "format_mismatch:{$file}";
|
||
continue;
|
||
}
|
||
|
||
$totalFiles++;
|
||
|
||
// 5. 串行处理：增量读取 + 解析 + 批量入库
|
||
$stats = $this->processFile($parser, $reader, $file, $excludeStatic);
|
||
$totalLines += $stats['lines'];
|
||
$totalFails += $stats['fails'];
|
||
} catch (\Throwable $e) {
|
||
Log::error("nginx_log_import: 处理文件异常 {$file} - " . $e->getMessage());
|
||
$errors[] = 'error:' . basename($file) . ':' . $e->getMessage();
|
||
}
|
||
}
|
||
|
||
return json_encode([
|
||
'files' => $totalFiles,
|
||
'lines' => $totalLines,
|
||
'fails' => $totalFails,
|
||
'errors' => $errors,
|
||
], JSON_UNESCAPED_UNICODE);
|
||
}
|
||
|
||
/**
|
||
* 处理单个文件：增量读取、解析、过滤、批量入库、savePosition.
|
||
*
|
||
* @return array{lines:int,fails:int}
|
||
*/
|
||
protected function processFile(
|
||
\app\common\service\NginxLogParserService $parser,
|
||
\app\common\service\NginxLogReaderService $reader,
|
||
string $file,
|
||
bool $excludeStatic
|
||
): array {
|
||
$lines = 0;
|
||
$fails = 0;
|
||
$samples = [];
|
||
$batch = [];
|
||
$lastLineHash = null;
|
||
$inode = 0;
|
||
$offset = 0;
|
||
|
||
// Reader 是 Generator，让其自然走完（maxLines / EOF 会自动 savePosition）
|
||
// 见 notepad Task 6 #2：不要中途 break，否则 maxLines/EOF 的 savePosition 都不执行
|
||
foreach ($reader->read($file, self::READ_MAX_LINES) as $line) {
|
||
// 跟踪最后处理的行（用于失败时 savePosition）
|
||
$lastLineHash = md5($line);
|
||
|
||
$parsed = $parser->parse($line);
|
||
if ($parsed === null) {
|
||
$fails++;
|
||
if (count($samples) < self::FAIL_SAMPLES_MAX) {
|
||
$samples[] = mb_substr($line, 0, 500);
|
||
}
|
||
continue;
|
||
}
|
||
|
||
// 静态资源排除（在 parse 成功后判断，避免与 parse 失败混淆）
|
||
if ($excludeStatic && $parser->isStaticResource($parsed['uri'])) {
|
||
continue;
|
||
}
|
||
|
||
$batch[] = $this->buildRow($parsed, $file);
|
||
|
||
if (count($batch) >= self::INSERT_CHUNK) {
|
||
$this->insertBatch($batch);
|
||
$lines += count($batch);
|
||
$batch = [];
|
||
}
|
||
}
|
||
|
||
// flush 剩余
|
||
if (!empty($batch)) {
|
||
$this->insertBatch($batch);
|
||
$lines += count($batch);
|
||
$batch = [];
|
||
}
|
||
|
||
// 二次 savePosition：把本批解析失败信息写到 position（Reader 已经写过 offset，这里只更新 fail 字段）
|
||
// 见 notepad Task 6 #3：Reader 默认写 failCount=0；本批若有失败需要覆盖
|
||
if ($fails > 0) {
|
||
$this->updateFailStats($reader, $file, $fails, $samples);
|
||
}
|
||
|
||
Log::info("nginx_log_import: 文件 {$file} 完成 lines={$lines} fails={$fails}");
|
||
|
||
return ['lines' => $lines, 'fails' => $fails];
|
||
}
|
||
|
||
/**
|
||
* 构造入库行：只保留 raw 表实际字段（过滤 parser 多余的 referer_domain/ua_type/ua_name）。
|
||
*
|
||
* raw 表字段（见 app/admin/scheme/NginxAccessLog.php）：
|
||
* file_path / remote_addr / remote_user / time_local / method / uri / query_string /
|
||
* http_version / status / body_bytes_sent / http_referer / http_user_agent /
|
||
* request_time / upstream_response_time / bytes_sent / country / province / city / create_time
|
||
*/
|
||
protected function buildRow(array $parsed, string $filePath): array
|
||
{
|
||
$now = time();
|
||
return [
|
||
'node_id' => \app\common\service\HostService::getNodeId(),
|
||
'file_path' => $filePath,
|
||
'remote_addr' => $parsed['remote_addr'],
|
||
'remote_user' => $parsed['remote_user'] ?? '',
|
||
'time_local' => $parsed['time_local'],
|
||
'method' => $parsed['method'],
|
||
'uri' => $parsed['uri'],
|
||
'query_string' => $parsed['query_string'] ?? '',
|
||
'http_version' => $parsed['http_version'] ?? '',
|
||
'status' => $parsed['status'],
|
||
'body_bytes_sent' => $parsed['body_bytes_sent'],
|
||
'http_referer' => $parsed['http_referer'] ?? '',
|
||
'http_user_agent' => $parsed['http_user_agent'] ?? '',
|
||
'request_time' => $parsed['request_time'],
|
||
'upstream_response_time' => $parsed['upstream_response_time'],
|
||
'bytes_sent' => $parsed['bytes_sent'],
|
||
'country' => null,
|
||
'province' => null,
|
||
'city' => null,
|
||
'create_time' => $now,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 批量入库（使用 model 的 insertAll，自动走默认连接 + 表前缀）。
|
||
*/
|
||
protected function insertBatch(array $batch): void
|
||
{
|
||
if (empty($batch)) {
|
||
return;
|
||
}
|
||
// NginxAccessLog::insertAll 走 model，schema 字段映射可控；
|
||
// chunk 由调用方控制（INSERT_CHUNK=500），避免单条 SQL 过大
|
||
NginxAccessLog::insertAll($batch);
|
||
}
|
||
|
||
/**
|
||
* 更新 position 的失败统计（offset/inode 不变，仅覆盖 fail 字段）。
|
||
*
|
||
* 注意：Reader 在 maxLines/EOF 已经 savePosition（failCount=0），
|
||
* 这里取当前 position 后用相同 offset 重写，仅更新 failCount / failSamples。
|
||
*/
|
||
protected function updateFailStats(
|
||
\app\common\service\NginxLogReaderService $reader,
|
||
string $file,
|
||
int $fails,
|
||
array $samples
|
||
): void {
|
||
$position = $reader->getLastPosition($file);
|
||
if ($position === null) {
|
||
return; // Reader 已经保存过，这里取不到说明异常，放弃覆盖
|
||
}
|
||
$samplesJson = empty($samples) ? null : json_encode($samples, JSON_UNESCAPED_UNICODE);
|
||
$reader->savePosition(
|
||
$file,
|
||
(int) $position['inode'],
|
||
(int) $position['offset'],
|
||
$position['last_line_hash'] ?? null,
|
||
$fails,
|
||
$samplesJson
|
||
);
|
||
}
|
||
|
||
/**
|
||
* 读取文件首个非空行（用于格式自检）.
|
||
*
|
||
* @return string|null 首个非空行（trim 后），空文件返回 null
|
||
*/
|
||
protected function readFirstNonEmptyLine(string $file): ?string
|
||
{
|
||
$fp = @fopen($file, 'rb');
|
||
if ($fp === false) {
|
||
return null;
|
||
}
|
||
try {
|
||
while (($line = fgets($fp)) !== false) {
|
||
$line = trim($line);
|
||
if ($line !== '') {
|
||
return $line;
|
||
}
|
||
}
|
||
return null;
|
||
} finally {
|
||
if (is_resource($fp)) {
|
||
fclose($fp);
|
||
}
|
||
}
|
||
}
|
||
}
|