全部產品
Search
文件中心

Elastic Compute Service:Python程式免登入管理ECS執行個體

更新時間:Jun 19, 2024

雲助手命令可以在多台ECS執行個體中批量執行Shell、Bat或者PowerShell指令碼。相對於通過SSH或者RDP遠程登入ECS執行個體進行營運操作,您可以使用雲助手免登入直接為ECS執行個體進行營運操作。本文介紹在Python開發環境中使用雲助手營運管理ECS執行個體。

前提條件

  • 目標ECS執行個體已安裝雲助手Agent。具體步驟,請參見安裝雲助手Agent

  • 已更新Python開發環境的SDK依賴aliyun-python-sdk-ecs至2.1.2或以上版本。更多詳情,請前往GitHub Repo Alibaba Cloud擷取最新版本。

操作步驟

  1. 根據ECS執行個體的配置,以及您需要執行的操作,編寫Shell、Bat或者PowerShell命令。

  2. 查詢滿足條件的ECS執行個體。

    ECS執行個體狀態必須為運行中Running)。查詢執行個體的Python SDK樣本請參見查詢ECS執行個體

  3. 擷取帳號的AccessKey,以及查詢地區ID。

    具體步驟,請參見地區和可用性區域建立AccessKey

  4. 為一台或多台ECS執行個體執行雲助手命令。

    範例程式碼如下:

    # coding=utf-8
    # If the Python sdk is not installed, run 'sudo pip install aliyun-python-sdk-ecs'.
    # Make sure you're using the latest sdk version.
    # Run 'sudo pip install --upgrade aliyun-python-sdk-ecs' to upgrade.
    
    from aliyunsdkcore.client import AcsClient
    from aliyunsdkcore.acs_exception.exceptions import ClientException
    from aliyunsdkcore.acs_exception.exceptions import ServerException
    from aliyunsdkecs.request.v20140526.RunCommandRequest import RunCommandRequest
    from aliyunsdkecs.request.v20140526.DescribeInvocationResultsRequest import DescribeInvocationResultsRequest
    import json
    import sys
    import base64
    import time
    import logging
    import os
    
    # Configure the log output formatter
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(name)s [%(levelname)s]: %(message)s",
                        datefmt='%m-%d %H:%M')
    
    logger = logging.getLogger()
    
    # 請確保代碼運行環境設定了環境變數 ALIBABA_CLOUD_ACCESS_KEY_ID 和 ALIBABA_CLOUD_ACCESS_KEY_SECRET。
    # 工程代碼泄露可能會導致 AccessKey 泄露,並威脅帳號下所有資源的安全性。以下程式碼範例使用環境變數擷取 AccessKey 的方式進行調用,僅供參考,建議使用更安全的STS方式
    # 將<RegionId>變數替換為您待查詢的ECS執行個體所屬的地區ID,阿里雲支援的地區資訊您可以調用DescribeRegions查看。
    access_key = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']             
    access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']  
    region_id = '<RegionId>'                  
    
    
    
    
    client = AcsClient(access_key, access_key_secret, region_id)
    
    def base64_decode(content, code='utf-8'):
        if sys.version_info.major == 2:
            return base64.b64decode(content)
        else:
            return base64.b64decode(content).decode(code)
    
    
    def get_invoke_result(invoke_id):
        request = DescribeInvocationResultsRequest()
        request.set_accept_format('json')
    
        request.set_InvokeId(invoke_id)
        response = client.do_action_with_exception(request)
        response_detail = json.loads(response)["Invocation"]["InvocationResults"]["InvocationResult"][0]
        status = response_detail.get("InvocationStatus","")
        output = base64_decode(response_detail.get("Output",""))
        return status,output
    
    
    def run_command(cmdtype,cmdcontent,instance_id,timeout=60):
        """
        cmdtype: 命令類型: RunBatScript;RunPowerShellScript;RunShellScript
        cmdcontent: 命令內容
        instance_id: 執行個體ID
        """
        try:
            request = RunCommandRequest()
            request.set_accept_format('json')
    
            request.set_Type(cmdtype)
            request.set_CommandContent(cmdcontent)
            request.set_InstanceIds([instance_id])
            # 執行命令的逾時時間,單位s,預設是60s,請根據執行的實際命令來設定
            request.set_Timeout(timeout)
            response = client.do_action_with_exception(request)
            invoke_id = json.loads(response).get("InvokeId")
            return invoke_id
        except Exception as e:
            logger.error("run command failed")
    
    
    
    def wait_invoke_finished_get_out(invoke_id,wait_count,wait_interval):
        for i in range(wait_count):
            status,output = get_invoke_result(invoke_id)
            if status not in ["Running","Pending","Stopping"]:
                return status,output
            time.sleep(wait_interval)
    
        logger.error("after wait %d times, still can not wait invoke-id %s finished")
        return "",""
    
    
    
    def run_task():
        # 設定雲助手命令的命令類型
        cmdtype = "RunShellScript"
        # 設定雲助手命令的命令內容
        cmdcontent = """
        #!/bin/bash
        yum check-update
        """
        # 設定逾時時間
        timeout = 60
        # 設定您的執行個體ID
        ins_id = "i-wz9bsqk9pa0d2oge****" 
        # 執行命令
        invoke_id = run_command(cmdtype,cmdcontent,ins_id,timeout)
        logger.info("run command,invoke-id:%s" % invoke_id)
    
        # 等待命令執行完成,迴圈查詢10次,每次間隔5秒,查詢次數和間隔請根據實際情況配置
        status,output = wait_invoke_finished_get_out(invoke_id,10,5)
        if status:
            logger.info("invoke-id execute finished,status: %s,output:%s" %(status,output))
    
    
    if __name__ == '__main__':
        run_task()