Best Practice on Testing System Event Processing Program for ECS Instances

ECS OpenAPIs for Convenient Testing

Simulated System Event Lifecycle

Sample Code

#  coding=utf-8# make sure the sdk version is 4.10.0 or upper, you can use command 'pip show aliyun-python-sdk-ecs' to check
# if the python sdk is not installed, use 'sudo pip install aliyun-python-sdk-ecs'
# if the python sdk is installed, use 'sudo pip install --upgrade aliyun-python-sdk-ecs'
import json
import logging
from aliyunsdkcore import client
from aliyunsdkecs.request.v20140526.CancelSimulatedSystemEventsRequest import CancelSimulatedSystemEventsRequest
from aliyunsdkecs.request.v20140526.CreateSimulatedSystemEventsRequest import CreateSimulatedSystemEventsRequest
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')
# your access key Id
ak_id = "YOU_ACCESS_KEY_ID"
# your access key secret
ak_secret = "YOU_ACCESS_SECRET"
region_id = "cn-beijing"
client = client.AcsClient(ak_id, ak_secret, region_id)
# send open api request
def _send_request(request):
request.set_accept_format('json')
try:
response_str = client.do_action_with_exception(request)
logging.info(response_str)
response_detail = json.loads(response_str)
return response_detail
except Exception as e:
logging.error(e)
def build_create_request(event_type, not_before, instance_ids):
request = CreateSimulatedSystemEventsRequest()
request.set_EventType(event_type)
request.set_NotBefore(not_before)
request.set_InstanceIds(instance_ids)
return request
def print_created_event_id(response):
error_code = response.get('Code')
if error_code is None:
event_id_list = response.get('EventIdSet').get('EventId')
print("Created %s simulated events: %s", len(event_id_list), event_id_list)
else:
print("Creating failed, error code: %s", error_code)
def get_created_event_id(response):
error_code = response.get('Code')
if error_code is None:
event_id_list = response.get('EventIdSet').get('EventId')
return event_id_list
else:
return []
def build_cancel_request(event_id):
request = CancelSimulatedSystemEventsRequest()
request.set_EventIds([event_id])
return request
if __name__ == '__main__':
request = build_create_request("SystemMaintenance.Reboot", "2018-09-01T00:00:00Z", ["i-2zegswzznxbp168sc5c9", "i-2zeimxypwhnj04sbgf5t"])
response = _send_request(request)
event_ids = get_created_event_id(response)
if event_ids:
print("Created %s simulated events: %s"%(len(event_ids), event_ids))
cancel_event_id = event_ids[0]
print("Now we cancel one event %s" % (cancel_event_id))
cancel_request = build_cancel_request(cancel_event_id)
cancel_response = _send_request(cancel_request)
if not cancel_response.get('Code'):
print("Cancel succeeded")
Created 2 simulated events: [u'e-2zec65b85gi9zwcv1kpz', u'e-2zec65b85gi9zwcv1kq0']
Wed, 22 Aug 2018 18:39:49 simulate_system_event.py[line:35] INFO {"EventIdSet":{"EventId":["e-2zec65b85gi9zwcv1kpz","e-2zec65b85gi9zwcv1kq0"]},"RequestId":"C1762464-CCC2-46EC-B233-92A4D9C1782C"}
Now we cancel one event e-2zec65b85gi9zwcv1kpz
Cancel succeeded
Wed, 22 Aug 2018 18:39:49 simulate_system_event.py[line:35] INFO {"RequestId":"44286901-1BC3-4BA0-AAAF-C3CF20578E0F"}

Constraints and Cautions

--

--

Follow me to keep abreast with the latest technology news, industry insights, and developer trends. Alibaba Cloud website:https://www.alibabacloud.com

Love podcasts or audiobooks? Learn on the go with our new app.

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store