云助手命令可以在多台ECS实例中批量执行Shell、Bat或者PowerShell脚本。相较于SSH/RDP模式,您可以使用云助手通过Python开发环境直接为云端的ECS实例进行运维操作。

前提条件

  • 目标ECS实例已安装云助手客户端。具体步骤,请参见云助手客户端
  • 已更新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.
    
    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')