From adf728adfc1b55ff7119aed9174041f7cf089a1a Mon Sep 17 00:00:00 2001 From: tradewind Date: Tue, 1 Sep 2026 15:53:29 +0800 Subject: [PATCH] #feature: add agent consumer delay monitoring --- .../AgentConsumerDelayMonitorCommand.php | 39 ++++ .../AgentConsumerDelayMonitorService.php | 98 +++++++++ app/Services/DingTalkService.php | 6 +- app/Services/ScheduledTaskService.php | 1 + config/logging.php | 8 + routes/console.php | 8 + .../AgentConsumerDelayMonitorServiceTest.php | 195 ++++++++++++++++++ 7 files changed, 352 insertions(+), 3 deletions(-) create mode 100644 app/Console/Commands/AgentConsumerDelayMonitorCommand.php create mode 100644 app/Services/AgentConsumerDelayMonitorService.php create mode 100644 tests/Unit/AgentConsumerDelayMonitorServiceTest.php diff --git a/app/Console/Commands/AgentConsumerDelayMonitorCommand.php b/app/Console/Commands/AgentConsumerDelayMonitorCommand.php new file mode 100644 index 0000000..8e77896 --- /dev/null +++ b/app/Console/Commands/AgentConsumerDelayMonitorCommand.php @@ -0,0 +1,39 @@ +check(); + + Log::channel('agent-consumer-monitor')->info('Agent 消费延迟检查完成', $result); + $this->info(sprintf( + '检查完成：eventname=%s，延迟=%s 分钟，%d 个新告警，%d 个恢复。', + $result['event_name'] ?? '-', + $result['delay_minutes'] ?? '-', + $result['alerted_count'], + $result['recovered_count'] + )); + + return self::SUCCESS; + } catch (\Throwable $e) { + Log::channel('agent-consumer-monitor')->error('Agent 消费延迟检查失败', [ + 'message' => $e->getMessage(), + ]); + $this->error($e->getMessage()); + + return self::FAILURE; + } + } +} diff --git a/app/Services/AgentConsumerDelayMonitorService.php b/app/Services/AgentConsumerDelayMonitorService.php new file mode 100644 index 0000000..fb27cd5 --- /dev/null +++ b/app/Services/AgentConsumerDelayMonitorService.php @@ -0,0 +1,98 @@ +database->connection('agentslave') + ->table('crm_event_consumer') + ->select(['event_name', 'created']) + ->where('status', '<>', 1) + ->orderByDesc('id') + ->first(); + + $eventName = $latestPending?->event_name; + $created = $latestPending ? CarbonImmutable::parse($latestPending->created) : null; + $delayMinutes = $created?->diffInMinutes($now, false); + $isDelayed = $delayMinutes !== null && $delayMinutes > self::DELAY_THRESHOLD_MINUTES; + + $previousState = $this->configService->get(self::STATE_CONFIG_KEY, []); + $previousState = is_array($previousState) ? $previousState : []; + $wasDelayed = $previousState !== []; + $alertedCount = 0; + $recoveredCount = 0; + + if ($isDelayed && ! $wasDelayed + && $this->dingTalkService->sendText($this->buildDelayAlert($eventName, $created, $delayMinutes, $now))) { + $this->configService->set(self::STATE_CONFIG_KEY, [ + 'event_name' => $eventName, + 'delayed_since' => $created->toDateTimeString(), + ], 'Agent 消费延迟告警状态'); + $alertedCount = 1; + } elseif (! $isDelayed && $wasDelayed + && $this->dingTalkService->sendText($this->buildRecoveryNotice($eventName, $delayMinutes, $now))) { + $this->configService->set(self::STATE_CONFIG_KEY, [], 'Agent 消费延迟告警状态'); + $recoveredCount = 1; + } + + return [ + 'checked_at' => $now->toDateTimeString(), + 'event_name' => $eventName, + 'consumer_time' => $created?->toDateTimeString(), + 'delay_minutes' => $delayMinutes === null ? null : (int) floor($delayMinutes), + 'alerted_count' => $alertedCount, + 'recovered_count' => $recoveredCount, + ]; + } + + private function buildDelayAlert( + string $eventName, + CarbonImmutable $created, + float $delayMinutes, + CarbonImmutable $now + ): string { + return implode("\n", [ + '⚠️ 【Agent 消费延迟告警】', + 'eventname: '.$eventName, + '消费到: '.$created->toDateTimeString(), + sprintf('延迟: %d 分钟', (int) floor($delayMinutes)), + '检查时间: '.$now->toDateTimeString(), + ]); + } + + private function buildRecoveryNotice( + ?string $eventName, + ?float $delayMinutes, + CarbonImmutable $now + ): string { + $lines = [ + '✅ 【Agent 消费延迟恢复】', + '检查时间: '.$now->toDateTimeString(), + ]; + + if ($eventName !== null && $delayMinutes !== null) { + $lines[] = '当前 eventname: '.$eventName; + $lines[] = sprintf('当前延迟: %d 分钟', (int) floor($delayMinutes)); + } else { + $lines[] = '当前无待消费消息'; + } + + return implode("\n", $lines); + } +} diff --git a/app/Services/DingTalkService.php b/app/Services/DingTalkService.php index 9f0dea1..8eee20a 100644 --- a/app/Services/DingTalkService.php +++ b/app/Services/DingTalkService.php @@ -18,7 +18,7 @@ class DingTalkService $this->secret = $config['secret'] ?? null; } - public function sendText(string $message, array $atMobiles = [], bool $atAll = false): void + public function sendText(string $message, array $atMobiles = [], bool $atAll = false): bool { if (empty($this->webhook)) { Log::warning('DingTalk webhook is not configured, skip sending alert. Alert content logged below.', [ @@ -27,10 +27,10 @@ class DingTalkService 'atAll' => $atAll, ]); - return; + return false; } - $this->sendTextToWebhook($this->webhook, $message, $atMobiles, $atAll, $this->secret); + return $this->sendTextToWebhook($this->webhook, $message, $atMobiles, $atAll, $this->secret); } public function sendTextToToken(string $token, string $message, array $atMobiles = [], bool $atAll = false): bool diff --git a/app/Services/ScheduledTaskService.php b/app/Services/ScheduledTaskService.php index 431592e..55411f5 100644 --- a/app/Services/ScheduledTaskService.php +++ b/app/Services/ScheduledTaskService.php @@ -154,6 +154,7 @@ class ScheduledTaskService 'daily-log-analysis' => 'SLS 日志分析 - 每日分析过去 24 小时日志', 'frequent-log-analysis' => 'SLS 日志分析 - 定期分析过去 6 小时日志', 'jenkins-monitor' => 'Jenkins 发布监控 - 检查新构建并发送通知', + 'agent-consumer-delay-monitor' => 'Agent 消费监控 - 按 eventname 检查消费延迟并发送通知', 'erp-request-report' => 'ERP 请求日报 - 汇总前一天 OpenAPI 请求并发送钉钉', 'scheduled-task-refresh' => '定时任务管理 - 刷新定时任务列表', 'logs-cleanup' => '日志清理 - 自动删除 7 天前的定时任务日志', diff --git a/config/logging.php b/config/logging.php index cb1d9b7..2b49204 100644 --- a/config/logging.php +++ b/config/logging.php @@ -143,6 +143,14 @@ return [ 'replace_placeholders' => true, ], + 'agent-consumer-monitor' => [ + 'driver' => 'daily', + 'path' => storage_path('logs/scheduled-tasks/agent-consumer-monitor.log'), + 'level' => env('LOG_LEVEL', 'debug'), + 'days' => 7, + 'replace_placeholders' => true, + ], + 'git-monitor' => [ 'driver' => 'daily', 'path' => storage_path('logs/scheduled-tasks/git-monitor.log'), diff --git a/routes/console.php b/routes/console.php index ec4e0c6..f9138cf 100644 --- a/routes/console.php +++ b/routes/console.php @@ -61,6 +61,14 @@ Schedule::command('jenkins:monitor') ->description('jenkins-monitor') ->when(fn () => \App\Services\ScheduledTaskService::isEnabled('jenkins-monitor')); +// Agent Consumer Monitor - 每分钟检查各 eventname 的消费延迟 +Schedule::command('agent-consumer:monitor-delay') + ->everyMinute() + ->withoutOverlapping(10) + ->runInBackground() + ->description('agent-consumer-delay-monitor') + ->when(fn () => \App\Services\ScheduledTaskService::isEnabled('agent-consumer-delay-monitor')); + // ERP OpenAPI 请求日报 - 每天早上 8 点统计前一天请求并发送钉钉 Schedule::command('erp-request-report:send') ->dailyAt('08:00') diff --git a/tests/Unit/AgentConsumerDelayMonitorServiceTest.php b/tests/Unit/AgentConsumerDelayMonitorServiceTest.php new file mode 100644 index 0000000..7190bed --- /dev/null +++ b/tests/Unit/AgentConsumerDelayMonitorServiceTest.php @@ -0,0 +1,195 @@ +mockLatestPendingQuery(); + $dingTalk = Mockery::mock(DingTalkService::class); + $config = Mockery::mock(ConfigService::class); + + $query->shouldReceive('first')->once()->andReturn((object) [ + 'event_name' => 'CASE_CREATE', + 'created' => '2026-09-01 14:52:00', + ]); + $config->shouldReceive('get') + ->once() + ->with(AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, []) + ->andReturn([]); + $dingTalk->shouldReceive('sendText') + ->once() + ->with(Mockery::on(fn (string $message) => str_contains($message, 'Agent 消费延迟告警') + && str_contains($message, 'CASE_CREATE') + && str_contains($message, '14:52:00') + && str_contains($message, '8 分钟'))) + ->andReturnTrue(); + $config->shouldReceive('set') + ->once() + ->with( + AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, + ['event_name' => 'CASE_CREATE', 'delayed_since' => '2026-09-01 14:52:00'], + 'Agent 消费延迟告警状态' + ); + + $result = (new AgentConsumerDelayMonitorService($database, $dingTalk, $config))->check(); + + $this->assertSame('CASE_CREATE', $result['event_name']); + $this->assertSame(8, $result['delay_minutes']); + $this->assertSame(1, $result['alerted_count']); + $this->assertSame(0, $result['recovered_count']); + } finally { + CarbonImmutable::setTestNow(); + } + } + + public function test_it_does_not_repeat_an_active_delay_alert(): void + { + CarbonImmutable::setTestNow('2026-09-01 15:00:00'); + + try { + [$database, $query] = $this->mockLatestPendingQuery(); + $dingTalk = Mockery::mock(DingTalkService::class); + $config = Mockery::mock(ConfigService::class); + $state = ['event_name' => 'CASE_CREATE', 'delayed_since' => '2026-09-01 14:50:00']; + + $query->shouldReceive('first')->once()->andReturn((object) [ + 'event_name' => 'DOCTOR_CREATE', + 'created' => '2026-09-01 14:52:00', + ]); + $config->shouldReceive('get') + ->once() + ->with(AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, []) + ->andReturn($state); + $dingTalk->shouldNotReceive('sendText'); + $config->shouldNotReceive('set'); + + $result = (new AgentConsumerDelayMonitorService($database, $dingTalk, $config))->check(); + + $this->assertSame('DOCTOR_CREATE', $result['event_name']); + $this->assertSame(0, $result['alerted_count']); + $this->assertSame(0, $result['recovered_count']); + } finally { + CarbonImmutable::setTestNow(); + } + } + + public function test_it_notifies_recovery_when_the_latest_pending_message_is_within_five_minutes(): void + { + CarbonImmutable::setTestNow('2026-09-01 15:00:00'); + + try { + [$database, $query] = $this->mockLatestPendingQuery(); + $dingTalk = Mockery::mock(DingTalkService::class); + $config = Mockery::mock(ConfigService::class); + + $query->shouldReceive('first')->once()->andReturn((object) [ + 'event_name' => 'CASE_CREATE', + 'created' => '2026-09-01 14:57:00', + ]); + $config->shouldReceive('get') + ->once() + ->with(AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, []) + ->andReturn(['event_name' => 'CASE_CREATE', 'delayed_since' => '2026-09-01 14:50:00']); + $dingTalk->shouldReceive('sendText') + ->once() + ->with(Mockery::on(fn (string $message) => str_contains($message, 'Agent 消费延迟恢复'))) + ->andReturnTrue(); + $config->shouldReceive('set') + ->once() + ->with(AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, [], 'Agent 消费延迟告警状态'); + + $result = (new AgentConsumerDelayMonitorService($database, $dingTalk, $config))->check(); + + $this->assertSame(0, $result['alerted_count']); + $this->assertSame(1, $result['recovered_count']); + } finally { + CarbonImmutable::setTestNow(); + } + } + + public function test_it_notifies_recovery_when_there_are_no_pending_messages(): void + { + CarbonImmutable::setTestNow('2026-09-01 15:00:00'); + + try { + [$database, $query] = $this->mockLatestPendingQuery(); + $dingTalk = Mockery::mock(DingTalkService::class); + $config = Mockery::mock(ConfigService::class); + + $query->shouldReceive('first')->once()->andReturnNull(); + $config->shouldReceive('get') + ->once() + ->with(AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, []) + ->andReturn(['event_name' => 'CASE_CREATE', 'delayed_since' => '2026-09-01 14:50:00']); + $dingTalk->shouldReceive('sendText')->once()->andReturnTrue(); + $config->shouldReceive('set') + ->once() + ->with(AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, [], 'Agent 消费延迟告警状态'); + + $result = (new AgentConsumerDelayMonitorService($database, $dingTalk, $config))->check(); + + $this->assertNull($result['event_name']); + $this->assertSame(1, $result['recovered_count']); + } finally { + CarbonImmutable::setTestNow(); + } + } + + public function test_it_retries_notifications_when_dingtalk_sending_fails(): void + { + CarbonImmutable::setTestNow('2026-09-01 15:00:00'); + + try { + [$database, $query] = $this->mockLatestPendingQuery(); + $dingTalk = Mockery::mock(DingTalkService::class); + $config = Mockery::mock(ConfigService::class); + + $query->shouldReceive('first')->once()->andReturn((object) [ + 'event_name' => 'CASE_CREATE', + 'created' => '2026-09-01 14:50:00', + ]); + $config->shouldReceive('get') + ->once() + ->with(AgentConsumerDelayMonitorService::STATE_CONFIG_KEY, []) + ->andReturn([]); + $dingTalk->shouldReceive('sendText')->once()->andReturnFalse(); + $config->shouldNotReceive('set'); + + $result = (new AgentConsumerDelayMonitorService($database, $dingTalk, $config))->check(); + + $this->assertSame(0, $result['alerted_count']); + } finally { + CarbonImmutable::setTestNow(); + } + } + + private function mockLatestPendingQuery(): array + { + $database = Mockery::mock(DatabaseManager::class); + $connection = Mockery::mock(Connection::class); + $query = Mockery::mock(Builder::class); + + $database->shouldReceive('connection')->once()->with('agentslave')->andReturn($connection); + $connection->shouldReceive('table')->once()->with('crm_event_consumer')->andReturn($query); + $query->shouldReceive('select')->once()->with(['event_name', 'created'])->andReturnSelf(); + $query->shouldReceive('where')->once()->with('status', '<>', 1)->andReturnSelf(); + $query->shouldReceive('orderByDesc')->once()->with('id')->andReturnSelf(); + + return [$database, $query]; + } +}