Nexus Repository Manager keep the last X docker images delete all other

clean_docker_nexus.sh

#!/bin/bash

REPO_URL="https://repository.xxx.net/repository/"
USER="admin"
PASSWORD="datpassword"

BUCKET="portal-docker"
KEEP_IMAGES=10

IMAGES=$(curl --silent -X GET -H 'Accept: application/vnd.docker.distribution.manifest.v2+json' -u ${USER}:${PASSWORD} "${REPO_URL}${BUCKET}/v2/_catalog" | jq .repositories | jq -r '.[]' )

echo ${IMAGES}

for IMAGE_NAME in ${IMAGES}; do
 echo ${IMAGE_NAME}

   TAGS=$(curl --silent -X GET -H 'Accept: application/vnd.docker.distribution.manifest.v2+json' -u ${USER}:${PASSWORD} "${REPO_URL}${BUCKET}/v2/${IMAGE_NAME}/tags/list" | jq .tags | jq -r '.[]' )
   
   TAG_COUNT=$(echo $TAGS | wc -w)
   
   let TAG_COUNT_DEL=${TAG_COUNT}-${KEEP_IMAGES}
   COUNTER=0
   
   echo "THERE ARE ${TAG_COUNT} IMAGES FOR ${IMAGE_NAME}"
   
   ## skip if smaller than keep
   if [ "${KEEP_IMAGES}" -gt "${TAG_COUNT}" ]
    then
      echo "There are only ${TAG_COUNT} Images for ${IMAGE_NAME} - nothing to delete"
      continue
   fi
   
   for TAG in ${TAGS}; do
    let COUNTER=COUNTER+1
    if [ "${COUNTER}" -gt "${TAG_COUNT_DEL}" ]
     then
      break;
    fi
    
    IMAGE_SHA=$(curl --silent -I -X GET -H 'Accept: application/vnd.docker.distribution.manifest.v2+json' -u ${USER}:${PASSWORD} "${REPO_URL}${BUCKET}/v2/${IMAGE_NAME}/manifests/$TAG" | grep Docker-Content-Digest | cut -d ":" -f3 | tr -d '\r')
    echo "DELETE ${TAG} ${IMAGE_SHA}";
    DEL_URL="${REPO_URL}${BUCKET}/v2/${IMAGE_NAME}/manifests/sha256:${IMAGE_SHA}"
    RET="$(curl --silent -k -X DELETE -H 'Accept: application/vnd.docker.distribution.manifest.v2+json' -u ${USER}:${PASSWORD} $DEL_URL)"
   
   done;
done;

Thanks for the script! im using it as a reference. Currently, i have implemented this script in my jenkins job to automatically delete a specific image tag after the job runs.

#!/bin/bash

REPO_URL="http://1.2.3.4:8081/repository/"
USERNAME="user"
PASSWORD="password"

BUCKET="dockerbucket"
KEEP_IMAGES=3



# IMAGE_DIR="nginx"
# IMAGE_NAME="nginx"
IMAGE_DIR="$1"
IMAGE_NAME="$2"
IMAGE_FULL_NAME=$IMAGE_DIR/$IMAGE_NAME
# echo "${IMAGE_FULL_NAME}"

TAGS=$(curl --silent -X GET -H 'Accept: application/vnd.docker.distribution.manifest.v2+json' -u ${USERNAME}:${PASSWORD} "${REPO_URL}${BUCKET}/v2/${IMAGE_FULL_NAME}/tags/list" | jq .tags | jq -r '.[]' | sort -r)


echo "$TAGS" | awk -v img="$IMAGE_NAME" '{print img ":" $0}'

TOTAL_TAGS=$(echo "$TAGS" | wc -l)
echo "total tags = $TOTAL_TAGS"

if [[ $TOTAL_TAGS -gt $KEEP_IMAGES ]]; then
    echo "Total tags ($TOTAL_TAGS) exceed KEEP_IMAGES ($KEEP_IMAGES). Deleting older tags."

    TAGS_TO_DELETE=$(echo "$TAGS" | tail -n +$((KEEP_IMAGES + 1)))

    while IFS= read -r TAG; do
        echo "Deleting image ${IMAGE_NAME}:$TAG"
        echo "Executing curl request..."

        IMAGE_SHA=$(curl --silent -I -X GET -H 'Accept: application/vnd.docker.distribution.manifest.v2+json' -u ${USERNAME}:${PASSWORD} "${REPO_URL}${BUCKET}/v2/${IMAGE_FULL_NAME}/manifests/$TAG" | grep Docker-Content-Digest | cut -d ":" -f3 | tr -d '\r')
        echo "DELETE ${TAG} ${IMAGE_SHA}";
        DEL_URL="${REPO_URL}${BUCKET}/v2/${IMAGE_FULL_NAME}/manifests/sha256:${IMAGE_SHA}"

        RET="$(curl --silent -k -X DELETE -H 'Accept: application/vnd.docker.distribution.manifest.v2+json' -u ${USERNAME}:${PASSWORD} $DEL_URL)"
      #   curl -s -X DELETE -u ${USERNAME}:${PASSWORD} "${REPO_URL}${BUCKET}/v2/${IMAGE_FULL_NAME}/manifests/$TAG"
        if [[ $? -ne 0 ]]; then
            echo "Failed to delete image tag: ${IMAGE_NAME}:$TAG"
        else
            echo "Successfully deleted image tag: ${IMAGE_NAME}:$TAG"
        fi

        echo "----------------------------------------"

    done <<< "$TAGS_TO_DELETE"
else
    echo "Total tags ($TOTAL_TAGS) are within or equal to KEEP_IMAGES ($KEEP_IMAGES). No deletion needed."
fi

I use this keeps last 5 images based on crated date desc, works great. I have no idea why they don't add it just to a job .. ugh.

import requests
import argparse
import json
from datetime import datetime, timezone

# Function to print usage message
def usage():
    print("Usage: nexus_docker_cleanup.py -u USER -p PASSWORD [-p PREFIX] [-k KEEP_IMAGES]")
    exit(1)

# Default values
KEEP_IMAGES = 5
PREFIX = "my/services/"

# Argument parser
parser = argparse.ArgumentParser(description='Clean up Docker images from Nexus repository.')
parser.add_argument('-u', '--user', required=True, help='User for repository authentication')
parser.add_argument('-p', '--password', required=True, help='Password for repository authentication')
parser.add_argument('--prefix', default=PREFIX, help='Filter images by prefix')
parser.add_argument('-k', '--keep_images', default=KEEP_IMAGES, type=int, help='Number of images to keep')
args = parser.parse_args()

USER = args.user
PASSWORD = args.password
KEEP_IMAGES = args.keep_images
PREFIX = args.prefix

REPO_URL = "https://nexus.myorg/repository/"
BUCKET = "docker-mcm-hosted"

def get_json_response(url, headers):
    response = requests.get(url, headers=headers, auth=(USER, PASSWORD))
    if response.status_code != 200:
        print(f"Error fetching URL {url}. Status code: {response.status_code}")
        return None
    return response.json()

def parse_isoformat(date_str):
    try:
        if 'Z' in date_str:
            date_str = date_str.replace('Z', '+00:00')
        if '+' in date_str or '-' in date_str:
            # Workaround to strip the isoformat to maximum microsecond precision (6 digits)
            base, frac_sec = date_str.split('.')
            frac_sec = frac_sec[:6]  # limit to 6 digits
            if '+' in date_str:
                offset = date_str.split('+')[1]
                date_str = f"{base}.{frac_sec}+{offset}"
            elif '-' in date_str:
                offset = date_str.split('-')[1]
                date_str = f"{base}.{frac_sec}-{offset}"
        else:
            date_str = date_str[:26] + "+00:00"
        dt = datetime.fromisoformat(date_str)

        # Ensure the datetime is timezone-aware
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except ValueError:
        print(f"Invalid isoformat string: '{date_str}'")
        return None

# Fetch list of images
images_response = get_json_response(f"{REPO_URL}{BUCKET}/v2/_catalog", 
                                    {'Accept': 'application/vnd.docker.distribution.manifest.v2+json'})

if images_response is None:
    exit(1)

IMAGES = images_response.get('repositories', [])

# Filter images based on the prefix
filtered_images = [image for image in IMAGES if image.startswith(PREFIX)]

print(f"\nFiltered list of images with prefix '{PREFIX}':")
print(filtered_images)

for IMAGE_NAME in filtered_images:
    print(IMAGE_NAME)
    
    tags_response = get_json_response(f"{REPO_URL}{BUCKET}/v2/{IMAGE_NAME}/tags/list", 
                                      {'Accept': 'application/vnd.docker.distribution.manifest.v2+json'})

    if tags_response is None:
        continue

    TAGS = tags_response.get('tags', [])
    
    image_tags_with_date = []
    
    for TAG in TAGS:
        manifest_response = get_json_response(f"{REPO_URL}{BUCKET}/v2/{IMAGE_NAME}/manifests/{TAG}", 
                                              {'Accept': 'application/vnd.docker.distribution.manifest.v2+json'})
        
        if manifest_response is None:
            continue
        
        config_digest = manifest_response.get('config', {}).get('digest')

        if config_digest:
            blob_response = get_json_response(f"{REPO_URL}{BUCKET}/v2/{IMAGE_NAME}/blobs/{config_digest}", 
                                              {'Accept': 'application/vnd.docker.distribution.manifest.v2+json'})
            
            if blob_response is None:
                continue
            
            creation_date = blob_response.get('created')

            if creation_date:
                parsed_date = parse_isoformat(creation_date)
                if parsed_date:
                    image_tags_with_date.append((TAG, parsed_date))

    # Sort tags by creation date descending
    image_tags_with_date.sort(key=lambda x: x[1], reverse=True)

    TAGS_TO_KEEP = image_tags_with_date[:KEEP_IMAGES]
    TAGS_TO_DELETE = image_tags_with_date[KEEP_IMAGES:]

    print(f"THERE ARE {len(TAGS)} IMAGES FOR {IMAGE_NAME}")
    
    for tag, creation_date in TAGS_TO_DELETE:
        delete_response = requests.head(f"{REPO_URL}{BUCKET}/v2/{IMAGE_NAME}/manifests/{tag}", 
                                        headers={'Accept': 'application/vnd.docker.distribution.manifest.v2+json'},
                                        auth=(USER, PASSWORD))

        if delete_response.status_code != 200:
            print(f"Error fetching manifest for {IMAGE_NAME}:{tag}. Status code: {delete_response.status_code}")
            continue
        
        image_sha = delete_response.headers.get('Docker-Content-Digest').split(':')[-1].strip()
        print(f"DELETE {tag} {image_sha} created at {creation_date}")
        
        del_url = f"{REPO_URL}{BUCKET}/v2/{IMAGE_NAME}/manifests/sha256:{image_sha}"

        del_response = requests.delete(del_url, 
                                       headers={'Accept': 'application/vnd.docker.distribution.manifest.v2+json'},
                                       auth=(USER, PASSWORD))
        
        if del_response.status_code != 202:
            print(f"Error deleting {IMAGE_NAME}:{tag} with SHA {image_sha}")
        else:
            print(f"Successfully deleted {IMAGE_NAME}:{tag} with SHA {image_sha}, created at {creation_date}")