-
-
Save tobywf/6eb494f4b46cef367540074512161334 to your computer and use it in GitHub Desktop.
from __future__ import absolute_import, print_function, unicode_literals | |
import boto3 | |
def clean_old_lambda_versions(): | |
client = boto3.client('lambda') | |
functions = client.list_functions()['Functions'] | |
for function in functions: | |
versions = client.list_versions_by_function(FunctionName=function['FunctionArn'])['Versions'] | |
for version in versions: | |
if version['Version'] != function['Version']: | |
arn = version['FunctionArn'] | |
print('delete_function(FunctionName={})'.format(arn)) | |
#client.delete_function(FunctionName=arn) # uncomment me once you've checked | |
if __name__ == '__main__': | |
clean_old_lambda_versions() |
Works great, thanks for sharing 👍
Slight variation, handling amounts of old versions exceeding the paging size of list_versions_by_function
:
from __future__ import absolute_import, print_function, unicode_literals
import boto3
def clean_old_lambda_versions():
client = boto3.client('lambda')
functions = client.list_functions()['Functions']
for function in functions:
while True:
versions = client.list_versions_by_function(FunctionName=function['FunctionArn'])['Versions']
if len(versions) == 1:
print('{}: done'.format(function['FunctionName']))
break
for version in versions:
if version['Version'] != function['Version']:
arn = version['FunctionArn']
print('delete_function(FunctionName={})'.format(arn))
# client.delete_function(FunctionName=arn) # uncomment me once you've checked
if __name__ == '__main__':
clean_old_lambda_versions()
I have improved the code further to get all lambdas if they're more than 50 and included checks to ignore lambda at edge.
link to repo
Awesome mate, glad it's proven so useful!
Awesome. Thanks for sharing.
I changed this a bit to remove all but the $LATEST version and the most recent version
from __future__ import absolute_import, print_function, unicode_literals
import boto3
# This script removes all versions except $LATEST and the newest version
# If this script tries to delete a version any alias is using,
# boto3 will throw an exception and the script will exit
def clean_old_lambda_versions():
client = boto3.client('lambda')
functions = client.list_functions()['Functions']
for function in functions:
while True:
versions = client.list_versions_by_function(FunctionName=function['FunctionArn'])['Versions']
numVersions = len(versions)
if numVersions <= 2:
print('{}: done'.format(function['FunctionName']))
break
for version in versions:
if version['Version'] != function['Version'] and numVersions > 2: # $LATEST
arn = version['FunctionArn']
print('delete_function(FunctionName={})'.format(arn))
# client.delete_function(FunctionName=arn) # uncomment me once you've checked
numVersions -= 1
if __name__ == '__main__':
clean_old_lambda_versions()
With pagination, using boto's built-in paginators, and here in this gist:
# from __future__ import absolute_import, print_function, unicode_literals
import boto3
def clean_old_lambda_versions():
lambda_client = boto3.client('lambda')
functions_paginator = lambda_client.get_paginator('list_functions')
version_paginator = lambda_client.get_paginator('list_versions_by_function')
for function_page in functions_paginator.paginate():
for function in function_page['Functions']:
print(function['FunctionName'])
for version_page in version_paginator.paginate(FunctionName=function['FunctionArn']):
for version in version_page['Versions']:
arn = version['FunctionArn']
if version['Version'] != function['Version']:
print(' 🥊 {}'.format(arn))
# lambda_client.delete_function(FunctionName=arn) # uncomment me once you've checked
else:
print(' 💚 {}'.format(arn))
if __name__ == '__main__':
clean_old_lambda_versions()
A slight variation to avoid ResourceConflictException if the functions contained aliases:
from __future__ import absolute_import, print_function, unicode_literals
import boto3
def clean_old_lambda_versions():
client = boto3.client('lambda')
functions = client.list_functions()['Functions']
for function in functions:
versions = client.list_versions_by_function(FunctionName=function['FunctionArn'])['Versions']
aliases = client.list_aliases(FunctionName=function['FunctionArn'])
alias_versions = [alias['FunctionVersion'] for alias in aliases['Aliases']]
for version in versions:
if version['Version'] != function['Version'] and not version['Version'] in alias_versions:
arn = version['FunctionArn']
print('delete_function(FunctionName={})'.format(arn))
#client.delete_function(FunctionName=arn) # uncomment me once you've checked
if __name__ == '__main__':
clean_old_lambda_versions()
Revised copy with versions and alias check, Keep latest and most recent version
from __future__ import absolute_import, print_function, unicode_literals
import boto3
def clean_old_lambda_versions(client):
functions = client.list_functions()['Functions']
for function in functions:
versions = client.list_versions_by_function(FunctionName=function['FunctionArn'])['Versions']
numVersions = len(versions)
aliases = client.list_aliases(FunctionName=function['FunctionArn'])
alias_versions = [alias['FunctionVersion'] for alias in aliases['Aliases']]
if numVersions <= 2:
print('{}: done'.format(function['FunctionName']))
else:
for version in versions:
if version['Version'] != function['Version'] and numVersions > 2 and not version['Version'] in alias_versions:
arn = version['FunctionArn']
print('delete_function(FunctionName={})'.format(arn))
client.delete_function(FunctionName=arn) # uncomment me once you've checke
numVersions -= 1
if __name__ == '__main__':
client = boto3.client('lambda',region_name='us-east-1')
clean_old_lambda_versions(client)
client = boto3.client('lambda',region_name='eu-west-1')
clean_old_lambda_versions(client)
client = boto3.client('lambda',region_name='us-west-1')
clean_old_lambda_versions(client)
More revisions. We don't use aliases, so this keeps the most recent two by version number.
import boto3
def clean_old_lambda_versions(client):
functions = client.list_functions()['Functions']
for function in functions:
arn = function['FunctionArn']
print(arn)
all_versions = []
versions = client.list_versions_by_function(
FunctionName=arn)
# Page through all the versions
while True:
page_versions = [int(v['Version']) for v in versions['Versions'] if not v['Version'] == '$LATEST']
all_versions.extend(page_versions)
try:
marker = versions['NextMarker']
except:
break
versions = client.list_versions_by_function(
FunctionName=arn, Marker=marker)
# Sort and keep the last 2
all_versions.sort()
print('Which versions must go?')
print(all_versions[0:-2])
print('Which versions will live')
print(all_versions[-2::])
for chopBlock in all_versions[0:-2]:
functionArn = '{}:{}'.format(arn, chopBlock)
print('When uncommented, will run: delete_function(FunctionName={})'.format(functionArn))
# I want to leave this commented in Git for safety so we don't run it unscrupulously
# client.delete_function(FunctionName=functionArn) # uncomment me once you've checked
if __name__ == '__main__':
client = boto3.client('lambda', region_name='us-east-1')
clean_old_lambda_versions(client)
Thank you!
@jeffisadams - Thanks but the markers are not working fine when we have more than 400+ lambdas
Revised copy with versions, alias check, keep latest and most recent version. I put code to manage pagination to get all versions and multiple regions.
``
# Remove Old Lambda Versions
from __future__ import absolute_import, print_function, unicode_literals
import boto3
NUM_VERSION = 5
def get_versions_function(client, function_name):
versions = client.list_versions_by_function(FunctionName=function_name)
while "NextMarker" in versions and versions["NextMarker"]:
tmp = client.list_versions_by_function(
FunctionName=function_name,
Marker=versions["NextMarker"])
versions['Versions'].extend(tmp['Versions'])
versions["NextMarker"] = tmp["NextMarker"] if "NextMarker" in tmp else None
return versions['Versions']
def get_functions(client):
functions = client.list_functions()
while "NextMarker" in functions and functions["NextMarker"]:
tmp = client.list_functions(
Marker=functions["NextMarker"])
functions['Functions'].extend(tmp['Functions'])
functions["NextMarker"] = tmp["NextMarker"] if "NextMarker" in tmp else None
return functions['Functions']
def clean_old_lambda_versions(client):
functions = get_functions(client)
for function in functions:
versions = get_versions_function(client, function['FunctionArn'])
numVersions = len(versions)
aliases = client.list_aliases(FunctionName=function['FunctionArn'])
alias_versions = [alias['FunctionVersion'] for alias in aliases['Aliases']]
if numVersions <= NUM_VERSION:
print('{}: done'.format(function['FunctionName']))
else:
for version in versions:
if (version['Version'] != function['Version']
and numVersions > NUM_VERSION
and not version['Version'] in alias_versions):
arn = version['FunctionArn']
print('delete_function(FunctionName={})'.format(arn))
client.delete_function(FunctionName=arn) # uncomment me once you've checke
numVersions -= 1
session = boto3.session.Session(profile_name='default')
regions = ['us-west-1', 'us-west-2', 'sa-east-1', 'ca-central-1', 'us-east-1', 'us-east-2']
for region in regions:
client = session.client('lambda', region_name=region)
clean_old_lambda_versions(client)
``
Thanks!!!
Hi. I’m still new to aws but I have a question. Is there any way to do some damage control if the script accidentally deletes all the lambda versions available?
I think it's great you're being cautious; it's also why I commented out the "danger-zone" function calls. There's several mitigations (although obviously you're on your own and I take no responsibility).
To mitigate the worst case, you should make sure you're in a position to be able to re-deploy the Lambda code and supporting infrastructure if needed. Whether this is via CI/CD or e.g. downloading the most current package manually.
Obviously, it's best not to delete the version you're using in the first place. The original script should already check and not delete the version currently running. You could also modify the script to only delete e.g. the oldest 10 versions - similar to this comment. This should free up enough space without getting dangerous, unless you somehow use old versions for a long time.
As you can see in this thread, the original script is just a starting point. Make sure you understand every line of it (or whatever script you start with). Modify it to fit your needs. Run it with print
statements and the delete call commented out. Get someone else to double-check the script and the dry-run (an extra set of eyes helps). Unfortunately, everyone's workflow is different, so I can't (and won't) help with specifics, but hopefully it's enough to get you started!
@andrescollao works
Came here looking for a nodejs solution. We needed to completely empty out the Layers.
import AWS from 'aws-sdk'
const lambda = new AWS.Lambda({
region: 'us-east-1'
});
async function deleteAll() {
const { Layers } = await lambda.listLayers().promise();
const promises = [];
for (let layer of Layers) {
const {
LayerName,
LatestMatchingVersion: { Version }
} = layer;
for (let i = parseInt(Version, 10); i >= 1; i--) {
const params = {
LayerName: LayerName,
VersionNumber: i
};
promises.push(lambda.deleteLayerVersion(params).promise());
}
}
return Promise.all(promises);
}
deleteAll()
.then(promises => {
await promises;
})
.catch(error => {
console.log('error', error);
});
Sorry for the dumb question but is this script supposed to be executed via lambda itself?
If yes, it doesn't help if the account has already breached the 75 GB limit, as we can't create new lambdas. Would you suggest manually deleting the lambda versions (until AWS Support grants the quota increase request)
More revisions. We don't use aliases, so this keeps the most recent two by version number.
import boto3 def clean_old_lambda_versions(client): functions = client.list_functions()['Functions'] for function in functions: arn = function['FunctionArn'] print(arn) all_versions = [] versions = client.list_versions_by_function( FunctionName=arn) # Page through all the versions while True: page_versions = [int(v['Version']) for v in versions['Versions'] if not v['Version'] == '$LATEST'] all_versions.extend(page_versions) try: marker = versions['NextMarker'] except: break versions = client.list_versions_by_function( FunctionName=arn, Marker=marker) # Sort and keep the last 2 all_versions.sort() print('Which versions must go?') print(all_versions[0:-2]) print('Which versions will live') print(all_versions[-2::]) for chopBlock in all_versions[0:-2]: functionArn = '{}:{}'.format(arn, chopBlock) print('When uncommented, will run: delete_function(FunctionName={})'.format(functionArn)) # I want to leave this commented in Git for safety so we don't run it unscrupulously # client.delete_function(FunctionName=functionArn) # uncomment me once you've checked if __name__ == '__main__': client = boto3.client('lambda', region_name='us-east-1') clean_old_lambda_versions(client)
Nice. I tried all above. And this Code were what I need thank you. Even care about the pagination. Cleanup everything except two latest version numbers
And a nodejs version to delete function versions (not layers as the earlier nodejs code does)
delete-old-lambda-versions.mjs
import AWS from 'aws-sdk';
//Delete all but 'keepversions' highest numbered numeric versions
const keepversions = 1;
const lambda = new AWS.Lambda({ region: 'us-east-1' });
async function main()
{
const listfunctions = await lambda.listFunctions().promise();
for(let func of listfunctions.Functions)
{
let versions = (await lambda.listVersionsByFunction({FunctionName: func.FunctionArn}).promise()).Versions;
versions.forEach(_ => _.versionnumber = parseInt(_.Version) ) //eliminate '$LATEST'
versions = versions.filter(_ => !isNaN(_.versionnumber))
//sort in reverse order
versions = versions.sort((lhs,rhs) => rhs.versionnumber - lhs.versionnumber);
let keep = versions.slice(0,keepversions);
let remove = versions.slice(keepversions);
console.log(`${func.FunctionArn}: Keeping ${JSON.stringify(keep.map(_ => _.Version))}, deleting ${JSON.stringify(remove.map(_ => _.Version))}`);
for(let version of remove)
await lambda.deleteFunction({ FunctionName: version.FunctionArn }).promise() //version FunctionArn embeds the version number
}
}
main().then(() => process.exit(0)).catch(e => { console.error(e); });
This worked for me to delete all but the live alias/most recent version:
import boto3
functions_paginator = client.get_paginator('list_functions')
version_paginator = client.get_paginator('list_versions_by_function')
for function_page in functions_paginator.paginate():
for function in function_page['Functions']:
aliases = client.list_aliases(FunctionName=function['FunctionArn'])
alias_versions = [alias['FunctionVersion'] for alias in aliases['Aliases']]
for version_page in version_paginator.paginate(FunctionName=function['FunctionArn']):
for version in version_page['Versions']:
arn = version['FunctionArn']
if version['Version'] != function['Version'] and version['Version'] not in alias_versions:
print(' 🥊 {}'.format(arn))
# client.delete_function(FunctionName=arn)
else:
print(' 💚 {}'.format(arn))
anybody know how to do this cleanup with terraform ?
anybody know how to do this cleanup with terraform ?
I don't think this can be done via Terraform (TF) without writing a custom provider in go-lang. Even then, it would be impractical since you'd have to first import each version into TF state before TF would destroy it.
If you're already managing the Lambdas via TF and applying the TF generates additional versions, you could consider setting the publish
attribute false: https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_function#publish . That will have TF disable the versioning system that's built in to the Lambda service. You'd probably want to make sure you have versioning controlled through some other means before you do that, though. E.g., you could store multiple revisions in S3 and direct the TF to deploy one version or another. In that scenario, rolling back to a prior version would just mean selecting that older revision from S3 and having the TF push that as the one/active version that Lambda is aware of.
More revisions. We don't use aliases, so this keeps the most recent two by version number.
import boto3 def clean_old_lambda_versions(client): functions = client.list_functions()['Functions'] for function in functions: arn = function['FunctionArn'] print(arn) all_versions = [] versions = client.list_versions_by_function( FunctionName=arn) # Page through all the versions while True: page_versions = [int(v['Version']) for v in versions['Versions'] if not v['Version'] == '$LATEST'] all_versions.extend(page_versions) try: marker = versions['NextMarker'] except: break versions = client.list_versions_by_function( FunctionName=arn, Marker=marker) # Sort and keep the last 2 all_versions.sort() print('Which versions must go?') print(all_versions[0:-2]) print('Which versions will live') print(all_versions[-2::]) for chopBlock in all_versions[0:-2]: functionArn = '{}:{}'.format(arn, chopBlock) print('When uncommented, will run: delete_function(FunctionName={})'.format(functionArn)) # I want to leave this commented in Git for safety so we don't run it unscrupulously # client.delete_function(FunctionName=functionArn) # uncomment me once you've checked if __name__ == '__main__': client = boto3.client('lambda', region_name='us-east-1') clean_old_lambda_versions(client)
Nice. I tried all above. And this Code were what I need thank you. Even care about the pagination. Cleanup everything except two latest version numbers
More revisions. We don't use aliases, so this keeps the most recent two by version number.
import boto3 def clean_old_lambda_versions(client): functions = client.list_functions()['Functions'] for function in functions: arn = function['FunctionArn'] print(arn) all_versions = [] versions = client.list_versions_by_function( FunctionName=arn) # Page through all the versions while True: page_versions = [int(v['Version']) for v in versions['Versions'] if not v['Version'] == '$LATEST'] all_versions.extend(page_versions) try: marker = versions['NextMarker'] except: break versions = client.list_versions_by_function( FunctionName=arn, Marker=marker) # Sort and keep the last 2 all_versions.sort() print('Which versions must go?') print(all_versions[0:-2]) print('Which versions will live') print(all_versions[-2::]) for chopBlock in all_versions[0:-2]: functionArn = '{}:{}'.format(arn, chopBlock) print('When uncommented, will run: delete_function(FunctionName={})'.format(functionArn)) # I want to leave this commented in Git for safety so we don't run it unscrupulously # client.delete_function(FunctionName=functionArn) # uncomment me once you've checked if __name__ == '__main__': client = boto3.client('lambda', region_name='us-east-1') clean_old_lambda_versions(client)
More revisions. We don't use aliases, so this keeps the most recent two by version number.
import boto3 def clean_old_lambda_versions(client): functions = client.list_functions()['Functions'] for function in functions: arn = function['FunctionArn'] print(arn) all_versions = [] versions = client.list_versions_by_function( FunctionName=arn) # Page through all the versions while True: page_versions = [int(v['Version']) for v in versions['Versions'] if not v['Version'] == '$LATEST'] all_versions.extend(page_versions) try: marker = versions['NextMarker'] except: break versions = client.list_versions_by_function( FunctionName=arn, Marker=marker) # Sort and keep the last 2 all_versions.sort() print('Which versions must go?') print(all_versions[0:-2]) print('Which versions will live') print(all_versions[-2::]) for chopBlock in all_versions[0:-2]: functionArn = '{}:{}'.format(arn, chopBlock) print('When uncommented, will run: delete_function(FunctionName={})'.format(functionArn)) # I want to leave this commented in Git for safety so we don't run it unscrupulously # client.delete_function(FunctionName=functionArn) # uncomment me once you've checked if __name__ == '__main__': client = boto3.client('lambda', region_name='us-east-1') clean_old_lambda_versions(client)
Nice. I tried all above. And this Code were what I need thank you. Even care about the pagination. Cleanup everything except two latest version numbers
Thanks for this code. I added one extra check to filter only functions with a name like substring. I used it today to clean up one function that had 110 versions. Definitely saved a lot of time!
I had the use case for this due to the deprecation of node versions < 18. I wrote a script to delete all old versions that were not the newest to make the AWS Trusted Advisor happy again. This is my code for people in need to delete for all lambdas all old versions but the newest:
import { LambdaClient, ListVersionsByFunctionCommand, DeleteFunctionCommand, FunctionConfiguration, ListFunctionsCommand } from '@aws-sdk/client-lambda';
const lambda = new LambdaClient({
region: 'eu-central-1'
});
const fetchAllVersions = async (functionName: string) => {
let marker: string | undefined;
let versions: FunctionConfiguration[] = [];
do {
const listCommand = new ListVersionsByFunctionCommand({
FunctionName: functionName,
Marker: marker
});
const partOfVersions = await lambda.send(listCommand);
versions = versions.concat(partOfVersions.Versions ?? []);
marker = partOfVersions.NextMarker;
} while(marker);
return versions;
};
const deleteOldVersionsOfLambda = async (functionName: string) => {
const versions = await fetchAllVersions(functionName);
const deletionList: string[] = versions?.filter(v => v.Version !== '$LATEST').map(v => v.FunctionArn) as string[];
for (const arn of deletionList) {
const deletionCommand = new DeleteFunctionCommand({
FunctionName: arn
});
await lambda.send(deletionCommand);
}
return null;
};
const getAllFunctionsAvailable = async () => {
const listFunctionsCommand = new ListFunctionsCommand({});
const functions = await lambda.send(listFunctionsCommand);
return functions.Functions?.map(f => f.FunctionName).filter(f => f != null);
};
const deleteAllOldVersions = async () => {
const allFunctionsNames = await getAllFunctionsAvailable();
for (const functionName of allFunctionsNames!) {
await deleteOldVersionsOfLambda(functionName!);
}
};
deleteAllOldVersions();
More revisions. We don't use aliases, so this keeps the most recent two by version number.
import boto3 def clean_old_lambda_versions(client): functions = client.list_functions()['Functions'] for function in functions: arn = function['FunctionArn'] print(arn) all_versions = [] versions = client.list_versions_by_function( FunctionName=arn) # Page through all the versions while True: page_versions = [int(v['Version']) for v in versions['Versions'] if not v['Version'] == '$LATEST'] all_versions.extend(page_versions) try: marker = versions['NextMarker'] except: break versions = client.list_versions_by_function( FunctionName=arn, Marker=marker) # Sort and keep the last 2 all_versions.sort() print('Which versions must go?') print(all_versions[0:-2]) print('Which versions will live') print(all_versions[-2::]) for chopBlock in all_versions[0:-2]: functionArn = '{}:{}'.format(arn, chopBlock) print('When uncommented, will run: delete_function(FunctionName={})'.format(functionArn)) # I want to leave this commented in Git for safety so we don't run it unscrupulously # client.delete_function(FunctionName=functionArn) # uncomment me once you've checked if __name__ == '__main__': client = boto3.client('lambda', region_name='us-east-1') clean_old_lambda_versions(client)
This Helped for my 88 lambda cleanup, thanks!
Thanks for writing such a nice code but how can I keep only 10 versions of lambda function and delete all versions using this code or is this possible using python?