Streamline: пользовательские видеоконвейеры с помощью Cloudflare Stream и Workers

Cloudflare Stream — это мощная платформа для вещания, которая для многих наших клиентов просто работает «из коробки». Но что, если вы хотите отображать динамические аннотации в прямом эфире или создать альтернативную версию размещенного видео с «вшитыми» субтитрами? Для этого вам потребуется запустить собственный видеоконвейер.

Сегодня мы выпускаем новую площадку для разработчиков, Streamline, которая демонстрирует, как можно создать систему для предоставления таких специализированных видеосервисов на платформе Cloudflare Developer Platform. Мы расскажем, как Streamline использует Workers, Containers и несколько медиапротоколов для модификации видео, а затем мгновенно публикует результат в виде прямой трансляции или нового размещенного видео. У вас также будет возможность опробовать это в своих проектах.

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

Cloudflare предоставляет необходимые нам примитивы. Containers — это долгоживущие среды выполнения, подходящие для обработки медиа. Durable Objects помогают с оркестрацией. Наконец, Workers идеально подходят для передачи сигналов управления и мониторинга.

Для Streamline мы создали медиадвижок, работающий в контейнере для обработки медиа в реальном времени. Контейнер управляется воркером (Worker), который предоставляет агенту или пользователю функции управления, предварительного просмотра и тестирования. Обработка будет продолжаться, даже если воркер отключится. Мы спроектировали Streamline с использованием модульных компонентов, чтобы в будущем медиадвижок можно было заменить на специализированные продукты для кодирования.

Архитектура

Развертывание Streamline состоит из двух компонентов: Media Engine (медиадвижок), который обрабатывает ввод/вывод и обработку медиа, и управляющее приложение (Application), которое создает, настраивает, отслеживает и останавливает медиасессии.

Media Engine

Media Engine состоит из двух компонентов:

• Контроллер (Controller). Это управляющая оболочка, написанная на Go, которая реализует HTTP-сервер, принимает входящие запросы и преобразует их в операции, выполняемые медиадвижком.
• Процессор (Processor), который выполняет непосредственную обработку медиа. Текущая реализация использует FFmpeg, но это внутренняя деталь реализации, а не часть пользовательского API.

Media Engine размещается в контейнере и обрабатывает весь ввод/вывод медиа, а также саму обработку. Он может получать воспроизведение RTMPS по сети из одного источника Stream Live и публиковать вывод RTMPS в другой источник Stream Live. Он может получать HLS-манифест Cloudflare Stream и его сегменты для использования размещенных видео в качестве входных данных. Он может принимать видеовход из источника, предоставленного управляющим приложением, например, с веб-камеры. Он может публиковать видео для предварительного просмотра через исходящий WebSocket в ретранслятор Durable Object. Приложение, которому нужен предварительный просмотр, может подключиться к этому ретранслятору через собственный WebSocket.

Приложение (Application)

Приложение построено с использованием Workers и может быть полнофункциональным браузерным приложением, агентом или встроенной системой. Оно состоит из:

• Пользовательского интерфейса (UI), включая клиентскую логику, политику идентификации и доступа. В этой статье в качестве конкретного примера используется браузерное приложение, поэтому оно также включает интерфейс браузера.
• Оркестратора, который координирует сессию, жизненный цикл контейнера и ретранслятор предварительного просмотра. Оркестратор реализован с помощью Durable Object.

Систему можно запускать локально во время разработки: в этом случае контейнер является просто локальным экземпляром Docker, а Durable Object не используется. В таком режиме есть только один пользователь, управляющему приложению не требуется авторизация для локального доступа, а предварительный просмотр видео может подключаться напрямую к WebSocket на localhost.

Когда эти компоненты развертываются в Cloudflare, авторизованный пользователь или агент может обратиться к воркеру, чтобы начать новую сессию. Это при необходимости запускает новый контейнер Streamline, автоматически управляет его жизненным циклом, предоставляет API для выполнения ряда операций по манипуляции видео и направляет входные и выходные данные обратно в Cloudflare Stream.

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

Жизненный цикл контейнера и управление сессиями

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

Контейнер Cloudflare автоматически переходит в спящий режим, если не получал входящих запросов в течение определенного интервала. Однако в нашем случае, как только конвейер запущен, он должен продолжать работу, даже если управляющее приложение отключилось и запросов нет. Мы можем реализовать это поведение, переопределив обратный вызов onActivityExpired() в контейнере. Если время истечения не наступило, мы продлеваем активность, в противном случае — уничтожаем контейнер.

API

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

Мы реализовали это, экспортировав два пакета из Streamline:

