refactor(ecs): Improve task definition deletion logic

ecs/delete_all_inactive_task_definitions.py

@@ -5,34 +5,73 @@
 # This script deletes all inactive task definitions in the ECS service in all AWS Regions.
 
 import boto3
+import sys
+import time  # Import time module for sleep
 
 
 def get_inactive_task_definition_arns(region):
     client = boto3.client("ecs", region_name=region)
-    response = client.list_task_definitions(status="INACTIVE")
-    return response.get("taskDefinitionArns", [])
+    arns = []
+    paginator = client.get_paginator("list_task_definitions")
+    for page in paginator.paginate(status="INACTIVE"):
+        arns.extend(page.get("taskDefinitionArns", []))
+    return arns
 
 
 def delete_task_definition(region, arn):
     client = boto3.client("ecs", region_name=region)
-    client.delete_task_definitions(taskDefinitions=[arn])
+    max_retries = 5
+    backoff = 1  # Initial backoff time in seconds
+    for attempt in range(1, max_retries + 1):
+        try:
+            client.delete_task_definitions(taskDefinitions=[arn])
+            print(f"Deleted task definition {arn}")
+            break  # Break the loop if deletion was successful
+        except client.exceptions.ClientException as e:
+            if "Throttling" in str(e):  # Check for throttling in the error message
+                print(f"Throttling exception when deleting {arn}: {e}, retrying in {backoff} seconds...")
+                time.sleep(backoff)
+                backoff *= 2  # Exponential backoff
+            else:
+                print(f"Client exception when deleting task definition {arn}: {e}")
+                break  # Break the loop for other client exceptions
+        except client.exceptions.ServerException as e:
+            if "Throttling" in str(e):  # Check for throttling in the error message
+                print(f"Throttling exception when deleting {arn}: {e}, retrying in {backoff} seconds...")
+                time.sleep(backoff)
+                backoff *= 2  # Exponential backoff
+            else:
+                print(f"Server exception when deleting task definition {arn}: {e}")
+                break  # Break the loop for other server exceptions
+        except Exception as e:
+            print(f"Unexpected error deleting task definition {arn}: {e}")
+            break  # Break the loop for any other unexpected errors
+
+
+def delete_inactive_task_definitions_in_region(region):
+    try:
+        arns = get_inactive_task_definition_arns(region)
+        if not arns:
+            print(f"No inactive task definitions found in region {region}")
+        else:
+            for arn in arns:
+                delete_task_definition(region, arn)
+    except Exception as e:
+        print(f"Error accessing region {region}: {e}")
 
 
 def delete_inactive_task_definitions_in_all_regions():
     ecs_regions = boto3.session.Session().get_available_regions("ecs")
     for region in ecs_regions:
-        try:
-            arns = get_inactive_task_definition_arns(region)
-            if not arns:
-                print(f"No inactive task definitions found in region {region}")
-            else:
-                for arn in arns:
-                    print(f"Deleting inactive task definition with ARN: {arn}")
-                    delete_task_definition(region, arn)
-        except Exception:
-            print(f"No access to region: {region}")
-            continue
+        delete_inactive_task_definitions_in_region(region)
 
 
 if __name__ == "__main__":
-    delete_inactive_task_definitions_in_all_regions()
+    if len(sys.argv) > 2:
+        print("Usage: python script.py [region]")
+        sys.exit(1)
+    elif len(sys.argv) == 2:
+        region = sys.argv[1]
+        delete_inactive_task_definitions_in_region(region)
+    else:
+        delete_inactive_task_definitions_in_all_regions()