CDN:[Function Compute console] Archive offline logs to OSS

1. Obtain bucket details

The Function Compute function requires your OSS bucket details. Follow these steps to obtain the bucket name, the public access endpoint, and the log storage directory name.

2. Create the Function Compute function and trigger

In Function Compute, you will create a function to perform the archiving and a trigger to start it.

1. Log on to the Function Compute console. In the left-side navigation pane, choose Functions.

2. On the Functions page, click Create Function, select Event Function, and then click Create Event Function.

3. When creating the event function, configure the following key parameters:

   • Basic Configuration-Function Name: This function name is required for subsequent operations. We recommend that you use cdn-log-dump.

   • Function Code > Runtime: The function is written in Python. Select Official Runtimes, Python, and then Python 3.10.

   • Advanced Settings > Environment Variable: To pass the OSS bucket details to the function, create the following environment variables:

     • target_oss_bucket: Bucket name

     • target_oss_endpoint: The endpoint for public access

     • target_oss_prefix: The name of the directory for storing log files

     For example, set target_oss_bucket to demo-bucket, target_oss_endpoint to oss-cn-hangzhou.aliyuncs.com, and target_oss_prefix to cdn_log.

4. After you configure the parameters, click Create.

5. In Function Details, click the Triggers tab, and then click Create Trigger.

6. Configure the key parameters for the trigger as follows, and then click OK.

   • Trigger Type: Select Alibaba Cloud CDN (Sync invocation).

   • Name: Enter a name for the trigger (Recommended: cdn-logs-triggers).

   • Event Name: Select LogFileCreated.

   • Domain Name: Enter an active accelerated domain name that is under the same Alibaba Cloud account.

   • Description: Enter a description for the trigger. We recommend that you use "CDN offline log file generation trigger".

   • Role: Select AliyunCDNEventNotificationRole.

7. After you configure the trigger parameters, click OK. If the No Default Role Is Created for CDN Trigger message appears, click Authorize Now and follow the on-screen instructions to create the default role. Otherwise, the trigger is created immediately.

8. In Function Details, click the Code tab, and in the online editor, enter the code below, which retrieves offline logs from Alibaba Cloud CDN and stores them in Object Storage Service (OSS).

   Archive task code

   # -*- 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 # The number of worker processes in each subtask
   BLOCK_SIZE = 6 * 1024 * 1024 # The size of each part
   BLOCK_NUM_INTERNAL = 18 # The default number of blocks in each subtask in case of an internal URL
   BLOCK_NUM = 10 # The default number of blocks in each subtask
   MAX_SUBTASKS = 49 # The number of worker processes to perform subtasks
   CHUNK_SIZE = 8 * 1024 # The size of each chunk
   SLEEP_TIME = 0.1 # The initial seconds to wait before retrying
   MAX_RETRY_TIME = 10 # The maximum number of retries
   def retry(func):
       """
       An executor for a function with a retry mechanism.
       :param func: (required, lambda) The function to be executed.
       :return: The execution result of 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):
       """
       Get the CRC64 and total length of the file.
       :param url: (required, string) The URL of the file.
       :return: CRC64, length
       """
       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):
       """
       A response class that supports chunked reading.
       """
       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):
       """
       Download a part from a URL and upload it to OSS.
       :param args: (bucket, object_name, upload_id, part_number, url, st, en)
       :bucket: (required, Bucket) The destination OSS bucket.
       :object_name: (required, string) The destination object name.
       :upload_id: (required, integer) The upload ID of this upload task.
       :part_number: (integer) The part number of this part.
       :url: (required, string) The URL of the file.
       :st, en: (required, integer) The byte range of this part, denoting [st, en].
       :return: (part_number, etag)
       :part_number: (integer) The part number of this part.
       :etag: (string) The etag of the upload_part result.
       """
       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):
       """
       Download a range of the file from a URL and upload it to OSS.
       :param event: (required, json) The event in JSON format.
       :param context: (required, FCContext) The context of the handler.
       :return: parts
       :parts: ([(integer, string)]) The part number and etag of each process.
       """
       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):
       """
       Synchronously invoke the same function to start a subtask.
       :param args: (object_name, upload_id, part_number, url, st, en, context)
       :object_name: (required, string) The destination object name.
       :upload_id: (required, integer) The upload ID of this upload task.
       :part_number: (integer) The part number of the first part in this subtask.
       :url: (required, string) The URL of the file.
       :st, en: (required, integer) The byte range of this subtask, denoting [st, en].
       :context: (required, FCContext) The context of the handler.
       :return: The return value of the invoked function.
       """
       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):
       """
       Calculate ceil(n / m) without floating point arithmetic.
       :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):
       """
       Download the file from a URL and upload it to OSS.
       :param url: (required, string) The URL of the file.
       :param oss_object_name: (required, string) The destination object name.
       :param context: (required, FCContext) The context of the handler.
       :return: actual_crc64, expect_crc64
       :actual_crc64: (string) The CRC64 of the uploaded file.
       :expect_crc64: (string) The CRC64 of the source file.
       """
       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):
       """
       Get the OSS object name.
       :param url: (required, string) The URL of the file.
       :return: (string) The OSS object name.
       """
       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. Click Deploy Code.

