AWS Python automation Fetch_EC2

#Python #AWS #PythonAutomation

AWS Python automation Fetch_EC2

Overview

AWS introduced a boto library for Python, Which provides powerful automation capabilities. with boto library, you can do lots of things like ec2, VPC creation, etc; however, there are other powerful tools that provide infrastructure as a service like Terraform, which means Python Boto library is very powerful for query executer, fetching data, so forth... You can find the Boto documentation here

Brief

This example will play around with EC2 and will make a production example of how Python helps you to automate AWS resources. Steps of What We Gonna Do:

  • Fetch EC2 instances status

  • Schedule the script to run every 10 seconds

Code Discovery "fetches ec2 data"

#1

import boto3

ec2_client = boto3.client('ec2')  # 1

reservations = ec2_client.describe_instances()  # 2
for reservation in reservations['Reservations']:  # 3
    instances = reservation['Instances']  # 4
    # print(instances)  # use print to test the result value
    for instance in instances:  # 5
        # print(instance['State'])  # print the status list
        # print(instance['State']['Name'])  # select name from status list
        print(f"Instance: {instance['InstanceId']} is {instance['State']['Name']}")  # 6

Explanation

Open boto documentation/EC2 section Boto3

  • #1 Select EC2 class as appears in docs

  • #2 Select describe_instances function discover response section to view what the function has and the result value

  • #3 Select the 'Reservations' list and iterate in it. 4: Select 'Instances' Dictionary from 'Reservations'

  • #5 Iterate on 'Instances' values 6: Print the result by selecting the value of instance and ignore the key.

#2

Another way using the describe_instance_status function, This function has more capabilities, view the docs for more describe_instance_status Docs

import boto3

ec2_client = boto3.client('ec2')  # 1
statuses = ec2_client.describe_instance_status( IncludeAllInstances=True )  # 2

for status in statuses['InstanceStatuses']:  # 3
  ins_status = status['InstanceStatus']['Status']  # 4
  sys_status = status['SystemStatus']['Status']  # 5
  state = status['InstanceState']['Name']  # 6
  print(f"Instance: {status['InstanceId']} status is {state} with instance status {ins_status} and system "
          f"status is {sys_status}")

Explanation

Open boto documentation/EC2 section Boto3

  • #1 Select EC2 class as appears in docs

  • #2 Select describe_instance_status function, With getting all instance statue by default not include "not running states" describe_instance_status Docs

  • #3 Select the 'InstanceStatuses' list and iterate in it.

  • #4 Select the 'InstanceStatus' Dictionary and specify the Status value in an ins_status variable

  • #5 Select the 'SystemStatus' Dictionary and specify the Status value in a sys_status variable

  • #6 Select the 'InstanceState' Dictionary and specify the Name value in a state variable

Code Discovery fetch ec2 data, and schedule script

import boto3
import schedule

ec2_client = boto3.client('ec2')

def check_instance_status():
    statuses = ec2_client.describe_instance_status(IncludeAllInstances=True)

    for status in statuses['InstanceStatuses']:
        ins_status = status['InstanceStatus']['Status']
        sys_status = status['SystemStatus']['Status']
        state = status['InstanceState']['Name']
        print(f"Instance: {status['InstanceId']} status is {state} with instance status {ins_status} and system "
              f"status is {sys_status}")
        print("##############################\n")  # just a break

# schedule function
schedule.every(5).seconds.do(check_instance_status)  # 1

# while loop
while True:  # 2
    schedule.run_pending()

Explanation

  • In this example, we just created a function that has the last code that we tried out.

  • Start while loop with true value to make sure the script is working all the time.

  • Start the schedule every 10 sec, with selecting the check_instance_status function. Also, you can use the following:

    schedule.every(5).minutes.do(check_instance_status)
    schedule.every().hours
    schedule.every().day.at("1:00") # 1 AM
    schedule.every().monday.at("12:00")