Logpush доставляет логи пакетами как можно быстрее, позволяя Cloudflare предоставлять информацию практически в режиме реального времени. Во время нашей недавней Birthday Week мы сделали Logpush доступным на всех тарифах self-serve и Transformersобщедоступным[[/L1].
Эти изменения вошли в число восьми основных обновлений Cloudflare Observability и отражают наше стремление сделать каждую функцию Cloudflare доступной для всех. В этом посте мы подробнее рассмотрим, как работают Transformers внутри конвейера экспорта Logpush. Команды Observability используют Logpush для экспорта логов Cloudflare в системы управления событиями и информацией о безопасности (SIEM), аналитические платформы и объектные хранилища. Прежде чем эти логи поступят в пункт назначения, командам может потребоваться удалить проверки работоспособности, исключить конфиденциальные поля или изменить структуру записей для дашбордов и расследований. Сегодня для этого часто требуется отдельная служба обработки, которую команды должны развертывать, масштабировать, контролировать и обслуживать.
Например, команда, собирающая логи с нескольких CDN, может добавить статическое поле провайдера для нормализации записей до того, как они поступят на ее аналитическую платформу. Команды также платят за экспорт, прием и хранение записей, которые они отбрасывают сразу по прибытии, поэтому их фильтрация перед доставкой может сократить объем данных в последующих системах и снизить затраты.
С помощью Transformers эта обработка переносится в рабочий процесс экспорта Logpush. Команды определяют SQL-запрос для фильтрации записей, маскирования конфиденциальных данных и изменения структуры вывода путем добавления, переименования, удаления или вложенности полей. Logpush применяет запрос перед экспортом, поэтому каждый пункт назначения получает данные в нужном ему виде. Мы сделали этот рабочий процесс простым для клиентов, встроив SQL-механизм трансформации в высокопроизводительный конвейер экспорта Logpush, сохранив при этом предсказуемость производительности и использования ресурсов.
Как работают Logpush Transformers
Transformer — это SQL-запрос, написанный для конкретного набора данных Logpush. Запрос ссылается на поля из этого набора данных и определяет, какие записи и поля должен экспортировать Logpush. Logpush применяет запрос к каждому пакету перед экспортом. Записи, не соответствующие условию WHERE, удаляются. Для оставшихся записей результат инструкции SELECT становится экспортированной записью. Transformers могут изменять структуру и содержимое каждой записи, а преобразованные результаты экспортируются в формате NDJSON.
Клиенты могут создавать, просматривать и управлять Transformers в панели управления Cloudflare через Transformer Studio или программно через API. После сохранения Transformer его можно прикрепить к совместимому заданию NDJSON Logpush, использующему тот же набор данных. Каждое обновление создает новую версию, поэтому клиенты могут отслеживать изменения Transformer с течением времени.
В следующем примере изменяется структура лога HTTP-запросов Cloudflare в соответствии со схемой, определенной командой и используемой у разных CDN-провайдеров. Преобразование между форматами сериализации не производится: как входные, так и выходные данные используют NDJSON.
Transformer переименовывает и вкладывает поля, маскирует часть IP-адреса клиента, добавляет информацию о провайдере и ошибках, удаляет неиспользуемые поля и отфильтровывает внутренний трафик и трафик проверки работоспособности:
Полученная запись NDJSON использует нормализованную схему:
Как Transformers выполняют SQL с помощью Apache DataFusion и Arrow
Конвейер данных Cloudflare обрабатывает более одного миллиарда событий в секунду. Служба доставки Logpush, которую мы называем Bufstream, доставляет более 110 миллионов записей Logpush в секунду в пункты назначения клиентов. Она работает на сотнях подов Kubernetes, каждый из которых запрашивает временной срез данных от Buftee, распределенного внутреннего брокера сообщений Cloudflare на базе логов. Эта модель распределяет работу горизонтально между заданиями Logpush. Объем заданий подчиняется распределению, похожему на закон Парето: большинство заданий работают с меньшими объемами, в то время как небольшое число высоконагруженных заданий обрабатывают миллионы логов каждую секунду. Transformers должны были соответствовать тем же операционным требованиям, что и остальная часть Logpush: надежная доставка, высокая пропускная способность и предсказуемое использование ресурсов без создания дополнительных сценариев сбоев на критическом пути экспорта.
Наш первоначальный прототип запускал определенные клиентом Transformers на WebAssembly в Cloudflare Workers, которые обеспечивают глобально распределенные изолированные вычисления. Однако вызов Worker добавлял бы сетевой вызов в путь экспорта Logpush. Эта удаленная зависимость могла бы приводить к временным тайм-аутам и повторным попыткам, снижая пропускную способность и увеличивая задержку доставки.
Мы решили оставить выполнение Transformer внутри процесса в нашей службе доставки, чтобы уделять первоочередное внимание надежности и пропускной способности доставки. Мы оценили несколько механизмов выполнения SQL и выбрали Apache DataFusion, который уже используется в Cloudflare Basin.
Apache DataFusion — это SQL-движок на базе Rust, созданный на основе Apache Arrow, с планировщиком SQL и векторизованным колоночным движком выполнения. В ходе бенчмарка выполнение подготовленного Transformer для синтетического пакета http_requests из 5000 строк из одной горутины Go заняло около 3 мс, что эквивалентно примерно 1,6 миллиона строк в секунду.
Прикрепление Transformer к заданию добавляет нагрузку на ЦП в его путь обработки. Bufstream использует параллелизм преобразований для каждого задания, поэтому высокопроизводительные задания могут выполнять пакеты преобразований параллельно. Встраивание DataFusion в Bufstream позволяет этим пакетам использовать существующую инфраструктуру доставки службы, включая управление планированием, пакетами и назначениями, без внедрения отдельного уровня трансформации или сетевого перехода.
Выбор DataFusion означал объединение Bufstream, написанного на Go, с движком, написанным на Rust. Следующей задачей было эффективное перемещение типизированных пакетов через эту границу.
Итоговая архитектура внутрипроцессного конвейера Transformer показана на диаграмме ниже.
От записей к столбцам
Bufstream получает сжатые пакеты Logpush, содержащие записи Cap'n Proto от Buftee. Для задания с подключенным Transformer Bufstream распаковывает пакет, анализирует подходящие логи и формирует типизированный объект Apache Arrow RecordBatch перед выполнением преобразования.
Apache Arrow — это независимый от языка колоночный формат в памяти, предназначенный для эффективной аналитики на современных ЦП и ГП. Он представляет плоские и вложенные данные в виде типизированных непрерывных столбцов. Arrow IPC предоставляет Bufstream и DataFusion стандартный способ обмена пакетами через границу интерфейса внешних функций (FFI) между Go и Rust, предоставляемый через интерфейс двоичных данных приложений C (ABI) и вызываемый из Go с помощью
, механизма Go для вызова кода C.
Среды выполнения Go и Rust не разделяют владение данными преобразования. Go передает указатель на свой входной буфер Arrow IPC в Rust только на время вызова cgo, и Rust не сохраняет этот указатель после возврата управления. Rust владеет преобразованным выходным буфером; Go копирует его в управляемую Go память, а затем явно освобождает выделенную память в Rust. Это делает владение и время жизни явными на границе языков, позволяя избежать ошибок управления памятью между средами выполнения.
Использование Arrow IPC позволяет обертке Rust DataFusion применять Arrow StreamReader для чтения типизированных значений RecordBatch, а не разбирать отдельные строки из промежуточного формата вроде JSON и заново создавать типизированные столбцы Arrow для выполнения. Типы полей, поддержка значений NULL и вложенные значения одинаково представляются с обеих сторон границы, поэтому DataFusion получает данные в своей нативной колонно-ориентированной модели.
Пересечение границы Go/Rust не обходится бесплатно: пакеты по-прежнему необходимо сериализовать, декодировать и обрабатывать в памяти. Arrow IPC сохраняет эту работу пакетной и позволяет избежать синтаксического анализа и преобразовывания для каждой записи. Благодаря этому граница лучше подходит для конвейера доставки Logpush с высокой пропускной способностью, сохраняя при этом проверку схемы на соответствие определению датасета Logpush.
Каждому пакету Arrow требуется схема, описывающая его поля и типы. Вместо того чтобы выводить эту схему из логов в конкретном пакете, Bufstream предоставляет сгенерированную схему из спецификации каждого датасета Logpush. Это дает DataFusion стабильное представление о доступных полях и типах, обеспечивая валидацию запросов. Таблица, указанная в предложении FROM трансформера, определяет, какая схема датасета используется в запросе. Например, этот трансформер запрашивает датасет http_requests:
Выполнение преобразования
Когда клиент сохраняет трансформер через API или Transformer Studio, Cloudflare сохраняет его SQL-запрос в R2. Bufstream извлекает запрос через многоуровневый кэш и преобразует его в многократно используемый логический план DataFusion, который определяет сканирование, фильтры, проекции и выражения. Bufstream подготавливает этот план один раз для каждого временного среза данных задания и повторно использует его для каждого пакета Arrow в этом срезе. Для каждого нового пакета Arrow обертка Rust связывает пакет в качестве входных данных таблицы для плана, выполняет запрос и возвращает пакеты Arrow, содержащие выбранные записи и поля.
Трансформеры спроектированы по принципу «безопасного сбоя». Если преобразование не может быть выполнено, Bufstream не сдвигает смещение доставки задания и не отправляет нетрансформированный пакет. Доставка сырых данных вместо трансформированных может нарушить допущения конвейеров ниже по потоку относительно схемы, фильтрации или маскирования, поэтому Bufstream предпочитает повторно пытаться выполнить неудачные преобразования.
Чтобы перманентно сбоящее задание не потребляло емкость, необходимую для здоровых заданий, Bufstream постепенно увеличивает задержки между повторными попытками и сообщает об ошибках в задании. Задания, которые остаются в состоянии сбоя в течение длительного периода времени, автоматически отключаются. Клиенты могут контролировать успешные и неудачные преобразования, объем данных до и после преобразования, а также продолжительность преобразования в представлении состояния трансформеров на вкладке «Состояние» (Health) в Logpush.
Возврат к конвейеру экспорта
После выполнения Bufstream преобразует вывод Arrow в логи NDJSON. Он пакетирует, сжимает с помощью gzip и выгружает эти логи в настроенный пункт назначения. Пункты назначения принимают данные с разной скоростью и могут временно замедляться. Bufstream регулирует количество горутин выгрузки, уважая ограничения параллелизма пункта назначения.
Встроив выполнение SQL в Bufstream, трансформеры используют существующую распределенную инфраструктуру, пакетную обработку и элементы управления пунктами назначения Logpush. Конвейер данных Cloudflare справляется с этой сложностью, в то время как клиенты создают и применяют трансформер всего за несколько шагов.
Начало работы
Откройте Transformer Studio на странице Logpush в панели управления Cloudflare и выполните следующие действия:
- Выберите датасет Logpush, который хотите преобразовать.
- Напишите свой SQL-запрос и предварительно просмотрите его на примере записи.
- Сохраните трансформер.
- Прикрепите его к соответствующему заданию Logpush.
Трансформеры доступны на тарифах Free, Pro, Business и Enterprise. Ознакомьтесь с документацией по трансформерам для получения сведений по настройке и с документацией по ценам Logpush для получения актуальной информации о ценах.