3. Create a custom role and policy

The Function Compute function needs permissions to access Object Storage Service (OSS). You can grant these permissions by attaching a custom role to the function. Follow these steps to create the required role and policy.

1. Log on to the Resource Access Management (RAM) console. In the left-side navigation pane, choose Permissions > Policies.

2. Click Create Policy and select the JSON tab.

3. In the following policy, replace BucketName with your bucket name, and replace all three occurrences of FC-NAME with the function name from Step 2 (we recommend that you use 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. Click OK and enter a Policy Name and a Remarks. Then, click OK again to create the policy (We recommend that you set Policy Name to AliyunCDNLogDumpAccess and Description to Permissions to manage CDN offline log dumps).

5. In the left-side navigation pane, choose Identities > Roles and click Create Role.

6. For Select Trusted Entity, select Alibaba Cloud Service. For Trusted Service, select Function Compute, and then click OK.

7. In the Create Role dialog box, enter a Role Name (we recommend that you use AliyunCDNLogDumpRole) and click OK to create the role.

8. On the Permissions tab, click Precise Authorization. For Policy Type, select Custom Policies, and for Policy Name, enter the name of the policy that you created in Step 4 (we recommend that you use AliyunCDNLogDumpAccess). Then, click OK.

9. On the Trust Policy tab, click Edit Trust Policy. In the JSON editor, enter the following trust policy, and then click OK.

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

4. Attach the role to the function

1. On the Functions tab in the Function Compute console, select the function that you created in Step 2, and click Configure.

2. On the function details page, click the Configure tab. In the Advanced Settings section, click Configure.

3. In Advanced Settings, find the Permissions - Function Role option and select the role that you created in Step 3. We recommend that you use AliyunCDNLogDumpRole. Then, click Deploy to bind the role to the Function Compute task.

5. Test the function (Optional)

The configuration is now complete. Because Alibaba Cloud CDN generates offline logs with a delay of about 24 hours, you cannot immediately verify that the setup works. To test the function right away, follow these optional steps.

1. On the Functions tab of the Function Compute console, select the function created in Step 2, and click Configure.

2. On the function details page, click the Test tab. For Test Event, select Create New Test Event. For Event Template, select CDN (LogFileCreated). For Event Name, enter Test_cdn_log_dump.

   The JSON editor is automatically populated with the CDN LogFileCreated event template, which includes fields such as eventName, eventSource, region, eventTime, traceId, userIdentity (aliUid), resource (for example, the domain is example.com), and eventParameter.

3. Use the parameters obtained below to modify the filePath parameter in the event template.

   How to obtain the filePath parameter for testing

   1. Go to the Offline Log Download page in the Alibaba Cloud CDN console.

   2. Select the accelerated domain name that you configured for the trigger, select yesterday's date, and then click Query.

   3. In the results, find a log file. Hover over its Download button, right-click, and select Copy Link.

4. Click Test Function. After the execution is complete, the returned result is null and the execution status is successful.

5. In the Object Storage Service (OSS) console, select the bucket that is used to store the Alibaba Cloud CDN logs.

6. Click Files and navigate to the directory that you configured to store the Alibaba Cloud CDN logs. You will see a folder named after the accelerated domain name. Inside it, a subfolder named after the date contains the archived log file. This confirms that the Function Compute task ran successfully.