Skip to content

Instantly share code, notes, and snippets.

@Johnsonkdavid
Created June 9, 2022 12:38
Show Gist options
  • Save Johnsonkdavid/35dbac44f132d5e120daa0b0c01d7981 to your computer and use it in GitHub Desktop.
Save Johnsonkdavid/35dbac44f132d5e120daa0b0c01d7981 to your computer and use it in GitHub Desktop.
Python code to clean up ECR images which is NOT using any ECS service, NOT comes in the top 10 latest images in the repo AND which is older than 30 days
import json
import boto3
from datetime import datetime, timedelta
from datetime import date
region=['eu-central-1','us-east-1']
current_date = date.today()
def lambda_handler(event, context):
ecs_used_images=[]
ecr_images=[]
old_images=[]
top_ten=[]
t2=[]
t3=[]
for regn in region:
client = boto3.client('ecs',region_name=regn)
client_ecr = boto3.client('ecr',region_name=regn)
repo_desc = client_ecr.describe_repositories()
repo_details=repo_desc['repositories']
for x in range(len(repo_details)):
sort_list = []
repo_name=repo_details[x]['repositoryName']
images_desc = client_ecr.describe_images(repositoryName=repo_name,maxResults=200)
image_details=images_desc['imageDetails']
for y in range(len(image_details)):
try:
image_tags=image_details[y]['imageTags']
ecr_images.extend(image_tags)
except:
continue
image_pushed_date=image_details[y]['imagePushedAt'].date()
time=image_details[y]['imagePushedAt'].strftime("%m/%d/%Y %H:%M:%S")
diff = current_date-image_pushed_date
day_diff = diff.days
sort_list.append((image_tags,time))
if day_diff >= 30:
old_images.extend(image_tags)
s_list_1=sorted(sort_list,key=lambda p: datetime.strptime(p[1], '%m/%d/%Y %H:%M:%S'), reverse=True)
sort_list_2=s_list_1[:10]
top_ten.extend(sort_list_2)
t1=(top_ten)
cluster_list = client.list_clusters()
for i in range(len(cluster_list['clusterArns'])):
cluster_arn=cluster_list['clusterArns'][i]
service_list = client.list_services(cluster=cluster_arn)
for j in range(len(service_list['serviceArns'])):
service_arn=service_list['serviceArns'][j]
service_desc = client.describe_services(cluster=cluster_arn,services=[service_arn])
for k in range(len(service_desc['services'])):
taskdef_arn=(service_desc['services'][k]['taskDefinition'])
taskdef_desc = client.describe_task_definition(taskDefinition=taskdef_arn)
image_details=taskdef_desc['taskDefinition']['containerDefinitions']
for l in range(len(image_details)):
try:
ecs_images=image_details[l]['image'].split(':')[1]
ecs_used_images.append(ecs_images)
except:
ecs_images=image_details[l]['image'].split('/')[1]
ecs_used_images.append(ecs_images)
un_used_images=(list(set(ecr_images) - set(ecs_used_images)))
old_images = [ element for element in old_images if element != 'latest']
deletable_images=(list(set(un_used_images) & set(old_images)))
t2= [ item[0] for item in top_ten]
t3 = [ item_0 for elem in t2 for item_0 in elem]
deletable_images_2=list(set(deletable_images) - set(t3))
ecr_cleanup(deletable_images_2)
def ecr_cleanup(deletable_images_2):
s=''
for regn_2 in region:
client_ecr_2 = boto3.client('ecr',region_name=regn_2)
repo_desc_2 = client_ecr_2.describe_repositories()
repo_details_2 = repo_desc_2['repositories']
for a in range(len(repo_details_2)):
repo_name_2=repo_details_2[a]['repositoryName']
images_desc_2 = client_ecr_2.describe_images(repositoryName=repo_name_2,maxResults=200)
image_details_2=images_desc_2['imageDetails']
for b in range(len(image_details_2)):
try:
image_tags_2=image_details_2[b]['imageTags']
except:
continue
temp = s.join(image_tags_2)
for deletable_images_list in deletable_images_2:
if temp == deletable_images_list:
print(deletable_images_list,repo_name_2)
image_delete = client_ecr_2.batch_delete_image(
repositoryName=repo_name_2,
imageIds=[
{
'imageTag': deletable_images_list
},
]
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment