Build an elastic and highly available audio and video processing system based on a serverless architecture

Simple video processing system

Prerequisites

• Function Compute: Activate Function Compute.

• Object Storage Service (OSS): Create a bucket in the console.

Procedure

The following steps describe how to build a serverless video conversion system. When a video is uploaded to the source directory of an OSS bucket, a function is triggered. The function converts the MP4 video file to the FLV format and saves the FLV file to the dest directory in the OSS bucket.

1. Create a function.

   Log on to the Function Compute console and select Event Function on the Create Function page. On the Create Event Function page, set the Function Name and the following options, and then click Create.

   • Elastic Configuration: Event functions support only elastic instances. You must configure the Specification Plan and Minimum Number Of Instances.

     Because video processing is a resource-intensive task, this topic sets vCPU to 4 cores, Memory to 8 GB, and Disk to 10 GB.

   • Function Code: Set the runtime to Built-in Runtime > Python > Python 3.10, the Code Upload Method to Use Sample Code, and the Execution Timeout to 7200 seconds.

   • Advanced Configuration: In the Permission category, select AliyunFcDefaultRole for Function Role.

   Use the default values for other configurations. For more information about creating a function, see Create a function.

2. Create an OSS trigger.

   1. On the Function Details page, click the Configurations tab. In the left-side navigation pane, click Triggers. Then, click Create Trigger.

   2. In the Create Trigger panel, for Trigger Type, select Two-way Integrated Trigger > Object Storage Service (OSS). Set the remaining configuration items and click OK.

      Configuration item	Operation	Example
      Name	Enter a custom name for the trigger.	oss-trigger
      Version Or Alias	The default value is LATEST. To create a trigger for another version or alias, you must first select the version from the Version Or Alias dropdown list on the function details page. For an introduction to versions and aliases, see Version management and Alias management.	LATEST
      Bucket Name	Select an existing OSS bucket.	testbucket
      File Prefix	Enter the prefix of the file names to match. We recommend that you configure a file prefix and suffix to avoid nested loop triggers that can cause extra fees. In addition, if different triggers for the same bucket specify the same event type, their prefixes and suffixes cannot be the same. For more information, see OSS trigger rules. Important The file prefix cannot start with /. Otherwise, the OSS trigger cannot be triggered.	source
      File Suffix	Enter the suffix of the file names to match. We strongly recommend that you configure a prefix and suffix to avoid nested loop triggers that can cause extra fees. In addition, if different triggers for the same bucket specify the same event type, their prefixes and suffixes cannot be the same. For more information, see OSS trigger rules.	mp4
      Triggering Event	Select one or more triggering events. For more information about Object Storage Service (OSS) event types, see OSS events. In this example, oss:ObjectCreated:PutObject is selected.	oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload
      Role Name	Select AliyunOSSEventNotificationRole. Note If you are creating this type of trigger for the first time, click OK, and then select Authorize Now in the pop-up dialog box.	AliyunOSSEventNotificationRole

      Important

      Note that you can configure only one OSS trigger for a single prefix in a bucket. If you configure multiple triggers, a Cannot specify overlapping prefix and suffix with same event type error occurs. To resolve this issue, use one of the following methods:

      • For Bucket Name, enter the name of another bucket.

      • Change the prefix name.

      • Delete the conflicting OSS trigger.

3. Write the function code.

   1. On the function details page, click the Code tab and write code in the editor.

      This topic uses Python as an example. The sample code is as follows.

      # -*- 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)
          # Special handling for m3u8
          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 transcoding fails, trigger the dest-fail function
              raise Exception(request_id + " transcode failure, detail: " + str(exc))
      
          finally:
              if os.path.exists(ts_dir):
                  shutil.rmtree(ts_dir)
      
              # Remove the .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 transcoding fails, trigger the 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.

4. Test the function code.

   You can configure the function's entry point parameter to simulate an OSS event parameter to test the code. In a real-world scenario, the function is automatically triggered when an OSS event occurs.

   1. On the function details page, click the Code tab. Then, click the icon to the right of Test Function and select Configure Test Parameters from the drop-down list.

   2. In the Configure Test Parameters panel, select the Create New Test Event or Edit Existing Test Event tab, enter the Event Name and event content, and click OK.

      The event template for this topic is as follows. Replace the bucket name, ETag, and video file name with your actual values.

      {
          "events": [
              {
                  "eventName": "oss:ObjectCreated:CompleteMultipartUpload",
                  "eventSource": "acs:oss",
                  "eventTime": "2022-08-13T06:45:43.000Z",
                  "eventVersion": "1.0",
                  "oss": {
                      "bucket": {
                          "name": "testbucket"
                      },
                      "object": {
                          "eTag": "688A7BF4F233DC9C88A80BF985AB****",
                          "key": "source/a.mp4"
                      }
                  },
                  "region": "cn-hangzhou"
              }
          ]
      }

      Note

      An ETag (entity tag) is created when an object is generated to identify the content of the object. You can find the ETag by clicking Details for the target object in the OSS file list.

      

      For more information about event parameters, see Configure function entry point parameters.

   3. Click Test Function. After the function runs successfully, you can find the converted video file at the dest/ path in the target bucket.

      

Video processing workflow system

Prerequisites

• Activate services and create a bucket

  • Function Compute: Activate Function Compute.

  • Object Storage Service (OSS): Create a bucket in the console.

  • CloudFlow: Activate CloudFlow.

  • File Storage NAS: Activate Apsara File Storage NAS.

  • Virtual Private Cloud: Activate VPC.

• Configure Resource Access Management (RAM) roles

  • AliyunFcDefaultRole: This role is required when you create a function. The role must have the AliyunFCInvocationAccess, AliyunFnFFullAccess, and AliyunOSSFullAccess permissions to invoke functions, manage workflows, and manage OSS.

  • AliyunOSSEventNotificationRole: OSS uses this role by default to send event notifications.

  • fnf-execution-default-role: This role is required to create and manage workflows. The role must have the AliyunFCInvocationAccess and AliyunFnFFullAccess permissions.

    In the Trust Policy for the role, grant the workflow service the AssumeRole permission, as shown below.

    {
      "Statement": [
        {
          "Action": "sts:AssumeRole",
          "Effect": "Allow",
          "Principal": {
            "Service": [
              "fnf.aliyuncs.com"
            ]
          }
        }
      ],
      "Version": "1"
    }

• Install and configure Serverless Devs

  Quick Start

Procedure