Б Бёрнс - Распределенные системы. Паттерны проектирования

Рис. 10.3. API контейнера-исполнителя

Такой механизм файлового API проще реализовать с помо-щью контейнера. Исполнитель в рамках очереди задач часто Глава 10. Системы на основе очередей задач 179

представляет собой простой сценарий командной оболочки, обращающийся к нескольким инструментам. Нецелесообразно поднимать для управления задачами целый веб-сервер — это приводит к усложнению архитектуры. Как и в случае с ис-точниками задач, большая часть контейнеров-исполнителей будет представлять собой специализированные контейнеры для определенных задач, но есть и обобщенные исполнители, применимые для решения нескольких разных задач. Рассмотрим пример контейнера-исполнителя, который скачивает файл из облачного хранилища, выполняет над ним сценарий ко-мандной оболочки, а затем копирует результат обратно в облачное хранилище. Такой контейнер может быть по большей части об-щим, но в качестве параметра ему может передаваться конкретный сценарий. Таким образом, большая часть кода работы с файлом может быть повторно использована многими пользователями/ очередями задач. Конечному пользователю необходимо лишь предоставить сценарий, содержащий специфику обработки файла. Общая инфраструктура очередей задач Что остается внедрить в повторно используемой реализации очереди, если реализации двух ранее описанных интерфейсов контейнеров у вас уже есть? Базовый алгоритм работы очереди задач довольно прост.

1. Загрузить из контейнера-источника доступные на данный момент задачи.

2. Уточнить состояние очереди задач на предмет того, какие задачи уже выполнены или еще выполняются.

3. Для каждой из нерешенных задач породить контейнеры-исполнители с соответствующим интерфейсом.

4. При успешном завершении контейнера-исполнителя зафи-ксировать, что задача выполнена.

180Часть III. Паттерны проектирования систем пакетных вычислений Этот алгоритм прост на словах, но в действительности его не так легко реализовать. К счастью, оркестратор Kubernetes имеет несколько возможностей, которые значительно упрощают его реализацию. А именно: в Kubernetes есть объект Job , который позволяет обеспечить надежную работу очереди задач. Объект Job можно настроить так, чтобы он запускал соответствующий контейнер-исполнитель либо разово, либо пока задача не будет успешно выполнена. Если контейнер-исполнитель настроить, чтобы он выполнялся до завершения задачи, то, даже когда машина в кластере откажет, задача в конце концов будет вы-полнена успешно.

Таким образом, построение очереди задач существенно упро-щается, так как оркестратор берет на себя ответственность за надежное исполнение задач.

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

Это значит, что мы можем реализовать очередь задач поверх ор-кестратора Kubernetes, не используя собственного хранилища. Все это существенно упрощает задачу построения инфраструк-туры очереди задач.

Следовательно, подробный алгоритм работы контейнера — дис-петчера очереди задач выглядит следующим образом.

Повторять бесконечно.

1. Получить список задач посредством интерфейса контейне-ра — источника задач.

2. Получить список заданий, обслуживающих данную очередь Глава 10. Системы на основе очередей задач 181

3. Выделить на основе этих списков перечень необработанных задач.

4. Для каждой необработанной задачи создать объект Job , который порождает соответствующий контейнер-испол-нитель.

Приведу сценарий на языке Python, реализующий такую оче-редь:

import requests

import json

from kubernetes import client, config

import time

namespace = "default"

def make_container(item, obj):

container = client.V1Container()

container.image = "my/worker-image"

container.name = "worker"

return container

def make_job(item):

response =

requests.get("http://localhost:8000/items/{}".format(item)) obj = json.loads(response.text)

job = client.V1Job()

job.metadata = client.V1ObjectMeta()

job.metadata.name = item

job.spec = client.V1JobSpec()

job.spec.template = client.V1PodTemplate() job.spec.template.spec = client.V1PodTemplateSpec() job.spec.template.spec.restart_policy = "Never" job.spec.template.spec.containers = [

make_container(item, obj)

]

return job

def update_queue(batch):

response = requests.get("http://localhost:8000/items") 182Часть III. Паттерны проектирования систем пакетных вычислений obj = json.loads(response.text)

items = obj['items']

ret = batch.list_namespaced_job(namespace, watch=False) for item in items:

found = False

for i in ret.items:

if i.metadata.name == item:

found = True

if not found:

# Функция создает объект Job, пропущена

# для краткости

job = make_job(item)

batch.create_namespaced_job(namespace, job)

config.load_kube_config()

batch = client.BatchV1Api()

while True:

update_queue(batch)

time.sleep(10)

Практикум. Реализация генератора миниатюр видеофайлов

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

Для реализации миниатюр понадобится два контейнера. Пер-вый — для источника задач. Проще всего будет размещать за-дачи на общем сетевом диске, подключенном, например, по NFS (Network File System, сетевая файловая система). Источник задач получает список файлов в этом каталоге и передает их вызывающей стороне.

Глава 10. Системы на основе очередей задач 183

Приведу простую программу на NodeJS:

const http = require('http');

const fs = require('fs');

const port = 8080;

const path = process.env.MEDIA_PATH;

const requestHandler = (request, response) => { console.log(request.url);

fs.readdir(path + '/*.mp4', (err, items) => { var msg = {

'kind': 'ItemList',

'apiVersion': 'v1',

'items': []

};

if (!items) {

return msg;

}

for (var i = 0; i < items.length; i++) {

msg.items.push(items[i]);

}

response.end(JSON.stringify(msg));

});

}

const server = http.createServer(requestHandler); server.listen(port, (err) => {

if (err) {

return console.log('Ошибка запуска сервера', err);

}

console.log(`сервер запущен на порте ${port}`) });

Данный источник определяет список фильмов, подлежащих обработке. Для извлечения миниатюр используется утилита ffmpeg.

184Часть III. Паттерны проектирования систем пакетных вычислений Можно создать контейнер, запускающий такую команду: ffmpeg -i ${INPUT_FILE} -frames:v 100 thumb.png Команда извлекает один из каждых 100 кадров (параметр -frames:v 100 ) и сохраняет его в формате PNG (например, thumb1.png , thumb2.png и т. д.).

Подобного рода обработку можно реализовать на основе суще-ствующего Docker-образа ffmpeg . Популярностью пользует-ся образ jrottenberg/ffmpeg ( https://hub.docker.com/r/jrottenberg/ ffmpeg/ ).

Определив простой контейнер-источник и еще более простой контейнер-исполнитель, нетрудно заметить, какие преиму-щества имеет обобщенная, контейнерно-ориентированная система управления очередью. Она существенно сокращает время между проектированием и реализацией очереди задач. Динамическое масштабирование исполнителей