Function Compute:サーバーレスアーキテクチャを使用した弾力的で高可用な音声・動画処理システムの構築

シンプルな動画処理システム

前提条件

• Function Compute: Function Compute をアクティブ化します。

• Object Storage ServiceOSS: OSS コンソールでバケットを作成する。

操作手順

1. サービスを作成します。

   1. Function Compute コンソールにログインします。左側のナビゲーションウィンドウで、サービスと関数 をクリックします。

   2. トップメニューバーでリージョンを選択します。サービス ページで、サービスの作成 をクリックします。

   3. サービスの作成パネルで、サービス名と説明を入力します。必要に応じてオプションを設定します。[OK] をクリックします。

      このトピックでは、サービスロールを AliyunFCDefaultRole に設定し、AliyunOSSFullAccess 権限ポリシーを追加します。

      詳細なサービス構成については、「サービスを作成」をご参照ください。

2. 関数を作成します。

   1. [関数管理] ページで、[関数の作成] をクリックします。

   2. [関数の作成] ページで、作成方法を選択します。以下のオプションを設定します。[作成] をクリックします。

      • 作成方法：組み込みランタイムで作成。

      • 基本設定: [関数名] や [ハンドラタイプ] を含む関数の基本情報を設定します。[ハンドラタイプ] を [イベント駆動ハンドラ] に設定します。

      • 関数コード: ランタイムとして Python 3.9 を選択します。[サンプルコードを使用] をコードアップロード方法として選択します。

      • 高度な構成: ビデオ処理には時間がかかり、ファイルサイズも異なるため、このトピックでは [vCPU 数] = 4、[メモリ] = 8 GB、[一時ディスクサイズ] = 10 GB、および [タイムアウト] = 7200 秒を選択します。

      その他のオプションにはデフォルト値を使用します。詳細な関数構成については、「関数を作成」をご参照ください。

3. OSS トリガーを作成します。

   1. 関数の詳細ページで、[トリガー] タブをクリックします。[バージョンまたはエイリアス] ドロップダウンリストから、トリガーのバージョンまたはエイリアスを選択します。[トリガーの作成] をクリックします。

   2. 「トリガーの作成」パネルで、必要な情報を入力します。[OK] をクリックします。

      構成項目	アクション	このトピックの例
      トリガーの種類	[オブジェクトストレージ OSS] を選択します。	Object Storage OSS
      名前	カスタムトリガー名を入力します。	oss-trigger
      バージョンまたはエイリアス	デフォルトは LATEST です。別のバージョンまたはエイリアスのトリガーを作成するには、まず関数の詳細ページにある [バージョンまたはエイリアス] ドロップダウンリストから選択します。詳細については、「バージョンを管理する」および「エイリアスを管理する」をご参照ください。	LATEST
      バケット名	既存の OSS バケットを選択します。	testbucket
      オブジェクトプレフィックス	オブジェクト名と一致するプレフィックスを入力します。ネストされたイベントループと追加料金を回避するために、プレフィックスとサフィックスの両方を設定することを推奨します。また、同じバケット内のトリガーでイベントタイプが同じ場合、プレフィックスとサフィックスは重複してはなりません。詳細については、「OSS トリガールール」をご参照ください。 重要 ファイルプレフィックスが / で始まる場合、OSS トリガーはトリガーされません。	source
      ファイル拡張子	オブジェクト名と一致するサフィックスを入力します。ネストされたイベントループと追加料金を回避するために、プレフィックスとサフィックスの両方を設定することを強く推奨します。また、同じバケット内のトリガーでイベントタイプが同じ場合、プレフィックスとサフィックスは重複してはなりません。詳細については、「OSS トリガールール」をご参照ください。	mp4
      トリガーイベント	1つ以上のトリガーイベントを選択します。Object Storage Service (OSS) のイベントタイプについては、「OSS イベント定義」をご参照ください。	oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload
      ロール名	[AliyunOSSEventNotificationRole] を選択します。 説明 このタイプのトリトリガーを初めて使用する場合、[OK] をクリックします。ダイアログボックスで、[Authorize now] を選択します。	AliyunOSSEventNotificationRole

4. 関数コードを記述します。

   1. 関数の詳細ページで、[関数コード] タブをクリックします。エディターにコードを記述します。

      この関数は MP4 動画を FLV フォーマットに変換し、FLV ファイルを OSS バケットの 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 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. [コードをデプロイ] をクリックします。

5. 関数コードをテストします。

   関数入力パラメーターを使用して OSS イベントパラメーターをシミュレートし、正確性を検証できます。実際には、OSS イベントが発生すると関数は自動的にトリガーされます。

   1. 「関数詳細」ページで、[関数コード] タブをクリックし、[関数をテスト] の右側にある アイコンをクリックします。ドロップダウンリストから [テストパラメーターを設定] を選択します。

   2. [テストパラメーターの設定] パネルで、[新しいテストイベントの作成] または [既存のテストイベントの編集] を選択します。[イベント名] とイベントコンテンツを入力します。[OK] をクリックします。

      このトピックで使用されるイベントコンテンツは以下のとおりです。イベントパラメーターの詳細については、「ステップ 2: 関数入力パラメーターを構成」をご参照ください。

      {
          "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. [関数のテスト] をクリックします。[関数コード] タブで結果を確認します。

動画処理ワークフローシステム

前提条件

• サービスをアクティブ化し、バケットを作成

  • Function Compute: Function Compute をアクティブ化します。

  • Object Storage ServiceOSS: OSS コンソールでバケットを作成する.

  • サーバーレス ワークフロー: サーバーレス ワークフローの有効化。

  • File Storage NAS: Apsara File Storage NAS を有効化する。

  • Virtual Private Cloud: Virtual Private Cloud の有効化。

• サービスロールを構成

  • AliyunFcDefaultRole: サービス作成時にこのロールを割り当てます。関数を呼び出し、ワークフローを管理し、OSS を管理するために、AliyunOSSFullAccess、AliyunFnFFullAccess、および AliyunFCInvocationAccess 権限をアタッチします。

  • AliyunOSSEventNotificationRole: OSS はデフォルトでこのロールを使用してイベント通知を送信します。

  • fnf-execution-default-role: ワークフローの作成と管理に必要です。AliyunFCInvocationAccess および AliyunFnFFullAccess 権限をアタッチします。

• Serverless Devs をインストールして構成

  • Serverless Devs と依存関係をインストールします。

  • Serverless Devs を構成します。

操作手順