This script looks for an EBS in the region "eu-west-1" with a tag whose key is "Application" and the value is passed as an argument, creating a snapshot of this EBS. In the same way, it searches for a snapshot by tag and deletes everything except the last one.
An example of running to create a snapshot:
python3 main.py create artem-test-app
An example of running to delete old snapshots:
python3 main.py cleanup artem-test-app
main.py:
from __future__ import print_function
from datetime import datetime
import sys
import boto3
import botocore
import urllib.request
ec2 = boto3.client('ec2', region_name='eu-west-1')
def create_snapshot(app):
print("Creating snapshot for " + str(app))
# Get Instance ID
instances_id = []
response = ec2.describe_instances(
Filters = [
{
'Name': 'tag:Application',
'Values': [
app,
]
},
{
'Name': 'instance-state-name',
'Values': ['running']
}
]
)
for reservation in response['Reservations']:
for instance in reservation['Instances']:
instances_id.append(instance['InstanceId'])
# Get Volumes ID
volumes_id = []
for instance_id in instances_id:
response = ec2.describe_instance_attribute(InstanceId = instance_id, Attribute = 'blockDeviceMapping')
for BlockDeviceMappings in response['BlockDeviceMappings']:
volumes_id.append(BlockDeviceMappings['Ebs']['VolumeId'])
# Create Volume Snapshot
for volume_id in volumes_id:
date = datetime.now()
response = ec2.create_snapshot(
Description = app + " snapshot created by Jenkins at " + str(date),
VolumeId = volume_id,
TagSpecifications=[
{
'ResourceType': 'snapshot',
'Tags': [
{
'Key': 'Name',
'Value': app + "-snapshot"
},
]
},
]
)
try:
print("Waiting until snapshot will be created...")
snapshot_id = response['SnapshotId']
snapshot_complete_waiter = ec2.get_waiter('snapshot_completed')
snapshot_complete_waiter.wait(SnapshotIds=[snapshot_id], WaiterConfig = { 'Delay': 30, 'MaxAttempts': 120})
except botocore.exceptions.WaiterError as e:
print(e)
def cleanup_snapshot(app):
print("Clean up step.")
print("Looking for all snapshots regarding " + str(app) + " application.")
# Get snapshots
response = ec2.describe_snapshots(
Filters = [
{
'Name': 'tag:Name',
'Values': [
app + "-snapshot",
]
},
],
OwnerIds = [
'self',
]
)
sorted_snapshots = sorted(response['Snapshots'], key=lambda k: k['StartTime'])
# Clean up snapshots and keep only the latest snapshot
for snapshot in sorted_snapshots[:-1]:
print("Deleting snapshot: " + str(snapshot['SnapshotId']))
response = ec2.delete_snapshot(
SnapshotId = snapshot['SnapshotId']
)
def main():
if sys.argv[1:] and sys.argv[2:]:
action = sys.argv[1]
app = sys.argv[2]
if action == 'create':
create_snapshot(app)
elif action == 'cleanup':
cleanup_snapshot(app)
else:
print("Wrong action: " + str(action))
else:
print("This script for create and cleanup snapshots")
print("Usage : python3 " + sys.argv[0] + " {create|cleanup} " + "{application_tag}")
print("Example: python3 " + sys.argv[0] + " create " + "ebs-snapshot-check")
if __name__ == '__main__':
main()