CDN:[Function Compute コンソール] オフラインログを OSS にアーカイブする

1. バケット詳細の取得

Function Compute の関数には、OSS バケットの詳細が必要です。次の手順に従って、バケット名、パブリックアクセスエンドポイント、およびログストレージディレクトリ名を取得します。

2. Function Compute の関数とトリガーの作成

Function Compute で、アーカイブを実行する関数と、それを開始するトリガーを作成します。

1. Function Compute コンソールにログインします。左側のナビゲーションウィンドウで、[関数] を選択します。

2. [関数] ページで、[関数の作成] をクリックし、[イベント関数] を選択してから、[イベント関数の作成] をクリックします。

3. イベント関数を作成する際に、次の主要なパラメーターを設定します：

   • [基本設定] - 関数名：この関数名は後続の操作で必要になります。cdn-log-dump を使用することを推奨します。

   • 関数コード > ランタイム：この関数は Python で記述されているため、[公式ランタイム]、[Python]、次に [Python 3.10] を選択します。

   • [詳細設定] > 環境変数：OSS バケットの詳細を関数に渡すために、次の環境変数を作成します：

     • target_oss_bucket：バケット名

     • target_oss_endpoint：パブリックアクセス用のエンドポイント

     • target_oss_prefix：ログファイルを保存するディレクトリの名前

     例：target_oss_bucket を demo-bucket に、target_oss_endpoint を oss-cn-hangzhou.aliyuncs.com に、target_oss_prefix を cdn_log に設定します。

4. パラメーターを設定した後、作成 をクリックします。

5. [関数詳細] で、[トリガー] タブをクリックし、[トリガーの作成] をクリックします。

6. トリガーの主要なパラメーターを次のように設定し、[OK] をクリックします。

   • [トリガータイプ]：[Alibaba Cloud CDN (同期呼び出し)] を選択します。

   • 名前：トリガーの名前を入力します (推奨：cdn-logs-triggers)。

   • [イベント名]：[LogFileCreated] を選択します。

   • ドメイン名：同じ Alibaba Cloud アカウント配下のアクティブな高速化ドメイン名を入力します。

   • 説明：トリガーの説明を入力します。「CDN オフラインログファイル生成トリガー」の使用を推奨します。

   • [ロール]：[AliyunCDNEventNotificationRole] を選択します。

7. トリガーのパラメーターを設定した後、OK をクリックします。[CDN トリガーのデフォルトロールが作成されていません] というメッセージが表示された場合は、[今すぐ承認] をクリックし、画面の指示に従ってデフォルトロールを作成します。それ以外の場合、トリガーはすぐに作成されます。

