Практическое руководство: добавление (расширенное)

Введение

Надежное добавление ресурсов — это основная функция каждой интеграции C2C. Это руководство содержит передовые методы и лучшие практики для создания надежной, отказоустойчивой и эффективной системы добавления, которая работает хорошо даже в сложных условиях.

Требования

Если вы еще не сделали этого, просмотрите руководство Реализация C2C: настройка перед продолжением. Вам понадобится access_token, полученный в процессе аутентификации и авторизации. Мы продолжим использовать тот же тестовый ресурс из руководства по базовому добавлению.

Расширенные параметры ресурсов

При создании ресурсов в Frame.io можно использовать несколько расширенных параметров для настройки поведения добавления. Параметр offset особенно важен для правильной интеграции.

Смещение — обработка приостановленных устройств

Предоставление точного значения offset крайне важно. Этот параметр указывает, когда создан медиафайл, и гарантирует, что ваше устройство не добавляет контент, которым не следует делиться. Когда устройство приостановлено в Frame.io, пользователь указывает, что медиа, созданные во время паузы, не должны добавляться. Дополнительную информацию см. в нашем руководстве по функции приостановки.

Дополнительные преимущества параметра offset

Параметр offset предоставляет еще одно значительное преимущество для систематизации медиафайлов в Frame.io. При добавлении контента, снятого в более раннюю дату, например когда пользователь выбирает фото, сделанное на прошлой неделе во время воспроизведения, параметр offset обеспечивает появление этого медиафайла в папках, соответствующих исходной дате съемки, а не текущей дате добавления. Эта хронологическая систематизация поддерживает логическую временную шкалу в структуре проекта Frame.io. Без параметра offset предыдущие медиафайлы неправильно группируются с сегодняшним контентом, что может вызвать путаницу у редакторов и других участников совместной работы. Можно предоставить пользователям выбор в этом вопросе с помощью вашего интерфейса. Если пользователи предпочитают систематизировать все добавленные материалы по текущей дате независимо от даты съемки, можно просто опустить параметр offset, поскольку по умолчанию он равен 0, если не указан.

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

Чтобы увидеть эту функцию в действии, приостановите ваше устройство в меню с тремя точками на вкладке «Подключения C2C».

Теперь попробуйте добавить ресурс:

$
{
>
curl -X POST https://api.frame.io/v2/devices/assets \
>
    --header 'Authorization: Bearer [access_token]' \
>
    --header 'Content-Type: application/json' \
>
    --header 'x-client-version: 2.0.0' \
>
    --data-binary @- <<'__JSON__' 
>
        {
>
            "name": "C2C_TEST_CLIP.mp4", 
>
            "filetype": "video/mp4", 
>
            "filesize": 21136250,
>
            "offset": 0
>
        }
>
__JSON__
>
} | python -m json.tool

Отобразится ошибка:

1
{
2
    "code": 409,
3
    "errors": [
4
        {
5
            "code": 409,
6
            "detail": "The channel you're uploading from is currently paused.",
7
            "status": 409,
8
            "title": "Channel Paused"
9
        }
10
    ],
11
    "message": "Channel Paused"
12
}

Если вы отмените приостановку устройства и повторите попытку с тем же запросом, ресурс будет создан.

Однако, если ресурс создан во время окна приостановки, необходимо задать offset, чтобы указать, когда он фактически создан:

$
{
>
curl -X POST https://api.frame.io/v2/devices/assets \
>
    --header 'Authorization: Bearer [access_token]' \
>
    --header 'Content-Type: application/json' \
>
    --header 'x-client-version: 2.0.0' \
>
    --data-binary @- <<'__JSON__' 
>
        {
>
            "name": "C2C_TEST_CLIP.mp4", 
>
            "filetype": "video/mp4", 
>
            "filesize": 21136250,
>
            "offset": 60
>
        }
>
__JSON__
>
} | python -m json.tool

Это указывает Frame.io на то, что ресурс создан 60 секунд назад (во время приостановки), что закономерно вызывает ошибку Канал приостановлен. Точные значения offset крайне важны для предотвращения добавления конфиденциального контента против воли пользователя, включая защищенную интеллектуальную собственность, конфиденциальные отснятые материалы или другие материалы с ограниченным доступом.

Добавление в определенный канал

Если у вашего устройства есть несколько каналов, можно указать, какой из них использовать:

$
{
>
curl -X POST https://api.frame.io/v2/devices/assets \
>
    --header 'Authorization: Bearer [access_token]' \
>
    --header 'Content-Type: application/json' \
>
    --header 'x-client-version: 2.0.0' \
>
    --data-binary @- <<'__JSON__' 
>
        {
>
            "name": "C2C_TEST_CLIP.mp4", 
>
            "filetype": "video/mp4", 
>
            "filesize": 21136250,
>
            "offset": -10,
>
            "channel": 2
>
        }
>
__JSON__
>
} | python -m json.tool

