Last active
October 28, 2022 20:14
-
-
Save revolutionisme/8ec785f8202f47da5517c295a28c7cb5 to your computer and use it in GitHub Desktop.
Remove Network interfaces created by AWS Lambda while undeploying stack
This file contains 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