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

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

前提条件

• Function Compute: Function Computeの有効化

• OSS: バケットの作成

手順

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

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

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

      関数は変換します。にmp4ファイル。flv形式で保存します。OSSバケットのdestディレクトリにあるflvファイル。 この例では、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: Function Computeの有効化

  • OSS: バケットの作成

  • CloudFlow: CloudFlowの有効化

  • NAS: NASの有効化

  • 仮想プライベートクラウド (VPC): VPCの有効化

• サービスロールの設定

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

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

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

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

  • Serverless DevsとDockerのインストール

  • Serverless Devsの設定

手順