Function Compute：構建基於Serverless架構的彈性高可用音視頻處理系統

簡單視頻處理系統

前提條件

• Function Compute：開通Function Compute服務。

• Object Storage ServiceOSS：控制台建立儲存空間。

操作步驟

1. 建立服務。

   1. 登入Function Compute控制台，在左側導覽列，單擊服務及函數。

   2. 在頂部功能表列，選擇地區，然後在服務列表頁面，單擊建立服務。

   3. 在建立服務面板，填寫服務名稱和描述，並按需設定配置項，然後單擊確定。

      本文設定服務角色為AliyunFCDefaultRole，並在權限原則中增加AliyunOSSFullAccess。

      關於建立服務的詳細配置資訊，請參見建立服務。

2. 建立函數。

   1. 在函數管理頁面，單擊建立函數。

   2. 在建立函數頁面，按需選擇建立函數的方式，配置以下配置項，然後單擊建立。

      • 建立函數方式：使用內建運行時建立。

      • 基本設定：配置函數的基本資料，包括函數名稱和請求處理常式類型。請求處理常式類型選擇處理事件請求。

      • 函數代碼：函數的運行環境選擇Python 3.9，代碼上傳方式選擇使用範例程式碼。

      • 進階配置：由於處理視頻檔案比較耗時，並且要考慮視頻檔案的大小，所以本文選擇vCPU規格為4核，記憶體規格為8 GB，臨時硬碟大小為10 GB，執行逾時時間為7200秒。

      其他配置項使用預設值，關於建立函數的詳細配置資訊，請參見建立函數。

3. 建立OSS觸發器。

   1. 在函數詳情頁面，單擊觸發器管理頁簽，從版本或別名下拉式清單選擇要建立觸發器的版本或別名，然後單擊建立觸發器。

   2. 在建立觸發程序面板，填寫相關資訊。然後單擊確定。

      配置項	操作	本文樣本
      觸發器類型	選擇Object Storage Service。	Object Storage Service
      名稱	填寫自訂的觸發器名稱。	oss-trigger
      版本或別名	預設值為LATEST，如果您需要建立其他版本或別名的觸發器，需先在函數詳情頁的版本或別名下拉式清單選擇該版本。關於版本和別名的簡介，請參見管理版本和管理別名。	LATEST
      Bucket 名稱	選擇已建立的OSS Bucket。	testbucket
      檔案首碼	輸入要匹配的檔案名稱的首碼。建議您設定檔的首碼和尾碼，避免觸發事件嵌套迴圈觸發引起額外費用。此外，一個Bucket的不同觸發器如果指定了相同的事件類型，則首碼和尾碼不能重複。詳細資料，請參見OSS觸發器觸發規則。 重要 檔案首碼不能以/開頭，否則會導致OSS觸發器無法被觸發。	source
      檔案尾碼	輸入要匹配的檔案名稱的尾碼。強烈建議您配置首碼和尾碼，避免觸發事件嵌套迴圈觸發引起額外費用。另外，一個Bucket的不同觸發器如果指定了相同的事件類型，則首碼和尾碼不能重複。詳細資料，請參見OSS觸發器觸發規則。	mp4
      觸發事件	選擇一個或多個觸發事件。關於Object Storage Service的事件類型，請參見OSS事件定義。	oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload
      角色名稱	選擇AliyunOSSEventNotificationRole。 說明 如果您第一次建立該類型的觸發器，則需要在單擊確定後，在彈出的對話方塊中選擇立即授權。	AliyunOSSEventNotificationRole

4. 編寫函數代碼。

   1. 在函數詳情頁面，單擊函數代碼頁簽，在代碼編輯器中編寫代碼。

      函數需要將MP4格式的視頻檔案轉為FLV格式，並將FLV格式視頻檔案儲存在OSS Bucket下的dest目錄。本文以Python語言為例，範例程式碼如下。

      # -*- 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 特殊處理
          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 檔案
              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. 單擊部署代碼。

5. 測試函數代碼。

   您可以通過配置函數入口參數類比OSS事件參數進行測試，驗證代碼的正確性。在實際操作過程中，當發生OSS事件時，函數會自動被觸發執行。

   1. 在函數詳情頁面，單擊函數代碼頁簽，然後單擊測試函數右側表徵圖，從下拉式清單中，選擇配置測試參數。

   2. 在配置測試參數面板，選擇建立新測試事件或編輯已有測試事件頁簽，填寫事件名稱和事件內容。然後單擊確定。

      本文配置的事件內容如下，關於event參數具體資訊，請參見步驟二：配置函數入口參數。

      {
          "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. 單擊測試函數，在函數代碼頁簽查看執行結果。

視頻處理工作流程系統

前提條件

• 開通服務並建立Bucket

  • Function Compute：開通Function Compute服務。

  • Object Storage ServiceOSS：控制台建立儲存空間。

  • Serverless 工作流程：開通Serverless工作流程。

  • Apsara File Storage NAS：開通檔案儲存體服務NAS。

  • Virtual Private Cloud：開通Virtual Private Cloud。

• 佈建服務角色

  • AliyunFcDefaultRole：建立服務時需要配置該角色，同時該角色需要配置許可權AliyunOSSFullAccess、AliyunFnFFullAccess、AliyunFCInvocationAccess，用於調用函數、管理工作流程和管理Object Storage Service。

  • AliyunOSSEventNotificationRole：OSS預設使用此角色發送事件通知。

  • fnf-execution-default-role：建立及管理工作流程所需要的角色，該角色需要配置許可權AliyunFCInvocationAccess、AliyunFnFFullAccess。

• Serverless Devs工具安裝及配置

  • 安裝Serverless Devs工具及依賴

  • 配置Serverless Devs

操作步驟