阿里云為您提供邊緣腳本CLI工具,可以將您本地編寫的邊緣腳本規則發布到模擬環境進行測試,并發布至生產環境,也可以對模擬環境和生產環境的邊緣腳本規則進行查詢、修改和刪除。本文為您介紹邊緣腳本CLI工具的使用方法。
在本地新建CLI工具腳本文件
在本地新建一個文件夾,例如命名為
cdn-api
。在文件夾內新建一個命名為
aliyun.ini
的文本文件,用于記錄阿里云RAM賬號的鑒權信息,文件內容如下:[Credentials] accesskeyid = $accesskeyid accesskeysecret = $accesskeysecret
在文件夾內新建一個命名為
es.py
的文本文件,用于記錄CLI工具的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)
CLI工具的使用說明
配置AccessKey ID和AccessKey Secret。
$python ./es.py config --id=AK_ID --secret=AK_SECRET $cat aliyun.ini [Credentials] accesskeyid = 更新后AK accesskeysecret = 更新后AK Secret
發布邊緣腳本規則至模擬環境或生產環境。
./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>"}'
規則字段說明,請參見下表:
字段名稱
是否必選
說明
enable
是
是否生效。取值:
on
off
pos
是
執行位置。取值:
head
foot
pri
是
執行優先級。不同執行位置的優先級各自獨立。取值范圍:0~999。
0表示優先級最高
999表示優先級最低。
rule
是
規則內容。
brk
否
本規則命中情況下,是否終止本階段剩余規則的執行。取值:
on
off
testip
否
測試客戶IP。默認為空。如果配置了該參數,則只有客戶端IP地址可觸發本條規則執行。
option
否
擴展項。當前支持擴展。
_es_dbg=signature
用于增加調試響應頭。說明為域名添加功能配置時會生成ConfigId(功能配置ID,唯一性),通過指定ConfigId可更新或刪除域名配置,通過ConfigId使用說明可以了解到ConfigId的生成、查詢和使用方法。
新增規則不需要指定configid。
修改規則需要指定configid,使用查詢接口可獲取到對應規則的configid。
您可以指定多條rule。
查詢模擬環境或生產環境下的邊緣腳本規則。
./es.py action=query_test_env domain=<domain> ./es.py action=query_product_env domain=<domain>
刪除模擬環境或生產環境下的邊緣腳本規則。
./es.py action=del_test_env domain=<domain> configid=<configid> ./es.py action=del_product_env domain=<domain> configid=<configid>
模擬環境的邊緣腳本規則執行正式發布或回滾。
./es.py action=publish_test_env domain=<domain> ./es.py action=rollback_test_env domain=<domain>
基于m3u8.es規則攔截所有.m3u8請求做示例,為您詳細介紹邊緣規則的編寫、保存、測試和發布的操作方法。具體操作,請參見示例:通過CLI工具使用邊緣腳本。
實時調試方式
配置實時調試。
您可以通過控制臺的WebIDE圖形化操作頁面進行
_es_dbg
配置,也可以使用如下命令進行配置。./es.py action=push_test_env domain=<domain> rule='{"pos":"<head|foot>","pri":"0-999","rule_path":"<the es code path>","enable":"<on|off>","configid":"<configid>", "option":"_es_dbg=123"}'
測試請求。
測試時請求參數中攜帶
_es_dbg
參數,參數值為配置的option中_es_dbg
的值,關注如下響應頭信息。Trace信息:
X-DEBUG-ES-TRACE-RULE-{規則ID}
,查看對應規則的執行流,格式為_行號_函數名(入參):返回值{_執行耗時}
。