如何使用云助手自动化管理实例

# coding=utf-8

# if the python sdk is not install using 'sudo pip install aliyun-python-sdk-ecs'

# if the python sdk is install using 'sudo pip install --upgrade aliyun-python-sdk-ecs'

# make sure the sdk version is 2.1.2, you can use command 'pip show aliyun-python-sdk-ecs' to check

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

# configuration 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-shanghai'

#zone_id = 'cn-shanghai-b'

access_key = 'LTAIXXXXXXXXXXXX'

acess_key_secrect = '4dZXXXXXXXXXXXXXXXXXXXXXXXX'

region_name = 'cn-hangzhou'

zone_id = 'cn-hangzhou-f'

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 open 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')