= QUEUE_BACKLOG_THRESHOLD 条, * 且连续 QUEUE_BACKLOG_CONSECUTIVE 次命中(每分钟检查一次)。 * 含义:job 已经派发进队列但没人消费——worker 进程挂了/没启动, * 或队列名、Redis 实例与派发方不一致。 * * 【规则B】任务长时间未启动(PENDING_*) * 判定:队列派发型模型(GPT-Image2、火山系列)中 status=pending * 且 created_at 早于 PENDING_TIMEOUT_MINUTES 分钟的任务数 > 0, * 且连续 PENDING_CONSECUTIVE 次命中。 * 含义:任务已创建却一直没被 worker 取走,用户端会一直显示 0 进度。 * 说明:即梦(jimeng_4.0)、NanoBanana 走内部排队/同步提交,等待属于正常,不参与本规则。 * * 【规则C】任务长时间处理中(PROCESSING_*) * 判定:status=processing 且 started_at 早于 PROCESSING_TIMEOUT_MINUTES 分钟的任务数 > 0, * 且连续 PROCESSING_CONSECUTIVE 次命中。 * 含义:任务已被 worker 取走但卡在第三方接口,或 worker 被强杀导致状态悬挂。 * * 【规则D】失败突增(FAILED_*) * 判定:最近 FAILED_WINDOW_MINUTES 分钟内 status=failed 的任务数 > FAILED_THRESHOLD, * 且连续 FAILED_CONSECUTIVE 次命中。 * 含义:第三方账号/额度异常(例如历史出现过的 USER_INACTIVE)或内容审核批量拦截。 * * 【通用行为】 * 1. 同一规则告警后,ALERT_COOLDOWN_MINUTES 分钟内不重复发送;超过冷却期仍异常则重复提醒一次; * 2. 指标恢复正常时发送一条"已恢复"通知(含本次异常持续时长); * 3. 所有通知都带环境(APP_ENV)、命中详情与排查建议; * 4. 连续次数、冷却状态记录在 Redis(ALERT_KEY_PREFIX + 规则名),ALERT_STATE_TTL 后自动过期; * 5. 钉钉通知直接复用项目已有的 sendNotice()(读取 .env 的 DD_WEB_HOOK); * 未配置 webhook 或发送失败都只记录日志,不影响命令执行。 * =========================================================================================== */ class CheckAiTaskHealthCommand extends Command { /** * 命令名与参数 * * @var string */ protected $signature = 'ai:check-task-health {--dry-run : 只输出判定结果,不发送钉钉也不写入状态}'; /** * 命令说明 * * @var string */ protected $description = '图片任务健康检查(队列积压/任务卡住/失败突增)并在异常时发送钉钉告警'; // ============================== 规则A:队列积压 ============================== /** 监控的 Redis 队列:队列名 => 展示名(Laravel Redis 队列的 key 为 queues:{队列名}) */ const MONITOR_QUEUES = [ '{GenerateGptImage2Pics}' => 'GPT-Image2', '{GenerateVolcPics}' => '火山API', ]; /** 队列积压条数阈值(达到即算命中) */ const QUEUE_BACKLOG_THRESHOLD = 200; /** 队列积压需要连续命中的次数(每分钟检查一次) */ const QUEUE_BACKLOG_CONSECUTIVE = 3; // ============================== 规则B:任务长时间未启动 ============================== /** pending 超时时间(分钟),超过即算命中 */ const PENDING_TIMEOUT_MINUTES = 10; /** pending 超时需要连续命中的次数 */ const PENDING_CONSECUTIVE = 2; // ============================== 规则C:任务长时间处理中 ============================== /** processing 超时时间(分钟),超过即算命中 */ const PROCESSING_TIMEOUT_MINUTES = 15; /** processing 超时需要连续命中的次数 */ const PROCESSING_CONSECUTIVE = 2; // ============================== 规则D:失败突增 ============================== /** 失败统计的滑动窗口(分钟) */ const FAILED_WINDOW_MINUTES = 10; /** 窗口内失败条数阈值(超过才告警) */ const FAILED_THRESHOLD = 20; /** 失败突增需要连续命中的次数 */ const FAILED_CONSECUTIVE = 2; /** 失败样例在告警中展示的条数 */ const FAILED_SAMPLE_LIMIT = 3; // ============================== 通用参数 ============================== /** 同一规则告警后的冷却时间(分钟),冷却期内不重复发送 */ const ALERT_COOLDOWN_MINUTES = 30; /** 告警状态在 Redis 中的 key 前缀 */ const ALERT_KEY_PREFIX = 'ai_task_alert:'; /** 告警状态保留时间(秒),避免 Redis 中残留脏状态 */ const ALERT_STATE_TTL = 604800; // 7 天 /** 使用的 Redis 连接,必须与队列保持一致 */ const ALERT_REDIS_CONNECTION = 'default'; /** 展示错误信息时截断的字符数 */ const ERROR_MESSAGE_LIMIT = 80; /** * 执行命令 * * @return int */ public function handle() { $dryRun = (bool) $this->option('dry-run'); $this->info('图片任务健康检查开始' . ($dryRun ? '(dry-run:不发送钉钉、不写入状态)' : '')); $this->line('环境:' . config('app.env') . ' 时间:' . now()->toDateTimeString()); try { // 采集 4 项规则 $rules = [ $this->buildQueueBacklogRule(), $this->buildPendingTimeoutRule(), $this->buildProcessingTimeoutRule(), $this->buildFailedSpikeRule(), ]; foreach ($rules as $rule) { $this->processRule($rule, $dryRun); } } catch (\Throwable $e) { // 检查本身异常不影响其它定时任务,只记录并输出 $this->error('图片任务健康检查执行失败:' . $e->getMessage()); dLog('generate')->error('图片任务健康检查执行失败', ['error' => $e->getMessage()]); return 1; } $this->info('图片任务健康检查结束'); return 0; } /** * 规则处理:连续计数、冷却、告警与恢复通知、状态落库 * * @param array $rule 规则数据 * @param bool $dryRun 是否演练模式(不发通知、不写状态) * @return void */ private function processRule(array $rule, bool $dryRun): void { $state = $this->loadState($rule['key']); $cooldownSeconds = self::ALERT_COOLDOWN_MINUTES * 60; $this->outputRuleResult($rule, $state); if ($rule['triggered']) { $state['consecutive'] = (int) $state['consecutive'] + 1; if ($state['consecutive'] === 1 && empty($state['first_hit_at'])) { $state['first_hit_at'] = time(); } if (!$state['alerting']) { // 未处于告警状态:连续命中达到配置次数才发送 if ($state['consecutive'] >= (int) $rule['consecutive']) { if ($dryRun) { $this->line(' → [dry-run] 已达到连续命中次数,将发送告警'); } else { $this->sendAlert($rule, false); $state['alerting'] = true; $state['alerted_at'] = time(); } } else { $this->line(' → 命中 ' . $state['consecutive'] . '/' . $rule['consecutive'] . ' 次,未达连续次数,暂不告警'); } } else { // 已在告警中:冷却结束后重复提醒 $elapsed = time() - (int) $state['alerted_at']; if ($elapsed >= $cooldownSeconds) { if ($dryRun) { $this->line(' → [dry-run] 冷却期已过且异常仍在持续,将重复发送告警'); } else { $this->sendAlert($rule, true); $state['alerted_at'] = time(); } } else { $this->line(' → 告警中,冷却剩余约 ' . (int) ceil(($cooldownSeconds - $elapsed) / 60) . ' 分钟'); } } } else { if ($state['alerting']) { if ($dryRun) { $this->line(' → [dry-run] 指标已恢复,将发送恢复通知'); } else { $this->sendRecovery($rule, $state); } $state['alerting'] = false; } $state['consecutive'] = 0; $state['first_hit_at'] = 0; } if (!$dryRun) { $state['last_detail'] = implode(' | ', $rule['detail']); $state['last_check_at'] = time(); $this->saveState($rule['key'], $state); } } /** * 规则A:队列积压 * * @return array */ private function buildQueueBacklogRule(): array { $detail = []; $hit = []; try { foreach (self::MONITOR_QUEUES as $queue => $label) { $length = (int) Redis::connection(self::ALERT_REDIS_CONNECTION)->llen('queues:' . $queue); $detail[] = $label . '队列(queues:' . $queue . '):' . $length . ' 条,阈值 ' . self::QUEUE_BACKLOG_THRESHOLD . ' 条'; if ($length >= self::QUEUE_BACKLOG_THRESHOLD) { $hit[] = $label . '=' . $length; } } } catch (\Throwable $e) { // Redis 不可用时无法判定队列长度,只提示不告警(避免误报) $detail[] = 'Redis 队列长度读取失败:' . $e->getMessage(); $hit = []; } return [ 'key' => 'queue_backlog', 'title' => '队列积压', 'rule_text' => '队列长度 >= ' . self::QUEUE_BACKLOG_THRESHOLD . ' 条,连续 ' . self::QUEUE_BACKLOG_CONSECUTIVE . ' 次命中', 'consecutive' => self::QUEUE_BACKLOG_CONSECUTIVE, 'triggered' => !empty($hit), 'detail' => $detail, 'hit_text' => implode(',', $hit), 'suggest' => '检查 worker 进程是否存活(supervisorctl status generate_gpt_image2_pics:*),确认队列名与 Redis 实例和派发方一致', ]; } /** * 规则B:任务长时间未启动 * * @return array */ private function buildPendingTimeoutRule(): array { $deadline = now()->subMinutes(self::PENDING_TIMEOUT_MINUTES); $query = DB::table('mp_generate_pic_tasks') ->where('status', MpGeneratePicTask::STATUS_PENDING) ->whereIn('model', $this->pendingWatchModels()) ->where('created_at', '<', $deadline); $count = (clone $query)->count(); $detail = ['pending 超过 ' . self::PENDING_TIMEOUT_MINUTES . ' 分钟的任务:' . $count . ' 条']; if ($count > 0) { $oldest = (clone $query)->orderBy('created_at', 'asc')->first(); if ($oldest) { $detail[] = '最早一条:id=' . $oldest->id . ',模型=' . $oldest->model . ',外部任务号=' . $oldest->task_id . ',创建于 ' . $oldest->created_at . '(已等待 ' . $this->minutesSince($oldest->created_at) . ' 分钟)'; } } return [ 'key' => 'pending_timeout', 'title' => '任务长时间未启动', 'rule_text' => 'pending 超过 ' . self::PENDING_TIMEOUT_MINUTES . ' 分钟的任务数 > 0,连续 ' . self::PENDING_CONSECUTIVE . ' 次命中', 'consecutive' => self::PENDING_CONSECUTIVE, 'triggered' => $count > 0, 'detail' => $detail, 'hit_text' => '超时 pending ' . $count . ' 条', 'suggest' => '任务已入队但未被消费,优先检查对应队列的 worker 与队列积压情况', ]; } /** * 规则C:任务长时间处理中 * * @return array */ private function buildProcessingTimeoutRule(): array { $deadline = now()->subMinutes(self::PROCESSING_TIMEOUT_MINUTES); $query = DB::table('mp_generate_pic_tasks') ->where('status', MpGeneratePicTask::STATUS_PROCESSING) ->whereNotNull('started_at') ->where('started_at', '<', $deadline); $count = (clone $query)->count(); $detail = ['processing 超过 ' . self::PROCESSING_TIMEOUT_MINUTES . ' 分钟的任务:' . $count . ' 条']; if ($count > 0) { $oldest = (clone $query)->orderBy('started_at', 'asc')->first(); if ($oldest) { $detail[] = '最早一条:id=' . $oldest->id . ',模型=' . $oldest->model . ',外部任务号=' . $oldest->task_id . ',开始于 ' . $oldest->started_at . '(已处理 ' . $this->minutesSince($oldest->started_at) . ' 分钟)'; } } return [ 'key' => 'processing_timeout', 'title' => '任务长时间处理中', 'rule_text' => 'processing 超过 ' . self::PROCESSING_TIMEOUT_MINUTES . ' 分钟的任务数 > 0,连续 ' . self::PROCESSING_CONSECUTIVE . ' 次命中', 'consecutive' => self::PROCESSING_CONSECUTIVE, 'triggered' => $count > 0, 'detail' => $detail, 'hit_text' => '超时 processing ' . $count . ' 条', 'suggest' => '检查第三方接口是否超时/限流,以及 worker 是否被强杀(supervisor 的 stopwaitsecs 是否小于任务超时时间)', ]; } /** * 规则D:失败突增 * * @return array */ private function buildFailedSpikeRule(): array { $deadline = now()->subMinutes(self::FAILED_WINDOW_MINUTES); $query = DB::table('mp_generate_pic_tasks') ->where('status', MpGeneratePicTask::STATUS_FAILED) ->where('updated_at', '>=', $deadline); $count = (clone $query)->count(); $detail = ['最近 ' . self::FAILED_WINDOW_MINUTES . ' 分钟内失败的任务:' . $count . ' 条,阈值 > ' . self::FAILED_THRESHOLD . ' 条']; if ($count > self::FAILED_THRESHOLD) { $samples = (clone $query)->orderBy('updated_at', 'desc')->limit(self::FAILED_SAMPLE_LIMIT)->get(['id', 'model', 'error_message', 'updated_at']); foreach ($samples as $sample) { $detail[] = '样例:id=' . $sample->id . ',模型=' . $sample->model . ',时间=' . $sample->updated_at . ',错误=' . mb_substr((string) $sample->error_message, 0, self::ERROR_MESSAGE_LIMIT); } } return [ 'key' => 'failed_spike', 'title' => '失败任务突增', 'rule_text' => '最近 ' . self::FAILED_WINDOW_MINUTES . ' 分钟内失败数 > ' . self::FAILED_THRESHOLD . ' 条,连续 ' . self::FAILED_CONSECUTIVE . ' 次命中', 'consecutive' => self::FAILED_CONSECUTIVE, 'triggered' => $count > self::FAILED_THRESHOLD, 'detail' => $detail, 'hit_text' => '失败 ' . $count . ' 条', 'suggest' => '检查第三方账号/额度状态与内容审核拦截,确认接口返回的错误码', ]; } /** * pending 规则监控的模型:仅"队列派发型"(GPT-Image2、火山系列) * * 即梦(jimeng_4.0)、NanoBanana 走内部排队/同步提交,长时间 pending 属正常,不参与告警。 * * @return array */ private function pendingWatchModels(): array { return array_merge(BaseConst::GPT_IMAGE2_MODELS, BaseConst::VOLC_PIC_MODELS); } /** * 控制台输出单条规则的判定结果 * * @param array $rule * @param array $state * @return void */ private function outputRuleResult(array $rule, array $state): void { $status = $rule['triggered'] ? '命中(连续 ' . ((int) $state['consecutive'] + 1) . '/' . $rule['consecutive'] . ')' : '正常'; $this->line('【' . $rule['title'] . '】' . $status); foreach ($rule['detail'] as $line) { $this->line(' - ' . $line); } } /** * 发送告警(钉钉 + 控制台 + 日志) * * @param array $rule * @param bool $repeat 是否为持续告警的重复提醒 * @return void */ private function sendAlert(array $rule, bool $repeat): void { $lines = [ ($repeat ? '【图片任务告警·持续】' : '【图片任务告警】') . $rule['title'], '环境:' . config('app.env'), '规则:' . $rule['rule_text'], '当前:' . $rule['hit_text'], ]; foreach ($rule['detail'] as $line) { $lines[] = ' - ' . $line; } if ($repeat) { $lines[] = '说明:该异常仍在持续,距上次通知已 ' . self::ALERT_COOLDOWN_MINUTES . ' 分钟'; } $lines[] = '建议:' . $rule['suggest']; $lines[] = '时间:' . now()->toDateTimeString(); $content = implode("\n", $lines); $this->error($content); dLog('generate')->error('图片任务健康告警:' . $rule['title'], [ 'rule' => $rule['key'], 'detail' => $rule['detail'], ]); logDB('generate', 'error', '图片任务健康告警:' . $rule['title'], [ 'rule' => $rule['key'], 'detail' => $rule['detail'], ]); $this->notify($content); } /** * 发送恢复通知(只在上一轮处于告警状态时发送) * * @param array $rule * @param array $state * @return void */ private function sendRecovery(array $rule, array $state): void { $lines = [ '【图片任务恢复】' . $rule['title'] . ' 已恢复正常', '环境:' . config('app.env'), ]; if (!empty($state['first_hit_at'])) { $lines[] = '本次异常持续约 ' . (int) ceil((time() - (int) $state['first_hit_at']) / 60) . ' 分钟'; } foreach ($rule['detail'] as $line) { $lines[] = ' - ' . $line; } $lines[] = '时间:' . now()->toDateTimeString(); $content = implode("\n", $lines); $this->info($content); dLog('generate')->info('图片任务健康恢复:' . $rule['title'], ['rule' => $rule['key']]); logDB('generate', 'info', '图片任务健康恢复:' . $rule['title'], ['rule' => $rule['key']]); $this->notify($content); } /** * 发送钉钉通知 * * 直接复用项目已有的 sendNotice()(读取线上 .env 的 DD_WEB_HOOK,默认 @所有人)。 * webhook 未配置或发送失败都只记录日志,绝不影响检查命令本身。 * * @param string $content * @return void */ private function notify(string $content): void { if (empty(env('DD_WEB_HOOK'))) { $this->warn('DD_WEB_HOOK 未配置,跳过钉钉发送'); dLog('generate')->warning('图片任务健康告警未发送:DD_WEB_HOOK 未配置', ['content' => $content]); return; } try { sendNotice($content); } catch (\Throwable $e) { $this->warn('钉钉发送失败:' . $e->getMessage()); dLog('generate')->error('图片任务健康告警发送失败', [ 'error' => $e->getMessage(), 'content' => $content, ]); } } /** * 读取规则状态(连续命中次数、是否告警中、首次命中时间等) * * @param string $ruleKey * @return array */ private function loadState(string $ruleKey): array { $default = [ 'consecutive' => 0, 'alerting' => false, 'alerted_at' => 0, 'first_hit_at'=> 0, ]; try { $raw = Redis::connection(self::ALERT_REDIS_CONNECTION)->get($this->stateKey($ruleKey)); if (empty($raw)) { return $default; } $state = json_decode($raw, true); if (!is_array($state)) { return $default; } return array_merge($default, $state); } catch (\Throwable $e) { dLog('generate')->warning('读取图片任务告警状态失败', [ 'rule' => $ruleKey, 'error' => $e->getMessage(), ]); return $default; } } /** * 保存规则状态 * * @param string $ruleKey * @param array $state * @return void */ private function saveState(string $ruleKey, array $state): void { try { Redis::connection(self::ALERT_REDIS_CONNECTION)->set( $this->stateKey($ruleKey), json_encode($state, JSON_UNESCAPED_UNICODE), 'EX', self::ALERT_STATE_TTL ); } catch (\Throwable $e) { dLog('generate')->warning('保存图片任务告警状态失败', [ 'rule' => $ruleKey, 'error' => $e->getMessage(), ]); } } /** * 规则状态在 Redis 中的 key * * @param string $ruleKey * @return string */ private function stateKey(string $ruleKey): string { return self::ALERT_KEY_PREFIX . 'state:' . $ruleKey; } /** * 计算某个时间点距今经过的分钟数 * * @param string $dateTime * @return int */ private function minutesSince(string $dateTime): int { $timestamp = strtotime($dateTime); return $timestamp ? (int) floor((time() - $timestamp) / 60) : 0; } }