日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

通過CLI工具使用邊緣腳本

阿里云為您提供邊緣腳本CLI工具,可以將您本地編寫的邊緣腳本規則發布到模擬環境進行測試,并發布至生產環境,也可以對模擬環境和生產環境的邊緣腳本規則進行查詢、修改和刪除。本文為您介紹邊緣腳本CLI工具的使用方法。

在本地新建CLI工具腳本文件

  1. 在本地新建一個文件夾,例如命名為cdn-api

  2. 在文件夾內新建一個命名為aliyun.ini的文本文件,用于記錄阿里云RAM賬號的鑒權信息,文件內容如下:

    [Credentials]
    accesskeyid = $accesskeyid
    accesskeysecret = $accesskeysecret 
  3. 在文件夾內新建一個命名為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工具的使用說明

  1. 配置AccessKey ID和AccessKey Secret。

    $python ./es.py config --id=AK_ID --secret=AK_SECRET
    $cat aliyun.ini
    
    [Credentials]
    accesskeyid = 更新后AK
    accesskeysecret = 更新后AK Secret

  2. 發布邊緣腳本規則至模擬環境或生產環境。

    ./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。

  3. 查詢模擬環境或生產環境下的邊緣腳本規則。

    ./es.py action=query_test_env domain=<domain>
    ./es.py action=query_product_env domain=<domain>
  4. 刪除模擬環境或生產環境下的邊緣腳本規則。

    ./es.py action=del_test_env domain=<domain> configid=<configid>
    ./es.py action=del_product_env domain=<domain> configid=<configid>                  
  5. 模擬環境的邊緣腳本規則執行正式發布或回滾。

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

基于m3u8.es規則攔截所有.m3u8請求做示例,為您詳細介紹邊緣規則的編寫、保存、測試和發布的操作方法。具體操作,請參見示例:通過CLI工具使用邊緣腳本

實時調試方式

  1. 配置實時調試。

    您可以通過控制臺的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"}'
  2. 測試請求。

    測試時請求參數中攜帶_es_dbg參數,參數值為配置的option中_es_dbg的值,關注如下響應頭信息。

    Trace信息:X-DEBUG-ES-TRACE-RULE-{規則ID},查看對應規則的執行流,格式為_行號_函數名(入參):返回值{_執行耗時}