Distributed Throttler в YTsaurus Flow

Распределённый троттлер — глобальный для всего пайплайна механизм ограничения скорости. На контроллере для каждого именованного троттлера держится token bucket; джобы запрашивают у контроллера квоту перед обработкой сообщений или перед произвольным пользовательским действием.

Квотироваться может либо число обработанных сообщений, либо суммарный размер их payload в байтах — отдельно или вместе.

Типичные задачи:

• ограничить суммарную нагрузку на внешний API со стороны всех джобов;
• равномерно распределить пропускную способность между партициями одного Computation;
• притормозить чтение из источника.

Приоритет по отставанию

Каждая джоба прикладывает к запросу за квотой свою временную метку — ту же, по которой во входном буфере определяется отставание. Берётся минимум по всем stabilized_event_timestamp + stream_delay из input streams и read_alignment_timestamp из source streams. Сервер ранжирует запросы по возрастанию этой метки: сначала отдаёт квоту тем, кто отстаёт сильнее. При перекосе нагрузки между партициями одного Computation отстающие автоматически получают больше квоты, а лидирующие — замедляются.

Взвешенные классы квоты

У именованного троттлера можно объявить классы с весами. Между классами, в которых есть ожидающие запросы, квота распределяется пропорционально весам и с полным использованием доступной пропускной способности. Если класс простаивает, его доля немедленно перераспределяется между активными классами. Вес задаёт долгосрочную долю полосы, а не строгий приоритет.

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

Если classes отсутствует, сохраняется прежнее поведение с одной очередью и прежняя семантика token bucket. Зарезервированный класс default всегда имеет вес 1.0. Отсутствующий, пустой или неизвестный quota_class_id на сервере попадает в default; так обслуживаются троттлеры, которым класс не задан. Класс относится ко всем запросам компьютейшена к этому троттлеру: если тот же id получен вручную через GetThrottlerOrThrow, эти запросы идут в том же классе, а не в default. Класс задаётся отдельно для каждого автоматического троттлера — input_rows_throttler_class_id и input_bytes_throttler_class_id, — и каждый проверяется против классов своего троттлера: неизвестный класс отклоняется валидацией до запуска пайплайна.

"dynamic_spec" = {
    "throttlers" = {
        "output_quota" = {
            "limit" = 1000.0;
            "period" = "1s";
            "request_period" = "500ms";
            "max_grant_amount" = 50;
            "classes" = {
                "vip" = {"weight" = 5.0;};
                "regular" = {"weight" = 3.0;};
                "bulk" = {"weight" = 1.0;};
            };
        };
    };
    "computations" = {
        "ReaderVip" = {
            "input_rows_throttler_id" = "output_quota";
            "input_rows_throttler_class_id" = "vip";
        };
    };
}

Веса как абсолютные скорости

Параметр use_class_weights_as_limit = %true делает скоростью бакета сумму объявленных весов классов, поэтому вес читается как единицы квоты в секунду, а не как доля от отдельно заданного limit. С весами 5.0, 3.0 и 1.0 троттлер выдаёт 9 единиц в секунду, и пока все классы забэкложены, каждый обслуживается со скоростью своего веса. Work conservation при этом сохраняется: пока остальные простаивают, класс по-прежнему выходит за свой вес, то есть вес — это гарантия при конкуренции, а не потолок.

Параметр требует хотя бы одного класса и несовместим с limit — если заданы оба, конфигурация отклоняется валидацией.

Зарезервированный класс default в сумму не входит. Запросы без класса на таком троттлере конкурируют с весом 1.0 поверх объявленной скорости и опускают каждый объявленный класс ниже его веса; это видно по сенсорам троттлера /classless_requests и /unknown_class_requests.

Задержка переключения между классами

Планировщик пересматривает активные классы после завершения текущего серверного чанка. max_grant_amount задаёт максимальный размер чанка в абсолютных единицах квоты; если его не задать, один запрос удерживает token bucket всё время своего prefetch-окна и ровно на столько же задерживает остальные классы. В наблюдаемую задержку также входят drain_period, RPC, локальный prefetch, empty_batch_backoff источника и commit Flow.

Веса и пропускная способность ридеров

Наблюдаемые доли совпадают с настроенными весами, только пока каждый измеряемый класс забэкложен и каждый ридер способен выбрать положенную ему долю. Итерация ридера несёт фиксированные накладные расходы, не зависящие от выданной скорости, поэтому при небольшом max_rows_per_batch высоковесный класс упирается в собственный потолок пропускной способности раньше, чем добирает свою долю; work conservation отдаёт остаток низковесным классам, и наблюдаемое распределение смещается в сторону равного. Масштабируйте max_rows_per_batch вместе с limit, чтобы накладные расходы итерации оставались малой долей времени ожидания квоты.

Реконфигурация на лету

Все параметры троттлера (limit, period, request_period, retrying_channel, rpc_timeout, classes, max_grant_amount, use_class_weights_as_limit) подхватываются без перезапуска пайплайна — достаточно обновить dynamic_spec. Кэшированный в пользовательском коде IThroughputThrottlerPtr остаётся валидным после смены конфига.

