Пару лет назад мы в Just Work делали несколько похожих проектов, которые должны были обрабатывать данные, получаемые из одного внешнего HTTP API. Это API, несмотря на согласованные повышенные лимиты, изредка банило наши ключи доступа за малейшее превышение. Из-за этого ответственность за соблюдение лимитов лежала на клиентах. В дальнейшем, проектов, использующих это API, должно было становиться все больше, и заказчика не устраивала перспектива разбираться с каждой реализацией по отдельности.

В итоге было решено сделать собственный прокси-сервер, который реализовывал бы контроль скорости и предоставлял бы асинхронный доступ к API.

Производительность

Основным языком программирования у нас является Python, но, несмотря на весь имеющийся опыт, у нас не было уверенности в том, что требуемую производительность обработки запросов можно будет реализовать с использованием вменяемого количества серверов. Так что пришлось провести маленькое исследование того, как различные фреймворки справляются с конкуррентными HTTP-запросами.

Больше всего опасений вызывал разбор HTTP-протокола и накладные расходы на обеспечение параллелизма. Исходя из всего выше описанного, были выбраны три асинхронных фреймворка (tornado, gevent и asyncio) и пара HTTP-клиентов (requests + libcurl). В итоге в категории "фреймворки" всех уделал gevent, который "шаманит" со стеком и тредами где-то во внутренностях интерпретатора питона, и libcurl, который помимо того, что написан на C, имеет еще один "козырь" в рукаве под названием "multi-curl". Это API библиотеки, которое позволяет libcurl выполнять несколько HTTP-запросов параллельно.

Реализация

После того, как удалось заставить работать multi-curl и gevent вместе, способ реализации стал очевиден: берем Celery в режиме gevent и каждую задачу отправляем в конвейер multi-curl, ждем пока запрос завершится и ставим еще одну Celery-задачу с ответом API в "очередь" ответов, указанную клиентом.

Забегая вперед, стоит сказать, что все проекты на тот момент использовали Python, и предоставляемый Celery протокол общения для прокси был скорее преимуществом: ведь Celery позволяет описывать цепочки задач, выполняемые друг за другом. Но и к клиентам "из других миров" мы были готовы, т.к. есть возможность трансформировать на лету входящее сообщение так, чтобы Celery считала его нормальной задачей. Похожим способом, например, пользуется библиотека celery-message-consumer. Но давайте вернемся к контролю за соблюдением лимитов.

Rate-Limiting

Для того, чтобы ограничить скорость отправки запросов к API, мы просто останавливаем получение сообщений из очереди RabbitMQ, соответствующей нужному эндпоинту. С помощью таймеров Celery мы возобновляем обработку задач через время, соответствующее слоту отслеживания лимитов. Задача, попавшая под rate-limit, возвращается в начало очереди с помощью AMQP-команды reject(requeue=True).

Поскольку для нас важно не выходить за рамки лимитов, мы начинаем снижать скорость обработки задач еще до полного исчерпания слота. По сути, вероятность получить "стоп" линейно возрастает с нуля на 50% загрузки слота до 100% при полной загрузке. С таким подходом, при наличии большого количества воркеров Celery, к исчерпанию слота мы приходим с практически нулевой скоростью, что дает ровную линию на графике обработки задач.

HTTPS и логи

Поскольку связка gevent + multi-curl всё еще не вылезает за пределы одного ядра CPU, нам захотелось перенести с этого ядра накладные расходы, связанные с шифрованием HTTPS-трафика. Для этого мы использовали nginx в режиме реверс-прокси, где апстримом выступало API, к которому мы отправляли запросы. Это также позволило нам получить подробные логи, которые нам сильно пригодились в дальнейшем.

В какой-то момент в продакшне все ответы API стали возвращать 404 ошибку. Когда стало очевидно, что это не связано с техническими работами или со сбоями "апстрима", мы стали изучать логи celery и nginx. Тогда-то и выяснилось, что 404 ошибку нам возвращает не наше API а какой-то чужой веб-сайт с совершенно левым HTTP-заголовком Server:. Если честно, я с таким ранее не сталкивался, поэтому просто сидел и смотрел в окно следующие 20 минут. Оказалось, что виновата "облачная погода": AWS, на котором был развернут API, погасил его контейнер и стартанул вместо него чужой веб-сайт с тем же IP-адресом. nginx, который тоже работал внутри AWS, ресолвил DNS "апстрима" только при старте, а потом просто возвращал ответы любого сервиса, который развернут на этом IP-адресе.

Решение было довольно простым: динамический proxy_pass и короткий resolver_timeout. В результате, при миграции API на другие адреса, прокси это замечает за 5 секунд.

Использование nginx избавило нас еще и от забот по обработке таймаутов и сетевых ошибок, поддержке keep-alive и переиспользованию SSL-соединений. В общем, идея оказалась удачной.

Маршрутизация в RabbitMQ

Для корректной работы прокси с клиентами через RabbitMQ нужно было учесть много нюансов.

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

• Во-вторых, запросы и ответы клиентов должны быть изолированы друг от друга: наверно, не очень хорошо получить ответ на запрос, который ты не посылал.

• В-третьих, "раскладка" очередей клиента не должна регулироваться прокси-сервисом, т.к. это зона ответственности разработчиков клиентского проекта.

Получилась следующая реализация:

1. Каждому клиенту при заведении создаются свои request_exchange и response_exchange.

2. После заведения нового клиента прокси автоматически создает и начинает обрабатывать очереди RabbitMQ для каждого активного эндпоинта. Очереди биндятся к request_exchange.

3. Клиент отправляет сообщение с запросом к API в request_exchange с routing key, соответствующим запрашиваемому url. Это обеспечивает попадание сообщения в нужную очередь, обрабатываемую воркерами прокси.

4. После успешного выполнения запроса прокси ставит в response_exchange сообщение с ответом API. Routing key сообщения задается в теле исходного сообщения-запроса за счет использования цепочек задач.

5. Клиент сам создает и биндит очереди обработки ответов к response_exchange.

Изоляция клиентов друг от друга реализована за счет добавления прав доступа в RabbitMQ: регулярками заданы имена exchange и очередей, в которые клиент может посылать сообщения и из которых может читать. Все собственные очереди клиента должны начинаться с соответствующего префикса.

Такая схема позволила обеспечить изоляцию, но одновременно сломала пару фичей Celery, таких как heartbeat, mingle и gossip. Если кто-то знает, как их вообще используют (хотя бы теоретически), напишите в комментариях, хочется знать, чего конкретно мы лишились.

В итоге

В итоге на нагрузочном тестировании удалось "разогнать" прокси до обработки 6000 запросов в секунду на одном ядре (к локалхосту, правда), при том что согласованный лимит API составлял всего 700 RPS. В таком виде система живет уже почти 3 года.

Хотелось ли переписать данный сервис на какой-либо другой язык вроде golang? Точно нет, пока затраты на хостинг из-за увеличившейся на 2 порядка нагрузки не перевесят затраты на написание всего задействованного функционала, предоставляемого Celery.