8. [関数詳細] で、[コード] タブをクリックし、オンラインエディターに以下のコードを入力します。このコードは、Alibaba Cloud CDN からオフラインログを取得し、Object Storage Service (OSS) に保存します。

   アーカイブタスクのコード

   # -*- coding: utf-8 -*-
   import os, time, json, requests, traceback, oss2, fc2
   from requests.exceptions import *
   from fc2.fc_exceptions import *
   from oss2.models import PartInfo
   from oss2.exceptions import *
   from multiprocessing import Pool
   from contextlib import closing
   MAX_PROCCESSES = 20 # 各サブタスクのワーカプロセスの数
   BLOCK_SIZE = 6 * 1024 * 1024 # 各パートのサイズ
   BLOCK_NUM_INTERNAL = 18 # 内部 URL の場合の各サブタスクのデフォルトのブロック数
   BLOCK_NUM = 10 # 各サブタスクのデフォルトのブロック数
   MAX_SUBTASKS = 49 # サブタスクを実行するワーカプロセスの数
   CHUNK_SIZE = 8 * 1024 # 各チャンクのサイズ
   SLEEP_TIME = 0.1 # リトライ前に待機する初期秒数
   MAX_RETRY_TIME = 10 # 最大リトライ回数
   def retry(func):
       """
       リトライメカニズムを持つ関数のエグゼキュータ。
       :param func: (必須, lambda) 実行される関数。
       :return: func の実行結果。
       """
       wait_time = SLEEP_TIME
       retry_cnt = 1
       while True:
           if retry_cnt > MAX_RETRY_TIME:
               return func()
           try:
               return func()
           except (ConnectionError, SSLError, ConnectTimeout, Timeout) as e:
               print(traceback.format_exc())
           except (OssError) as e:
               if 500 <= e.status < 600:
                   print(traceback.format_exc())
               else:
                   raise Exception(e)
           except (FcError) as e:
               if (500 <= e.status_code < 600) or (e.status_code == 429):
                   print(traceback.format_exc())
               else:
                   raise Exception(e)
           print('Retrying (attempt %d)...' % retry_cnt)
           time.sleep(wait_time)
           wait_time *= 2
           retry_cnt += 1
   def get_info(url):
       """
       ファイルの CRC64 と合計長を取得します。
       :param url: (必須, string) ファイルの URL。
       :return: CRC64, 長さ
       """
       with retry(lambda : requests.get(url, {}, stream = True)) as r:
           return r.headers['x-oss-hash-crc64ecma'], int(r.headers['content-length'])
   class Response(object):
       """
       チャンク読み取りをサポートするレスポンスクラス。
       """
       def __init__(self, response):
           self.response = response
           self.status = response.status_code
           self.headers = response.headers
       def read(self, amt = None):
           if amt is None:
               content = b''
               for chunk in self.response.iter_content(CHUNK_SIZE):
                   content += chunk
               return content
           else:
               try:
                   return next(self.response.iter_content(amt))
               except StopIteration:
                   return b''
       def __iter__(self):
           return self.response.iter_content(CHUNK_SIZE)
   def migrate_part(args):
       """
       URL からパートをダウンロードし、OSS にアップロードします。
       :param args: (bucket, object_name, upload_id, part_number, url, st, en)
       :bucket: (必須, Bucket) 送信先の OSS バケット。
       :object_name: (必須, string) 送信先のオブジェクト名。
       :upload_id: (必須, integer) このアップロードタスクのアップロード ID。
       :part_number: (integer) このパートのパート番号。
       :url: (必須, string) ファイルの URL。
       :st, en: (必須, integer) このパートのバイト範囲。[st, en] を示す。
       :return: (part_number, etag)
       :part_number: (integer) このパートのパート番号。
       :etag: (string) upload_part 結果の etag。
       """
       bucket = args[0]
       object_name = args[1]
       upload_id = args[2]
       part_number = args[3]
       url = args[4]
       st = args[5]
       en = args[6]
       try:
           headers = {'Range' : 'bytes=%d-%d' % (st, en)}
           resp = Response(retry(lambda : requests.get(url, headers = headers, stream = True)))
           result = retry(lambda : bucket.upload_part(object_name, upload_id, part_number, resp))
           return (part_number, result.etag)
       except Exception as e:
           print(traceback.format_exc())
           raise Exception(e)
   def do_subtask(event, context):
       """
       URL からファイルの範囲をダウンロードし、OSS にアップロードします。
       :param event: (必須, json) JSON 形式のイベント。
       :param context: (必須, FCContext) ハンドラのコンテキスト。
       :return: parts
       :parts: ([(integer, string)]) 各プロセスのパート番号と etag。
       """
       oss_endpoint = os.environ.get('target_oss_endpoint')
       oss_bucket_name = os.environ.get('target_oss_bucket')
       access_key_id = context.credentials.access_key_id
       access_key_secret = context.credentials.access_key_secret
       security_token = context.credentials.security_token
       auth = oss2.StsAuth(access_key_id, access_key_secret, security_token)
       bucket = oss2.Bucket(auth, oss_endpoint, oss_bucket_name)
       object_name = event['object_name']
       upload_id = event['upload_id']
       part_number = event['part_number']
       url = event['url']
       st = event['st']
       en = event['en']
       if part_number == 1:
           return [migrate_part((bucket, object_name, upload_id, part_number, url, st, en))]
       pool = Pool(MAX_PROCCESSES)
       tasks = []
       while st <= en:
           nxt = min(en, st + BLOCK_SIZE - 1)
           tasks.append((bucket, object_name, upload_id, part_number, url, st, nxt))
           part_number += 1
           st = nxt + 1
       parts = pool.map(migrate_part, tasks)
       pool.close()
       pool.join()
       return parts
   def invoke_subtask(args):
       """
       同じ関数を同期的に呼び出してサブタスクを開始します。
       :param args: (object_name, upload_id, part_number, url, st, en, context)
       :object_name: (必須, string) 送信先のオブジェクト名。
       :upload_id: (必須, integer) このアップロードタスクのアップロード ID。
       :part_number: (integer) このサブタスクの最初のパートのパート番号。
       :url: (必須, string) ファイルの URL。
       :st, en: (必須, integer) このサブタスクのバイト範囲。[st, en] を示す。
       :context: (必須, FCContext) ハンドラのコンテキスト。
       :return: 呼び出された関数の戻り値。
       """
       object_name = args[0]
       upload_id = args[1]
       part_number = args[2]
       url = args[3]
       st = args[4]
       en = args[5]
       context = args[6]
       account_id = context.account_id
       access_key_id = context.credentials.access_key_id
       access_key_secret = context.credentials.access_key_secret
       security_token = context.credentials.security_token
       region = context.region
       service_name = context.service.name
       function_name = context.function.name
       endpoint = 'http://%s.%s-internal.fc.aliyuncs.com' % (account_id, region)
       client = fc2.Client(
           endpoint = endpoint,
           accessKeyID = access_key_id,
           accessKeySecret = access_key_secret,
           securityToken = security_token
       )
       payload = {
           'object_name' : object_name,
           'upload_id' : upload_id,
           'part_number' : part_number,
           'url' : url,
           'st' : st,
           'en' : en,
           'is_children' : True
       }
       if part_number == 1:
           return json.dumps(do_subtask(payload, context))
       ret = retry(lambda : client.invoke_function(service_name, function_name, payload = json.dumps(payload)))
       return ret.data
   def divide(n, m):
       """
       浮動小数点演算なしで ceil(n / m) を計算します。
       :param n, m: (integer)
       :return: (integer) ceil(n / m)。
       """
       ret = n // m
       if n % m > 0:
           ret += 1
       return ret
   def migrate_file(url, oss_object_name, context):
       """
       URL からファイルをダウンロードし、OSS にアップロードします。
       :param url: (必須, string) ファイルの URL。
       :param oss_object_name: (必須, string) 送信先のオブジェクト名。
       :param context: (必須, FCContext) ハンドラのコンテキスト。
       :return: actual_crc64, expect_crc64
       :actual_crc64: (string) アップロードされたファイルの CRC64。
       :expect_crc64: (string) ソースファイルの CRC64。
       """
       crc64, total_size = get_info(url)
       oss_endpoint = os.environ.get('target_oss_endpoint')
       oss_bucket_name = os.environ.get('target_oss_bucket')
       access_key_id = context.credentials.access_key_id
       access_key_secret = context.credentials.access_key_secret
       security_token = context.credentials.security_token
       auth = oss2.StsAuth(access_key_id, access_key_secret, security_token)
       bucket = oss2.Bucket(auth, oss_endpoint, oss_bucket_name)
       upload_id = retry(lambda : bucket.init_multipart_upload(oss_object_name)).upload_id
       pool = Pool(MAX_SUBTASKS)
       st = 0
       part_number = 1
       tasks = []
       block_num = BLOCK_NUM_INTERNAL if '-internal.aliyuncs.com' in oss_endpoint else BLOCK_NUM
       block_num = min(block_num, divide(divide(total_size, BLOCK_SIZE), MAX_SUBTASKS + 1))
       while st < total_size:
           en = min(total_size - 1, st + block_num * BLOCK_SIZE - 1)
           tasks.append((oss_object_name, upload_id, part_number, url, st, en, context))
           size = en - st + 1
           cnt = divide(size, BLOCK_SIZE)
           part_number += cnt
           st = en + 1
       subtasks = pool.map(invoke_subtask, tasks)
       pool.close()
       pool.join()
       parts = []
       for it in subtasks:
           for part in json.loads(it):
               parts.append(PartInfo(part[0], part[1]))
       res = retry(lambda : bucket.complete_multipart_upload(oss_object_name, upload_id, parts))
       return str(res.crc), str(crc64)
   def get_oss_object_name(url):
       """
       OSS オブジェクト名を取得します。
       :param url: (必須, string) ファイルの URL。
       :return: (string) OSS オブジェクト名。
       """
       prefix = os.environ.get('target_oss_prefix')
       tmps = url.split('?')
       if len(tmps) != 2:
           raise Exception('Invalid URL: %s' % url)
       urlObject = tmps[0]
       if urlObject.count('/') < 3:
           raise Exception('Invalid URL: %s' % url)
       objectParts = urlObject.split('/')
       objectParts = [prefix] + objectParts[len(objectParts) - 3 : len(objectParts)]
       return '/'.join(objectParts)
   def handler(event, context):
       evt = json.loads(event)
       if list(evt.keys()).count('is_children'):
           return json.dumps(do_subtask(evt, context))
       url = evt['events'][0]['eventParameter']['filePath']
       if not (url.startswith('http://') or url.startswith('https://')):
           url = 'https://' + url
       oss_object_name = get_oss_object_name(url)
       st_time = int(time.time())
       wait_time = SLEEP_TIME
       retry_cnt = 1
       while True:
           actual_crc64, expect_crc64 = migrate_file(url, oss_object_name, context)
           if actual_crc64 == expect_crc64:
               break
           print('Migrated object CRC64 does not match. Expected: %s, Actual: %s' % (expect_crc64, actual_crc64))
           if retry_cnt > MAX_RETRY_TIME:
               raise Exception('Maximum number of retries exceeded.')
           print('Retrying (attempt %d)...' % retry_cnt)
           time.sleep(wait_time)
           wait_time *= 2
           retry_cnt += 1
       print('Success! Total time: %d s.' % (int(time.time()) - st_time))
   

