CDN:[Function Compute console] オフラインログを OSS にアーカイブ

1. バケット情報の取得

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

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

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

1. Function Compute コンソールにログインします。左側メニューで、[Functions] を選択します。

2. [Functions] ページで [Create Function] をクリックし、[Event Function] を選択してから [Create Event Function] をクリックします。

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

   • [Basic Settings] - 関数名：この関数名は後の操作で必要です。cdn-log-dump の使用を推奨します。

   • [Function Code] > [Runtime]：この関数は Python で記述されています。[Official Runtimes]、[Python]、[Python 3.10] を選択します。

   • [Advanced Settings] > 環境変数：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. [Function Details] ページで、[Triggers] タブをクリックし、[Create Trigger] をクリックします。

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

   • [Trigger Type]：[Alibaba Cloud CDN (Sync)] を選択します。

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

   • [Event Name]：[LogFileCreated] を選択します。

   • ドメイン名：同じ Alibaba Cloud アカウント内にある有効なアクセラレーションドメイン名を入力します。

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

   • [Role]：[AliyunCDNEventNotificationRole] を選択します。

7. トリガーのパラメーターを設定した後、OK をクリックします。"No Default Role Is Created for CDN Trigger" というメッセージが表示された場合は、[Authorize Now] をクリックし、画面の指示に従ってデフォルトのロールを作成します。表示されない場合、トリガーはすぐに作成されます。

8. [Function Details] ページで [Code] タブをクリックし、オンラインエディターに次のコードを入力します。このコードは Alibaba Cloud CDN からオフラインログを取得し、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. [Deploy Code] をクリックします。

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

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

1. Resource Access Management (RAM) コンソールにログインします。左側メニューで、[Permissions] > [Policies] を選択します。

2. [Create Policy] をクリックし、[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 をクリックし、[Policy Name] と 備考 を入力します。その後、再度 OK をクリックして ポリシーを作成します (ポリシー名には AliyunCDNLogDumpAccess、説明には CDN オフラインログダンプを管理する権限 を設定することを推奨します) 。

5. 左側メニューで、[Identities] > [Roles] を選択し、[Create Role] をクリックします。

6. [Select Trusted Entity] で [Alibaba Cloud Service] を選択します。[信頼されたサービス] で [Function Compute] を選択し、OK をクリックします。

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

8. [Permissions] タブで、[Grant Permission] をクリックします。[Policy Type] で カスタムポリシー を選択し、[Policy Name] に先ほど作成したポリシーの名前 (推奨：AliyunCDNLogDumpAccess) を入力します。その後、OK をクリックします。

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

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

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

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

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

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

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

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

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

2. 関数の詳細ページで、テスト タブをクリックします。[Test Event] で [Create New Test Event] を選択します。[Event Template] で [CDN (LogFileCreated)] を選択します。[Event Name] に 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 コンソールの [Download Offline Logs] ページに移動します。

   2. トリガーに設定したアクセラレーションドメイン名を選択し、昨日の日付を選択して、[Query] をクリックします。

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

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

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

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