如何使用云助手自动化管理实例
使用云助手自动化管理实例
运维 ECS 实例的目的是保持 ECS 实例的最佳状态以及确保排错的效率，但是手动维护会花费您大量的时间和精力，因此阿里云研制了 云助手，用以解决如何自动化、批量处理日常维护任务。本文举例如何使用云助手 API，为 ECS 实例执行相应命令，达到自动化运维 ECS 实例的目的。
命令类型介绍
目前，云助手支持如下三种命令类型。
前提条件
您需要确保目标 ECS 实例的网络类型为 专有网络（VPC）。
目标 ECS 实例的状态必须为 运行中（Running）。
目标 ECS 实例必须预先安装云助手客户端。您可以参阅 阿里云助手 安装并使用云助手客户端。
执行类型为 PowerShell 的命令时，您需要确保目标 Windows 实例已经配置了 PowerShell 模块。
以下示例在命令行工具中完成，您需要确保您已经安装了阿里云命令行工具 CLI（Command-Line Interface）。
Windows 实例参阅 在线安装命令行工具和 SDK。
Linux 实例参阅 在线安装命令行工具和 SDK。
您需要 升级 SDK。
修改 CLI 配置：
下载文件 aliyunOpenApiData.py。
使用下载的文件替换路径中 %python_install_path%\Lib\site-packages\aliyuncli 中的文件 aliyunOpenApiData.py。
关于如何配置阿里云 CLI，参阅文档 配置命令行工具和 SDK。
操作步骤
以下举例说明怎么在阿里云 CLI 中通过 API 使用云助手，为 ECS 实例执行相应命令。以执行一条 echo 123 命令为例。
在本地计算机的 CMD、PowerShell 或者 Shell 中运行 aliyuncli ecs CreateCommand --CommandContent ZWNobyAxMjM= --Type RunShellScript --Name test --Description test 创建命令（CreateCommand）。
运行 aliyuncli ecs InvokeCommand --InstanceIds your-vm-instance-id1 instance-id2 --CommandId your-command-id --Timed false 执行命令（InvokeCommand）。
注意：
InstanceIds 为您的 ECS 实例 ID，支持多台 ECS 实例，最多 100 台。
Timed 表示是否为周期性任务，Timed True 表示是周期性任务，Timed False表示不是周期性任务。
当您的任务为周期性任务时，即参数 Timed 取值为 True 时，您需要通过参数 Frequency 指定周期，例如 0 */20 * * * * 表示周期为每 20 分钟。更多关于 Cron 表达式详情，请参阅 Cron 表达式取值说明。
返回结果为所有的目标 ECS 实例返回一个共同的 InvokeId。您可以使用该 InvokeId 查询命令的执行情况。
（可选）运行 aliyuncli ecs DescribeInvocations --InstanceId your-vm-instance-id --InvokeId your-invoke-id 查看命令执行状态（DescribeInvocations）。其中，InvokeId 是 第二步 为 ECS 实例执行命令时返回的执行 ID。
返回参数 InvokeStatus 为 Finished 时仅表示命令进程 执行完成，不代表一定有预期的命令效果，您需要通过 DescribeInvocationResults 中的参数 Output 查看实际的具体执行结果。
（可选）运行 aliyuncli ecs DescribeInvocationResults --InstanceId your-vm-instance-id --InvokeId your-invoke-id 查看指定 ECS 实例的命令的实际执行结果（DescribeInvocationResults）。其中，InvokeId 是 第二步 为 ECS 实例执行命令时返回的执行 ID。
在 创建命令（CreateCommand） 时，您还可以为命令设置如下请求参数。
通过 Python SDK 使用云助手的完整代码示例
您也可以通过 阿里云 SDK 使用云助手。关于如何配置阿里云 SDK，参阅文档 配置命令行工具和 SDK。以下为通过 Python SDK 使用云助手的完整代码示例。
# 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')
以上就是如何使用云助手自动化管理实例的详细内容，更多请关注其它相关文章！