9. [コードのデプロイ] をクリックします。

3. カスタムロールとポリシーの作成

Function Compute の関数は、Object Storage Service (OSS) にアクセスするための権限が必要です。これらの権限は、カスタムロールを関数にアタッチすることで付与できます。必要なロールとポリシーを作成するには、次の手順に従います。

1. Resource Access Management (RAM) コンソールにログインします。左側のナビゲーションウィンドウで、[権限] > [ポリシー] を選択します。

2. [ポリシーの作成] をクリックし、[JSON] タブを選択します。

3. 次のポリシーで、BucketName をご利用のバケット名に置き換え、3つの FC-NAME をすべてステップ 2 の関数名 (推奨：cdn-log-dump) に置き換えます。

   {
     "Version": "1",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": "oss:PutObject",
         "Resource": "acs:oss:*:*:BucketName/*"
       },
       {
         "Effect": "Allow",
         "Action": "fc:InvokeFunction",
         "Resource": [
           "acs:fc:*:*:services/FC-NAME/functions/FC-NAME",
           "acs:fc:*:*:services/FC-NAME.*/functions/*"
         ]
       }
     ]
   }

4. OK をクリックし、[ポリシー名] と 備考 を入力します。その後、もう一度 OK をクリックしてポリシーを作成します (ポリシー名には AliyunCDNLogDumpAccess、説明には「CDN オフラインログダンプを管理する権限」を設定することを推奨します)。