Если не указано, каналом по умолчанию является 0. Для большинства интеграций не потребуется изменять это значение.

Запрос пользовательского количества фрагментов

По умолчанию серверная часть Frame.io делит файлы на фрагменты размером примерно 25 МБ. Для сетей с высокой перегрузкой стоит предпочесть меньшие фрагменты. Можно запросить определенное количество фрагментов с помощью параметра parts:

$
{
>
curl -X POST https://api.frame.io/v2/devices/assets \
>
    --header 'Authorization: Bearer [access_token]' \
>
    --header 'Content-Type: application/json' \
>
    --header 'x-client-version: 2.0.0' \
>
    --data-binary @- <<'__JSON__'
>
        {
>
            "name": "C2C_TEST_CLIP.mp4",
>
            "filetype": "video/mp4",
>
            "filesize": 21136250,
>
            "offset": 0,
>
            "parts": 4
>
        }
>
__JSON__
>
} | python -m json.tool

Ответ будет включать четыре URL-адреса добавления:

1
{
2
    ...
3
    "upload_urls": [
4
        "https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-01-path]",
5
        "https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-02-path]",
6
        "https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-03-path]",
7
        "https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-04-path]"
8
    ],
9
    ...
10
}

Размер фрагмента составит:

1
math.ceiling(float(21136250) / float(4))
2
# 5284063 bytes

Последний фрагмент составит 5 284 061 байт (рассчитывается как 21 136 250-5 284 063*3). При запросе пользовательского количества фрагментов учитывайте ограничения многочастного добавления AWS S3:

• Каждый фрагмент должен быть не менее 5 МиБ (5 242 880 байт), за исключением финального фрагмента
• Максимальное количество фрагментов — 10 000

Если запрос нарушает эти ограничения, отображается 500: INTERNAL SERVER ERROR:

{
    "code": 500,
    "errors": [
        {
            "code": 500,
            "detail": "There was a problem with your request",
            "status": 500,
            "title": "Something went wrong"
        }
    ],
    "message": "Something went wrong"
}

Всегда проверяйте, соответствует ли пользовательское количество фрагментов требованиям S3.

Эффективное добавление

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

Повторное использование/пул подключений TCP

Установка зашифрованных подключений требует значительных накладных расходов на согласование. Для эффективной работы повторно используйте подключения TCP при выполнении нескольких запросов. Большинство библиотек HTTP предоставляют абстракцию клиента или сеанса, которая поддерживает постоянные подключения.

Процесс согласования для нового подключения HTTPS включает криптографические рукопожатия и проверку сертификатов. Повторно используя подключения, вы выполняете эту ресурсоемкую процедуру только один раз, а не для каждого запроса.

Чтобы показать повторное использование подключений с curl, сначала создайте новый ресурс в Frame.io, как описано в базовом руководстве по добавлению.

Затем разделите файл на отдельные фрагменты для тестирования:

$
head -c 10568125 ~/Downloads/C2C_TEST_CLIP.mp4 > "C2C_TEST_CLIP-Chunk01"
$
tail -c 10568125 ~/Downloads/C2C_TEST_CLIP.mp4 > "C2C_TEST_CLIP-Chunk02"

Теперь добавьте оба фрагмента через одно подключение TCP, используя параметр --next curl:

$
curl -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-1-path] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk01 \
>
--next -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-2-path] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk02

Сравните это с отдельными подключениями:

$
curl -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-1-path] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk01 \
>
&& curl -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-2-path] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk02

При тестировании повторное использование подключений обычно улучшает производительность на 15–20% для последовательных операций добавления.

Параллельные операции добавления

Для еще большей пропускной способности добавляйте несколько фрагментов одновременно:

$
curl -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-1-path] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk01 \
>
& \
>
curl -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[part-2-path]\
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk02 \
>
&

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

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

Комбинирование двух подходов

Для максимальной эффективности объединяйте пул подключений с параллельными операциями добавления. Создавайте несколько процессов, каждый из которых использует пул подключений для своей собственной последовательности загрузок:

$
curl -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[asset01-chunk01] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk01 \
>
--next -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[asset01-chunk02] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk02 \
>
& \
>
curl -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[asset02-chunk01] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk01 \
>
--next -X PUT https://frameio-uploads-production.s3-accelerate.amazonaws.com/parts/[asset02-chunk02] \
>
        --include \
>
        --header 'content-type: video/mp4' \
>
        --header 'x-amz-acl: private' \
