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.
   
   import json
   import logging
   import os
   import time
   import datetime
   import base64
   from aliyunsdkcore import client
   from aliyunsdkecs.request.v20140526.CreateCommandRequest import CreateCommandRequest
   from aliyunsdkecs.request.v20140526.InvokeCommandRequest import InvokeCommandRequest
   from aliyunsdkecs.request.v20140526.DescribeInvocationResultsRequest import DescribeInvocationResultsRequest
   
   # Configure the log output formatter, if you want to save the output to file,
   # append ",filename='ecs_invoke.log'" after datefmt.
   
   logging.basicConfig(level=logging.INFO,
                       format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
                       datefmt='%a, %d %b %Y %H:%M:%S',filename='aliyun_assist_openapi_test.log', filemode='w')
   #access_key = 'Your AccessKey ID'
   #access_key_secret = 'Your AccessKey secret'
   #region_name = 'cn-hangzhou'
   #zone_id = 'cn-hangzhou-f'
   
   access_key = 'LTAIXXXXXXXXXXXX'  # 请根据实际情况填写
   access_key_secret = '4dZXXXXXXXXXXXXXXXXXXXXXXXX'  # 请根据实际情况填写
   region_name = 'cn-hangzhou'  # 请根据实际情况填写
   zone_id = 'cn-hangzhou-f'  # 请根据实际情况填写
   
   clt = client.AcsClient(access_key, access_key_secret, region_name)
   
   def create_command(command_content, type, name, description):
       request = CreateCommandRequest()
       request.set_CommandContent(command_content)
       request.set_Type(type)
       request.set_Name(name)
       request.set_Description(description)
       response = _send_request(request)
       if response is None:
           return None
       command_id = response.get('CommandId')
       return command_id;
   
   def invoke_command(instance_id, command_id, timed, cronat):
       request = InvokeCommandRequest()
       request.set_Timed(timed)
       InstanceIds = [instance_id]
       request.set_InstanceIds(InstanceIds)
       request.set_CommandId(command_id)
       request.set_Frequency(cronat)
       response = _send_request(request)
       invoke_id = response.get('InvokeId')
       return invoke_id;
   
   def get_task_output_by_id(instance_id, invoke_id):
       logging.info("Check instance %s invoke_id is %s", instance_id, invoke_id)
       request = DescribeInvocationResultsRequest()
       request.set_InstanceId(instance_id)
       request.set_InvokeId(invoke_id)
       response = _send_request(request)
       invoke_detail = None
       output = None
       if response is not None:
           result_list = response.get('Invocation').get('InvocationResults').get('InvocationResult')
           for item in result_list:
               invoke_detail = item
               output = base64.b64decode(item.get('Output'))
               break;
           return output;
   
   def execute_command(instance_id):
       command_str = 'yum check-update'
       command_id = create_command(base64.b64encode(command_str), 'RunShellScript', 'test', 'test')
       if(command_id is None):
           logging.info('create command failed')
           return
   
       invoke_id = invoke_command(instance_id, command_id, 'false', '')
       if(invoke_id is None):
           logging.info('invoke command failed')
           return
   
       time.sleep(15)
   
       output = get_task_output_by_id(instance_id, invoke_id)
       if(output is None):
           logging.info('get result failed')
           return
   
       logging.info("output: %s is \n", output)
   
   # Send API request.
   def _send_request(request):
       request.set_accept_format('json')
       try:
           response_str = clt.do_action(request)
           logging.info(response_str)
           response_detail = json.loads(response_str)
           return response_detail
       except Exception as e:
           logging.error(e)
   
   if __name__ == '__main__':
       execute_command('i-bp17zhpbXXXXXXXXXXXXX')