Function Compute:Build an Elastic, Highly Available Audio and Video Processing System Using Serverless Architecture

Prerequisites

Before you begin, make sure you have:

• Activated Function Compute. See Activate Function Compute.

• An OSS bucket. See Create a bucket in the OSS console.

Deploy the simple video processing system

Step 1: Create a service

1. Log on to the Function Compute console. In the left navigation pane, click Services & Functions.

2. In the top menu bar, select a region. On the Services page, click Create Service.

3. In the Create Service panel, enter a service name and description, then click OK. For this example, set the service role to AliyunFCDefaultRole and attach the AliyunOSSFullAccess permission policy. For full service configuration options, see Create a service.

Step 2: Create a function

1. On the Function Management page, click Create Function.

2. On the Create Function page, configure the following settings, then click Create. Video processing is time-intensive and file sizes vary, so the advanced configuration values above are recommended for this use case. For full function configuration options, see Create a function.

   Setting	Value
   Creation method	Create with built-in runtime
   Handler type	Event-driven handler
   Runtime	Python 3.9
   Code upload method	Use sample code
   vCPU count	4
   Memory	8 GB
   Temporary disk size	10 GB
   Timeout	7200 s

Step 3: Create an OSS trigger

1. On the function details page, click the Triggers tab. From the Version or alias drop-down list, select a version or alias. Click Create Trigger.

2. In the Create Trigger panel, configure the following settings, then click OK.

   Important

   If the object prefix starts with /, the OSS trigger will not fire.

   Setting	Action	Example
   Trigger type	Select Object Storage OSS.	Object Storage OSS
   Name	Enter a custom name.	oss-trigger
   Version or alias	Defaults to LATEST. For other versions or aliases, select one first from the function details page. See Manage versions and Manage aliases.	LATEST
   Bucket name	Select an existing OSS bucket.	testbucket
   Object prefix	Enter a prefix to match object names. Set both a prefix and suffix to avoid nested event loops and extra charges. For triggers in the same bucket with identical event types, prefixes and suffixes must not overlap. See OSS trigger rules.	source
   File Extension	Enter a suffix to match object names. Set both a prefix and suffix to avoid nested event loops and extra charges.	mp4
   Trigger events	Select one or more triggering events. See OSS event definitions.	oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload
   Role name	Select AliyunOSSEventNotificationRole. If this is your first trigger of this type, click OK in the dialog that appears to authorize it.	AliyunOSSEventNotificationRole

Step 4: Write function code

