You can use Cloud Assistant to simultaneously run scripts on multiple ECS instances. The scripts can be shell, bat, or PowerShell scripts. Compared with SSH or RDP, you can use Cloud Assistant to perform O&M operations directly on ECS instances in Python.

Prerequisites

Procedure

  1. Compile the shell, bat, or PowerShell scripts based on the instance configurations and the operations that you want to perform.
  2. Find ECS instances that meet the specified requirements.
    The target instances must be in the Running (Running) state. For more information about examples of SDK for Python, see Manage instances.
  3. Obtain the AccessKey pair of the account and query the region ID.
    For more information, see Regions and zones and Create an AccessKey.
  4. Run a Cloud Assistant script on one or more ECS instances.
    Example code:
    # 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 Access Key Id'
    #acess_key_secrect = 'Your Access Key Secrect'
    #region_name = 'cn-hangzhou'
    #zone_id = 'cn-hangzhou-f'
     access_key = 'LTAIXXXXXXXXXXXX'  # Enter your actual Accesskey ID. acess_key_secrect = '4dZXXXXXXXXXXXXXXXXXXXXXXXX'  # Enter your actual AccessKey secret.
    region_name = 'cn-hangzhou'  # Enter your actual region ID.
    zone_id = 'cn-hangzhou-f'  # Enter your actual zone ID.
     clt = client.AcsClient(access_key, acess_key_secrect, 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')