CDN:EdgeScript CLI kullanarak betikleri yönetin

Proje dosyalarını oluşturun

1. Proje için bir klasör oluşturun, örneğin cdn-api.

2. Klasörün içinde, RAM kullanıcısı kimlik bilgilerinizi saklamak için aliyun.ini adlı bir dosya oluşturun:

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

3. Aynı klasörün içinde es.py adlı bir dosya oluşturun ve aşağıdaki Python betiğini yapıştırın:

   #!/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)

Kimlik bilgilerini yapılandırın

AccessKey çiftinizi aliyun.ini dosyasında saklamak için aşağıdaki komutu çalıştırın:

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

Kimlik bilgilerinin kaydedildiğini doğrulayın:

cat aliyun.ini

Beklenen çıktı:

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

Betik yayınlama

Canlıya geçmeden önce test etmek için bir betiği hazırlama ortamına gönderin:

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

Bir betiği doğrudan üretim ortamına göndermek için:

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

rule parametresi aşağıdaki alanları kabul eder:

Tek bir komutta birden çok rule parametresi belirterek birden fazla betiği aynı anda gönderebilirsiniz.

Yapılandırma kimlikleri hakkında

Bir alan adı için betik oluşturduğunuzda otomatik olarak bir yapılandırma kimliği (configid) üretilir. Betik oluştururken bunu sağlamanız gerekmez. Mevcut bir betiği değiştirmek için önce betiği sorgulayarak yapılandırma kimliğini alın, ardından rule alanına "configid":"<configid>" ekleyin. Ayrıntılar için bkz. ConfigId Kullanım Notları.

Betikleri sorgulama

Bir alan adı için tüm EdgeScript betiklerini listeleyin:

# Staging environment
./es.py action=query_test_env domain=<domain>

# Production environment
./es.py action=query_product_env domain=<domain>

Betik silme

Bir betiği yapılandırma kimliğine göre silin. Yapılandırma kimliğini bilmiyorsanız önce bir sorgu çalıştırın.

# Staging environment
./es.py action=del_test_env domain=<domain> configid=<configid>

# Production environment
./es.py action=del_product_env domain=<domain> configid=<configid>

Hazırlama betiklerini üretime taşıma

Betikleri hazırlama ortamında doğruladıktan sonra bunları üretime taşıyın:

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

Betikleri geri alma

Hazırlama ortamını önceki yapılandırmasına geri alın:

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