Hyperf对接报表 帆布报表涉及敏感财务数据时，如何在 HyperF 框架中设计报表访问的审计日志系统？请说明日志采集、存储和告警的整体方案

选型： hyperf/async-queue 异步落库 + monolog/monolog 结构化日志 + elasticsearch/elasticsearch 存储检索 + hyperf/signal 告警 --- 架构总览 请求 └─ AuditMiddleware# 无侵入采集，异步投递└─ AsyncQueue# 削峰，不阻塞主流程├─ ES Writer# 结构化存储，支持全文检索├─ MySQL Writer# 合规留存，不可篡改└─ AlertEngine# 规则匹配 → 钉钉/邮件告警--- 一、审计中间件 — 无侵入采集<?php // app/Middleware/AuditMiddleware.php namespace App\Middleware;use App\Job\AuditLogJob;use Hyperf\AsyncQueue\Driver\DriverFactory;use Hyperf\Context\Context;use Psr\Http\Message\ResponseInterface;use Psr\Http\Message\ServerRequestInterface;use Psr\Http\Server\MiddlewareInterface;use Psr\Http\Server\RequestHandlerInterface;class AuditMiddleware implements MiddlewareInterface{publicfunction__construct(privatereadonlyDriverFactory$queue){}publicfunctionprocess(ServerRequestInterface$request, RequestHandlerInterface$handler): ResponseInterface{$start=microtime(true);$response=$handler->handle($request);$auth=Context::get('auth',[]);// 异步投递，主流程零延迟$this->queue->get('audit')->push(new AuditLogJob(['trace_id'=>Context::get('trace_id', uniqid('tr_',true)),'user_id'=>$auth['user_id']??0,'tenant_id'=>$auth['tenant_id']??0,'roles'=>$auth['roles']??[],'action'=>$this->resolveAction($request),'uri'=>(string)$request->getUri(),'method'=>$request->getMethod(),'params'=>$this->safeParams($request),'ip'=>$request->getServerParams()['remote_addr']??'','status'=>$response->getStatusCode(),'duration_ms'=>(int)((microtime(true)-$start)*1000),'timestamp'=>date('c'),]));return$response;}// 脱敏：移除敏感字段 privatefunctionsafeParams(ServerRequestInterface$request): array{$body=(array)($request->getParsedBody()??[]);$query=$request->getQueryParams();$blocked=['password','token','secret','card_no','id_card'];$mask=fn($v,$k)=>in_array($k,$blocked)?'***':$v;returnarray_map($mask, array_merge($query,$body), array_keys(array_merge($query,$body)));}privatefunctionresolveAction(ServerRequestInterface$request): string{returnmatch(true){str_contains($request->getUri()->getPath(),'/export')=>'report.export', str_contains($request->getUri()->getPath(),'/view')=>'report.view', default=>'report.access',};}}--- 二、异步 Job — 双写 ES + MySQL<?php // app/Job/AuditLogJob.php namespace App\Job;use Elastic\Elasticsearch\ClientBuilder;use Hyperf\AsyncQueue\Job;use Hyperf\DbConnection\Db;class AuditLogJob extends Job{public int$maxAttempts=3;publicfunction__construct(privatereadonlyarray$log){}publicfunctionhandle(): void{// 并发双写，互不阻塞\Hyperf\Coroutine\parallel([fn()=>$this->writeEs(), fn()=>$this->writeMysql(), fn()=>$this->checkAlert(),]);}privatefunctionwriteEs(): void{$client=ClientBuilder::create()->setHosts([env('ES_HOST','localhost:9200')])->build();$client->index(['index'=>'audit-'.date('Y.m'), // 按月滚动索引'id'=>$this->log['trace_id'],'body'=>$this->log,]);}privatefunctionwriteMysql(): void{Db::table('audit_logs')->insert([...$this->log,'roles'=>json_encode($this->log['roles']),'params'=>json_encode($this->log['params']),]);}privatefunctioncheckAlert(): void{(new AlertEngine($this->log))->evaluate();}}--- 三、告警引擎 — 规则驱动<?php // app/Audit/AlertEngine.php namespace App\Audit;use Hyperf\DbConnection\Db;class AlertEngine{// 告警规则：[描述, 判断闭包, 级别]private array$rules;publicfunction__construct(privatereadonlyarray$log){$this->rules=[['大批量导出', fn()=>$this->isBulkExport(),'warning'],['非工作时间访问', fn()=>$this->isOffHours(),'info'],['高频访问', fn()=>$this->isHighFreq(),'warning'],['异地登录', fn()=>$this->isAnomalyIp(),'critical'],];}publicfunctionevaluate(): void{foreach($this->rules as[$desc,$check,$level]){if($check()){$this->fire($desc,$level);}}}privatefunctionisBulkExport(): bool{return$this->log['action']==='report.export'&&($this->log['params']['limit']??0)>10000;}privatefunctionisOffHours(): bool{$hour=(int)date('G', strtotime($this->log['timestamp']));return$hour<8||$hour>20;}privatefunctionisHighFreq(): bool{$key="audit:freq:{$this->log['user_id']}:".date('YmdH');$count=redis()->incr($key);redis()->expire($key,3600);return$count>100;//1小时内超100次}privatefunctionisAnomalyIp(): bool{$lastIp=Db::table('audit_logs')->where('user_id',$this->log['user_id'])->orderByDesc('id')->skip(1)->value('ip');return$lastIp&&$lastIp!==$this->log['ip']&&$this->ipRegion($lastIp)!==$this->ipRegion($this->log['ip']);}privatefunctionipRegion(string$ip): string{// 接入 ip2region 或调用内网 IP 归属服务returnsubstr($ip,0, strrpos($ip,'.'));// 简化：取 /24 段}privatefunctionfire(string$desc, string$level): void{$payload=['level'=>$level,'desc'=>$desc,'user_id'=>$this->log['user_id'],'tenant_id'=>$this->log['tenant_id'],'ip'=>$this->log['ip'],'action'=>$this->log['action'],'time'=>$this->log['timestamp'],'trace_id'=>$this->log['trace_id'],];// 写告警表（持久化） Db::table('audit_alerts')->insert([...$payload,'notified'=>0]);// 钉钉 Webhook（critical 立即推送）if($level==='critical'){$this->dingTalk($payload);}}privatefunctiondingTalk(array$payload): void{$webhook=env('DINGTALK_WEBHOOK');$text="[{$payload['level']}] {$payload['desc']}\n"."用户: {$payload['user_id']} | IP: {$payload['ip']}\n"."时间: {$payload['time']} | TraceID: {$payload['trace_id']}";\Hyperf\Support\make(\GuzzleHttp\Client::class)->post($webhook,['json'=>['msgtype'=>'text','text'=>['content'=>$text]],]);}}--- 四、日志查询接口 — ES 检索<?php // app/Controller/AuditController.php namespace App\Controller;use Elastic\Elasticsearch\ClientBuilder;use Hyperf\HttpServer\Annotation\Controller;use Hyperf\HttpServer\Annotation\Get;use Hyperf\HttpServer\Contract\RequestInterface;#[Controller(prefix: '/audit')]class AuditController{#[Get('/logs', options: ['permission' => 'audit:read'])]publicfunctionlogs(RequestInterface$request): array{$p=$request->query();$must=[];isset($p['user_id'])&&$must[]=['term'=>['user_id'=>$p['user_id']]];isset($p['tenant_id'])&&$must[]=['term'=>['tenant_id'=>$p['tenant_id']]];isset($p['action'])&&$must[]=['term'=>['action'=>$p['action']]];isset($p['keyword'])&&$must[]=['multi_match'=>['query'=>$p['keyword'],'fields'=>['uri','params','ip'],]];if(isset($p['start'],$p['end'])){$must[]=['range'=>['timestamp'=>['gte'=>$p['start'],'lte'=>$p['end']]]];}$result=ClientBuilder::create()->setHosts([env('ES_HOST')])->build()->search(['index'=>'audit-*','body'=>['query'=>['bool'=>['must'=>$must]],'sort'=>[['timestamp'=>'desc']],'from'=>(int)($p['page']??0)*20,'size'=>20,],]);return['total'=>$result['hits']['total']['value'],'list'=>array_column($result['hits']['hits'],'_source'),];}}--- 五、MySQL 审计表 — 合规留存 CREATE TABLE audit_logs(idBIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, trace_id VARCHAR(40)NOT NULL, user_id INT UNSIGNED NOT NULL, tenant_id INT UNSIGNED NOT NULL, roles JSON, action VARCHAR(64)NOT NULL, uri VARCHAR(512)NOT NULL, method VARCHAR(8), params JSON,ipVARCHAR(45), status SMALLINT, duration_ms INT, timestamp DATETIME(3)NOT NULL, INDEX idx_user(user_id, timestamp), INDEX idx_tenant(tenant_id, timestamp), INDEX idx_action(action, timestamp))ENGINE=InnoDBROW_FORMAT=COMPRESSED -- 压缩存储，财务日志量大COMMENT='审计日志，禁止业务删改';--- 六、整体数据流 请求 → AuditMiddleware(同步采集，<1ms)↓ 异步投递 audit队列(Redis)↓ 消费 AuditLogJob::handle()├─ parallel协程1 → ES(毫秒级写入，支持全文检索)├─ parallel协程2 → MySQL(合规留存，不可篡改)└─ parallel协程3 → AlertEngine ├─ 规则1: 批量导出 → 写alert表 ├─ 规则2: 非工作时间 → 写alert表 ├─ 规则3: 高频(Redis计数)→ 写alert表 └─ 规则4: 异地IP(critical)→ 钉钉即时推送 --- 七、关键设计决策 ┌───────────────┬──────────────────────┬──────────────────────────────────────────────┐ │ 问题 │ 决策 │ 原因 │ ├───────────────┼──────────────────────┼──────────────────────────────────────────────┤ │ 日志写入方式 │ 异步队列 │ 主流程零延迟，削峰防 DB 压力 │ ├───────────────┼──────────────────────┼──────────────────────────────────────────────┤ │ 双写 ES+MySQL │ ES 检索 + MySQL 合规 │ ES 支持全文/范围查询，MySQL 满足审计留存要求 │ ├───────────────┼──────────────────────┼──────────────────────────────────────────────┤ │ 告警触发时机 │ Job 内同步判断 │ 避免二次队列延迟，critical 级秒级推送 │ ├───────────────┼──────────────────────┼──────────────────────────────────────────────┤ │ 高频检测 │ Redis incr + expire │ 原子操作，协程安全，无锁 │ ├───────────────┼──────────────────────┼──────────────────────────────────────────────┤ │ 参数脱敏 │ 中间件层统一处理 │ 敏感字段不进日志，合规要求 │ └───────────────┴──────────────────────┴──────────────────────────────────────────────┘