云助手命令可以在多台ECS实例中批量执行Shell、Bat或者PowerShell脚本。相较于SSH/RDP模式,您可以使用云助手通过Python开发环境直接为云端的ECS实例进行运维操作。
操作步骤
- 根据ECS实例的配置,以及您需要执行的操作,编写Shell、Bat或者PowerShell命令。
- 查询满足条件的ECS实例。
ECS实例状态必须为
运行中(
Running
)。查询实例的Python SDK示例请参见
查询ECS实例。
- 获取账号AccessKey,以及查询地域ID。
- 为一台或多台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')