1. On the function details page, click the Function Code tab and write your code in the editor. The function below converts MP4 videos to FLV format and saves the output to the dest directory in the same OSS bucket.

   # -*- coding: utf-8 -*-
   import logging
   import oss2
   import os
   import json
   import subprocess
   import shutil
   
   logging.getLogger("oss2.api").setLevel(logging.ERROR)
   logging.getLogger("oss2.auth").setLevel(logging.ERROR)
   LOGGER = logging.getLogger()
   
   
   def get_fileNameExt(filename):
       (_, tempfilename) = os.path.split(filename)
       (shortname, extension) = os.path.splitext(tempfilename)
       return shortname, extension
   
   
   def handler(event, context):
       LOGGER.info(event)
       evt = json.loads(event)
       evt = evt["events"]
       oss_bucket_name = evt[0]["oss"]["bucket"]["name"]
       object_key = evt[0]["oss"]["object"]["key"]
   
       output_dir = "dest"
       dst_format = "flv"
       shortname, _ = get_fileNameExt(object_key)
       creds = context.credentials
       auth = oss2.StsAuth(creds.accessKeyId,
                           creds.accessKeySecret, creds.securityToken)
       oss_client = oss2.Bucket(auth, 'oss-%s-internal.aliyuncs.com' %
                                context.region, oss_bucket_name)
   
       exist = oss_client.object_exists(object_key)
       if not exist:
           raise Exception("object {} is not exist".format(object_key))
   
       input_path = oss_client.sign_url('GET', object_key, 6 * 3600)
       # m3u8 special handling
       rid = context.request_id
       if dst_format == "m3u8":
           return handle_m3u8(rid, oss_client, input_path, shortname, output_dir)
       else:
           return handle_common(rid, oss_client, input_path, shortname, output_dir, dst_format)
   
   
   def handle_m3u8(request_id, oss_client, input_path, shortname, output_dir):
       ts_dir = '/tmp/ts'
       if os.path.exists(ts_dir):
           shutil.rmtree(ts_dir)
       os.mkdir(ts_dir)
       transcoded_filepath = os.path.join('/tmp', shortname + '.ts')
       split_transcoded_filepath = os.path.join(
           ts_dir, shortname + '_%03d.ts')
       cmd1 = ['ffmpeg', '-y', '-i', input_path, '-c:v',
               'libx264', transcoded_filepath]
   
       cmd2 = ['ffmpeg', '-y', '-i', transcoded_filepath, '-c', 'copy', '-map',  '0',  '-f', 'segment',
               '-segment_list', os.path.join(ts_dir, 'playlist.m3u8'), '-segment_time', '10', split_transcoded_filepath]
   
       try:
           subprocess.run(
               cmd1, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
   
           subprocess.run(
               cmd2, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
   
           for filename in os.listdir(ts_dir):
               filepath = os.path.join(ts_dir, filename)
               filekey = os.path.join(output_dir, shortname, filename)
               oss_client.put_object_from_file(filekey, filepath)
               os.remove(filepath)
               print("Uploaded {} to {}".format(filepath, filekey))
   
       except subprocess.CalledProcessError as exc:
           # if transcode fail, trigger invoke dest-fail function
           raise Exception(request_id + " transcode failure, detail: " + str(exc))
   
       finally:
           if os.path.exists(ts_dir):
               shutil.rmtree(ts_dir)
   
           # remove ts file
           if os.path.exists(transcoded_filepath):
               os.remove(transcoded_filepath)
   
       return {}
   
   
   def handle_common(request_id, oss_client, input_path, shortname, output_dir, dst_format):
       transcoded_filepath = os.path.join('/tmp', shortname + '.' + dst_format)
       if os.path.exists(transcoded_filepath):
           os.remove(transcoded_filepath)
       cmd = ["ffmpeg", "-y", "-i", input_path, transcoded_filepath]
       try:
           subprocess.run(
               cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
   
           oss_client.put_object_from_file(
               os.path.join(output_dir, shortname + '.' + dst_format), transcoded_filepath)
       except subprocess.CalledProcessError as exc:
           # if transcode fail, trigger invoke dest-fail function
           raise Exception(request_id + " transcode failure, detail: " + str(exc))
       finally:
           if os.path.exists(transcoded_filepath):
               os.remove(transcoded_filepath)
       return {}

2. Click Deploy code.

Step 5: Test the function

Simulate OSS event parameters using function input to verify correctness. In production, the function triggers automatically when an OSS event occurs.

1. On the Function Code tab, click the icon to the right of Test Function and select Configure Test Parameters.

2. In the Configure test parameters panel, select Create new test event or Edit existing test event. Enter an event name and the following event content, then click OK.

   {
       "events": [
           {
               "eventName": "oss:ObjectCreated:CompleteMultipartUpload",
               "eventSource": "acs:oss",
               "eventTime": "2022-08-13T06:45:43.000Z",
               "eventVersion": "1.0",
               "oss": {
                   "bucket": {
                       "arn": "acs:oss:cn-hangzhou:123456789:testbucket",
                       "name": "testbucket",
                       "ownerIdentity": "164901546557****"
                   },
                   "object": {
                       "deltaSize": 122539,
                       "eTag": "688A7BF4F233DC9C88A80BF985AB****",
                       "key": "source/a.mp4",
                       "size": 122539
                   },
                   "ossSchemaVersion": "1.0",
                   "ruleId": "9adac8e253828f4f7c0466d941fa3db81161****"
               },
               "region": "cn-hangzhou",
               "requestParameters": {
                   "sourceIPAddress": "140.205.XX.XX"
               },
               "responseElements": {
                   "requestId": "58F9FF2D3DF792092E12044C"
               },
               "userIdentity": {
                   "principalId": "164901546557****"
               }
           }
       ]
   }

   For details about event parameters, see Configure function input parameters.

3. Click Test Function and view the result on the Function Code tab.

Prerequisites

Before you begin, make sure you have:

• Activated Function Compute. See Activate Function Compute.

• An OSS bucket. See Create a bucket in the OSS console.

• Activated Serverless Workflow. See Activate Serverless Workflow.

• Activated Apsara File Storage NAS. See Activate Apsara File Storage NAS.

• Activated Virtual Private Cloud (VPC). See Activate Virtual Private Cloud.

• Configured the following Resource Access Management (RAM) roles:

  • AliyunFCDefaultRole — Assign when creating the service. Attach AliyunOSSFullAccess, AliyunFnFFullAccess, and AliyunFCInvocationAccess to enable calling functions, managing workflows, and accessing OSS.

  • AliyunOSSEventNotificationRole — Used by OSS to send event notifications.

  • fnf-execution-default-role — Required to create and manage Serverless Workflow workflows. Attach AliyunFCInvocationAccess and AliyunFnFFullAccess.

• Installed and configured Serverless Devs. See Install Serverless Devs and dependencies and Configure Serverless Devs.

Deploy the video processing workflow system

This solution uses Serverless Devs to deploy multiple functions and a Serverless Workflow in one step.

Step 1: Initialize the application

Run the following command to initialize the project:

s init video-process-flow -d video-process-flow

Configure the following items when prompted. Adjust values as needed.

Step 2: Deploy the project

cd video-process-flow && s deploy -y

Successful deployment output looks like this:

[2023-08-31 13:22:21] [INFO] [S-CORE] - Project video-demo-flow successfully to execute

fc-video-demo-split:
  region:   cn-hangzhou
  service:
    name: video-process-flow-wg76
  function:
    name:       split
    runtime:    python3
    handler:    index.handler
    memorySize: 3072
    timeout:    600
fc-video-demo-transcode:
  region:   cn-hangzhou
  service:
    name: video-process-flow-wg76
  function:
    name:       transcode
    runtime:    python3
    handler:    index.handler
    memorySize: 3072
    timeout:    600
fc-video-demo-merge:
  region:   cn-hangzhou
  service:
    name: video-process-flow-wg76
  function:
    name:       merge
    runtime:    python3
    handler:    index.handler
    memorySize: 3072
    timeout:    600
fc-video-demo-after-process:
  region:   cn-hangzhou
  service:
    name: video-process-flow-wg76
  function:
    name:       after-process
    runtime:    python3
    handler:    index.handler
    memorySize: 512
    timeout:    120
fc-oss-trigger-trigger-fnf:
  region:   cn-hangzhou
  service:
    name: video-process-flow-wg76
  function:
    name:       trigger-fnf
    runtime:    python3
    handler:    index.handler
    memorySize: 128
    timeout:    120
  triggers:
    -
      type: oss
      name: oss-t
video-demo-flow:
  RegionId: cn-hangzhou
  Name:     video-process-flow

Verify the deployment

1. Log on to the OSS console. Navigate to the source directory in testBucket and upload an MP4 video file.

2. Log on to the Serverless Workflow console. On the Workflows page, click the target workflow. On the Executions tab, click the execution name to view the workflow execution flow and status.

   

3. When the execution status shows Succeeded, go to the dest directory in testBucket to view the transcoded files. If the transcoded files appear, the video processing system is working correctly.