CDN:Gunakan CLI EdgeScript untuk mengelola skrip

Buat file proyek

1. Buat folder untuk proyek, misalnya, cdn-api.

2. Di dalam folder tersebut, buat file bernama aliyun.ini untuk menyimpan kredensial Pengguna RAM Anda:

   [Credentials]
   accesskeyid = <your-access-key-id>
   accesskeysecret = <your-access-key-secret>

3. Di dalam folder yang sama, buat file bernama es.py dan tempel skrip Python berikut:

   #!/usr/bin/python
   # -*- coding:utf-8 -*-
   
   import sys, os
   import urllib, urllib2
   import base64
   import hmac
   import hashlib
   from hashlib import sha1
   import time
   import uuid
   import json
   from optparse import OptionParser
   import ConfigParser
   import traceback
   import urllib2
   
   access_key_id = '';
   access_key_secret = '';
   cdn_server_address = 'https://cdn.aliyuncs.com'
   CONFIGFILE = os.getcwd() + '/aliyun.ini'
   CONFIGSECTION = 'Credentials'
   cmdlist = '''
   
      1. Publish the ES rule to the simulated environment or production environment
         ./es.py action=push_test_env domain=<domain> rule='{"pos":"<head|foot>","pri":"0-999","rule_path":"<the es code path>","enable":"<on|off>"}'
         ./es.py action=push_product_env domain=<domain> rule='{"pos":"<head|foot>","pri":"0-999","rule_path":"<the es code path>","enable":"<on|off>","configid":"<configid>"}'
   
      2. Query the ES rule in the simulated environment or production environment
         ./es.py action=query_test_env domain=<domain>
         ./es.py action=query_product_env domain=<domain>
   
      3. Delete the ES rule in the simulated environment or production environment
         ./es.py action=del_test_env domain=<domain> configid=<configid>
         ./es.py action=del_product_env domain=<domain> configid=<configid>
   
      4. Publish the ES rule from the simulated to production environment, or Rollback the ES rule in the simulated environment
         ./es.py action=publish_test_env domain=<domain>
         ./es.py action=rollback_test_env domain=<domain>
   '''
   
   def percent_encode(str):
       res = urllib.quote(str.decode(sys.stdin.encoding).encode('utf8'), '')
       res = res.replace('+', '%20')
       res = res.replace('*', '%2A')
       res = res.replace('%7E', '~')
       return res
   
   def compute_signature(parameters, access_key_secret):
       sortedParameters = sorted(parameters.items(), key=lambda parameters: parameters[0])
   
       canonicalizedQueryString = ''
       for (k,v) in sortedParameters:
           canonicalizedQueryString += '&' + percent_encode(k) + '=' + percent_encode(v)
   
       stringToSign = 'GET&%2F&' + percent_encode(canonicalizedQueryString[1:])
   
       h = hmac.new(access_key_secret + "&", stringToSign, sha1)
       signature = base64.encodestring(h.digest()).strip()
       return signature
   
   def compose_url(user_params):
       timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
   
       parameters = { \
           'Format'        : 'JSON', \
           'Version'       : '2018-05-10', \
           'AccessKeyId'   : access_key_id, \
           'SignatureVersion'  : '1.0', \
           'SignatureMethod'   : 'HMAC-SHA1', \
           'SignatureNonce'    : str(uuid.uuid1()), \
           'Timestamp'         : timestamp, \
       }
   
       for key in user_params.keys():
           parameters[key] = user_params[key]
   
       signature = compute_signature(parameters, access_key_secret)
       parameters['Signature'] = signature
       url = cdn_server_address + "/?" + urllib.urlencode(parameters)
       return url
   
   def make_request(user_params, quiet=False):
       url = compose_url(user_params)
   
       try:
           req = urllib2.Request(url)
           r = urllib2.urlopen(req)
       except urllib2.HTTPError, err:
           print "Response Code:\n============="
           print err
           body=err.read()
           body_json = json.loads(body)
           body_str =json.dumps(body_json,indent=4)
           print "\nResponse Info:\n=============="
           print body_str
           return
   
       if r.getcode() == 200:
           print "Response Code:\n=============\n200 OK"
       print "\nResponse Info:\n=============="
       body=r.read()
       body_json = json.loads(body)
       body_str =json.dumps(body_json,indent=4)
       print body_str
   
   def configure_accesskeypair(args, options):
       if options.accesskeyid is None or options.accesskeysecret is None:
           print("config miss parameters, use --id=[accesskeyid] --secret=[accesskeysecret]")
           sys.exit(1)
       config = ConfigParser.RawConfigParser()
       config.add_section(CONFIGSECTION)
       config.set(CONFIGSECTION, 'accesskeyid', options.accesskeyid)
       config.set(CONFIGSECTION, 'accesskeysecret', options.accesskeysecret)
       cfgfile = open(CONFIGFILE, 'w+')
       config.write(cfgfile)
       cfgfile.close()
   
   def setup_credentials():
       config = ConfigParser.ConfigParser()
       try:
           config.read(CONFIGFILE)
           global access_key_id
           global access_key_secret
           access_key_id = config.get(CONFIGSECTION, 'accesskeyid')
           access_key_secret = config.get(CONFIGSECTION, 'accesskeysecret')
       except Exception, e:
           print traceback.format_exc()
           print("can't get access key pair, use config --id=[accesskeyid] --secret=[accesskeysecret] to setup")
           sys.exit(1)
   
   def parse_args(user_params):
       req_args = {}
   
       if user_params['action'] == 'push_test_env' or user_params['action'] == 'push_product_env':
           if not user_params.has_key('domain') or not user_params.has_key('rule'):
               parser.print_help()
               sys.exit(0)
   
           data = []
           for rule in user_params['rule']:
               rule_cfg = {
                   'functionId' : 180,
                   'functionName' : 'edge_function',
                   'functionArgs' : []
               }
               for k in rule:
                   arg_cfg = {}
                   if k == 'configid':
                       rule_cfg['configId'] = int(rule[k])
                   elif k == 'rule_path':
                       try:
                           f = open(rule[k], "r")
                           code = f.read()
                       except IOError:
                           print "io error"
                           sys.exit(0)
                       else:
                           f.close()
                       arg_cfg['argName'] = 'rule'
                       arg_cfg['argValue'] = code
                       rule_cfg['functionArgs'].append(arg_cfg)
                   else:
                       arg_cfg['argName'] = k
                       arg_cfg['argValue'] = rule[k]
                       rule_cfg['functionArgs'].append(arg_cfg)
               data.append(rule_cfg)
           rule_str = json.dumps(data)
   
           if user_params['action'] == 'push_test_env':
               req_args = {'Action':'SetCdnDomainStagingConfig', 'DomainName':user_params['domain'], 'Functions':rule_str}
           else:
               req_args = {'Action':'BatchSetCdnDomainConfig', 'DomainNames':user_params['domain'], 'Functions':rule_str}
   
       elif user_params['action'] == 'query_test_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DescribeCdnDomainStagingConfig', 'DomainName':user_params['domain'], 'FunctionNames':'edge_function'}
   
       elif user_params['action'] == 'query_product_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DescribeCdnDomainConfigs', 'DomainName':user_params['domain'], 'FunctionNames':'edge_function'}
   
       elif user_params['action'] == 'del_test_env':
           if not user_params.has_key('domain') or not user_params.has_key('configid'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DeleteSpecificStagingConfig', 'DomainName':user_params['domain'], 'ConfigId':user_params['configid']}
   
       elif user_params['action'] == 'del_product_env':
           if not user_params.has_key('domain') or not user_params.has_key('configid'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DeleteSpecificConfig', 'DomainName':user_params['domain'], 'ConfigId':user_params['configid']}
   
       elif user_params['action'] == 'publish_test_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'PublishStagingConfigToProduction', 'DomainName':user_params['domain'], 'FunctionName':'edge_function'}
   
       elif user_params['action'] == 'rollback_test_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'RollbackStagingConfig', 'DomainName':user_params['domain'], 'FunctionName':'edge_function'}
   
       else:
           parser.print_help()
           sys.exit(0)
   
       return req_args
   
   if __name__ == '__main__':
       parser = OptionParser("%s Action=action Param1=Value1 Param2=Value2 %s\n" % (sys.argv[0], cmdlist))
       parser.add_option("-i", "--id", dest="accesskeyid", help="specify access key id")
       parser.add_option("-s", "--secret", dest="accesskeysecret", help="specify access key secret")
   
       (options, args) = parser.parse_args()
       if len(args) < 1:
           parser.print_help()
           sys.exit(0)
   
       if args[0] == 'help':
           parser.print_help()
           sys.exit(0)
       if args[0] != 'config':
           setup_credentials()
       else: #it's a configure id/secret command
           configure_accesskeypair(args, options)
           sys.exit(0)
   
       user_params = {}
       idx = 1
       if sys.argv[1].lower().startswith('action='):
           _, value = sys.argv[1].split('=')
           user_params['action'] = value
           idx = 2
       else:
           parser.print_help()
           sys.exit(0)
   
       for arg in sys.argv[idx:]:
           try:
               key, value = arg.split('=', 1)
               if key == 'rule': # push_test_env / push_product_env
                   if not user_params.has_key('rule'):
                       user_params['rule'] = []
                   user_params['rule'].append(json.loads(value))
               else:
                   user_params[key.strip()] = value
           except ValueError, e:
               print(e.read().strip())
               raise SystemExit(e)
   
       req_args = parse_args(user_params)
       make_request(req_args)

Konfigurasi kredensial

Jalankan perintah berikut untuk menyimpan pasangan AccessKey Anda di aliyun.ini:

python ./es.py config --id=<your-access-key-id> --secret=<your-access-key-secret>

Verifikasi bahwa kredensial telah tersimpan:

cat aliyun.ini

Output yang diharapkan:

[Credentials]
accesskeyid = <your-access-key-id>
accesskeysecret = <your-access-key-secret>

Terbitkan skrip

Dorong skrip ke lingkungan staging untuk mengujinya sebelum dipublikasikan:

./es.py action=push_test_env domain=<domain> rule='{"pos":"head","pri":"10","rule_path":"./m3u8.es","enable":"on"}'

Untuk mendorong skrip langsung ke lingkungan produksi:

./es.py action=push_product_env domain=<domain> rule='{"pos":"head","pri":"10","rule_path":"./m3u8.es","enable":"on","configid":"12345"}'

Parameter rule menerima bidang-bidang berikut:

Anda dapat menentukan beberapa parameter rule dalam satu perintah untuk mendorong beberapa skrip sekaligus.

Tentang ID konfigurasi

ID konfigurasi (configid) dibuat secara otomatis saat Anda membuat skrip untuk nama domain. Anda tidak perlu menyediakannya saat membuat skrip. Untuk memodifikasi skrip yang sudah ada, lakukan kueri terlebih dahulu untuk mendapatkan ID konfigurasinya, lalu sertakan "configid":"<configid>" dalam bidang rule. Untuk detail selengkapnya, lihat Catatan penggunaan ConfigId.

Kueri skrip

Daftar semua skrip EdgeScript untuk nama domain:

# Lingkungan staging
./es.py action=query_test_env domain=<domain>

# Lingkungan produksi
./es.py action=query_product_env domain=<domain>

Hapus skrip

Hapus skrip berdasarkan ID konfigurasinya. Lakukan kueri terlebih dahulu jika Anda tidak mengetahui ID konfigurasinya.

# Lingkungan staging
./es.py action=del_test_env domain=<domain> configid=<configid>

# Lingkungan produksi
./es.py action=del_product_env domain=<domain> configid=<configid>

Tingkatkan skrip staging ke produksi

Setelah memvalidasi skrip di lingkungan staging, tingkatkan ke produksi:

./es.py action=rollback_test_env domain=<domain>

Kembalikan skrip

Kembalikan lingkungan staging ke konfigurasi sebelumnya:

./es.py action=rollback_test_env domain=<domain>