Distributed Media Processing: Dynamic Node Scaling and Coordination Practices

Introduction: Transcoding Pressure Under Sudden Traffic

Video platforms and content delivery networks often experience significant fluctuations in processing tasks over time. During holidays, major sporting events, or trending events, user upload volumes can surge to several times the normal level within minutes. Fixed-size transcoding clusters tend to suffer from task backlogs during these periods, resulting in long playback delays for users after uploading.

There are two primary strategies to handle such traffic spikes. The first is vertical scaling, which involves upgrading the CPU, memory, and disk configurations of individual machines. This approach is costly and has a physical ceiling on single-machine specifications. The second is horizontal scaling, which dynamically adds compute nodes during peak periods and reclaims them during off-peak hours. The latter has become the industry standard. This article focuses on horizontal scaling, detailing how new nodes quickly join a cluster, securely claim tasks, and the boundary conditions to maintain during the scaling process.

Characteristics of Sudden Traffic and Scaling Trigger Conditions

Before deciding to scale, it is essential to describe the traffic using quantifiable metrics. Common metrics fall into three categories:

1. Queue Backlog Depth: The number of tasks waiting to be transcoded. It directly reflects whether the current processing capacity can keep up with the input rate.
2. Task Arrival Rate: The number of new transcoding requests per unit of time. It indicates the trend of traffic changes.
3. Average Task Duration and P99 Latency: These metrics determine how long it takes for a new node to become effective and how long task leases should be set.

Scaling trigger conditions should be derived from a combination of these metrics rather than relying on a single number. For instance, a rule could be established to trigger scaling when the queue backlog exceeds twice the volume of tasks that all online nodes can process in 5 minutes, and the average CPU utilization exceeds 70%. These thresholds (2x and 70%) are merely examples; actual values must be determined based on stress testing results.

It is important to note that scaling incurs a startup delay. Container provisioning, image downloading, and node registration typically take tens of seconds to several minutes. If a traffic spike lasts only a few dozen seconds, the scaled capacity might only become available after the traffic has subsided. In such cases, coverage should be ensured by reserving a minimum number of nodes or scaling out in advance, rather than relying on post-hoc scaling.

Elastic Node Onboarding Process

A new node joining a cluster generally requires three steps: registration, configuration delivery, and task claiming.

1. Service Registration and Discovery

Upon startup, a new node first reports its identifier, available compute resources, and supported encoding formats to a registry center. Common registry centers include etcd, Consul, and ZooKeeper. When selecting one, several dimensions should be compared: consistency model, operational complexity, compatibility with the existing tech stack, and behavior during network partitions. The registry center itself must be highly available; deploying at least 3 nodes is recommended so a majority can still be maintained if one node fails.

Nodes should send heartbeat signals periodically. A recommended configuration is a 10-second heartbeat interval; if no heartbeat is received for 3 consecutive intervals, the node is deemed disconnected and removed. Following removal, the node’s unconfirmed tasks are re-enqueued due to lease expiration. The heartbeat interval requires a trade-off: if too long, failed nodes occupy tasks for too long; if too short, network jitter may cause false positives.

2. Configuration Delivery and Credential Management

After successful registration, the control plane delivers transcoding templates, encoding parameters, and storage access credentials. Configurations should be versioned, and nodes should record the configuration version used during task processing to facilitate post-mortem troubleshooting.

Credential management requires extreme caution. The recommended practice is: credentials should be issued by a secret service, with an expiration time matching the estimated duration of a single task (e.g., no more than 1 hour). The scope of permissions should be strictly limited to the specific bucket or prefix required by the task, granting only read access to the source file and write access to the target path, without granting delete or bucket-level management permissions. Credentials must never be hardcoded in code or images; they should be obtained at runtime via environment variables or a secret service.

3. Task Pulling and Leasing

Dynamic nodes are well-suited for a pull mode, where nodes actively claim tasks based on their current load. In a push mode, the scheduler needs to accurately track the real-time status of each node, which is difficult for short-lived nodes as their states change rapidly. Pull mode allows nodes to decide for themselves when to claim tasks, resulting in simpler control logic and more even load distribution.

When claiming a task, the queue delivers it in the form of a lease with a visibility timeout. The node must confirm task completion before the timeout expires. If the node crashes before the timeout, the task is re-enqueued and handled by another node. The visibility timeout should be greater than the P99 latency of a single task, for example, set to twice the P99 latency. For exceptionally long-running tasks, nodes must renew the lease periodically during processing; otherwise, the task might be duplicated and claimed by another node.