• @cloudflare/streamline/client: определяет высокоуровневый API на основе сессий.
• @cloudflare/streamline/: предоставляет базовый класс Durable Object, связанный с контейнером. Он маршрутизирует запросы API, реализует описанный ниже сервер ретрансляции предварительного просмотра и предоставляет хуки для безопасности и политики доступа.

При удаленном развертывании ожидается, что управляющий воркер импортирует @streamline/cloudflare и определит конкретный подкласс Durable Object, предоставляемый контейнером, который можно использовать для логики и хранения данных, специфичных для приложения.

В локальном режиме, где нет Durable Object, фронтенд определяет тонкий слой адаптера, который поддерживает API на основе сессий, но подключается напрямую к локальному экземпляру Docker без контроля доступа и т. д.

В примере ниже показано, как управляющее приложение может использовать API для доступа к Streamline, подготовки сессии и запуска конвейера обработки видео.

config — это JSON-объект, который определяет конвейер обработки, который должен быть выполнен (описан подробнее в следующих разделах).

В таблице ниже представлен полный список всех вызовов API.

Метод клиента

Функция

createStreamline()

Создает новый экземпляр Streamline.

streamline.sessions.create()

Создает новую сессию обработки.

streamline.sessions.resume(id)

Повторно подключается к существующей сессии.

session.start(config)

Запускает новый конвейер обработки.

session.ingest(chunk)

Отправляет фрагмент видеоданных в режиме «веб-камеры».

session.annotation(png)

Обновляет прозрачный слой аннотаций.

session.metrics()

Получает метрики текущего сеанса.

session.stop()

Останавливает обработку в текущем сеансе.

Определение и запуск конвейера обработки видео

Метод session.start() создает и запускает конвейер обработки. Он принимает единственный аргумент — объект конфигурации JSON, определяющий выполняемую обработку:

• Входные данные
• Операции
• Выходные данные

В приведенном ниже примере запускается конвейер, который принимает трансляцию RTMP (протокол передачи сообщений в реальном времени) в качестве входных данных (например, поток от Stream Live input, получающий входящую прямую трансляцию), накладывает изображение с прозрачностью и отправляет результат в пункт назначения RTMP (например, на другой Stream Live input для записи или трансляции). Это позволяет приложению Worker создавать модифицированную версию прямой трансляции в режиме реального времени.

Входные данные видео по запросу через HLS

Streamline также может принимать потоковое видео через HLS (HTTP live streaming), например, видео, размещенное на Cloudflare Stream. В приведенном ниже примере показано, как приложение Worker может запустить конвейер, который принимает видео из Stream, считывает встроенные субтитры и отображает их в виде текста поверх видео, а затем отправляет результат через RTMP, например, на Stream Live Input для трансляции или записи модифицированной версии.

Отправка видео в Streamline

Часто бывает полезно быстро просмотреть конвейер обработки, отправляя видеоданные напрямую в Streamline, например, с веб-камеры. Агент или приложение для встроенного устройства также может использовать эту возможность, например, для отправки кадров с заводских камер для анализа ИИ или для объединения нескольких потоков с камер в составное изображение.

В приведенном ниже примере создается конвейер, который ожидает входные данные от приложения Worker и создает выходной видеопоток для предварительного просмотра, доступный через WebSocket (подробнее о видео для предварительного просмотра через WebSocket мы поговорим ниже). Он применяет два фильтра и «аннотацию» — наложение, заданное в виде PNG-изображения, которое можно обновлять во время обработки, например, для реализации анимированной графики.

Приведенный выше фрагмент кода только запускает конвейер. Управляющий Worker еще не отправляет никаких медиаданных в Streamline. Мы обсудим функцию openViewer() ниже.

Приложение Worker отправляет видеоданные в Streamline с помощью вызова session.ingest(). В приведенном ниже примере показано, как приложение веб-браузера может получать фрагменты с веб-камеры и пересылать их в Streamline.

Анимированное наложение

Слой аннотаций можно обновить с помощью вызова session.annotation(). В приведенном ниже примере показано, как приложение Worker может сделать снимок холста и отправить его в Streamline. Это можно делать в цикле анимации, хотя частота обновления на практике может быть ограничена размером PNG-изображений наложения, доступной пропускной способностью и вычислительной мощностью.

Получение видео для предварительного просмотра из Streamline

Streamline также может создавать выходной видеопоток для предварительного просмотра, если указать output: { mode: 'websocket' }.

Streamline использует WebSockets для доставки видео предварительного просмотра с низкой задержкой обратно в управляющее приложение: контейнер публикует фрагменты fMP4 в Durable Object, который пересылает их на ретранслятор вывода, доступный через WebSocket по URL-адресу /relay/view относительно источника приложения. Приложение должно подключить WebSocket к этому URL-адресу, после чего оно будет получать видеоданные по мере их поступления из Streamline. В приведенном ниже фрагменте кода показано, как приложение веб-браузера может отображать поток видео предварительного просмотра.

