CDN:Uso de la CLI de EdgeScript para gestionar scripts

Creación de los archivos del proyecto

1. Cree una carpeta para el proyecto; por ejemplo, cdn-api.

2. Dentro de la carpeta, cree un archivo llamado aliyun.ini para almacenar las credenciales de su usuario de RAM:

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

3. En la misma carpeta, cree un archivo llamado es.py y pegue el siguiente script de Python:

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

Configuración de credenciales

Ejecute el siguiente comando para almacenar su par de AccessKey en aliyun.ini:

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

Compruebe que las credenciales se guardaron correctamente:

cat aliyun.ini

Salida esperada:

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

Publicación de un script

Envíe un script al entorno de prueba para validarlo antes de ponerlo en producción:

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

Para enviar un script directamente al entorno de producción:

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

El parámetro rule admite los siguientes campos:

Puede especificar varios parámetros rule en un solo comando para enviar múltiples scripts a la vez.

Acerca de los ID de configuración

Un ID de configuración (configid) se genera automáticamente cuando crea un script para un nombre de dominio. No necesita proporcionar uno al crear un script. Para modificar un script existente, primero consúltelo para obtener su ID de configuración y luego incluya "configid":"<configid>" en el campo rule. Para más detalles, consulte Usage notes on ConfigId.

Consulta de scripts

Liste todos los scripts de EdgeScript para un nombre de dominio:

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

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

Eliminación de un script

Elimine un script mediante su ID de configuración. Ejecute una consulta primero si desconoce el ID de configuració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>

Promoción de scripts de prueba a producción

Después de validar los scripts en el entorno de prueba, promuévalos a producción:

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

Reversión de scripts

Revierta el entorno de prueba a su configuración anterior:

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

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

X-DEBUG-ES-TRACE-RULE-{Script ID}