Below is a simplified pseudo-code implementation of a node, illustrating the basic structure of registration, heartbeat, leasing, acknowledgment, and retry. Functions like get_cpu_utilization and object_exists_and_valid are provided by the monitoring and storage SDKs in use and serve as placeholders here.

import logging
import tempfile
import time
import uuid
import requests
logger = logging.getLogger("transcode_worker")
HEARTBEAT_INTERVAL = 10   # Seconds; removed if no heartbeat for 3 consecutive intervals
POLL_INTERVAL = 5         # Seconds; wait time when load is too high or queue is empty
MAX_LOAD = 0.8            # Maximum CPU utilization
MAX_RETRY = 3             # Maximum retries per request
class TranscodeWorker:
    def __init__(self, capacity, registry_url, queue_url, token):
        self.worker_id = str(uuid.uuid4())
        self.capacity = capacity
        self.registry_url = registry_url
        self.queue_url = queue_url
        self.session = requests.Session()
        # Credentials read from env vars or secret service, not hardcoded
        self.session.headers["Authorization"] = f"Bearer {token}"
    def _request(self, method, url, **kwargs):
        last_error = None
        for attempt in range(1, MAX_RETRY + 1):
            try:
                resp = self.session.request(method, url, timeout=5, **kwargs)
                resp.raise_for_status()
                return resp
            except requests.RequestException as exc:
                last_error = exc
                logger.warning("attempt=%d url=%s error=%s", attempt, url, exc)
                time.sleep(2 ** attempt)  # Exponential backoff
        raise last_error
    def register(self):
        # Use PUT with worker_id as resource key; identical results for duplicate registrations, ensuring idempotency
        payload = {"capacity": self.capacity, "status": "IDLE"}
        self._request("PUT", f"{self.registry_url}/{self.worker_id}", json=payload)
        logger.info("registered worker_id=%s", self.worker_id)
    def heartbeat(self):
        self._request("POST", f"{self.registry_url}/{self.worker_id}/heartbeat")
    def lease_task(self):
        # Server returns task and sets visibility timeout; unconfirmed tasks re-enqueue after timeout
        resp = self._request("POST", f"{self.queue_url}/lease",
                             json={"worker_id": self.worker_id})
        return resp.json() if resp.content else None
    def ack(self, task_id):
        self._request("POST", f"{self.queue_url}/{task_id}/ack")
    def execute(self, task):
        # Idempotency check: if target object exists and checksum matches, ack directly to avoid duplicate transcoding
        if object_exists_and_valid(task.target_uri, task.expected_checksum):
            self.ack(task.id)
            return
        workdir = tempfile.mkdtemp()
        try:
            src = download_from_object_storage(task.source_uri, workdir)
            out = run_ffmpeg_pipeline(src, task.params, workdir)
            upload_to_object_storage(out, task.target_uri)  # Ack only after successful upload
        finally:
            remove_tree(workdir)  # Clean local cache regardless of success or failure
        self.ack(task.id)
    def run(self):
        self.register()
        last_beat = 0.0
        while True:
            if time.time() - last_beat >= HEARTBEAT_INTERVAL:
                self.heartbeat()
                last_beat = time.time()
            if get_cpu_utilization() >= MAX_LOAD:
                time.sleep(POLL_INTERVAL)
                continue
            task = self.lease_task()
            if task is None:
                time.sleep(POLL_INTERVAL)
                continue
            try:
                self.execute(task)
            except Exception:
                # Do not ack the task; it will be redelivered after lease expiration
                logger.exception("task failed task_id=%s", task.id)

There are several points to note regarding this code. First, when a task fails, it is not acknowledged; instead, the lease expires and the task is automatically redelivered. Second, if a task fails repeatedly, the queue should transfer it to a dead-letter queue. For example, if the same task fails delivery more than 3 times, it should be moved to a dead-letter queue and trigger an alert, preventing corrupted tasks from looping indefinitely and occupying nodes. Third, the heartbeat is sent at the top of the loop, meaning it may be delayed if a single task takes too long. In a production environment, the heartbeat should be placed in an independent thread, or the lease should be renewed periodically during long tasks.

Computation and Storage Separation and Data Security

The premise of dynamic scaling is that nodes can be destroyed at any time. Therefore, raw videos, intermediate segments, and final outputs should all reside in object storage, while nodes only keep temporary files locally during processing. Once processing is complete, the temporary directory must be cleaned up.

This architecture introduces certain risks that must be mitigated.

The first is data consistency. The read-write consistency of object storage depends on the specific product. After a node writes a result, if another service reads it immediately, it might encounter a stale version or an incomplete object. Solutions include verifying the ETag or checksum after writing is complete, and writing to a new object path rather than overwriting the existing object to ensure data integrity.