Function Compute:Build an elastic and highly available audio and video processing system in a serverless architecture

Simple video processing system

Prerequisites

• Function Compute: Activate Function Compute.

• OSS: Create buckets.

Procedure

1. Create a service.

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

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

   3. In the Create Service panel, enter a service name and description, configure the parameters based on your business requirements, and then click OK.

      In this example, the service role is set to AliyunFCDefaultRole and AliyunOSSFullAccess is added to the policy.

      For more information about how to create a service, see Create a service.

2. Create a function.

   1. On the Functions page, click Create Function.

   2. On the Create Function page, select a method to create the function, configure the following parameters, and then click Create.

      • Method to create the function: Use Built-in Runtime.

      • Basic Settings: Configure the basic information about the function, including Function Name and Handler Type. Set Handler Type to Event Handler.

      • Code: Set Runtime to Python 3.9 and Code Upload Method to Use Sample Code.

      • Advanced Settings: The processing of video files is time-consuming. In this example, vCPU Capacity is set to 4 vCPUs, Memory Capacity is set to 8 GB, Size of Temporary Disk is set to 10 GB, and Execution Timeout Period is set to 7200 seconds. The preceding settings are configured based on the size of videos to process.

      Retain the default values for other parameters. For more information about how to create a function, see Create a function.

3. Create an OSS trigger.

   1. On the function details page, click the Triggers tab, select a version or alias from the Version or Alias drop-down list, and then click Create Trigger.

   2. In the Create Trigger panel, specify related parameters and click OK.

      Parameter	Description	Example
      Trigger Type	Select the type of the trigger. In this example, OSS is selected.	OSS
      Name	Enter the trigger name.	oss-trigger
      Version or Alias	Specify the version or alias. The default value is LATEST. If you want to create a trigger for another version or alias, select a version or alias from the Version or Alias drop-down list on the function details page. For more information about versions and aliases, see Manage versions and Manage aliases.	LATEST
      Bucket Name	Select the bucket that you created.	testbucket
      Object Prefix	Enter the prefix of the object names that you want to match. To prevent unexpected costs that are generated by nested loops, we recommend that you configure the prefix and suffix of the object name. If you specify the same event type for different triggers of a bucket, the prefix or suffix must be unique. For more information, see Rules for triggering native OSS triggers. Important The object prefix cannot start with a forward slash (/). Otherwise, the OSS trigger cannot be triggered.	source
      Object Suffix	Enter the suffix of the object name that you want to match. We recommend that you configure Object Prefix and Object Suffix to prevent additional costs that are generated by nested loops. If you specify the same event type for different triggers in a bucket, the object prefix or suffix must be unique. For more information, see Rules for triggering native OSS triggers.	mp4
      Trigger Event	Select one or more trigger events from the drop-down list. For more information about the event types of OSS, see OSS events.	oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload
      Role Name	Specify the name of a resource access management (RAM) role. In this example, AliyunOSSEventNotificationRole is selected. Note After you configure the preceding parameters, click OK. The first time you create a trigger of this type, click Authorize Now in the message that appears.	AliyunOSSEventNotificationRole

4. Write function code.

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

      The function converts .mp4 files to the .flv format and stores the .flv files in the dest directory of an OSS bucket. In this example, Python is used. The following sample code provides an example:

      # -*- 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 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 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.

5. Test the function code.

   You can configure function input parameters to simulate OSS events and verify the code. In actual operations, the function is automatically triggered when a specified OSS event occurs.

   1. On the function details page, click the Code tab, 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, click the Create New Test Event or Modify Existing Test Event tab, configure Event Name and the event content, and then click OK.

      The following sample code provides an example of event configurations. For more information about the event parameter, see Step 2: Configure input parameters of the function.

      {
          "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****"
                  }
              }
          ]
      }

   3. Click Test Function. On the Code tab, view the execution result.

Video processing workflow system

Prerequisites

• Activate Function Compute and create an OSS bucket

  • Function Compute: Activate Function Compute.

  • OSS: Create buckets.

  • CloudFlow: Activate CloudFlow.

  • NAS: Activate NAS.

  • Virtual Private Cloud (VPC): Activate VPC.

• Configure a service role

  • AliyunFcDefaultRole: The default role of Function Compute. You must configure this role when you create a service. Attach the AliyunOSSFullAccess, AliyunFnFFullAccess, and AliyunFCInvocationAccess permissions to the AliyunFcDefaultRole to invoke functions, manage workflows, and manage OSS.

  • AliyunOSSEventNotificationRole: The default role that is used by OSS to send event notifications.

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

• Install and configure Serverless Devs

  • Install Serverless Devs and Docker

  • Configure Serverless Devs

Procedure