>
        --data-binary @C2C_TEST_CLIP-Chunk02 \
>
&

Отслеживание хода выполнения добавления

Ваша интеграция должна предоставлять пользователям базовую индикацию хода выполнения. Детализация на уровне фрагментов допустима. Для добавления трех фрагментов ход выполнения по мере завершения операции с каждым фрагментом может увеличиваться следующим образом: 0% → 33% → 66% → 100%.

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

Надежные операции добавления

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

Создание средства добавления промышленного уровня требует учета дополнительных факторов, помимо простого отслеживания ошибок в отдельных запросах.

Создание очереди операций добавлений

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

Рассмотрим архитектуру с двумя очередями:

1. Очередь мультимедиа для регистрации локальных файлов в Frame.io.
2. Очередь фрагментов для добавления отдельных фрагментов файлов.

Упрощенная реализация:

1
# Where we are going to queue new files.
2
FILE_QUEUE = Queue()
3
4
# Where we are going to queue new chunks.
5
CHUNK_QUEUE = Queue()
6
7
# The http session that will handle TCP 
8
# connection pooling for us.
9
HTTP_SESSION = http.Session()
10
11
def take_picture():
12
    """Snaps a picture for the user."""
13
14
    image = MY_DEVICE.capture()
15
    file_path = MY_DEVICE.write_image(image)
16
    FILE_QUEUE.add(file_path)
17
18
def task_register_assets():
19
    """
20
    Pulls snapped pictures from the FILE_QUEUE, registers
21
    with Frame.io, and adds the chunks to the CHUNK_QUEUE.
22
    """
23
    while True:
24
        # Get the latest file added to the queue and register 
25
        # a C2C Asset for it.
26
        new_file = FILE_QUEUE.get()
27
        asset = c2c.crete_asset_for_file(HTTP_SESSION, new_file)
28
29
        # Calculate the size for each chunk
30
        chunk_size = c2c.calculate_chunk_size(asset, new_file)
31
32
        # Create a message for each chunk with it's parameters 
33
        # and add it to the 
34
        # queue.
35
        chunk_start = 0
36
        for chunk_url in asset.upload_urls:
37
            message = {
38
                "file_path": new_file,
39
                "chunk_url": chunk_url,
40
                "chunk_start": chunk_start,
41
                "chunk_size": chunk_size,
42
            }
43
44
            # Put the message in the queue and 
45
            CHUNK_QUEUE.put(message)
46
            chunk_start += chunk_size
47
48
def task_upload_chunk():
49
    """Takes a chunks and uploads them."""
50
51
    while True:
52
        info = CHUNK_QUEUE.get()
53
        c2c.upload_chunk(HTTP_SESSION, info)
54
55
def launch_upload_tasks():
56
    """Lauches our Frame.io upload tasks."""
57
    # Create a list to hold all of our tasks.
58
    tasks = list()
59
60
    # Create one task for registering assets.
61
    asset_task = run_task_in_thread(task_register_assets)
62
    tasks.append(asset_task)
63
64
    # Create 2 tasks per CPU core for uploading chunks.
65
    for _ in range(0, GET_CPU_COUNT() * 2):
66
        chunk_task = run_task_in_thread(task_upload_chunk)
67
        tasks.append(chunk_task)
68
69
    # Run these tasks until shutdown
70
    run_forever(tasks)

Постоянная очередь с сохранением при перегрузках

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

Это требует сохранения состояния очереди в хранилище между циклами питания. Встроенная база данных, такая как SQLite, обеспечивает отличную основу для этой функциональности.

Реализация постоянной очереди должна поддерживать эти ключевые операции:

• Добавление новых созданных файлов в очередь добавления.
• Отслеживание успешного создания ресурсов в Frame.io.
• Запись неудачных попыток создания ресурсов из-за ошибок.
• Хранение информации о фрагментах файлов для задач добавления.
• Получение следующего фрагмента для добавления.
• Отметка фрагментов как успешно добавленных.
• Регистрация неудачных попыток добавления фрагментов.
• Предоставление информации о состоянии файла для отображения пользователю.

Адаптация нашего предыдущего примера для использования системы постоянного хранения данных:

1
# Our persistence layer for queuing uploads, potentially using SQLite 
2
# or another embedded database
3
C2C_UPLOAD_STORE = NewC2CUploadStore()
4
5
# HTTP session for connection pooling
6
HTTP_SESSION = http.Session()
7
8
def take_picture():
9
    """Captures an image and adds it to the upload queue."""
10
    image = MY_DEVICE.capture()
11
    file_path = MY_DEVICE.write_image(image)
12
13
    # Register the file with our persistent store
