Forked from revolutionisme/net_interfaces_cleanup.py
Created
October 12, 2022 09:18
-
-
Save filipeandre/03066d9cc98c35dc4d2e8dad3908f3c4 to your computer and use it in GitHub Desktop.
Remove Network interfaces created by AWS Lambda while undeploying stack
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import boto3 | |
from botocore.exceptions import ClientError | |
from time import sleep | |
# Fix this wherever your custom resource handler code is | |
from common import cfn_custom_resources as csr | |
import sys | |
MAX_RETRIES = 3 | |
client = boto3.client('ec2') | |
def handler(event, context): | |
vpc_id = event['ResourceProperties']['VPCID'] | |
if not csr.__is_valid_event(event, context): | |
csr.send(event, context, FAILED, validate_response_data(result)) | |
return | |
elif event['RequestType'] == 'Create' or event['RequestType'] == 'Update': | |
result = {'result': 'Don\'t trigger the rest of the code'} | |
csr.send(event, context, csr.SUCCESS, csr.validate_response_data(result)) | |
return | |
try: | |
# Get all network intefaces for given vpc which are attached to a lambda function | |
interfaces = client.describe_network_interfaces( | |
Filters=[ | |
{ | |
'Name': 'description', | |
'Values': ['AWS Lambda VPC ENI*'] | |
}, | |
{ | |
'Name': 'vpc-id', | |
'Values': [vpc_id] | |
}, | |
], | |
) | |
failed_detach = list() | |
failed_delete = list() | |
# Detach the above found network interfaces | |
for interface in interfaces['NetworkInterfaces']: | |
detach_interface(failed_detach, interface) | |
# Try detach a second time and delete each simultaneously | |
for interface in interfaces['NetworkInterfaces']: | |
detach_and_delete_interface(failed_detach, failed_delete, interface) | |
if not failed_detach or not failed_delete: | |
result = {'result': 'Network interfaces detached and deleted successfully'} | |
csr.send(event, context, csr.SUCCESS, csr.validate_response_data(result)) | |
else: | |
result = {'result': 'Network interfaces couldn\'t be deleted completely'} | |
csr.send(event, context, csr.FAILED, csr.validate_response_data(result)) | |
# print(response) | |
except Exception: | |
print("Unexpected error:", sys.exc_info()) | |
result = {'result': 'Some error with the process of detaching and deleting the network interfaces'} | |
csr.send(event, context, csr.FAILED, csr.validate_response_data(result)) | |
def detach_interface(failed_detach, interface): | |
try: | |
if interface['Status'] == 'in-use': | |
detach_response = client.detach_network_interface( | |
AttachmentId=interface['Attachment']['AttachmentId'], | |
Force=True | |
) | |
# Sleep for 1 sec after every detachment | |
sleep(1) | |
print(f"Detach response for {interface['NetworkInterfaceId']}- {detach_response}") | |
if 'HTTPStatusCode' not in detach_response['ResponseMetadata'] or \ | |
detach_response['ResponseMetadata']['HTTPStatusCode'] != 200: | |
failed_detach.append(detach_response) | |
except ClientError as e: | |
print(f"Exception details - {sys.exc_info()}") | |
def detach_and_delete_interface(failed_detach, failed_delete, interface, retries=0): | |
detach_interface(failed_detach, interface) | |
sleep(retries + 1) | |
try: | |
delete_response = client.delete_network_interface( | |
NetworkInterfaceId=interface['NetworkInterfaceId']) | |
print(f"Delete response for {interface['NetworkInterfaceId']}- {delete_response}") | |
if 'HTTPStatusCode' not in delete_response['ResponseMetadata'] or \ | |
delete_response['ResponseMetadata']['HTTPStatusCode'] != 200: | |
failed_delete.append(delete_response) | |
except ClientError as e: | |
print(f"Exception while deleting - {str(e)}") | |
print() | |
if retries <= MAX_RETRIES: | |
if e.response['Error']['Code'] == 'InvalidNetworkInterface.InUse' or \ | |
e.response['Error']['Code'] == 'InvalidParameterValue': | |
retries = retries + 1 | |
print(f"Retry {retries} : Interface in use, deletion failed, retrying to detach and delete") | |
detach_and_delete_interface(failed_detach, failed_delete, interface, retries) | |
else: | |
raise RuntimeError("Code not found in error") | |
else: | |
raise RuntimeError("Max Number of retries exhausted to remove the interface") | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment