CDN：[Konsol FC] Gunakan Function Compute untuk menyimpan log offline di Object Storage Service (OSS)

1. Dapatkan konfigurasi bucket

Saat Anda membuat fungsi tugas di FC, Anda harus menyediakan informasi OSS untuk penyimpanan log. Oleh karena itu, Anda harus terlebih dahulu mendapatkan nama bucket, nilai Endpoint untuk akses internet, dan nama direktori untuk menyimpan file log. Ikuti langkah-langkah berikut untuk mengambil informasi ini:

2. Buat tugas Function Compute

Inti dari solusi penyimpanan otomatis ini adalah menggunakan FC sebagai penjadwal dan pelaksana. Oleh karena itu, Anda harus mengonfigurasi pemicu dan fungsi tugas yang sesuai di FC.

1. Pergi ke Konsol Function Compute 3.0. Di panel navigasi di sebelah kiri, pilih Functions.

2. Di tab Functions, klik Create Function, pilih Event Function, lalu klik Create Event Function.

3. Saat membuat fungsi peristiwa, konfigurasikan hanya parameter kunci yang memengaruhi eksekusi fungsi dengan benar.

   • Basic Configurations - Function Name: Anda akan menggunakan nama fungsi ini di langkah-langkah selanjutnya. Kami sarankan Anda menggunakan cdn-log-dump.

   • Function Code - Runtime Environment: Karena fungsi tugas adalah kode Python, pilih Built-in Runtime, Python, dan Python 3.10.

   • More Configurations - Environment Variables: Fungsi tugas perlu mendapatkan informasi bucket. Oleh karena itu, Anda harus melewati informasi konfigurasi bucket dalam variabel lingkungan. Buat tiga variabel lingkungan dan masukkan parameter yang sesuai:

     • target_oss_bucket: Nama Bucket

     • target_oss_endpoint: Nilai Endpoint untuk akses internet

     • target_oss_prefix: Nama Direktori untuk Menyimpan File Log

     

4. Setelah mengonfigurasi parameter, klik Create untuk membuat fungsi.

5. Di halaman Function Details, klik tab Triggers, lalu klik Create Trigger.

   

6. Ikuti petunjuk ini untuk mengonfigurasi parameter kunci untuk pemicu. Klik OK.

   • Trigger Type: Pilih CDN Synchronous Call.

   • Name: Masukkan nama pemicu. Kami sarankan Anda menggunakan cdn-logs-triggers.

   • Triggering Event: Pilih LogFileCreated.

   • Domain Name: Anda harus memasukkan nama domain yang dipercepat CDN yang berada di bawah Akun Alibaba Cloud yang sama dan berjalan dengan baik.

   • Description: Masukkan deskripsi untuk pemicu. Kami sarankan Anda menggunakan: Pemicu pembuatan file log offline CDN.

   • Role Name: Pilih AliyunCDNEventNotificationRole.

7. Setelah Anda mengonfigurasi parameter pemicu, klik OK. Jika pesan The Default Role For The CDN Trigger Has Not Been Created muncul, klik Authorize Now dan ikuti petunjuk untuk membuat peran default. Jika pesan ini tidak muncul, pemicu dibuat langsung.

   

8. Di halaman Function Details, klik tab Code. Di kompiler online, masukkan kode berikut untuk menarik log offline dari CDN dan menyimpannya di OSS.

   Kode Tugas Penyimpanan

   # 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 # Jumlah proses pekerja di setiap subtugas
   BLOCK_SIZE = 6 * 1024 * 1024 # Ukuran setiap bagian
   BLOCK_NUM_INTERNAL = 18 # Jumlah blok default di setiap subtugas jika menggunakan url internal
   BLOCK_NUM = 10 # Jumlah blok default di setiap subtugas
   MAX_SUBTASKS = 49 # Jumlah proses pekerja untuk melakukan subtugas
   CHUNK_SIZE = 8 * 1024 # Ukuran setiap chunk
   SLEEP_TIME = 0.1 # Detik awal untuk menunggu percobaan ulang
   MAX_RETRY_TIME = 10 # Jumlah percobaan ulang maksimum
   
   def retry(func):
       """
       Mengembalikan hasil dari fungsi lambda func dengan percobaan ulang.
       :param func: (wajib, lambda) fungsi.
       :return: Hasil dari 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('Mencoba ulang %d kali...' % retry_cnt)
           time.sleep(wait_time)
           wait_time *= 2
           retry_cnt += 1
   
   def get_info(url):
       """
       Mendapatkan CRC64 dan panjang total file.
       :param url: (wajib, string) alamat url file.
       :return: CRC64, panjang
       """
       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):
       """
       Kelas tanggapan untuk mendukung pembacaan berdasarkan chunk.
       """
       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):
       """
       Mengunduh satu bagian dari url lalu mengunggahnya ke OSS.
       :param args: (bucket, object_name, upload_id, part_number, url, st, en)
       :bucket: (wajib, Bucket) bucket OSS tujuan.
       :object_name: (wajib, string) object_name tujuan.
       :upload_id: (wajib, integer) upload_id dari tugas unggah ini.
       :part_number: (integer) part_number dari bagian ini.
       :url: (wajib, string) alamat url file.
       :st, en: (wajib, integer) rentang byte dari bagian ini, yang menunjukkan [st, en].
       :return: (part_number, etag)
       :part_number: (integer) part_number dari bagian ini.
       :etag: (string) etag dari hasil upload_part.
       """
       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):
       """
       Mengunduh rentang file dari url lalu mengunggahnya ke OSS.
       :param event: (wajib, json) format json dari event.
       :param context: (wajib, FCContext) konteks penanganan.
       :return: parts
       :parts: ([(integer, string)]) part_number dan etag dari setiap proses.
       """
       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):
       """
       Memanggil fungsi yang sama secara sinkron untuk memulai subtugas.
       :param args: (object_name, upload_id, part_number, url, st, en, context)
       :object_name: (wajib, string) object_name tujuan.
       :upload_id: (wajib, integer) upload_id dari tugas unggah ini.
       :part_number: (integer) part_number dari bagian pertama dalam subtugas ini.
       :url: (wajib, string) alamat url file.
       :st, en: (wajib, integer) rentang byte dari subtugas ini, yang menunjukkan [st, en].
       :context: (wajib, FCContext) konteks penanganan.
       :return: nilai kembali dari fungsi yang dipanggil.
       """
       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):
       """
       Menghitung ceil(n / m) tanpa aritmetika floating point.
       :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):
       """
       Mengunduh file dari url lalu mengunggahnya ke OSS.
       :param url: (wajib, string) alamat url file.
       :param oss_object_name: (wajib, string) object_name tujuan.
       :param context: (wajib, FCContext) konteks penanganan.
       :return: actual_crc64, expect_crc64
       :actual_crc64: (string) CRC64 dari unggahan.
       :expect_crc64: (string) CRC64 dari file sumber.
       """
       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):
       """
       Mendapatkan nama objek OSS.
       :param url: (wajib, string) alamat url file.
       :return: (string) nama objek OSS.
       """
       prefix = os.environ.get('target_oss_prefix')
       tmps = url.split('?')
       if len(tmps) != 2:
           raise Exception('Url tidak valid : %s' % url)
       urlObject = tmps[0]
       if urlObject.count('/') < 3:
           raise Exception('Url tidak valid : %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('CRC64 objek migrasi tidak cocok, diharapkan: %s, aktual: %s' % (expect_crc64, actual_crc64))
           if retry_cnt > MAX_RETRY_TIME:
               raise Exception('Waktu percobaan ulang maksimum terlampaui.')
           print('Mencoba ulang %d kali...' % retry_cnt)
           time.sleep(wait_time)
           wait_time *= 2
           retry_cnt += 1
       print('Berhasil! Total waktu: %d s.' % (int(time.time()) - st_time))

9. Klik Deploy Code untuk menyelesaikan konfigurasi fungsi.

3. Buat peran khusus dan kebijakan akses

FC memerlukan izin untuk mengakses OSS. Untuk menyederhanakan proses otorisasi, FC mendukung asosiasi peran. Ikuti langkah-langkah berikut untuk mengonfigurasi peran yang memungkinkan fungsi penyimpanan log offline mengakses OSS.

1. Buka konsol Resource Access Management (RAM). Di panel navigasi di sebelah kiri, pilih Permission Management > Policies.

2. Klik Create Policy. Di halaman Buat Kebijakan, klik tab Script.

3. Dalam kebijakan berikut, ganti BucketName dengan nama bucket Anda, dan ganti ketiga instance FC-NAME dengan nama fungsi dari Langkah 2. Kami sarankan Anda menggunakan 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. Klik OK. Masukkan Policy Name dan Description, lalu klik OK lagi untuk menyelesaikan proses Create Access Policy. (Kami sarankan menggunakan AliyunCDNLogDumpAccess sebagai Nama Kebijakan dan Mengelola izin untuk penyimpanan log CDN sebagai Deskripsi).

5. Di panel navigasi di sebelah kiri, pilih Identity Management > Roles. Di halaman Peran, klik Create Role.

6. Untuk Trusted Entity Type, pilih Alibaba Cloud Account. Untuk Trusted Alibaba Cloud Account, pilih Current Alibaba Cloud Account Xxxxxxx. Lalu, klik OK.

7. Di panel Create Role, masukkan Role Name. Kami sarankan Anda menggunakan AliyunCDNLogDumpRole. Lalu, klik OK untuk membuat peran.

8. Di halaman detail peran, klik tab Permissions, lalu klik Grant Permission. Atur Grant Scope ke Custom Policy. Untuk Policy Name, masukkan nama kebijakan yang Anda buat di Langkah 4. Kami sarankan Anda menggunakan AliyunCDNLogDumpAccess. Lalu, klik OK.

9. Klik tab Trust Policy, lalu klik Edit Trust Policy. Di editor Script, masukkan kebijakan kepercayaan berikut, lalu klik OK.

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

Pada titik ini, Anda telah menyelesaikan konfigurasi peran dan izin. Selanjutnya, Anda perlu mengikat peran ini ke tugas FC.

4. Ikat peran ke tugas Function Compute

1. Di konsol FC, pada tab Fungsi, pilih fungsi yang Anda buat di Langkah 2, lalu klik Configuration.

2. Di tab Configuration, pilih Advanced Configuration, lalu klik tombol Configure yang sesuai.

   

3. Dalam konfigurasi lanjutan, temukan opsi Izin - Peran Fungsi. Pilih peran yang Anda buat di Langkah 3. Kami sarankan Anda menggunakan AliyunCDNLogDumpRole. Lalu, klik Deploy untuk mengikat peran ke tugas FC.

   

5. Uji tugas Function Compute (opsional)

Setelah Anda menyelesaikan empat langkah pertama, seluruh proses untuk menyimpan log offline CDN di OSS telah dikonfigurasi. Namun, karena ada penundaan sekitar 24 jam dalam pembuatan log offline, Anda tidak dapat segera memverifikasi bahwa tugas FC yang dikonfigurasi berjalan dengan benar. Anda dapat menguji tugas FC yang dikonfigurasi dengan mengikuti langkah-langkah berikut.

1. Di konsol FC, pada tab Fungsi, pilih fungsi yang Anda buat di Langkah 2, lalu klik Configuration.

2. Di tab Test, untuk Test Request Operation, pilih Create New Test Event. Untuk Event Template, pilih CDN(LogFileCreated). Untuk Event Name, masukkan Test_cdn_log_dump.

   

3. Gunakan parameter yang diperoleh dalam langkah-langkah berikut untuk mengganti parameter filePath di template acara.

   Cara Mendapatkan Parameter filePath untuk Pengujian

   1. Pergi ke konsol Unduhan Log Offline CDN.

   2. Pilih nama domain yang dipercepat yang dikonfigurasi di pemicu. Pilih hari sebelum tanggal saat ini, lalu klik Query.

   3. Pilih file. Untuk menyalin tautan unduhan, arahkan kursor ke tombol Download, klik kanan, dan pilih Salin Tautan.

4. Klik Test Function. Setelah eksekusi selesai, Anda dapat melihat bahwa nilai kembali adalah null dan statusnya berhasil.

   

5. Di konsol Bucket OSS, pilih bucket yang digunakan untuk menyimpan log CDN.

6. Pilih File List dan pergi ke direktori yang dikonfigurasi untuk menyimpan log CDN. Anda dapat melihat folder bernama setelah nama domain yang dipercepat. Di dalam folder ini, ada subfolder bernama setelah tanggal, yang berisi file yang ditentukan dalam pengujian. Ini menunjukkan bahwa tugas FC telah berhasil memproses penyimpanan log offline CDN.