Как работать с очередью заданий с помощью kubernetes с масштабированием

Question

Мне нужна масштабируемая обработка очереди, основанная на Docker / Python Worker. Моя мысль пошла к Кубернетес. Тем не менее, я не уверен насчет лучшего контроллера / сервиса.

На основе функций Azure я получаю входящий http-трафик, добавляя простые сообщения в очередь хранилища. Эти сообщения должны быть обработаны и результаты возвращены в очередь результатов.

Чтобы обработать эти сообщения очереди, я разработал код Python, зацикливающий очередь и работающий над этими заданиями. После каждого успешного цикла сообщение будет удаляться из исходной очереди, а результат записывается в очередь результатов. Когда очередь пуста, код существует.

Итак, я создал образ докера, который запускает код Python. Если запущено более одного контейнера, очередь, очевидно, работает быстрее. Я также внедрил новые службы Azure Kubernetes для масштабирования. Будучи новичком в kubernetes, я читал о парадигме задания для обработки очереди, пока задание не будет готово. Мой простой шаблон yaml выглядит так:

apiVersion: batch/v1
kind: Job
metadata:
  name: myjob
spec:
  parallelism: 4
  template:
    metadata:
      name: myjob
    spec:
      containers:
      - name: c
        image: repo/image:tag

Моя проблема в том, что задание не может быть перезапущено.

Обычно очередь заполняется некоторыми записями, а затем какое-то время ничего не происходит. Затем снова могут появиться большие очереди, над которыми нужно работать как можно быстрее. Конечно, я хочу запустить работу снова, но это кажется невозможным. Кроме того, я хочу уменьшить площадь до минимума, если в очереди ничего нет.

Итак, мой вопрос: какую архитектуру / конструкции я должен использовать для этого сценария, и есть ли простые примеры yaml для этого?

Answer 1

Это распространенный шаблон, и существует несколько способов создания решения.

Распространенное решение состоит в том, чтобы приложение с набором работников всегда опрашивало вашу очередь (это может быть ваш скрипт на python, но вам нужно сделать его сервисом), и в общем случае вы захотите использовать Kubernetes Deployment возможно с Horizontal Pod Autoscaler на основе некоторых показателей для вашей очереди или ЦП.

В вашем случае вы захотите сделать свой скрипт демоном и опросить очередь, если есть какие-либо элементы (я полагаю, вы уже обрабатываете условия гонки с параллелизмом). Затем разверните этот демон с помощью развертывания в Kubernetes, после чего вы сможете увеличивать и уменьшать показатели или расписание на основе.

Уже есть планировщики заданий для разных языков. Очень популярным является Airflow , у которого уже есть возможность иметь «рабочих», но это может быть излишним для одного скрипта Python.

Answer 2

Это может быть «глупый / хакерский» ответ, но он простой, надежный, и я уже несколько месяцев использую его в производственной системе.

У меня есть похожая система, где у меня есть очередь, которая иногда очищается, а иногда захлопывается. Я написал свой обработчик очереди аналогичным образом, он обрабатывает одно сообщение в очереди за раз и завершается, если очередь пуста. Он настроен для работы в задании Kubernetes.

Хитрость заключается в следующем: я создал CronJob для регулярного запуска одного нового экземпляра задания, и задание допускает бесконечный параллелизм. Если очередь пуста, она немедленно завершается («уменьшается»). Если очередь захлопнулась, а последнее задание еще не завершено, запускается другой экземпляр («масштабируется»).

Нет необходимости возиться с запросом очереди и масштабированием набора состояний или чего-либо еще, и никакие ресурсы не потребляются, если очередь пуста. Возможно, вам придется настроить интервал CronJob, чтобы точно настроить скорость реакции на заполнение очереди, но она должна реагировать довольно хорошо.