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 多节点部署注意事项
307 lines
14 KiB
PHP
307 lines
14 KiB
PHP
<?php
|
||
|
||
namespace base\common\service;
|
||
|
||
use app\admin\model\NginxAccessLog;
|
||
use app\admin\model\NginxStatHour;
|
||
use app\admin\model\NginxStatReferer;
|
||
use app\admin\model\NginxStatUa;
|
||
use app\admin\model\NginxStatUrl;
|
||
use think\facade\Db;
|
||
|
||
/**
|
||
* Nginx 访问日志聚合统计 Service（Base 内核）.
|
||
*
|
||
* 设计要点：
|
||
* - 幂等：delete-then-insert 模式（不用 INSERT ON DUPLICATE KEY UPDATE，避免翻倍）
|
||
* - 事务：Db::startTrans / commit / rollback 包裹所有 DELETE + INSERT
|
||
* - top N 截断后不保留长尾（明确接受，不在 sysconfig 控制"是否截断"）
|
||
* - 依赖倒置：内部 use app 入口 model（NginxStatHour 等），不 use Base model
|
||
* - 只读 raw 表（ul_nginx_access_log），不做 nginx log line 解析（那是 T5 Parser 的事）
|
||
*/
|
||
class NginxLogAggregatorServiceBase
|
||
{
|
||
/**
|
||
* 聚合指定日期的指定小时.
|
||
*
|
||
* 流程（每张表生成 全局行 node_id='' + 各节点行）：
|
||
* 1. stat_hour：DELETE WHERE stat_date=:d AND stat_hour=:h → INSERT 全局 1 行 + 每节点 1 行
|
||
* 2. stat_url / stat_referer / stat_ua：DELETE WHERE stat_date=:d（整天重跑）→ INSERT 全局 topN + 每节点 topN
|
||
*
|
||
* @param int $statDate 日期 YYYYMMDD（如 20260728）
|
||
* @param int $statHour 小时 0-23
|
||
* @return array {stat_hour_rows:int, stat_url_rows:int, stat_referer_rows:int, stat_ua_rows:int}
|
||
*
|
||
* @throws \InvalidArgumentException 日期/小时非法
|
||
* @throws \Throwable 事务内任何错误（已 rollback 后 rethrow）
|
||
*/
|
||
public function aggregateHour(int $statDate, int $statHour): array
|
||
{
|
||
if ($statHour < 0 || $statHour > 23) {
|
||
throw new \InvalidArgumentException("stat_hour must be 0-23, got: {$statHour}");
|
||
}
|
||
|
||
// 时间窗口
|
||
$dateStr = $this->formatDateStr($statDate);
|
||
$dayStart = strtotime("{$dateStr} 00:00:00");
|
||
if ($dayStart === false) {
|
||
throw new \InvalidArgumentException("invalid stat_date: {$statDate}");
|
||
}
|
||
$dayEnd = $dayStart + 86400;
|
||
$hourStart = $dayStart + $statHour * 3600;
|
||
$hourEnd = $hourStart + 3600;
|
||
|
||
// topN 配置（topn=0 时不限制）
|
||
$topn = (int) sysconfig('nginx_log', 'topn_default', 10);
|
||
|
||
// 表名（带前缀）。框架默认连接是 main（不是 mysql），按 default 取连接 prefix
|
||
$conn = (string) config('database.default', 'main');
|
||
$prefix = (string) config("database.connections.{$conn}.prefix", 'ul_');
|
||
$rawTable = $prefix . 'nginx_access_log';
|
||
$hourTable = $prefix . 'nginx_stat_hour';
|
||
$urlTable = $prefix . 'nginx_stat_url';
|
||
$refererTable = $prefix . 'nginx_stat_referer';
|
||
$uaTable = $prefix . 'nginx_stat_ua';
|
||
|
||
Db::startTrans();
|
||
try {
|
||
// DELETE（逻辑不变）：stat_hour 按小时，stat_url/referer/ua 按整天
|
||
NginxStatHour::where('stat_date', $statDate)
|
||
->where('stat_hour', $statHour)
|
||
->delete();
|
||
NginxStatUrl::where('stat_date', $statDate)->delete();
|
||
NginxStatReferer::where('stat_date', $statDate)->delete();
|
||
NginxStatUa::where('stat_date', $statDate)->delete();
|
||
|
||
$statHourRows = 0;
|
||
$statUrlRows = 0;
|
||
$statRefererRows = 0;
|
||
$statUaRows = 0;
|
||
|
||
// 批次 A：全局行（node_id=''，跨所有节点聚合，UV 精确）
|
||
$global = $this->insertAggregatesForScope(
|
||
'', null,
|
||
$statDate, $statHour, $hourStart, $hourEnd, $dayStart, $dayEnd, $topn,
|
||
$rawTable, $hourTable, $urlTable, $refererTable, $uaTable
|
||
);
|
||
$statHourRows += $global['stat_hour'];
|
||
$statUrlRows += $global['stat_url'];
|
||
$statRefererRows += $global['stat_referer'];
|
||
$statUaRows += $global['stat_ua'];
|
||
|
||
// 批次 B：按节点行（循环每个 node_id）
|
||
// 用 day 窗口取节点列表：保证 day 窗口表（url/referer/ua）的 per-node 行不会因
|
||
// 某节点"本小时无活动"而被 DELETE 后丢失；stat_hour 的 per-node 行由
|
||
// HAVING COUNT(*)>0 自然过滤掉本小时无活动的节点。
|
||
$nodeRows = Db::query(
|
||
"SELECT DISTINCT node_id FROM `{$rawTable}` WHERE time_local >= :day_start AND time_local < :day_end AND node_id != ''",
|
||
['day_start' => $dayStart, 'day_end' => $dayEnd]
|
||
);
|
||
foreach ($nodeRows as $nodeRow) {
|
||
$nid = (string) $nodeRow['node_id'];
|
||
$perNode = $this->insertAggregatesForScope(
|
||
$nid, $nid,
|
||
$statDate, $statHour, $hourStart, $hourEnd, $dayStart, $dayEnd, $topn,
|
||
$rawTable, $hourTable, $urlTable, $refererTable, $uaTable
|
||
);
|
||
$statHourRows += $perNode['stat_hour'];
|
||
$statUrlRows += $perNode['stat_url'];
|
||
$statRefererRows += $perNode['stat_referer'];
|
||
$statUaRows += $perNode['stat_ua'];
|
||
}
|
||
|
||
Db::commit();
|
||
|
||
return [
|
||
'stat_hour_rows' => $statHourRows,
|
||
'stat_url_rows' => $statUrlRows,
|
||
'stat_referer_rows' => $statRefererRows,
|
||
'stat_ua_rows' => $statUaRows,
|
||
];
|
||
} catch (\Throwable $e) {
|
||
Db::rollback();
|
||
throw $e;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 为单个作用域（全局或某节点）执行 4 段 INSERT.
|
||
*
|
||
* - 全局作用域：$nodeId='' 且 $nodeFilter=null（不追加 node WHERE，跨所有节点聚合）
|
||
* - 节点作用域：$nodeId=节点ID 且 $nodeFilter=节点ID（追加 AND node_id=:node_filter）
|
||
*
|
||
* stat_hour：1 行（无 GROUP BY，HAVING COUNT(*)>0）
|
||
* stat_url / stat_referer / stat_ua：topN（各自 GROUP BY + LIMIT）
|
||
*
|
||
* @param string $nodeId INSERT 的 node_id 列值（全局为 ''）
|
||
* @param string|null $nodeFilter null=全局（不追加 WHERE）；非 null=追加节点过滤
|
||
* @return array {stat_hour:int, stat_url:int, stat_referer:int, stat_ua:int}
|
||
*/
|
||
protected function insertAggregatesForScope(
|
||
string $nodeId,
|
||
?string $nodeFilter,
|
||
int $statDate,
|
||
int $statHour,
|
||
int $hourStart,
|
||
int $hourEnd,
|
||
int $dayStart,
|
||
int $dayEnd,
|
||
int $topn,
|
||
string $rawTable,
|
||
string $hourTable,
|
||
string $urlTable,
|
||
string $refererTable,
|
||
string $uaTable
|
||
): array {
|
||
// 节点过滤片段：全局行无；节点行追加 AND node_id = :node_filter
|
||
$nodeWhere = $nodeFilter === null ? '' : 'AND node_id = :node_filter';
|
||
|
||
// hour 窗口绑定（stat_hour 专用）
|
||
$hourParams = [
|
||
'node_id' => $nodeId,
|
||
'stat_date' => $statDate,
|
||
'stat_hour' => $statHour,
|
||
'hour_start' => $hourStart,
|
||
'hour_end' => $hourEnd,
|
||
];
|
||
if ($nodeFilter !== null) {
|
||
$hourParams['node_filter'] = $nodeFilter;
|
||
}
|
||
|
||
// day 窗口绑定（stat_url/referer/ua 共用）
|
||
$dayParams = [
|
||
'node_id' => $nodeId,
|
||
'stat_date' => $statDate,
|
||
'day_start' => $dayStart,
|
||
'day_end' => $dayEnd,
|
||
];
|
||
if ($nodeFilter !== null) {
|
||
$dayParams['node_filter'] = $nodeFilter;
|
||
}
|
||
|
||
// 1. stat_hour：1 行（无 GROUP BY）
|
||
$hourSql = "INSERT INTO `{$hourTable}`
|
||
(node_id, stat_date, stat_hour, pv, uv, total_bytes, status_2xx, status_3xx, status_4xx, status_5xx, status_other, avg_request_time)
|
||
SELECT :node_id, :stat_date, :stat_hour,
|
||
COUNT(*) AS pv,
|
||
COUNT(DISTINCT remote_addr) AS uv,
|
||
COALESCE(SUM(body_bytes_sent), 0) AS total_bytes,
|
||
COALESCE(SUM(CASE WHEN status BETWEEN 200 AND 299 THEN 1 ELSE 0 END), 0) AS status_2xx,
|
||
COALESCE(SUM(CASE WHEN status BETWEEN 300 AND 399 THEN 1 ELSE 0 END), 0) AS status_3xx,
|
||
COALESCE(SUM(CASE WHEN status BETWEEN 400 AND 499 THEN 1 ELSE 0 END), 0) AS status_4xx,
|
||
COALESCE(SUM(CASE WHEN status BETWEEN 500 AND 599 THEN 1 ELSE 0 END), 0) AS status_5xx,
|
||
COALESCE(SUM(CASE WHEN status < 200 OR status > 599 THEN 1 ELSE 0 END), 0) AS status_other,
|
||
COALESCE(AVG(request_time), 0) AS avg_request_time
|
||
FROM `{$rawTable}`
|
||
WHERE time_local >= :hour_start AND time_local < :hour_end {$nodeWhere}
|
||
HAVING COUNT(*) > 0";
|
||
$statHourRows = (int) Db::execute($hourSql, $hourParams);
|
||
|
||
// 2. stat_url：topN（GROUP BY uri）
|
||
$urlLimit = $topn > 0 ? 'LIMIT ' . $topn : '';
|
||
$urlSql = "INSERT INTO `{$urlTable}`
|
||
(node_id, stat_date, uri, pv, uv, total_bytes, avg_request_time)
|
||
SELECT :node_id, :stat_date, uri,
|
||
COUNT(*) AS pv,
|
||
COUNT(DISTINCT remote_addr) AS uv,
|
||
COALESCE(SUM(body_bytes_sent), 0) AS total_bytes,
|
||
COALESCE(AVG(request_time), 0) AS avg_request_time
|
||
FROM `{$rawTable}`
|
||
WHERE time_local >= :day_start AND time_local < :day_end {$nodeWhere}
|
||
GROUP BY uri
|
||
ORDER BY pv DESC
|
||
{$urlLimit}";
|
||
$statUrlRows = (int) Db::execute($urlSql, $dayParams);
|
||
|
||
// 3. stat_referer：topN（从 http_referer 提取 domain）
|
||
$refererLimit = $topn > 0 ? 'LIMIT ' . $topn : '';
|
||
$refererSql = "INSERT INTO `{$refererTable}`
|
||
(node_id, stat_date, referer_domain, pv, uv)
|
||
SELECT :node_id, :stat_date,
|
||
CASE
|
||
WHEN http_referer IS NULL OR http_referer = '' OR http_referer = '-' THEN '-'
|
||
ELSE SUBSTRING_INDEX(
|
||
SUBSTRING_INDEX(
|
||
REPLACE(REPLACE(http_referer, 'https://', ''), 'http://', ''),
|
||
'/', 1
|
||
),
|
||
'?', 1
|
||
)
|
||
END AS referer_domain,
|
||
COUNT(*) AS pv,
|
||
COUNT(DISTINCT remote_addr) AS uv
|
||
FROM `{$rawTable}`
|
||
WHERE time_local >= :day_start AND time_local < :day_end {$nodeWhere}
|
||
GROUP BY referer_domain
|
||
ORDER BY pv DESC
|
||
{$refererLimit}";
|
||
$statRefererRows = (int) Db::execute($refererSql, $dayParams);
|
||
|
||
// 4. stat_ua：topN（从 http_user_agent 分类）
|
||
$uaLimit = $topn > 0 ? 'LIMIT ' . $topn : '';
|
||
$uaSql = "INSERT INTO `{$uaTable}`
|
||
(node_id, stat_date, ua_type, ua_name, pv, uv)
|
||
SELECT :node_id, :stat_date,
|
||
CASE
|
||
WHEN http_user_agent LIKE '%bot%'
|
||
OR http_user_agent LIKE '%spider%'
|
||
OR http_user_agent LIKE '%crawl%'
|
||
OR http_user_agent LIKE '%slurp%'
|
||
OR http_user_agent LIKE '%bingpreview%'
|
||
OR http_user_agent LIKE '%facebookexternalhit%'
|
||
OR http_user_agent LIKE '%twitterbot%' THEN 'spider'
|
||
WHEN http_user_agent IS NULL OR http_user_agent = '' OR http_user_agent = '-' THEN 'unknown'
|
||
WHEN http_user_agent LIKE '%Mozilla%'
|
||
OR http_user_agent LIKE '%Chrome%'
|
||
OR http_user_agent LIKE '%Safari%'
|
||
OR http_user_agent LIKE '%Firefox%'
|
||
OR http_user_agent LIKE '%Edg%'
|
||
OR http_user_agent LIKE '%Opera%'
|
||
OR http_user_agent LIKE '%MSIE%'
|
||
OR http_user_agent LIKE '%Trident%' THEN 'browser'
|
||
ELSE 'unknown'
|
||
END AS ua_type,
|
||
CASE
|
||
WHEN http_user_agent LIKE '%Googlebot%' THEN 'Googlebot'
|
||
WHEN http_user_agent LIKE '%Baiduspider%' THEN 'Baiduspider'
|
||
WHEN http_user_agent LIKE '%bingbot%' THEN 'Bingbot'
|
||
WHEN http_user_agent LIKE '%DuckDuckBot%' THEN 'DuckDuckBot'
|
||
WHEN http_user_agent LIKE '%YandexBot%' THEN 'YandexBot'
|
||
WHEN http_user_agent LIKE '%Edg/%' THEN 'Microsoft Edge'
|
||
WHEN http_user_agent LIKE '%OPR/%' OR http_user_agent LIKE '%Opera%' THEN 'Opera'
|
||
WHEN http_user_agent LIKE '%Firefox/%' THEN 'Firefox'
|
||
WHEN http_user_agent LIKE '%Chrome/%' THEN 'Chrome'
|
||
WHEN http_user_agent LIKE '%Safari/%' THEN 'Safari'
|
||
WHEN http_user_agent LIKE '%MSIE%' OR http_user_agent LIKE '%Trident%' THEN 'Internet Explorer'
|
||
WHEN http_user_agent IS NULL OR http_user_agent = '' OR http_user_agent = '-' THEN 'Unknown'
|
||
ELSE 'Other'
|
||
END AS ua_name,
|
||
COUNT(*) AS pv,
|
||
COUNT(DISTINCT remote_addr) AS uv
|
||
FROM `{$rawTable}`
|
||
WHERE time_local >= :day_start AND time_local < :day_end {$nodeWhere}
|
||
GROUP BY ua_type, ua_name
|
||
ORDER BY pv DESC
|
||
{$uaLimit}";
|
||
$statUaRows = (int) Db::execute($uaSql, $dayParams);
|
||
|
||
return [
|
||
'stat_hour' => $statHourRows,
|
||
'stat_url' => $statUrlRows,
|
||
'stat_referer' => $statRefererRows,
|
||
'stat_ua' => $statUaRows,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 将 int 日期 YYYYMMDD 格式化为 "YYYY-MM-DD".
|
||
*/
|
||
protected function formatDateStr(int $statDate): string
|
||
{
|
||
$y = intdiv($statDate, 10000);
|
||
$m = intdiv($statDate % 10000, 100);
|
||
$d = $statDate % 100;
|
||
return sprintf('%04d-%02d-%02d', $y, $m, $d);
|
||
}
|
||
}
|