Вес и размер серверного чанка меняются без перестроения prefetch-клиента. Исключение — use_class_weights_as_limit: тогда вес входит в скорость, поэтому изменение веса, меняющее размер prefetch-окна, приводит к перестроению клиента, а слишком малое для этого изменение оставляет клиента и уже полученную им квоту на месте. Новая скорость в любом случае применяется на сервере. Смена input_rows_throttler_class_id или input_bytes_throttler_class_id влияет на последующие RPC; уже полученная локально или находящаяся в полёте квота остаётся учтённой в прежнем классе.

Классы квоты и watermark alignment

Доступность квоты не отменяет watermark alignment: чтение разрешено, только если одновременно выполнены выходные лимиты, alignment и ограничение окна чтения. Поэтому класс с большим весом не ускорит чтение, если источник придерживает alignment, — вес распределяет только квоту. Настройка самого alignment описана в watermark strategy.

Что будет, если контроллер недоступен

Клиент троттлера переоткрывает соединение через retrying_channel. Пока контроллер не отвечает, локальный prefetch-кэш доедается, и Throttle() блокируется в ожидании квоты — джоба прогресса не делает. Когда контроллер возвращается, ожидание разрешается само. Если же контроллер не отвечает дольше, чем суммарный бюджет ретраев retrying_channel, работа с троттлером выкинет исключение и джоба упадёт.

Что будет, если квоты не хватает

Если суммарный спрос джоб на много порядков превышает возможную выдачу, отдельные RequestQuota могут долго стоять в серверной очереди и в итоге не уложиться в свой rpc_timeout × retry_attempts. В таком случае Throttle() тоже выкинет исключение, что приведет к падению джобы.

Fallback на локальный лимит ни в одном из этих случаев сейчас не поддержан.

Конфигурация

Все настройки — в DynamicSpec. Троттлеры объявляются в корне. В примере ниже external_api_quota — пользовательский id троттлера, RequestEnricher — имя computation:

"dynamic_spec" = {
    "throttlers" = {
        "external_api_quota" = {
            "limit" = 100.0;
        };
    };
    "computations" = {
        "RequestEnricher" = {
            "input_rows_throttler_id" = "external_api_quota";
        };
    };
}

Поля TDynamicThrottlerSpec:

Источник: yt/yt/flow/library/cpp/common/spec.h

Конфигурация именованного распределённого троттлера.

{
    "enable_exponential_retry_backoffs" = %true;
    "retry_attempts" = 100;
    "retry_timeout" = 600000;
}

Применение

Есть два варианта — автоматический и ручной. Они не конфликтуют: можно пользоваться одним или сразу обоими.

Автоматический: ограничить скорость входного батча

В TDynamicComputationSpec есть два поля — input_rows_throttler_id и input_bytes_throttler_id. Если они заданы, перед каждой итерацией Process компьютейшен ждёт квоту:

• по числу сообщений во входном батче (input_rows_throttler_id);
• по сумме byte_size сообщений батча — это системный размер сериализованного представления, без учёта компрессии/шифрования на сетевом уровне (input_bytes_throttler_id).

Ожидание оформляется отдельным span'ом Input.Throttle в трейсинге компьютейшена (видно в Jaeger и на графиках "Epoch parts time" в UI).

Id должен быть объявлен в dynamic_spec/throttlers. Можно задать оба поля сразу — тогда итерация ждёт обе квоты, — но они должны ссылаться на разные троттлеры: один троттлер меряет либо число сообщений, либо байты, и общий token bucket складывал бы штуки с байтами. Такая спека отклоняется валидацией.

Ручной: GetThrottlerOrThrow(id)из пользовательского кода

Если автоматического троттлинга по входному батчу недостаточно (например, нужен rate-limit на каждый внешний запрос), process function получает троттлер из IRuntimeContext. Метод context->GetThrottlerOrThrow(id) возвращает IThroughputThrottlerPtr — стандартный YT-троттлер, у которого вызывают Throttle(amount) и ждут результат. Если id нет в throttlers, GetThrottlerOrThrow бросает исключение; context->TryGetThrottler(id) в этом случае возвращает nullptr — удобно, когда сам факт объявления троттлера в dynamic_spec включает троттлинг для соответствующей сущности (например, троттлер на каждую модель или внешний сервис).

Возвращённый указатель стабилен в рамках всей жизни джоба: фабрика подменяет внутренний клиент при Reconfigure, поэтому process function может сохранить его в Init и использовать дальше без переполучения. Это не относится к nullptr от TryGetThrottler: он означает лишь «сейчас не объявлен», и после добавления троттлера в dynamic_spec его нужно запросить заново.

Сейчас этот механизм поддержан только в C++.

Мониторинг

У дашборда Flow есть отдельный таб Distributed throttler. Основные сенсоры:

• серверные (контроллер): value.rate — фактическая выдача квоты, queue_size — сколько запросов ждут, wait_time.max — хвост ожидания.
• клиентские (воркер, на каждом Computation): consumed.rate / released.rate, wait_time.max — время, которое джоба простояла в Throttle(), request_count.rate — частота prefetch-RPC.

См. также

• Spec, DynamicSpec и Config
• Computation