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

Simple video processing system

Prerequisites

• Function Compute is activated. For more information, see Step 1: Activate Function Compute.

• Object Storage Service is activated. For more information, see 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. Creates 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 of 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. Make sure to consider the sizes of your videos when configuring the preceding parameters.

      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 the 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 name that you want to match. We recommend that you configure Object Prefix and Object Suffix to prevent unexpected costs generated by nested loops. If you specify the same event type for different triggers of a bucket, the prefix and suffix cannot be the same. For more information, see Triggering rules. Important The object prefix cannot start with a forward slash (/). Otherwise, the OSS trigger will not function.	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 unexpected costs generated by nested loops. If you specify the same event type for different triggers of a bucket, the prefix and suffix cannot be the same. For more information, see Triggering rules.	mp4
      Trigger Event	Select one or more trigger events. 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. If this is the first time that 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 and then 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 your 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 next to Test Function, and then 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, and specify Event Name and the event content. Then, click OK.

      The following sample code provides an example of event configurations. For more information about the event parameters, 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 is activated. For more information, see Step 1: Activate Function Compute.

  • Object Storage Service is activated. For more information, see Create buckets.

  • CloudFlow is activated. For more information, visit Activate CloudFlow.

  • File Storage NAS is activated. For more information, visit Activate NAS.

  • Virtual Private Cloud is activated. For more information, visit 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 this role 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 this role.

• Install and configure Serverless Devs

  • Serverless Devs and dependencies are installed. For more information, see Install Serverless Devs and Docker.

  • Serverless Devs is configured. For more information, see Configure Serverless Devs.

Procedure