5. 左側のナビゲーションウィンドウで、[ID] > [ロール] を選択し、[ロールの作成] をクリックします。

6. [信頼できるエンティティの選択] で [Alibaba Cloud サービス] を選択します。[信頼できるサービス] で [Function Compute] を選択し、OK をクリックします。

7. [ロールの作成] ダイアログボックスで、[ロール名] を入力し (推奨：AliyunCDNLogDumpRole)、OK をクリックしてロールを作成します。

8. [権限] タブで、[正確な権限付与] をクリックします。[ポリシータイプ] で カスタムポリシー を選択し、[ポリシー名] にステップ 4 で作成したポリシーの名前 (推奨：AliyunCDNLogDumpAccess) を入力します。その後、OK をクリックします。

9. [信頼ポリシー] タブで、[信頼ポリシーの編集] をクリックします。[JSON] エディターに次の信頼ポリシーを入力し、OK をクリックします。

   {
     "Statement": [
       {
         "Action": "sts:AssumeRole",
         "Effect": "Allow",
         "Principal": {
           "Service": [
             "fc.aliyuncs.com"
           ]
         }
       }
     ],
     "Version": "1"
   }

4. 関数へのロールのアタッチ

1. Function Compute コンソールの [関数] タブで、ステップ 2 で作成した関数を選択し、設定 をクリックします。