Производственный плеер MediaSource должен ставить фрагменты в очередь, пока SourceBuffer.updating имеет значение true. При локальной разработке браузер или другое управляющее приложение просто открывает WebSocket-соединение напрямую с локальным контейнером.

В настоящее время поддерживаемые операции

В объекте конфигурации, передаваемом в session.start() в примерах выше, pipeline представляет собой массив операций из набора, поддерживаемого базовым медиадвижком. Порядок операций в настоящее время жестко задан движком; порядок, указанный в массиве, не имеет значения. Список поддерживаемых в настоящее время операций и порядок их применения приведены ниже.

Имя операции

Функция

filter

Применяет операции фильтрации, например, размытие, насыщенность.

overlay

Накладывает изображение, на которое ссылается URL-адрес, или двоичный PNG, указанный отдельно в вызове annotation().

subtitle

Вжигает субтитры.

encode

Задает параметры кодирования вывода.

Безопасность

Поскольку это Cloudflare, важно, чтобы безопасность была частью дизайна, а не дополнением в конце. Мы должны гарантировать, что только авторизованные пользователи могут создавать новый сеанс или брать под контроль существующий, и что сеансы изолированы друг от друга. Мы должны относиться к ключам ввода/вывода Stream RTMPS как к секретам, которые не должны быть раскрыты управляющему приложению. Мы должны гарантировать, что использование ресурсов ограничено.

Развертывание владельца остается приватным с использованием интеграции Access в Workers. Настроенный владелец и другие разрешенные пользователи могут редактировать одни и те же общие профили и начинать сеанс, пока синглтон простаивает. Worker проверяет сеанс Access перед принятием запросов на управление и привязывает активный сеанс к проверенному принципалу. Одновременно может выполняться только один сеанс, и другой принципал не может остановить или заменить активный сеанс.

Ключи Stream Live Input хранятся в секретах Worker или как общие переопределения «только для записи» в хранилище Durable Object. Они никогда не возвращаются API настроек и не помещаются в хранилище браузера. Управляющее приложение указывает входные и выходные данные RTMPS, ссылаясь на именованный профиль. Worker разрешает профиль перед обращением к контейнеру.

Поток видео предварительного просмотра имеет две учетные данные с разными целями. Сервисный токен Cloudflare Access аутентифицирует рабочую нагрузку контейнера для конечной точки издателя. Случайная возможность для каждого сеанса разрешает публикацию только для текущего активного ретранслятора. Сервисный токен внедряется исходящим Worker контейнера и никогда не попадает в память контейнера. Первоначальное развертывание использует временный обход Access, специфичный для пути, в то время как возможность для каждого сеанса остается принудительной; после развертывания и успешного дымового тестирования развертывание заменяет обход на Service Auth.

Развертывание владельца намеренно является приватным и маршрутизируется как синглтон. Это не модель безопасности для публичного многопользовательского сервиса.

Песочница и открытый исходный код

Мы хотим, чтобы вы попробовали Streamline и начали создавать! Поэтому вместе с этой публикацией мы выпускаем систему как открытый исходный код и развертываем публичную песочницу.

Контейнер Streamline можно запустить локально или развернуть в своей учетной записи. Он экспортирует API Worker для использования вашим управляющим приложением.

Также существует пример приложения Worker с веб-интерфейсом на Astro, который демонстрирует функциональность Streamline на нескольких распространенных примерах использования, включая наложения, декодирование субтитров, фильтры и режим «картинка в картинке». Предусмотрена функциональность зондирования, которая предоставляет метрики производительности и системную трассировку; она может быть полезна для отладки системы при разработке новых функций. Пример приложения можно запустить на локальном сервере Astro или настроить для развертывания за Cloudflare Access, чтобы вы могли контролировать, кто имеет доступ к вашему экземпляру Streamline.

Оба репозитория доступны с открытым исходным кодом на GitHub компании Cloudflare:

• https://github.com/cloudflare/streamline — Медиа-движок и экспорт пакетов для приложений.
• https://github.com/cloudflare/streamline-demo — Пример приложения Worker, фронтенд на Astro, профили развертывания и инструменты Access.

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

Вы можете попробовать общедоступную демо-версию по адресу:

Что дальше

Streamline демонстрирует один из способов объединения существующих управляемых сервисов, таких как Stream, с низкоуровневыми примитивами для создания высоконастраиваемых медиа-конвейеров. В этой итерации Streamline использует Container CPU для обработки медиа, что создает «узкое место» при более высоком качестве или частоте кадров.

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

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