За Cloudflare сидит больше 20% интернета, команда Data отвечает за конвейер данных, который перегоняет эти данные (и не только их) из глобальной сети в центральные дата-центры. Сеть Cloudflare — распределённая система из десятков тысяч серверов. Через неё и наш конвейер идут данные для Cloudflare Logs, SQL API, observability, внутренней аналитики и многого другого.
В мае этого года конвейер доставлял примерно два миллиарда событий в секунду. Это и трафик клиентов, и данные внутренних команд. Из года в год мы решаем одну и ту же задачу: трафик растёт экспоненциально, а железо — линейно. Чтобы расходы на ресурсы не вышли из-под контроля, приходится реагировать. За пять лет мы сделали ряд улучшений, которые равносильны отсрочке двух с половиной лет расширения парка серверов. В этой статье они описаны по порядку: разделение монолитной схемы данных в 2022 году, смена сжатия в 2024-м, перестройка репликации в конвейере весной 2026-го, децентрализация сэмплирования летом 2026-го и добавление агрегаций в конвейер осенью 2026-го.
Исходная архитектура
Конвейер начинается с внутреннего сервиса Logfwdr. Он собирает данные из сильно распределённых источников (например, Workers и FL). Дальше данные уходят на набор центральных приёмников, а те складывают их в нашу буферную систему Buftee. Потребители забирают данные из Buftee и раздают по конечным точкам, например в ClickHouse, на который мы сильно опираемся.
Упрощённая схема. Logfwdr принимает события от источников и пересылает их на центральные приёмники. Приёмники ставят их в очередь буферной системы, откуда забирают данные нижестоящие потребители.
Разбираем монолит
В 2022 году мы разобрали нашу единственную схему данных. Раньше был один поток со всеми полями всех продуктов, и внутренние потребители Cloudflare вынуждены были принимать всё подряд, а затем разделять записи или отбрасывать лишнее фильтрами. Тогда большинство наших продуктов данных строилось на метаданных HTTP, безопасности и CDN. В 2021 году схема сообщения выглядела так:
Часть полей относилась к одному продукту, а часть вообще никак не была связана друг с другом. Гипотеза была такой: ни одной команде не нужны 100% полей монолитной схемы. Значит, можно делать небольшие сообщения только с теми полями, которые нужны каждому продукту. Часть полей при этом дублировалась, некоторые даже в каждом сообщении. Но замена общего потока схемами под продукты всё равно сократила использование пропускной способности магистрали на 35–40%. Это убрало примерно год роста сжатого трафика.
График показывает обрыв пропускной способности в сжатых байтах в секунду после миграции с большой монолитной схемы.
Сжатие
К 2024 году трафик продолжал расти, и перегрузки сети мы разбирали каждый день. В ответ мы сменили алгоритм сжатия с LZ4 на zstd. Раньше мы уже пробовали zstd, но тогда заметили рост нагрузки на процессор, поэтому внимательно следили за метриками, чтобы убедиться, что компромисс того стоит.
Сначала мы включили zstd на нескольких серверах. К нашему облегчению, заметной разницы в загрузке CPU не было.
Версия 2024.6.2 была до смены сжатия, а 2024.7.1 — версия со сменой на zstd.
Увереннее выкатили на весь парк, и влияние на процессоры снова было небольшим. В итоге сжатый трафик сократился почти на 40%. Несжатые байты росли с прежней скоростью, то есть через трубу того же размера можно пропускать больше чем в 1,5 раза больше трафика. Иначе говоря, по оценке эквивалентного роста, это отбросило нас примерно на 10 месяцев назад.
График показывает фактическое сокращение числа сжатых байт в секунду.
В 2026 году мы попробовали три стратегии, чтобы сократить несжатые байты (а значит, и сжатые).
Балансировка нагрузки
Чтобы аналитика переживала региональные катастрофы, наши системы хранят избыточные данные в нескольких центральных регионах (данные, подпадающие под Customer Metadata Boundaries, хранятся строго в своём регионе). Исторически самым простым способом устойчивости к катастрофам была двойная запись: Logfwdr отправлял две полные копии каждого нерегионального аналитического события — по одной в каждый центральный регион. Каждый регион независимо прогонял полную копию до ClickHouse. В каждом регионе мы ещё применяем сэмплирование — очень масштабируемую технику, которая работает при любых объёмах данных и любом трафике. Мы прореживаем высокообъёмный трафик, а редкие события сохраняем.
Пока вторичные регионы использовали всего несколько потоков, с этим можно было жить. Но когда мы распространили отказоустойчивость почти на все аналитические потоки, трафик Logfwdr везде удвоился. Отправлять каждое событие дважды по глобальным каналам при экспоненциальном росте трафика было невыносимо.
С похожей задачей мы уже справились для Logpush. Там Logfwdr считает центральные дата-центры единым общим пулом приёмников: каждое сообщение уходит только в один регион, а если машина или площадка недоступна, происходит переключение на другой. Эта модель уже показала себя во время крупных инцидентов: трафик переезжал из одной площадки в другую без участия человека. Автоматическое переключение возможно, потому что доставка Logpush не хранит состояния: клиентским точкам назначения всё равно, какой центральный дата-центр отправил логи.
С аналитикой было сложнее. В отличие от логов без состояния, кластерам ClickHouse в обоих регионах всё равно нужна полная копия нерегиональных данных, чтобы запросы можно было мгновенно и независимо переключать. Прорыв — разделить транспорт внутри конвейера и избыточность базы:
1. Общий пул приёма. Мы применили модель Logpush к аналитическим потокам. Logfwdr отправляет каждое событие по глобальным каналам ровно один раз — на любой доступный центральный приёмник.
2. Поздняя репликация из центральных регионов. Вместо репликации трафика на источнике репликация происходит как можно позже в конвейере. Данные попадают в наш распределённый буферный слой Buftee, а удалённые вставщики (inserters) передают межрегиональные копии вниз по потоку.
3. Устойчивость к сбоям. Если центральный регион падает, Logfwdr сразу направляет весь нерегиональный трафик в уцелевший. Если связь между дата-центрами моргнула, каждый кластер продолжает принимать свою долю без противодавления (backpressure), а удалённые вставщики догоняют данные из буфера хранения Buftee, когда магистраль восстановится.
До: Logfwdr отправляет каждое нерегиональное событие в дата-центры обоих регионов. После: событие отправляется один раз, а вторую копию для ClickHouse делает удалённый вставщик после Buftee.
Для перенесённых наборов данных переключение убрало примерно 43 ГБ/с сжатого транзитного трафика, добавив лишь около 7 ГБ/с межрегиональной репликации в конвейере. Эту асимметрию объясняют два фактора:
Репликация идёт после сэмплирования. В старой модели Logfwdr отправляли по глобальным каналам две копии сырых несэмплированных событий. В новой каждый Logfwdr отправляет одну копию на центральные приёмники, те сразу применяют сэмплирование и только потом кладут данные в Buftee, так что по сети идёт уже сжатый поток.
Лучшее сжатие за счёт равномерного разбиения. На источнике данные раздроблены между тысячами отдельных серверов. После централизации потоки собираются и разбиваются на однородные пакеты по более широким временным окнам. Такая однородность позволяет алгоритмам сжатия достигать куда более высоких коэффициентов, чем на транзитных пакетах Logfwdr.
В сумме чистое сокращение по глобальным сетям составило 36 ГБ/с. Сначала мы раскатили это на два набора данных: HTTP-аналитику и Ready-Analytics, наш внутренний продукт «аналитика в один клик».
Первое переключение — HTTP-аналитика, сокращение примерно на 20 ГБ/с (50%+).
У наборов вроде Ready-Analytics трафик сразу упал на 43% (с ~28 до ~16 ГБ/с).
С весны мы переводим на эту архитектуру с балансировкой всё больше наборов данных.
Сэмплирование с самого начала
Ещё одна стратегия — расширить наш механизм сэмплирования и вынести критичную обработку ближе к началу конвейера. Сэмплирование уже применялось в центральных сервисах, в частности в приёмниках, и снимало нагрузку с нижестоящих систем. Мы можем эффективнее расходовать ресурсы, если отправляем вниз меньше данных (без потери качества).
Решение простое: сэмплировать ещё и в Logfwdr, сохранив сэмплирование в центральных приёмниках. Алгоритм в двух словах: у каждого набора данных есть ключ, который команды-производители выбрали для индексации и сэмплирования. При сэмплировании мы сначала перемешиваем записи по каждому ключу, чтобы убрать смещение. Затем для каждой группы записей берём выборку размером с квадратный корень из общего числа записей. Получается репрезентативная выборка, неважно, сэмплировали один раз или дважды. Чтобы получить оценки, мы масштабируем расчёты на коэффициент сэмплирования. Например, если из 100 записей выбрано 10, коэффициент 10/100, а несмещённая оценка исходного количества: 10 × 10 = 100.
Чтобы убедиться, что подход работает, мы сравнили число строк, записанных в ClickHouse, с оценённой суммой общего количества. Брали разные ключи при разных профилях трафика, чтобы охватить спектр от «реже всего встречающихся» ключей до «чаще всего».
Синий цвет — данные, сэмплированные один раз (контроль), красный — сэмплированные дважды. Оценка получается та же, хотя строк в базе меньше.
В итоге у ключей с большим объёмом данных коэффициент сэмплирования уменьшается, а у ключей с малым объёмом сэмплирование сравнимо с прежним. Выборка остаётся репрезентативной, а доставляется и хранится меньше событий. Качество данных не страдает, хотя количество уменьшается.
Если присмотреться и использовать степенные законы, видно, что дважды сэмплированные данные (красный) растут примерно как x^0,319, а сэмплированные один раз (синий, контроль) — как x^0,485. Иначе говоря, ключ с в 100 раз большим объёмом данных при однократном сэмплировании хранил бы примерно в 10 раз больше строк, а теперь хранит только примерно в 4 раза больше. В конкретных цифрах: 11,6 млн событий представлены 18 тыс. строк в ClickHouse вместо 79 тыс. Двойное сэмплирование ослабляет хранение для «самых частых» ключей, а для «самых редких» сохраняет прежнюю детализацию каждого события.
График со степенными законами показывает выравнивание зависимости между оценённым объёмом и числом сохранённых строк. Показатель x^0,25 мы не наблюдали, потому что двойное сэмплирование идёт отдельно на разных машинах и в разных пакетах, а центральные приёмники сэмплируют остаток после объединения части этих пакетов.
Сначала мы применили эту стратегию к трём нашим крупнейшим наборам данных. В каждом объём сжатых данных, которые мы изначально слали по сети на центральные приёмники, сократился минимум вдвое. Это 25 ГБ/с в сумме, или больше 2 ПБ в сутки.
В каждом наборе данных, к которому применили двойное сэмплирование, видно резкое различие до и после по сжатым байтам в секунду.
Агрегация
Cloudflare предлагает несколько аналитических наборов данных на основе свёрток (rollups): агрегатов сырых событий, которые считаются при приёме. Это обратный компромисс по сравнению с сэмплированием: мы не отвечаем приблизительно на любой вопрос, а отвечаем точно на фиксированный набор вопросов, если заранее выбрали измерения и метрики.
Например, в наборе HTTP мы считаем число уникальных IP-адресов за минуту, а не сами адреса. Другой пример: мы суммируем число запросов GET, POST, PUT и так далее для домена за эту минуту, сгруппировав их по HTTP-методу. Ни одно из этих вычислений не требует ждать хвоста конвейера, чтобы начаться.
Исторически агрегация происходила ближе к концу конвейера, в вставщиках, которые читают из Buftee и вставляют в ClickHouse. Децентрализация агрегации давала возможность распределить нагрузку и уменьшить объём данных, которые мы стягиваем в центр.
Поэтому мы сделали функциональность агрегации, которая работает на каждом шаге и даёт повторные возможности объединять совпадающие частичные агрегаты по мере схождения данных. Как и при сэмплировании, записи группируются по ключу, который определяет, какие записи можно объединить. Мы агрегируем на каждом экземпляре Logfwdr. Затем агрегируем снова, когда частично агрегированные события приходят на центральные приёмники, которые отправляют их на отдельные партиционированные буферы. Эти буферы собирают записи от нескольких приёмников, поэтому мы агрегируем ещё раз после того, как вставщики ClickHouse прочитали их пакетами. И наконец агрегируем в самом ClickHouse с помощью агрегирующих движков таблиц.
Схема показывает пример потока событий, агрегируемых по всему конвейеру.
Первым набором, на который мы это раскатили, был HTTP-трафик. Только в этом наборе сжатый трафик упал с 9+ ГБ/с до менее чем 2 ГБ/с. Хотя это сейчас набор с самой высокой кардинальностью, агрегации всё равно дали примерно 80% сокращения на сети.
Сжатые байты в секунду во времени для набора HTTP-свёрток. Оранжевый — скорость, когда агрегация происходила только по достижении вставщиков, зелёный — после переноса агрегации раньше по конвейеру.
Агрегация эффективна для наборов, которые легко сворачиваются, особенно для метрик использования, где важна точность сумм и счётчиков. Она же даёт командам-производителям ещё один способ снизить расход ресурсов помимо сэмплирования. Как и Ready-Analytics, мы сделали универсальный внутренний продукт «агрегирующая аналитика в один клик» под названием Ready-Aggregates. По мере развития этой возможности в конвейере мы ждём, что экономия будет расти.
Итоги
Лучшее доказательство экспоненциального роста — как быстро трафик съедает запас.
Возьмём Ready-Analytics, где применили два из трёх механизмов: балансировку нагрузки и сэмплирование. В середине мая балансировка сбросила трафик на 50%, с ~26 до ~13 ГБ/с. Через два месяца он снова вырос до ~20 ГБ/с. Органический рост стёр около половины сокращения. Затем мы создали ещё запас, снизив пропускную способность этого набора до однозначных значений, ~8 ГБ/с, когда раскатили сэмплирование.
Суммарная скорость Ready-Analytics на обоих переключениях. Зелёный — фактический трафик, жёлтые «призрачные» кривые — реконструкция трафика без балансировки (май) и без сэмплирования (июль).
Без этих двух вмешательств один Ready-Analytics превысил бы 60 ГБ/с к сентябрю. Обрабатывая меньше данных в центральных дата-центрах, мы к тому же экономим много процессоров и памяти. Графики показывают влияние балансировки и сэмплирования в Logfwdr на CPU и память только приёмников Ready-Analytics: сокращение примерно на 50%, вернулось около 750 процессоров и 3,5 ТиБ памяти.
По всему конвейеру с весны до осени балансировка нагрузки (~38 ГБ/с сэкономлено), сэмплирование (~25 ГБ/с на крупнейших наборах) и агрегация (пока 7+ ГБ/с) убрали с нашей магистрали около 70 ГБ/с сжатого трафика. Эти обрывы кривой дают немедленное облегчение.
Сейчас пропускная способность примерно на восемь месяцев отстаёт от ранее прогнозировавшейся траектории роста.
Если посмотреть на пять лет вперёд, изменения, сделанные только в этом году, дадут не меньший эффект, чем краткосрочная выгода. Мы вдвое уменьшили объём данных, который пришлось бы обрабатывать через полдесятилетия. Структурные и алгоритмические оптимизации, которые мы сделали и продолжим делать, необходимы, чтобы потребности в железе оставались устойчивыми.
Долгосрочный эффект сделанной в этом году работы: наш прогноз объёма данных на 2031 год сократился вдвое.