2. 関数詳細ページで、設定 タブをクリックします。高度設定 セクションで、設定 をクリックします。

3. [詳細設定] で、[権限] - [関数ロール] オプションを見つけ、ステップ 3 で作成したロールを選択します。AliyunCDNLogDumpRole の使用を推奨します。その後、[デプロイ] をクリックして、ロールを Function Compute タスクにバインドします。

5. 関数のテスト (任意)

これで設定は完了です。Alibaba Cloud CDN はオフラインログを約 24 時間の遅延で生成するため、設定が機能するかどうかをすぐに確認することはできません。すぐに関数をテストするには、次の任意の手順に従います。

1. Function Compute コンソールの [関数] タブで、ステップ 2 で作成した関数を選択し、設定 をクリックします。

2. 関数詳細ページで、テスト タブをクリックします。[テストイベント] で、[新しいテストイベントの作成] を選択します。[イベントテンプレート] で、[CDN (LogFileCreated)] を選択します。[イベント名] に [Test_cdn_log_dump] と入力します。

   JSON エディターには、CDN の LogFileCreated イベントテンプレートが自動的に入力されます。これには、eventName、eventSource、region、eventTime、traceId、userIdentity (aliUid)、resource (例：ドメインは example.com)、および eventParameter などのフィールドが含まれます。

3. 以下で取得したパラメーターを使用して、イベントテンプレートの filePath パラメーターを変更します。

   テスト用の filePath パラメーターの取得方法

   1. Alibaba Cloud CDN コンソールの [オフラインログのダウンロード] ページに移動します。

   2. トリガーに設定した高速化ドメイン名を選択し、昨日の日付を選択して、[クエリ] をクリックします。

   3. 結果からログファイルを見つけます。その ダウンロード ボタンにカーソルを合わせ、右クリックして [リンクをコピー] を選択します。

4. [関数のテスト] をクリックします。実行が完了すると、返される結果は null で、実行ステータスは成功になります。

5. Object Storage Service (OSS) コンソールで、Alibaba Cloud CDN ログの保存に使用するバケットを選択します。

6. [ファイル] をクリックし、Alibaba Cloud CDN ログを保存するように設定したディレクトリに移動します。高速化ドメイン名にちなんで名付けられたフォルダが表示されます。その中には、日付にちなんで名付けられたサブフォルダに、アーカイブされたログファイルが含まれています。これにより、Function Compute タスクが正常に実行されたことが確認できます。