CDN:[Consola de Function Compute] Archivar registros sin conexión en OSS

1. Obtener los detalles del bucket

Obtenga los detalles de su bucket de OSS: el nombre del bucket, el endpoint de acceso público y el nombre del directorio de almacenamiento de registros.

2. Crear la función y el trigger en Function Compute

Cree una función de Function Compute para realizar el archivado y un trigger para iniciarla.

1. Inicie sesión en la consola de Function Compute. En el panel de navegación izquierdo, elija Functions.

2. En la página Functions, haga clic en Create Function, seleccione Event Function y luego haga clic en Create Event Function.

3. Al crear la función de evento, configure los siguientes parámetros clave:

   • Basic Configurations-Function Name: este nombre es obligatorio para las operaciones posteriores. Se recomienda utilizar cdn-log-dump.

   • Function Code > Runtime: la función está escrita en Python. Seleccione Official Runtimes, Python y luego Python 3.10.

   • Advanced Settings > Environment Variable: para pasar los detalles del bucket de OSS a la función, cree las siguientes variables de entorno:

     • target_oss_bucket: nombre del bucket

     • target_oss_endpoint: endpoint de acceso público

     • target_oss_prefix: nombre del directorio para almacenar los archivos de registro

     Por ejemplo, establezca target_oss_bucket en demo-bucket, target_oss_endpoint en oss-cn-hangzhou.aliyuncs.com y target_oss_prefix en cdn_log.

4. Después de configurar los parámetros, haga clic en Create.

5. En Function Details, haga clic en la pestaña Triggers y luego en Create Trigger.

6. Configure los parámetros clave del trigger de la siguiente manera y haga clic en OK.

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

   • Name: introduzca un nombre para el trigger (se recomienda cdn-logs-triggers).

   • Trigger Event: seleccione LogFileCreated.

   • Domain Name: introduzca un nombre de dominio acelerado activo que pertenezca a la misma cuenta de Alibaba Cloud.

   • Description: introduzca una descripción para el trigger. Se recomienda usar "CDN offline log file generation trigger".

   • Role: seleccione AliyunCDNEventNotificationRole.

7. Una vez configurados los parámetros del trigger, haga clic en OK. Si aparece el mensaje No Default Role Is Created for CDN Trigger, haga clic en Authorize Now y siga las instrucciones en pantalla para crear el rol predeterminado. De lo contrario, el trigger se crea de inmediato.

8. En Function Details, haga clic en la pestaña Code. En el editor en línea, introduzca el siguiente código, el cual recupera los registros sin conexión de Alibaba Cloud CDN y los almacena en Object Storage Service (OSS).

   Código de la tarea de archivado

   # -*- 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. Haga clic en Deploy.

3. Crear un rol y una política personalizados

La función de Function Compute necesita permisos para acceder a Object Storage Service (OSS). Otorgue estos permisos mediante la creación de un rol y una política personalizados.

1. Inicie sesión en la consola de Resource Access Management (RAM). En el panel de navegación izquierdo, elija Permissions > Policies.

2. Haga clic en Create Policy y seleccione la pestaña JSON.

3. En la siguiente política, reemplace BucketName por el nombre de su bucket y sustituya las tres apariciones de FC-NAME por el nombre de la función del paso 2 (se recomienda usar 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. Haga clic en OK e introduzca un Policy Name y un campo Remarks. Después, vuelva a hacer clic en OK para crear la política (se recomienda establecer Policy Name en AliyunCDNLogDumpAccess y Description en Permisos para gestionar volcados de registros sin conexión de CDN).

5. En el panel de navegación izquierdo, elija Identities > Roles y haga clic en Create Role.

6. En Select Trusted Entity, seleccione Alibaba Cloud Service. En Trusted Service, elija Function Compute y haga clic en OK.

7. En el cuadro de diálogo Create Role, introduzca un Role Name (se recomienda usar AliyunCDNLogDumpRole) y haga clic en OK para crear el rol.

8. En la pestaña Permissions, haga clic en Precise Authorization. En Policy Type, seleccione Custom Policies y, en Policy Name, introduzca el nombre de la política creada en el paso 4 (se recomienda usar AliyunCDNLogDumpAccess). Luego, haga clic en OK.

9. En la pestaña Trust Policy, haga clic en Edit Trust Policy. En el editor JSON, introduzca la siguiente política de confianza y haga clic en OK.

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

4. Asignar el rol a la función

1. En la pestaña Functions de la consola de Function Compute, seleccione la función creada en el paso 2 y haga clic en Configure.

2. En la página de detalles de la función, haga clic en la pestaña Configurations. En la sección Advanced Settings, haga clic en Configure.

3. Dentro de Advanced Settings, busque la opción Permissions - Function Role y seleccione el rol creado en el paso 3. Se recomienda utilizar AliyunCDNLogDumpRole. Después, haga clic en Deploy para vincular el rol a la tarea de Function Compute.

5. Probar la función (opcional)

Alibaba Cloud CDN genera los registros sin conexión con un retraso aproximado de 24 horas, por lo que no es posible comprobar la configuración de inmediato. Si desea probar la función al momento, siga estos pasos opcionales.

1. En la pestaña Functions de la consola de Function Compute, seleccione la función creada en el paso 2 y haga clic en Configure.

2. En la página de detalles de la función, haga clic en la pestaña Test. En Test Event, seleccione Create New Test Event. En Event Template, elija CDN (LogFileCreated). En Event Name, introduzca Test_cdn_log_dump.

   El editor JSON se llena automáticamente con la plantilla del evento LogFileCreated de CDN. Esta plantilla incluye campos como eventName, eventSource, region, eventTime, traceId, userIdentity (aliUid), resource (por ejemplo, si el dominio es example.com) y eventParameter.

3. Utilice los parámetros obtenidos a continuación para modificar el parámetro filePath de la plantilla del evento.

   Cómo obtener el parámetro **filePath** para las pruebas

   1. Vaya a la página Offline Log Download en la consola de Alibaba Cloud CDN.

   2. Seleccione el nombre de dominio acelerado que configuró para el trigger, elija la fecha de ayer y haga clic en Query.

   3. En los resultados, localice un archivo de registro. Pase el cursor sobre su botón Download, haga clic derecho y seleccione Copy Link.

4. Haga clic en Test Function. Cuando finalice la ejecución, el resultado devuelto será null y el estado de ejecución indicará que se completó correctamente.

5. En la consola de Object Storage Service (OSS), seleccione el bucket que utiliza para almacenar los registros de Alibaba Cloud CDN.

6. Haga clic en Files y vaya al directorio que configuró para almacenar los registros de Alibaba Cloud CDN. Verá una carpeta con el nombre del dominio acelerado. Dentro de ella, una subcarpeta con la fecha contendrá el archivo de registro archivado. Esto confirma que la tarea de Function Compute se ejecutó correctamente.