14
    C2C_UPLOAD_STORE.add_file(file_path)
15
16
def task_register_assets():
17
    """
18
    Processes files from persistent storage and registers
19
    them with Frame.io for upload.
20
    """
21
    while True:
22
        # Get the next available file from our store
23
        file_record = C2C_UPLOAD_STORE.get_file()
24
25
        try:
26
            # Register the asset with Frame.io
27
            asset = c2c.create_asset_for_file(HTTP_SESSION, file_record)
28
            chunk_size = c2c.calculate_chunk_size(asset, file_record)
29
30
            # Create entries for each chunk in our persistent store
31
            chunk_start = 0
32
            for chunk_url in asset.upload_urls:
33
                message = {
34
                    "file_path": file_record,
35
                    "chunk_url": chunk_url,
36
                    "chunk_start": chunk_start,
37
                    "chunk_size": chunk_size,
38
                }
39
                
40
                C2C_UPLOAD_STORE.new_chunk(message)
41
                chunk_start += chunk_size
42
                
43
        except BaseException as error:
44
            # Record the error in our persistent store
45
            C2C_UPLOAD_STORE.file_asset_create_error(file_record, error)
46
        else:
47
            # Mark the asset as successfully created
48
            C2C_UPLOAD_STORE.file_asset_created(file_record)
49
50
def task_upload_chunk():
51
    """Uploads individual file chunks from the persistent queue."""
52
    while True:
53
        # Get the next chunk, marking it as "in progress" to prevent 
54
        # other tasks from processing it simultaneously
55
        chunk_record = C2C_UPLOAD_STORE.get_chunk()
56
57
        try:
58
            c2c.upload_chunk(HTTP_SESSION, chunk_record)
59
        except BaseException as error:
60
            # Record the error for potential retry
61
            C2C_UPLOAD_STORE.chunk_error(chunk_record, error)
62
        else:
63
            # Mark successful completion
64
            C2C_UPLOAD_STORE.chunk_success(chunk_record)
65
66
def launch_upload_tasks():
67
    """Launches Frame.io upload processing tasks."""
68
    tasks = []
69
70
    # Asset registration task
71
    asset_task = run_task_in_thread(task_register_assets)
72
    tasks.append(asset_task)
73
74
    # Multiple parallel chunk upload tasks
75
    worker_count = GET_CPU_COUNT() * 2
76
    for _ in range(worker_count):
77
        chunk_task = run_task_in_thread(task_upload_chunk)
78
        tasks.append(chunk_task)
79
80
    # Run indefinitely
81
    run_forever(tasks)

С таким подходом постоянного хранения интеграция становится устойчивой к прерываниям питания. Когда устройство перезапускается, оно просто продолжает обработку с последнего сохраненного состояния. Эта архитектура также обеспечивает основу для реализации более продвинутых функций, таких как отслеживание ошибок и обнаружение зависших операций добавления.

Отслеживание ошибок добавление

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

1. Понижать приоритет проблемных операций добавления, чтобы они не блокировали всю очередь.
2. Предоставлять пользователям точную информацию о состоянии.
3. Обеспечивать административное вмешательство при постоянных проблемах.

При возникновении фатальной ошибки отметьте элемент, чтобы предотвратить ненужные повторные попытки.

Управление зависшими операциями добавления

Реализуйте защиту от зависших на неопределенное время операций добавления. Установите максимальную продолжительность (например, 30 минут), по истечении которой задача добавления фрагмента должна быть завершена и перезапущена. Это предотвращает сценарии, когда все процессы добавления блокируются не отвечающими операциями.

Восстановление после скрытых сбоев

Сбои системы, потеря питания или завершение процесса могут помешать нормальному составлению отчетов об ошибках. При получении элементов из очереди записывайте время извлечения. Если элемент остается в состоянии «в обработке» дольше разумного порога (например, 30 минут) без сообщения о выполнении или сбое, автоматически возвращайте его в доступный пул для обработки другим процессом.

Минимализация вредоносных операций добавления

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

• После множественных сбоев понизьте приоритет элемента, чтобы можно было пропустить новый контент.
• Отслеживайте как явные ошибки, так и количество попыток обработки.
• Следуйте передовым практикам подключения и авторизации, чтобы различать временные проблемы среды и внутренние проблемы файлов.
• Реализуйте ограничения повторных попыток с эскалацией (например, 10 повторных попыток для каждой операции в рамках 3 попыток выполнения задачи, итого 30 попыток).
• Предоставьте пользовательский интерфейс для ручного сброса проблемных операций добавления после устранения проблем среды.

Вредоносные операции добавления могут возникать в следующих случаях:

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

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

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

Очистка очереди

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

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

Дальнейшие шаги

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