Clean up Old Lambda Layer#

A utility script to manage AWS Lambda Layer versions by automatically cleaning up old and unused layer versions.

Purpose:

AWS Lambda Layers can accumulate many versions over time, consuming storage and making layer management difficult. This script helps maintain a clean Lambda Layer environment by:

  • Keeping the N most recent versions of each layer

  • Removing versions older than a specified retention period

  • Supporting dry-run mode for safe verification before actual deletion

  • Providing detailed information about deleted versions including size and creation date

Example:

clean_up_old_lambda_layer.py
  1# -*- coding: utf-8 -*-
  2
  3"""
  4AWS Lambda Layer Version Cleanup Utility
  5
  6This module provides functionality to clean up old AWS Lambda Layer versions to optimize storage usage.
  7It allows you to maintain the N most recent versions of each layer while removing older versions
  8based on their creation date.
  9
 10Key Features:
 11
 12- Lists all Lambda layers filtering by runtime family (e.g., Python)
 13- Keeps a specified number of most recent layer versions
 14- Deletes versions older than a specified retention period
 15- Supports dry run mode for safety
 16- Handles pagination for AWS API calls
 17
 18Requirements:
 19
 20- boto3
 21
 22Usage:
 23
 24.. code-block:: python
 25
 26    boto_ses = boto3.Session(profile_name="your_profile")
 27    lbd_client = boto_ses.client("lambda")
 28
 29    # List all Python runtime layers
 30    layers = list_layers(lbd_client, runtime_family="python")
 31
 32    # Clean up each layer
 33    for layer_name in layers:
 34        delete_old_layer_version(
 35            lbd_client=lbd_client,
 36            layer_name=layer_name,
 37            keep_n_latest=3,
 38            retention_period_days=90,
 39            real_run=False  # Set to True for actual deletion
 40        )
 41"""
 42
 43import typing as T
 44from itertools import islice
 45from datetime import datetime, timezone, timedelta
 46
 47import boto3
 48
 49# from rich import print as rprint
 50
 51
 52def list_layers(
 53    lbd_client,
 54    runtime_family: str = "python",
 55) -> T.List[str]:
 56    """
 57    List all Lambda layers that are compatible with the specified runtime family.
 58
 59    :param lbd_client: Boto3 Lambda client instance
 60    :param runtime_family: Runtime family to filter layers (e.g., "python", "node")
 61
 62    :return: List of layer names compatible with the specified runtime
 63    """
 64    paginator = lbd_client.get_paginator("list_layers")
 65    response_iterator = paginator.paginate(
 66        PaginationConfig={
 67            "MaxItems": 1000,
 68            "PageSize": 50,
 69        }
 70    )
 71    layer_name_list = list()
 72    for response in response_iterator:
 73        # rprint(response)  # for debug only
 74        for layer_data in response.get("Layers"):
 75            for runtime in layer_data.get("LatestMatchingVersion", {}).get(
 76                "CompatibleRuntimes", []
 77            ):
 78                if runtime.startswith(runtime_family):
 79                    layer_name_list.append(layer_data["LayerName"])
 80                    break
 81    return layer_name_list
 82
 83
 84def iter_layer_version(
 85    lbd_client,
 86    layer_name: str,
 87) -> T.Iterable[T.Dict[str, T.Any]]:
 88    """
 89    Generate an iterator of all versions for a specific Lambda layer.
 90
 91    :param lbd_client: Boto3 Lambda client instance
 92    :param layer_name: Name of the Lambda layer
 93
 94    :return: iterator of Dict[str, Any], Layer version data including Version, CreatedDate, etc.
 95    """
 96    paginator = lbd_client.get_paginator("list_layer_versions")
 97    response_iterator = paginator.paginate(
 98        LayerName=layer_name,
 99        PaginationConfig={"MaxItems": 1000, "PageSize": 50},
100    )
101    for response in response_iterator:
102        # rprint(response)  # for debug only
103        yield from response.get("LayerVersions")
104
105
106def take(n: int, iterable: T.Iterable):
107    "Return first n items of the iterable as a list."
108    return list(islice(iterable, n))
109
110
111MAGNITUDE_OF_DATA = {
112    i: v for i, v in enumerate(["B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"])
113}
114
115
116def repr_data_size(
117    size_in_bytes: int,
118    precision: int = 2,
119) -> str:
120    """
121    Return human readable string represent of a file size. Doesn't support
122    size greater than 1YB.
123
124    For example:
125    - 100 bytes => 100 B
126    - 100,000 bytes => 97.66 KB
127    - 100,000,000 bytes => 95.37 MB
128    - 100,000,000,000 bytes => 93.13 GB
129    - 100,000,000,000,000 bytes => 90.95 TB
130    - 100,000,000,000,000,000 bytes => 88.82 PB
131    - and more ...
132
133    Magnitude of data::
134
135        1000         kB    kilobyte
136        1000 ** 2    MB    megabyte
137        1000 ** 3    GB    gigabyte
138        1000 ** 4    TB    terabyte
139        1000 ** 5    PB    petabyte
140        1000 ** 6    EB    exabyte
141        1000 ** 7    ZB    zettabyte
142        1000 ** 8    YB    yottabyte
143    """
144    if size_in_bytes < 1024:
145        return "%s B" % size_in_bytes
146
147    index = 0
148    while 1:
149        index += 1
150        size_in_bytes, mod = divmod(size_in_bytes, 1024)
151        if size_in_bytes < 1024:
152            break
153    template = "{0:.%sf} {1}" % precision
154    s = template.format(size_in_bytes + mod / 1024.0, MAGNITUDE_OF_DATA[index])
155    return s
156
157
158def delete_old_layer_version(
159    lbd_client,
160    layer_name: str,
161    keep_n_latest: int = 3,
162    retention_period_days: int = 90,
163    utc_now: T.Optional[datetime] = None,
164    real_run: bool = False,
165) -> None:
166    """
167    Delete old versions of a Lambda layer based on specified criteria.
168
169    :param lbd_client: Boto3 Lambda client instance
170    :param layer_name: Name of the Lambda layer to clean up
171    :param keep_n_latest: Number of most recent versions to keep regardless of age
172    :param retention_period_days: Delete versions older than this many days
173    :param utc_now: Current UTC time (useful for testing)
174    :param real_run: If False, only print what would be deleted without actual deletion
175
176    The function will:
177
178    1. Always keep the N most recent versions specified by ``keep_n_latest``
179    2. Delete remaining versions that are older than ``retention_period_days``
180    3. Print size and creation date information for deleted versions
181    4. Only perform actual deletions if ``real_run`` is True
182    """
183    if utc_now is None:
184        utc_now = datetime.utcnow().replace(tzinfo=timezone.utc)
185    ago = utc_now - timedelta(days=retention_period_days)
186
187    # Get iterator for all versions and skip the N latest ones
188    layer_version_data_iterator = iter_layer_version(
189        lbd_client=lbd_client,
190        layer_name=layer_name,
191    )
192    # print(list(layer_version_data_iterator))
193    take(keep_n_latest, layer_version_data_iterator)
194
195    # Process remaining versions
196    for layer_version_data in layer_version_data_iterator:
197        # rprint(layer_version_data)  # for debug only
198        created_date = datetime.strptime(
199            layer_version_data["CreatedDate"], "%Y-%m-%dT%H:%M:%S.%f%z"
200        )
201        if created_date < ago:
202            version = layer_version_data["Version"]
203            _response = lbd_client.get_layer_version(
204                LayerName=layer_name,
205                VersionNumber=version,
206            )
207            code_size = _response["Content"]["CodeSize"]
208            code_size_for_human = repr_data_size(code_size)
209            print(
210                f"delete: {layer_name}:{version} (created at = {created_date}, size = {code_size_for_human})"
211            )
212            if real_run:
213                lbd_client.delete_layer_version(
214                    LayerName=layer_name,
215                    VersionNumber=layer_version_data["Version"],
216                )
217
218
219if __name__ == "__main__":
220    # Example usage of the module
221    boto_ses = boto3.Session(profile_name="bmt_app_devops_us_east_1")
222    lbd_client = boto_ses.client("lambda")
223
224    # List all Python runtime layers
225    layer_name_list = list_layers(
226        lbd_client=lbd_client,
227        runtime_family="python",
228    )
229
230    # Clean up each layer
231    for layer_name in layer_name_list:
232        delete_old_layer_version(
233            lbd_client=lbd_client,
234            layer_name=layer_name,
235            keep_n_latest=3,
236            retention_period_days=90,
237            real_run=False,  # Set to True for actual deletion
238        )