Про асинхронность в Cloud Functions

Сегодня я хотел бы разобрать модель работы облачных функций в Яндекс Облаке.

Функции в облаке запускаются синхронно и в один момент времени обрабатывают ровно один запрос.

к сведению

В конце 2023 года появилась возможность запускать функции асинхронно. Подробнее можно почитать в отдельном посте.

Это значит, что вы можете быть уверены, что все переменные, созданные внутри обработчика будут относиться именно к текущему контексту. Т.е. если вы в обработчике идете в базу, чтобы получить текущего пользователя, это будет именно тот пользователь, чей id пришел в запросе.

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

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

Теперь подробнее разберем, что значит синхронность запуска функции.

Для этого возьмем простую функцию на Python, которая будет делать только одно: ждать 1 секунду и возвращать ответ Ok.

import time

def handler(event, context):
    time.sleep(1)
    return {
        'statusCode': 200,
        'body': 'Ok',
    }

Теперь, если мы вызовем ее при помощи curl, то через 1 секунду получим ответ Ok.

curl 'https://functions.yandexcloud.net/d4eic18eqks97gmifd3l'

Но, что будет, если мы будем ждать ответа не 1 секунду, 500мс?

curl -m .5 'https://functions.yandexcloud.net/d4eic18eqks97gmifd3l'

В ответ мы увидим вот что:

curl: (28) Operation timed out after 504 milliseconds with 0 bytes received

А в логах функции:

Что же произошло?

Рантайм функций увидел, что клиент разорвал соединение и больше не ждет ответа, значит функция не сможет вернуть результат своего выполнения, вызвавшему её коду, а значит можно больше не тратить деньги клиента Облака (ведь облачные функции тарифицируются именно за время исполнения) и прервать функцию. В логах такое поведение можно найти по коду ответа 499. Этот нестандартный код ответа говорит, что клиент не дождался и отменил запрос.

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

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

Что же это значит для разработчика, который пишет приложение в serverless модели? Нужно постараться не использовать синхронные вызовы одной функции из другой.

Каскадный таймаут

Потому что когда наступит таймаут у внешней функции, то и вызванная из неё функция завершится с ошибкой. И тут либо аккуратно следить за настройками функций, поддерживая таймауты у внешних чуть больше, чем у внутренних, либо использовать асинхронные подходы.

К тому же, вызывая одну функцию из другой, вы заплатите вдвойне, т.к. будет работать два экземпляра рантаймов. Так же придется следить за квотами.

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

Асинхронность в функциях.

Первое, о чем вы могли подумать: «Точно просто вместо обычной функции-обработчика сделаем её асинхронной!»

const {
  setTimeout,
} = require('node:timers/promises');

module.exports.handler = async function (event, context) {
    await setTimeout(1000);
    return {
        statusCode: 200,
        body: 'ok',
    };
};

Но это не поможет. Да обработчик стал асинхронным, теперь мы можем использовать в нём синтаксический сахар async/await, но вызов функции так и останется синхронным. То есть, если вы вызывая функцию по http, не дождетесь окончания ее исполнения, то она точно так же как и в случае с синхронным обработчиком вернёт 499 код ошибки.

Так что же делать, если вы хотите выполнить в облачной функции какую-то длительную работу и не хотите ждать ее завершения, и вас бы устроило, чтобы например результат был записан в базу данных?

Для этого можно использовать очереди сообщений, таким образом развязать постановку задачи и ее выполнение. А заодно и распараллелить выполнение задач.

Как положить сообщение в очередь?

У нас есть как минимум два похода:

1. Мы можем напрямую взаимодействовать с очередью через AWS SDK (т.к. YMQ поддерживает протокол AWS SQS);

const AWS = require('aws-sdk');
const QUEUE_URL = process.env.QUEUE_URL ?? "";
const AWS_ACCESS_KEY_ID = process.env.AWS_ACCESS_KEY_ID ?? "";
const AWS_SECRET_ACCESS_KEY = process.env.AWS_SECRET_ACCESS_KEY ?? "";

const sqs = new AWS.SQS({
    region: "ru-central1",
    endpoint: "https://message-queue.api.cloud.yandex.net",
    credentials: {
        accessKeyId: AWS_ACCESS_KEY_ID,
        secretAccessKey: AWS_SECRET_ACCESS_KEY,
    }
});

function messageSendResultHandler(err, data) {
    if (err) {
        console.log("Error", err);
    } else {
        console.log("Success", data.MessageId);
    }
}

module.exports.handler = async function (event, context) {
    sqs.sendMessage({
        MessageBody: JSON.stringify({"foo": "bar"}),
        QueueUrl: QUEUE_URL
    }, messageSendResultHandler);

    return {
        statusCode: 200,
        body: 'ok',
    };
};

Пример отправки сообщения в YMQ

2. Воспользоваться интеграцией API Gateway.

После этого нужно будет лишь настроить триггер, который будет обрабатывать сообщения из очереди и готово.

Таким образом, если у вас есть долгая операция и требование от внешней системы быстро отвечать на запросы, вы сможете выполнить оба условия. В первой функции быстро складывать данные в очередь задач и возможно проверять по ключу в базе готова ли задача, а другой отдельной асинхронной функции (кстати, это может быть даже и не функция, а виртуальная машина, в том случае, если вам не хватает ограничений serverless среды и нужны, например большие ресурсы, вычисления на GPU или доступ к сервисам из внутренней VPC) выполнять отложенные задачи.

Ограничения YMQ.

У вас может возникнуть желание применить описанный выше подход и для FIFO-очередей. К сожалению в текущей реализации триггеров вы не сможете создать триггер для FIFO-очереди YMQ. В таком случае у вас есть опции использовать для разбора очереди приложение развернутое на ВМ, в нём вы сможете читать любые YMQ очереди. Или же, если вы хотите оставаться serverless, вы можете использовать Yandex Data Streams.

Это сервис, который реализует API совместимое с AWS Kinesis. Вы точно так же, как и в случае с YMQ, сможете писать в него при помощи либо AWS SDK, либо интеграции с API Gateway. Yandex Data Streams предоставит вам гарантии порядка обработки сообщений записанных в стрим. Вам лишь нужно будет определить ключ, по которому будут шардироваться сообщения и помнить, что порядок гарантируется только в рамках сегмента, который задается ключом.

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