Inspector: мониторинг из кода

Asynq Inspector создаётся тем же коннект-оптом, что клиент и сервер: inspector := asynq.NewInspector(asynq.RedisClientOpt{Addr: "localhost:6379"}) — и работает независимо от воркеров: его можно встроить в админ-сервис, CLI-утилиту или тест. Базовая операция — GetQueueInfo(queue): метод возвращает QueueInfo со срезом очереди целиком — счётчики Pending, Active, Scheduled, Retry, Archived, Completed и Aggregating, суммарный Size, память, которую очередь занимает в Redis, и счётчики обработок: Processed и Failed за текущие сутки плюс накопительные ProcessedTotal и FailedTotal. Отдельного слова заслуживает поле Latency — возраст самой старой задачи в pending: это самый честный сигнал затыка, потому что растёт он именно тогда, когда очередь не успевает, а не когда задач много. Поле Paused показывает, стоит ли очередь на паузе — да, очередью тоже можно управлять, и об этом ниже. Поле MemoryUsage честно помечено в документации как приближение: значение получается сэмплированием, так что сравнивайте очереди между собой, а не байты с бухгалтерией. Рядом лежит Timestamp — момент снятия среза: в логах метрик он спасает, когда скрипт отстал от реальности. Свежая операция — UpdateTaskPayload из версии 0.26: метод меняет payload задачи, пока она сидит в scheduled; активную или ожидающую в pending он не тронет и вернёт ошибку. Список имён всех очередей даёт метод Queues() — с него удобно начинать обход в скрипте мониторинга.

Мониторинг очереди живёт на трёх числах. Первое — Latency из QueueInfo: возраст самой старой задачи в ожидании; порог задаётся бизнесом — для писем минуты, для платежей секунды. Второе — история: метод History(queue, n) возвращает DailyStats за последние n дней — Processed и Failed по суткам; на этих числах строятся графики пропускной способности и сравнение «вчерашний пик против сегодняшнего». DailyStats несёт ровно четыре поля: имя очереди, Processed, Failed и дата — достаточно для графика, ничего лишнего для путаницы. Обратите внимание: Processed включает и успешные, и упавшие задачи, — долю Failed считайте в своём скрипте, готовой её никто не отдаёт. Третье — соотношения внутри среза: растущий Retry при плоском Processed означает, что зависимый сервис деградирует; растущий Archived — что фатальные ошибки из четырнадцатого урока пошли потоком; большое Aggregating — что группы из десятого урока перестали собираться. Пороговые значения вычисляются у вас, а не в библиотеке: Inspector честно отдаёт сырые числа, а «много» для писем и «много» для платежей — разные вещи. Практичная связка: cron-задача из девятого урока каждые пять минут вызывает GetQueueInfo и шлёт метрики в вашу систему наблюдаемости — тем самым очередь становится частью обычного дашборда, а не отдельным миром, в который заглядывают только при инциденте.

За срезом очереди стоят списки задач: ListPendingTasks, ListActiveTasks, ListScheduledTasks, ListRetryTasks, ListArchivedTasks, ListCompletedTasks и ListAggregatingTasks — по одному на состояние. Все методы принимают имя очереди и опции пагинации: asynq.PageSize(n) и asynq.Page(n), где страницы считаются с единицы; без опций вернутся первые тридцать задач — этого хватает для взгляда, но не для обхода, поэтому в скриптах пагинацию листают до конца. Для aggregating-состояния метод отдельный и принимает ещё имя группы: ListAggregatingTasks(queue, group) показывает задачи конкретной пачки из десятого урока. Рядом со списками задач есть и SchedulerEntries — зарегистрированные периодические задачи из девятого урока с их cron-выражениями и временем следующего запуска; для диагностики «а запланировано ли вообще что-то» это первый метод, куда смотреть. Каждый элемент — TaskInfo с полным досье: ID, Type, Payload, состояние, MaxRetry и Retried, LastErr с текстом последней ошибки, LastFailedAt, NextProcessAt для отложенных и ретраев, Timeout и Deadline, поле IsOrphaned для задач, чей воркер умер посреди исполнения. Одиночная задача по ID — GetTaskInfo(queue, id): дежурный сценарий «пользователь дал ID из письма — скажите, что с задачей» реализуется в одну строку. Errors тоже продуманы: несуществующая очередь вернёт ошибку с ErrQueueNotFound, отсутствующая задача — с ErrTaskNotFound, и обе проверяются errors.Is — это не «случайные» ошибки, а часть контракта API.

Вмешательство — вторая половина API, и у каждой операции есть чёткие границы по состояниям. RunTask(queue, id) переводит задачу в pending немедленно: работает для scheduled, retry и archived — так запускают вручную отложенное, ретраи и задачи из архива после починки данных; для pending и active метод вернёт ошибку — задача и так в работе. ArchiveTask(queue, id) отправляет в архив задачи из pending, scheduled и retry — мягкая остановка для постановок, которые стали не нужны; активную задачу так остановить нельзя. DeleteTask(queue, id) удаляет задачу совсем: pending, scheduled, retry и archived — но не active. Общая логика границ проста: активную задачу нельзя ни удалить, ни архивировать, потому что воркер уже держит её в руках — для неё есть только отмена исполнения, о которой ниже. Ошибки те же: ErrTaskNotFound для уже исчезнувших задач — в ручных операциях это нормальный исход при гонке с выполнением. Типичный цикл обхода в скрипте: Page(i) в цикле, пока возвращаемая страница полна, — как только она короче PageSize, вы дошли до конца состояния. Для массовых операций есть All-варианты: RunAllArchivedTasks, ArchiveAllScheduledTasks, DeleteAllRetryTasks и другие, знакомые по десятому уроку групповым RunAllAggregatingTasks — они возвращают число обработанных задач и так же уважают активные; логируйте это число, «архивировано 1240 задач» честнее молчаливой массовой операции. Отдельно стоит DeleteQueue(queue, force): с force=false удаляет пустую очередь, с force=true — даже непустую, лишь бы в ней не было активных задач; пригодится при выводе очереди из эксплуатации.

Единственная операция над активными задачами — CancelProcessing(id): она принимает только идентификатор, без очереди, и работает сигналом, а не командой. Внутри — публикация в Redis pubsub-канал, который слушают все ваши серверы: каждый держит реестр исполняемых задач, находит нужный ID и отменяет контекст хендлера у себя. Документация называет метод best-effort, и это важно: результат вызова говорит лишь о том, что сигнал отправлен, — задача могла уже закончиться, сигнал мог прийти на сервер, где её нет, хендлер мог игнорировать контекст. Что происходит с задачей после отмены, определяется общими правилами: контекст отменён — хендлер, слушающий ctx.Done(), возвращает управление, воркер засчитывает неудачу с ошибкой контекста, и задача ретраится по обычной политике; хендлер, игнорирующий контекст, доделает свою работу, как в одиннадцатом уроке. Отсюда честные ожидания: CancelProcessing — это «попросить остановиться», а не «убить»; для повторов, которые должны прекратиться совсем, есть ArchiveTask после отмены или SkipRetry в самом хендлере. И помните: отмена не снимает задачу с очереди — она лишь завершает текущую попытку.