Logpush доставляет логи партиями как можно быстрее, что позволяет Cloudflare предоставлять информацию почти в режиме реального времени. Во время нашей недавней Birthday Week мы сделали Logpush доступным на всех тарифных планах с самообслуживанием и объявили общую доступность Transformers.
Эти изменения были среди восьми крупных обновлений Cloudflare Observability и отражают нашу приверженность сделать каждую функцию Cloudflare доступной для всех. В этой статье подробно рассмотрено, как работают Transformers внутри конвейера экспорта Logpush. Команды наблюдаемости используют Logpush для экспорта логов Cloudflare в системы управления информацией и событиями безопасности (SIEM), аналитические платформы и объектное хранилище. Прежде чем эти логи достигнут пункта назначения, командам может потребоваться удалить проверки состояния, исключить конфиденциальные поля или изменить форму записей для дашбордов и расследований. Сегодня это часто требует отдельного сервиса обработки, который команды должны развернуть, масштабировать, мониторить и поддерживать.
Например, команда, собирающая логи с нескольких CDN, может добавить статическое поле provider, чтобы нормализовать записи до их поступления в аналитическую платформу. Команды также платят за экспорт, прием и хранение записей, которые они отбрасывают сразу после получения, поэтому фильтрация до доставки может сократить объем данных и снизить затраты.
С Transformers эта обработка перемещается в рабочий процесс экспорта Logpush. Команды определяют SQL-запрос для фильтрации записей, редактирования конфиденциальных данных и изменения вывода путем добавления, переименования, удаления или вложения полей. Logpush применяет запрос до экспорта, поэтому каждое место назначения получает данные в нужном формате. Мы упростили этот рабочий процесс для клиентов, встроив движок SQL-преобразований в высокопроизводительный конвейер экспорта Logpush при сохранении предсказуемости производительности и использования ресурсов.
Transformer — это SQL-запрос, написанный для конкретного набора данных Logpush. Запрос ссылается на поля этого набора данных и определяет, какие записи и поля Logpush должен экспортировать. Logpush применяет запрос к каждой партии перед экспортом. Записи, которые не соответствуют условию WHERE, удаляются. Для оставшихся записей результат выражения SELECT становится экспортируемой записью. Transformers могут изменять структуру и содержание каждой записи, а преобразованные результаты экспортируются в формате NDJSON.
Клиенты могут создавать, предварительно просматривать и управлять Transformers в панели Cloudflare через Transformer Studio или программно через API. После сохранения Transformer можно прикрепить к совместимой задаче Logpush в формате NDJSON, использующей тот же набор данных. Каждое обновление создает новую версию, поэтому клиенты могут просмотреть, как Transformer изменялся со временем.
Приведенный пример преобразует лог HTTP-запроса Cloudflare в соответствие со схемой, определяемой командой и используемой во всех поставщиках CDN. Он не выполняет конвертацию между форматами сериализации: и вход, и выход используют NDJSON. Transformer переименовывает и вкладывает поля, редактирует часть IP-адреса клиента, добавляет информацию о провайдере и ошибках, удаляет неиспользуемые поля и фильтрует внутренний и проверочный трафик. В результате полученная запись в NDJSON использует нормализованную схему.
Как Transformers выполняют SQL с помощью Apache DataFusion и Arrow
Конвейер обработки данных Cloudflare принимает более миллиарда событий в секунду. Сервис доставки Logpush, называемый нами Bufstream, доставляет более 110 миллионов записей Logpush в секунду к точкам назначения клиентов. Он работает на сотнях подов Kubernetes, каждый из которых обрабатывает временной срез данных из Buftee, распределенного брокера сообщений Cloudflare на основе логов. Такая модель горизонтально распределяет работу между задачами Logpush. Объем задач следует распределению, похожему на распределение Парето: большинство задач работают с меньшими объемами, в то время как небольшое число высоконагруженных задач обрабатывает миллионы логов в секунду. Transformers должен был соответствовать тем же эксплуатационным требованиям, что и остальной Logpush: надежная доставка, высокая пропускная способность и предсказуемое использование ресурсов без введения дополнительных сценариев отказа в горячем пути экспорта.
Наш первоначальный прототип запускал определяемые клиентом WebAssembly Transformers в Cloudflare Workers, что обеспечивает глобально распределенное изолированное вычисление. Однако вызов Worker вставлял сетевой запрос в путь экспорта Logpush. Эта удаленная зависимость могла ввести временные тайм‑ауты и повторные попытки, снижая пропускную способность доставки и увеличивая задержки.
Мы решили выполнять Transformers в процессе внутри нашего сервиса доставки, чтобы приоритизировать надежность и пропускную способность. Мы оценили несколько движков выполнения SQL и выбрали Apache DataFusion, который уже используется в Cloudflare Basin. Apache DataFusion — это движок запросов на Rust, построенный на Apache Arrow, с планировщиком SQL и векторизованным столбцовым исполнением. В бенчмарке выполнение подготовленного Transformer против синтетической партии http_requests из 5000 строк из одного горутины Go заняло около 3 мс, что эквивалентно примерно 1.6 миллионам строк в секунду.
Присоединение Transformer к задаче добавляет нагрузку CPU в ее путь обработки. Bufstream использует параллелизм трансформации на задачу, поэтому высоконагруженные задачи могут выполнять преобразование партий одновременно. Встраивание DataFusion в Bufstream позволяет этим партиям использовать существующую инфраструктуру доставки сервиса, включая планирование, пакетирование и управление точками назначения, без введения отдельного уровня трансформации или сетевого перехода.
Выбор DataFusion означал необходимость соединить Bufstream, написанный на Go, с движком на Rust. Следующей задачей стало эффективное перемещение типизированных партий через эту границу. Окончательный дизайн in-process конвейера Transformer показан ниже. Bufstream получает сжатые партии Logpush, содержащие записи Cap’n Proto из Buftee. Для задачи с прикрепленным Transformer Bufstream распаковывает партию, парсит подходящие логи и строит типизированный Apache Arrow RecordBatch перед выполнением трансформации.
Apache Arrow — это независимый от языка столбцовый формат в памяти, предназначенный для эффективной аналитики на современных CPU и GPU. Он представляет плоские и вложенные данные в виде типизированных смежных колонок. Arrow IPC предоставляет Bufstream и DataFusion стандартный способ обмена партиями через границу интерфейса между языками (FFI), экспонируемый через C ABI и вызываемый из Go с помощью cgo, механизма Go для вызова C-кода. Рантаймы Go и Rust не разделяют владение данными трансформации. Go передает указатель на свой входной буфер Arrow IPC в Rust только на время вызова cgo, и Rust не сохраняет этот указатель после возврата. Rust владеет буфером преобразованного вывода; Go копирует его в память, управляемую Go, а затем явно освобождает выделение Rust. Это делает владение и время жизни данных явными через границу языков, избегая ошибок управления памятью между рантаймами.
Использование Arrow IPC означает, что оболочка DataFusion на Rust может использовать Arrow StreamReader для чтения типизированных значений RecordBatch, а не парсить отдельные строки из промежуточного формата, такого как JSON, и заново собирать типизированные колонки Arrow для выполнения. Типы полей, допускаемость null и вложенные значения представлены единообразно по обе стороны границы, поэтому DataFusion получает данные в своей нативной столбцовой модели.
Пересечение границы Go/Rust не бесплатно: партии все равно нужно сериализовать, декодировать и управлять ими в памяти. Arrow IPC делает эту работу ориентированной на партии и избегает построчного парсинга и конвертации. Это делает границу более подходящей для высокопроизводительного конвейера доставки Logpush при сохранении валидации схемы в соответствии со спецификацией набора данных Logpush.
Каждая Arrow-партия требует схемы, описывающей ее поля и типы. Вместо вывода схемы из логов конкретной партии Bufstream поставляет сгенерированную схему из спецификации набора данных Logpush. Это дает DataFusion стабильный вид доступных полей и типов, позволяя проводить проверку запросов. Таблица, указанная в разделе FROM Transformer, выбирает, какую схему набора данных использует запрос. Например, этот Transformer запрашивает набор данных http_requests.
Когда клиент сохраняет Transformer через API или Transformer Studio, Cloudflare хранит его SQL-запрос в R2. Bufstream получает запрос через многоуровневый кэш и переводит его в переиспользуемый логический план DataFusion, определяющий сканирование, фильтры, проекции и выражения. Bufstream подготавливает этот план один раз на временной срез данных задачи и переиспользует его для каждой Arrow-партии в этом срезе. Для каждой новой Arrow-партии оболочка на Rust связывает партию как вход таблицы плана, выполняет запрос и возвращает Arrow-партии, содержащие выбранные записи и поля.
Transformers спроектированы с подходом fail closed. Если трансформацию невозможно выполнить, Bufstream ни сдвигает смещение доставки задачи, ни отправляет нетрансформированную партию. Доставка необработанных данных вместо преобразованных могла бы нарушить предположения схемы, фильтрации или редактирования downstream-пайплайнов, поэтому Bufstream предпочитает повторно пытаться выполнить неудавшиеся трансформации.
Чтобы не позволить постоянно падающей задаче потреблять ресурсы, необходимые здоровым задачам, Bufstream постепенно увеличивает задержки между повторами и сообщает об ошибках по задаче. Задачи, которые остаются в состоянии ошибки длительное время, автоматически отключаются. Клиенты могут отслеживать успешные и неудачные трансформации, объем данных до и после трансформации и продолжительность трансформации в представлении Transformer Health на вкладке Health в Logpush.
После выполнения Bufstream конвертирует Arrow-выход в NDJSON-логи. Он пакетирует, сжимает gzip и загружает эти логи в настроенное место назначения. Точки назначения принимают данные с разной скоростью и могут временно замедляться. Bufstream регулирует количество горутин для загрузки, соблюдая ограничения по конкурентности направления.
Внедрив выполнение SQL в Bufstream, Transformers используют существующую распределенную инфраструктуру Logpush, пакетирование и управление точками назначения. Конвейер данных Cloudflare обрабатывает эту сложность, а клиенты создают и применяют Transformer всего в несколько шагов.
Откройте Transformer Studio на странице Logpush в панели Cloudflare и выполните следующие шаги:
- Выберите набор данных Logpush, который вы хотите трансформировать.
- Напишите SQL-запрос и предварительно просмотрите его на примере записи.
- Прикрепите его к подходящей задаче Logpush.
Transformers доступны на тарифах Free, Pro, Business и Enterprise. См. документацию по Transformers для деталей настройки и документацию по ценам Logpush для актуальной информации о стоимости.
Подпишитесь, чтобы получать уведомления о новых публикациях.