You can translate the content of this page by selecting a language in the select box.
AWS Lambda to auto start stop Ec2 instance on schedule using python and boto3
Use this lambda function to auto start stop all Ec2 instances based on schedule from tags.
#Auto Shutodown - Start EC2 instances based on tags
import boto3
import os
import json
import croniter
import datetime
# Enter the region your instances are in. Include only the region without specifying Availability Zone; e.g., 'us-east-1'
region = 'us-west-2'
EC2_STATUS_CODE_RUNNING = 16
EC2_STATUS_CODE_STOPPED = 80
def lambda_handler(event, context):
ec2 = boto3.client(‘ec2’, region_name=region)
#auto_start_stop_tag = ‘tc:uptime_schedule_gmt’
auto_start_tag = ‘tc:start_time_schedule_gmt_24h_format’
auto_shutdown_tag = ‘tc:shutdown_time_schedule_gmt_24h_format’
instances_to_shutdown = []
instances_to_start = []
# Query ec2 machines for auto_start_stop_tag,
#instances_with_schedules = get_instance_schedules(auto_start_stop_tag)
instances_with_start_schedules = get_instance_schedules(auto_start_tag)
instances_with_shutdown_schedules = get_instance_schedules(auto_shutdown_tag)
print(“instances_with_start_schedules: %s” % instances_with_start_schedules)
print(“instances_with_shutdown_schedules: %s” % instances_with_shutdown_schedules)
If you are looking for an all-in-one solution to help you prepare for the AWS Cloud Practitioner Certification Exam, look no further than this AWS Cloud Practitioner CCP CLFC01 book below.
for instance_id, values in instances_with_start_schedules.items():
now = datetime.datetime.now()
print(“now: %s” % now)
iterator = croniter.croniter(values[‘Schedule’], now)
next_run_time = iterator.get_next(datetime.datetime)
print(“next_run_time: %s” % next_run_time)
duration_until_next_run_time = next_run_time – now
print(“duration_until_next_start_time: %s” % duration_until_next_run_time)
duration_of_one_hour = datetime.timedelta(hours=1)
if duration_until_next_run_time <= duration_of_one_hour and values[‘State’][‘Code’] == EC2_STATUS_CODE_STOPPED: print(“true”) print(“instance_to_stop.append(%s)” % instance_id) instances_to_start.append(instance_id) for instance_id, values in instances_with_shutdown_schedules.items(): now = datetime.datetime.now() print(“now: %s” % now) iterator = croniter.croniter(values[‘Schedule’], now) next_run_time = iterator.get_next(datetime.datetime) print(“next_run_time: %s” % next_run_time) duration_until_next_run_time = next_run_time – now print(“duration_until_next_shutdown_time: %s” % duration_until_next_run_time) duration_of_one_hour = datetime.timedelta(hours=1) if duration_until_next_run_time <= duration_of_one_hour and values[‘State’][‘Code’] == EC2_STATUS_CODE_RUNNING: print(“instance_to_shutdown.append(%s)” % instance_id) instances_to_shutdown.append(instance_id) if len(instances_to_shutdown) > 0:
ec2.stop_instances(InstanceIds=instances_to_shutdown)
print(‘stopped your instances: ‘ + str(instances_to_shutdown))
send_shutdown_notification(instances_to_shutdown, “STOPPED”)
if len(instances_to_start) > 0:
ec2.start_instances(InstanceIds=instances_to_start)
print(‘started your instances: ‘ + str(instances_to_start))
send_start_notification(instances_to_start, “STARTED”)
def send_shutdown_notification(instances, event):
instances_json_object = {“instances”:instances, “event”:event}
instances_json_string = json.dumps(instances_json_object)
instances_json_bytes = instances_json_string.encode(‘utf-8’)
lambda_arn = os.environ[‘LAMBDA_NOTIFICATION_SHUTDOWN_ARN’]
lambda_client = boto3.client(“lambda”)
lambda_client.invoke(
FunctionName=lambda_arn,
InvocationType=’Event’,
LogType=’None’,
Payload=instances_json_bytes
)
def send_start_notification(instances, event):
instances_json_object = {“instances”:instances, “event”:event}
instances_json_string = json.dumps(instances_json_object)
instances_json_bytes = instances_json_string.encode(‘utf-8’)
lambda_arn = os.environ[‘LAMBDA_NOTIFICATION_START_ARN’]
lambda_client = boto3.client(“lambda”)
lambda_client.invoke(
FunctionName=lambda_arn,
InvocationType=’Event’,
LogType=’None’,
Payload=instances_json_bytes
)
def get_instance_schedules(tag_name):
# When passed a tag key, tag value this will return a list of InstanceIds that were found.
ec2client = boto3.client(‘ec2’)
response = ec2client.describe_instances(
Filters=[
{
‘Name’: ‘tag-key’,
‘Values’: [tag_name]
}
]
)
instancelist = {}
for reservation in (response[“Reservations”]):
for instance in reservation[“Instances”]:
tag_value = ”
for tag in instance[‘Tags’]:
if tag[‘Key’] == tag_name:
tag_value = tag[‘Value’]
break
instancelist[instance[“InstanceId”]] = {‘Schedule’:tag_value,’State’:instance[‘State’]}
return instancelist
