Elastic Compute Service:インスタンスにログインせずにPythonでECSインスタンスを管理する

手順

1. 、インスタンスの構成と実行する操作に基づいて、シェル、バッチ、またはPowerShellコマンドをコンパイルします。

   サンプルコマンドの詳細については、「インスタンス設定の表示」および「インスタンス設定とアプリケーションのインストールの変更」をご参照ください。

2. 指定された要件を満たすインスタンスを検索します。

   インスタンスは実行中 (実行中) 状態である必要があります。 ECS SDK For Pythonを使用してインスタンスを照会する方法の詳細については、「ECSインスタンスの照会」をご参照ください。

3. アカウントのAccessKeyペアを取得し、リージョンIDを照会します。

   詳細については、「リージョンとゾーン」および「AccessKeyペアの作成」をご参照ください。

4. 1つ以上のECSインスタンスでCloud Assistantコマンドを実行します。

   サンプルコード：

   # 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()
   
   # Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are configured in the code runtime. 
   # If the project code is leaked, the AccessKey pair may be leaked and the security of resources within your account may be compromised. The following sample code shows how to use environment variables to obtain an AccessKey pair and use the AccessKey pair to call API operations. The sample code is for reference only. We recommend that you use Security Token Service (STS) tokens, which provide higher security.
   # Replace <RegionId> with the actual region ID. You can call the DescribeRegions operation to query the most recent region list. 
   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: the type of the command. Valid values: RunBatScript, RunPowerShellScript, and RunShellScript.
       cmdcontent: the content of the command.
       instance_id: the ID of the instance.
       """
       try:
           request = RunCommandRequest()
           request.set_accept_format('json')
   
           request.set_Type(cmdtype)
           request.set_CommandContent(cmdcontent)
           request.set_InstanceIds([instance_id])
           # Specify the execution timeout period of the command. Unit: seconds. Default value: 60. Specify this parameter based on the actual command.
           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():
       # Specify the type of the Cloud Assistant command.
       cmdtype = "RunShellScript"
       # Specify the content of the Cloud Assistant command.
       cmdcontent = """
       #!/bin/bash
       yum check-update
       """
       # Specify the timeout period.
       timeout = 60
       # Specify the ID of your instance.
       ins_id = "i-wz9bsqk9pa0d2oge****" 
       # Run the command.
       invoke_id = run_command(cmdtype,cmdcontent,ins_id,timeout)
       logger.info("run command,invoke-id:%s" % invoke_id)
   
       # Wait for the command to be run. Query the execution status of the command 10 times at an interval of 5 seconds. You can also specify the number of times the command execution status is queried and the query interval based on your business requirements.
       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()