AWS Lambda to auto start stop Ec2 instance on schedule using python and boto3

AWS EC2 Start Stop with boto3

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.

aws ec2 auto start stop lambda


#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)


AI Unraveled: Demystifying Frequently Asked Questions on Artificial Intelligence (OpenAI, ChatGPT, Google Bard, Generative AI, Discriminative AI, xAI, LLMs, GPUs, Machine Learning, NLP, Promp Engineering)

#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 CLF-C02 book

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
)

Djamgatech: Build the skills that’ll drive your career into six figures: Get Djamgatech.

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’]}


Ace the Microsoft Azure Fundamentals AZ-900 Certification Exam: Pass the Azure Fundamentals Exam with Ease

return instancelist

error: Content is protected !!