#feature: add agent consumer delay monitoring

2026-09-01 15:53:29 +08:00
parent c45d68e201
commit adf728adfc
7 changed files with 352 additions and 3 deletions
@@ -0,0 +1,39 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Services\AgentConsumerDelayMonitorService;
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\Log;
+
+class AgentConsumerDelayMonitorCommand extends Command
+{
+    protected $signature = 'agent-consumer:monitor-delay';
+
+    protected $description = '检查 Agent 消息消费延迟并发送钉钉告警或恢复通知';
+
+    public function handle(AgentConsumerDelayMonitorService $service): int
+    {
+        try {
+            $result = $service->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;
+        }
+    }
+}
@@ -0,0 +1,98 @@
+<?php
+
+namespace App\Services;
+
+use Carbon\CarbonImmutable;
+use Illuminate\Database\DatabaseManager;
+
+class AgentConsumerDelayMonitorService
+{
+    public const STATE_CONFIG_KEY = 'agent_consumer_delay_monitor.alerts';
+
+    private const DELAY_THRESHOLD_MINUTES = 5;
+
+    public function __construct(
+        private readonly DatabaseManager $database,
+        private readonly DingTalkService $dingTalkService,
+        private readonly ConfigService $configService
+    ) {}
+
+    public function check(): array
+    {
+        $now = CarbonImmutable::now();
+        $latestPending = $this->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);
+    }
+}
@@ -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
@@ -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 天前的定时任务日志',
@@ -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'),
@@ -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')
@@ -0,0 +1,195 @@
+<?php
+
+namespace Tests\Unit;
+
+use App\Services\AgentConsumerDelayMonitorService;
+use App\Services\ConfigService;
+use App\Services\DingTalkService;
+use Carbon\CarbonImmutable;
+use Illuminate\Database\Connection;
+use Illuminate\Database\DatabaseManager;
+use Illuminate\Database\Query\Builder;
+use Mockery;
+use Tests\TestCase;
+
+class AgentConsumerDelayMonitorServiceTest extends TestCase
+{
+    public function test_it_alerts_when_the_latest_pending_message_is_more_than_five_minutes_old(): 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: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];
+    }
+}