Function Compute:柔軟で可用性の高いオーディオおよびビデオ処理システムをサーバーレスアーキテクチャで構築する

シンプルなビデオ処理システム

前提条件

• Function Computeが有効化されています。 詳細については、「手順1: Function Computeの有効化」をご参照ください。

• Object Storage Serviceが有効化されています。 詳細については、「バケットの作成」をご参照ください。

手順

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

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

   2. 上部のナビゲーションバーで、リージョンを選択します。 [サービス] ページで、[サービスの作成] をクリックします。

   3. [サービスの作成] パネルで、サービス名と説明を入力し、ビジネス要件に基づいてパラメーターを設定し、OKをクリックします。

      この例では、サービスロールはAliyunFCDefaultRoleに設定され、AliyunOSSFullAccessがポリシーに追加されています。

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

2. 関数を作成します。

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

   2. On the関数の作成ページで、関数を作成する方法を選択し、次のパラメーターを設定し、作成.

      • 関数を作成するメソッド: 組み込みランタイムを使用。

      • 基本設定: [関数名] や [ハンドラータイプ] など、関数の基本情報を設定します。 [ハンドラータイプ] を [イベントハンドラー] に設定します。

      • Code: ランタイムをPython 3.9に設定し、コードのアップロード方法をサンプルコードに設定します。

      • 詳細設定: ビデオファイルの処理には時間がかかります。 この例では、vCPU容量は4 vCPUに設定され、メモリ容量は8 GBに設定され、一時ディスクのサイズは10 GBに設定され、実行タイムアウト期間は7200秒に設定されます。 上記のパラメーターを設定するときは、ビデオのサイズを考慮してください。

      他のパラメーターのデフォルト値を保持します。 関数の作成方法の詳細については、「関数の作成」をご参照ください。

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

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

   2. [トリガーの作成] パネルで、関連パラメーターを指定し、[OK] をクリックします。

      パラメーター	説明	例
      トリガータイプ	トリガーのタイプを選択します。 この例では、OSSが選択されています。	OSS
      名前	トリガー名を入力します。	oss-trigger
      バージョンまたはエイリアス	バージョンまたはエイリアスを指定します。 デフォルト値はLATESTです。 別のバージョンまたはエイリアスのトリガーを作成する場合は、[関数の詳細] ページの [バージョン] または [エイリアス] ドロップダウンリストからそのバージョンまたはエイリアスを選択します。 バージョンとエイリアスの詳細については、「バージョンの管理」および「エイリアスの管理」をご参照ください。	LATEST
      バケット名	作成したバケットを選択します。	testbucket
      オブジェクトプレフィックス	一致させるオブジェクト名のプレフィックスを入力します。 ネストされたループによって予期しないコストが発生しないように、オブジェクトプレフィックスとオブジェクトサフィックスを設定することを推奨します。 バケットの異なるトリガーに同じイベントタイプを指定した場合、プレフィックスとサフィックスを同じにすることはできません。 詳細については、「トリガールール」をご参照ください。 重要 オブジェクトプレフィックスの先頭にスラッシュ (/) を付けることはできません。 それ以外の場合、OSSトリガーは機能しません。	source
      オブジェクト接尾辞	一致させたいオブジェクト名のサフィックスを入力します。 ネストされたループによって予期しないコストが発生しないように、オブジェクトプレフィックスとオブジェクトサフィックスを設定することを推奨します。 バケットの異なるトリガーに同じイベントタイプを指定した場合、プレフィックスとサフィックスを同じにすることはできません。 詳細については、「トリガールール」をご参照ください。	mp4
      トリガーイベント	1つ以上のトリガーイベントを選択します。 OSSのイベントタイプの詳細については、「OSSイベント」をご参照ください。	oss:ObjectCreated:PutObject、oss:ObjectCreated:PostObject、oss:ObjectCreated:CompleteMultipartUpload
      ロール名	リソースアクセス管理 (RAM) ロール名を指定します。 この例では、AliyunOSSEventNotificationRoleが選択されています。 説明 上記のパラメーターを設定したら、[OK] をクリックします。 このタイプのトリガーを初めて作成する場合は、表示されるメッセージの [今すぐ許可] をクリックします。	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 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. [デプロイ] をクリックします。

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の有効化とOSSバケットの作成

  • Function Computeが有効化されています。 詳細については、「手順1: Function Computeの有効化」をご参照ください。

  • Object Storage Serviceが有効化されています。 詳細については、「バケットの作成」をご参照ください。

  • CloudFlowが有効化されています。 詳細については、をご覧ください。 CloudFlowの有効化

  • File Storage NASが有効化されています。 詳細については、をご覧ください。 NASの有効化

  • Virtual Private Cloudが有効化されています。 詳細については、をご覧ください。 VPCの有効化

• サービスロールの設定

  • AliyunFcDefaultRole: Function Computeのデフォルトのロール。 サービスを作成するときに、このロールを設定する必要があります。 AliyunOSSFullAccess、AliyunFnFFullAccess、およびAliyunFCInvocationAccess権限をこのロールに付与して、関数の呼び出し、ワークフローの管理、およびOSSの管理を行います。

  • AliyunOSSEventNotificationRole: OSSがイベント通知を送信するために使用するデフォルトのロール。

  • fnf-execution-default-role: ワークフローの作成と管理に必要なロール。 このロールには、AliyunFCInvocationAccessおよびAliyunFnFFullAccess権限を割り当てる必要があります。

• Serverless Devsのインストールと構成

  • サーバーレスのDevsと依存関係がインストールされます。 詳細については、「Serverless DevsとDockerのインストール」をご参照ください。

  • サーバーレスDevsが設定されています。 詳細については、「Serverless Devsの設定」をご参照ください。

手順