云服务器 ECS：Python程序免登录管理ECS实例

操作步骤

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()