CDN:[console do Function Compute] Arquivar logs offline no OSS

1. Obter detalhes do bucket

Obtenha os detalhes do seu bucket do OSS: nome do bucket, endpoint de acesso público e nome do diretório de armazenamento de logs.

2. Criar a função e o gatilho no Function Compute

Crie uma função no Function Compute para realizar o arquivamento e um gatilho para iniciá-la.

1. Faça login no console do Function Compute. No painel de navegação à esquerda, escolha Functions.

2. Na página Functions, clique em Create Function, selecione Event Function e, em seguida, clique em Create Event Function.

3. Ao criar a função de evento, configure os seguintes parâmetros principais:

   • Basic Configurations - Function Name: Este nome é necessário para operações subsequentes. Recomendamos o uso de cdn-log-dump.

   • Function Code > Runtime: A função é escrita em Python. Selecione Official Runtimes, Python e, em seguida, Python 3.10.

   • Advanced Settings > Environment Variable: Para passar os detalhes do bucket do OSS para a função, crie as seguintes variáveis de ambiente:

     • target_oss_bucket: Nome do bucket

     • target_oss_endpoint: Endpoint de acesso público

     • target_oss_prefix: Nome do diretório para armazenamento de arquivos de log

     Por exemplo, defina target_oss_bucket como demo-bucket, target_oss_endpoint como oss-cn-hangzhou.aliyuncs.com e target_oss_prefix como cdn_log.

4. Após configurar os parâmetros, clique em Create.

5. Em Function Details, clique em na aba Triggers e, em seguida, clique em Create Trigger.

6. Configure os parâmetros principais do gatilho conforme descrito abaixo e clique em OK.

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

   • Name: Insira um nome para o gatilho (Recomendado: cdn-logs-triggers).

   • Trigger Event: Selecione LogFileCreated.

   • Domain Name: Insira um nome de domínio acelerado ativo que esteja na mesma conta Alibaba Cloud.

   • Description: Insira uma descrição para o gatilho. Recomendamos "Gatilho de geração de arquivo de log offline do CDN".

   • Role: Selecione AliyunCDNEventNotificationRole.

7. Após configurar os parâmetros do gatilho, clique em OK. Se a mensagem No Default Role Is Created for CDN Trigger aparecer, clique em Authorize Now e siga as instruções na tela para criar a função padrão. Caso contrário, o gatilho será criado imediatamente.

8. Em Function Details, clique em na aba Code e, no editor online, insira o código abaixo, que recupera logs offline do Alibaba Cloud CDN e os armazena no Object Storage Service (OSS).

   Código da tarefa de arquivamento

   # -*- 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. Clique em Deploy.

3. Criar uma função e política personalizadas

A função do Function Compute precisa de permissões para acessar o Object Storage Service (OSS). Conceda essas permissões criando uma função e uma política personalizadas.

1. Faça login no console do Resource Access Management (RAM). No painel de navegação à esquerda, escolha Permissions > Policies.

2. Clique em Create Policy e selecione a aba JSON.

3. Na política abaixo, substitua BucketName pelo nome do seu bucket e substitua todas as três ocorrências de FC-NAME pelo nome da função da Etapa 2 (recomendamos o uso de 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. Clique em OK e insira um Policy Name e Remarks. Em seguida, clique em OK novamente para criar a política (Recomendamos definir o Policy Name como AliyunCDNLogDumpAccess e a Description como Permissões para gerenciar dumps de logs offline do CDN).

5. No painel de navegação à esquerda, escolha Identities > Roles e clique em Create Role.

6. Em Select Trusted Entity, selecione Alibaba Cloud Service. Em Trusted Service, selecione Function Compute e clique em OK.

7. Na caixa de diálogo Create Role, insira um Role Name (recomendamos o uso de AliyunCDNLogDumpRole) e clique em OK para criar a função.

8. Na aba Permissions, clique em Precise Authorization. Em Policy Type, selecione Custom Policies e, em Policy Name, insira o nome da política criada na Etapa 4 (recomendamos o uso de AliyunCDNLogDumpAccess). Em seguida, clique em OK.

9. Na aba Trust Policy, clique em Edit Trust Policy. No editor JSON, insira a seguinte política de confiança e clique em OK.

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

4. Anexar a função à função do Function Compute

1. Na aba Functions do console do Function Compute, selecione a função criada na Etapa 2 e clique em Configure.

2. Na página de detalhes da função, clique em na aba Configurations. Na seção Advanced Settings, clique em Configure.

3. Em Advanced Settings, localize a opção Permissions - Function Role e selecione a função criada na Etapa 3. Recomendamos o uso de AliyunCDNLogDumpRole. Em seguida, clique em Deploy para vincular a função à tarefa do Function Compute.

5. Testar a função (Opcional)

O Alibaba Cloud CDN gera logs offline com um atraso de cerca de 24 horas, portanto, não é possível verificar a configuração imediatamente. Para testar a função agora mesmo, utilize as etapas opcionais a seguir.

1. Na aba Functions do console do Function Compute, selecione a função criada na Etapa 2 e clique em Configure.

2. Na página de detalhes da função, clique em na aba Test. Em Test Event, selecione Create New Test Event. Em Event Template, selecione CDN (LogFileCreated). Em Event Name, insira Test_cdn_log_dump.

   O editor JSON é preenchido automaticamente com o modelo de evento CDN LogFileCreated, que inclui campos como eventName, eventSource, region, eventTime, traceId, userIdentity (aliUid), resource (por exemplo, o domínio é example.com) e eventParameter.

3. Utilize os parâmetros obtidos abaixo para modificar o parâmetro filePath no modelo de evento.

   Como obter o parâmetro filePath para teste

   1. Acesse a página Offline Log Download no console do Alibaba Cloud CDN.

   2. Selecione o nome de domínio acelerado configurado para o gatilho, selecione a data de ontem e clique em Query.

   3. Nos resultados, localize um arquivo de log. Passe o mouse sobre o botão Download, clique em com o botão direito e selecione Copy Link.

4. Clique em Test Function. Após a conclusão da execução, o resultado retornado será null e o status da execução será bem-sucedido.

5. No console do Object Storage Service (OSS), selecione o bucket usado para armazenar os logs do Alibaba Cloud CDN.

6. Clique em Files e navegue até o diretório configurado para armazenar os logs do Alibaba Cloud CDN. Você verá uma pasta com o nome do domínio acelerado. Dentro dela, uma subpasta com a data contém o arquivo de log arquivado. Isso confirma que a tarefa do Function Compute foi executada com sucesso.