AWS EC2 and RDS daily snapshots status using BOTO3 Python Library

by Posted on

Applying filters to describe_snapshots or describe_db_snapshots functions and getting the desired results are really hard to implement. The best method for fetching snapshot status is fetching all the snapshots without a date filter and manipulate them with the python script itself using date functions. In such situations we might need to store and handle tokens for pagination when there is huge number of snapshots available for processing.

describe_db_snapshots return syntax.

Script for fetching today’s RDS snapshots with pagination

import datetime
from datetime import date
from time import sleep
snapdate = str(date.isoformat(date.today()))
client = boto3.client('rds')

isnext = None
while True:
  if isnext:
   response = client.describe_db_snapshots(Marker = isnext)
  else:
   response = client.describe_db_snapshots()

  snapshots = response['DBSnapshots']

  for snap in snapshots:
   snap_ids = snap['DBSnapshotIdentifier']
   snap_date = datetime.datetime.strftime(snap['SnapshotCreateTime'], '%Y-%m-%d')
   if snapdate == snap_date:
    msg=str(snap["SnapshotCreateTime"].strftime('%Y-%m-%d %H:%M:%S'))+"="+ snap["DBInstanceIdentifier"]+"="+snap["DBInstanceIdentifier"]+"="+snap["DBSnapshotIdentifier"]+"="+snap["Status"]+"="+str(snap["PercentProgress"])+"=dbsnapshot"
    print(msg)
    sleep(0.05)
    sleep(0.05)
  try:
   isnext = response['Marker']
  except KeyError:
   break

describe_snapshots return syntax

Script for fetching today’s EC2 snapshots with pagination (with Name Tags)

import boto3
import datetime
from datetime import date
import socket
from time import sleep
snapdate = str(date.isoformat(date.today()))
client = boto3.client('ec2')
isnext = None
i = 0
while True:
 if isnext:
  response = client.describe_snapshots(NextToken = isnext,OwnerIds=['xxxxx'])
 else:
  response = client.describe_snapshots(OwnerIds=['xxxxxx'])
 snapshots = response['Snapshots']

 for snap in snapshots:
  snap_date = datetime.datetime.strftime(snap['StartTime'], '%Y-%m-%d')
  if snapdate == snap_date:
   for tag in snap["Tags"]:
    if tag["Key"] == "Name":
     msg=str(snap["StartTime"].strftime('%Y-%m-%d %H:%M:%S'))+"="+ tag["Value"]+"="+snap["VolumeId"]+"="+snap["SnapshotId"]+"="+snap["State"]+"="+snap["Progress"]+"=ec2snapshot"
     print(msg)
     sleep(0.05)
     sleep(0.05)
 try:
  isnext = response['NextToken']
 except KeyError:
  break

 

Published by Sumesh MS