Skip to content

Commit

Permalink
WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
athornton committed Dec 31, 2024
1 parent 55e2a40 commit e5635d0
Show file tree
Hide file tree
Showing 7 changed files with 303 additions and 140 deletions.
57 changes: 51 additions & 6 deletions src/rsp_reaper/config.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,7 @@
"""Configuration for a reaper for a particular container registry."""

from dataclasses import dataclass

from .models.registry_category import RegistryCategory

from pathlib import Path

@dataclass
class RegistryAuth:
Expand All @@ -13,14 +11,61 @@ class RegistryAuth:
username: str
password: str

class LatestSemverKeepers:
minor: int | None
patch: int | None
build: int | None

class OlderSemverKeepers:
major: int | None
minor: int | None
patch: int | None
build: int | None

@dataclass
class SemverKeepers:
"""Within each of latest_major and older, how many minor versions,
how many patch versions within each of those, and how many builds of each
of those, to keep. Older also has a major version number. For instance,
older.major might be 3, and then when version 5.0 came out, you would
keep some images for the 2.x.y, 3.x.y, and 4.x.y series, but no 1.x images.
"""
latest_major: LatestSemverKeepers

@dataclass
class ContainerRegistryConfig:
"""Configuration for a particular container registry."""
class RSPKeepers:
"""Aliases are never purged.
"""
release: int | None
weekly: int | None
daily: int | None
release_candidate: int | None
experimental: int | None
unknown: int | None

@dataclass
class KeepPolicy:
"""How many of each image category to keep. `-1` or `None` means
"don't reap that category at all". `0` means "purge them all".
"""
untagged: int | None
semver: SemverKeepers | None
rsp: RSPKeepers | None

@dataclass
class RegistryConfig:
namespace: str
repository: str
registry: str
category: RegistryCategory
category: str
keep: KeepPolicy
project: str | None = None
auth: RegistryAuth | None = None
dry_run: bool = True
debug: bool = True
input_file: Path | None = None

@dataclass
class Config:
"""Configuration for multiple registries."""
registries: list[RegistryConfig]
137 changes: 118 additions & 19 deletions src/rsp_reaper/models/image.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

import semver

from .rsptag import RSPImageTag
from .rsptag import RSPImageTag, RSPImageTagCollection, RSPImageType

DATEFMT = "%Y-%m-%dT%H:%M:%S.%f%z"
LATEST_TAGS = ("latest", "latest_release", "latest_weekly", "latest_daily")
Expand All @@ -18,10 +18,11 @@


class ImageVersionClass(Enum):
"""Images are versioned with either RSP tags or semver tags."""
"""Tagged images are versioned with either RSP tags or semver tags."""

RSP = "rsp"
SEMVER = "semver"
UNTAGGED = "untagged"


@total_ordering
Expand All @@ -40,7 +41,7 @@ class Image:
date: datetime.datetime | None = None
id: int | None = None
rsp_image_tag: RSPImageTag | None = None
semver_tag: semver.Version | None = None
semver_tag: semver.VersionInfo | None = None
version_class: ImageVersionClass | None = None

def __eq__(self, other: object) -> bool:
Expand All @@ -65,27 +66,71 @@ def _compare(self, other: object) -> int:
int or NotImplemented
0 if equal, -1 if self is less than other, 1 if self is greater
than other, `NotImplemented` if they're not comparable.
Notes
-----
Because we're using this to sort images, which should not have both
RSP and Semver tags, but can certainly be untagged, we are going to
play pretty fast and loose with NotImplemented. Effectively, untagged
images and images whose tag types we cannot parse into some meaningful
order will get shoved to the bottom of the list.
"""
if not isinstance(other, Image):
return NotImplemented

if self.digest == other.digest:
# Tags are not relevant. It's the same image.
return 0

# If they have the same type of tags, sort on those
if self.rsp_image_tag is not None and other.rsp_image_tag is not None:
return self._compare_rsp_image_tags(other.rsp_image_tag)
# If they have the same type of tags, sort on those, and if the tags
# don't tell us, sort on date.

if self.semver_tag is not None and other.semver_tag is not None:
return self._compare_semver_tags(other.semver_tag)
return self._compare_by_class(other)

# Untagged sorts to the bottom
if self.tags and not other.tags:
return 1
if other.tags and not self.tags:
return -1
def _compare_by_class(self, other_image: Self) -> int:
myclass = self.version_class
otherclass = other_image.version_class

# Both untagged? Sort by date
return self._compare_dates(other)
if myclass == otherclass:
if myclass == ImageVersionClass.RSP:
return self._compare_rsptags(other_image)
elif myclass == ImageVersionClass.SEMVER:
return self._compare_semver(other_image)
else:
return self._compare_dates(other_image)
else:
# Untagged sorts to the bottom
if (
myclass != ImageVersionClass.UNTAGGED
and otherclass == ImageVersionClass.UNTAGGED
):
return 1
elif (
myclass == ImageVersionClass.UNTAGGED
and otherclass != ImageVersionClass.UNTAGGED
):
return -1
# Can't compare a Semver and an RSP image
return NotImplemented

def _compare_rsptags(self, other_image: Self) -> int:
if self.rsp_image_tag is None:
raise ValueError(f"{self} rsp_image_tag cannot be None")
if other_image.rsp_image_tag is None:
raise ValueError(f"{other_image} rsp_image_tag cannot be None")
return self.rsp_image_tag.compare(other_image.rsp_image_tag)

def _compare_semver(self, other_image: Self) -> int:
if self.semver_tag is None:
raise ValueError(f"{self} semver_tag cannot be None")
if other_image.semver_tag is None:
raise ValueError(f"{other_image} semver_tag cannot be None")
if self.semver_tag == other_image.semver_tag:
return 0
if self.semver_tag < other_image.semver_tag:
return -1
else:
return 1

def _compare_rsp_image_tags(self, other_tag: RSPImageTag) -> int:
if self.rsp_image_tag is None:
Expand All @@ -96,7 +141,7 @@ def _compare_rsp_image_tags(self, other_tag: RSPImageTag) -> int:
return 1
return 0

def _compare_semver_tags(self, other_tag: semver.Version) -> int:
def _compare_semver_tags(self, other_tag: semver.VersionInfo) -> int:
if self.semver_tag is None:
raise ValueError("semver_tag is None")
if self.semver_tag < other_tag:
Expand All @@ -116,12 +161,21 @@ def _compare_dates(self, other: Self) -> int:
return -1
if other.date and not self.date:
return 1
# Give up
return NotImplemented
return self._compare_digests(other)

def _compare_digests(self, other: Self) -> int:
if self.digest == other.digest:
return 0
if self.digest < other.digest:
return -1
return 1

def to_dict(self) -> JSONImage:
# Differs from asdict, in that set and datetime aren't
# JSON-serializable, so we make them a list and a string.
#
# We will just drop the semver/RSP tag fields, and rebuild them
# on load.
self_dict = asdict(self)
list_tags: list[str] = []
if self.tags:
Expand All @@ -136,6 +190,43 @@ def to_dict(self) -> JSONImage:
def to_json(self) -> str:
return json.dumps(self.to_dict())

def apply_best_tag(self) -> None:
"""Choose the best tag (preferring RSP to semver) for an image."""
collection = RSPImageTagCollection.from_tag_names(
list(self.tags), aliases=set(), cycle=None
)
self.rsp_image_tag = collection.best_tag()
if self.rsp_image_tag is not None:
self.semver_tag = self.rsp_image_tag.version
if self.rsp_image_tag.image_type == RSPImageType.UNKNOWN:
self.semver_tag = self._semver_from_tags()
if self.semver_tag is None:
self.semver_tag = self._generate_semver()

def _semver_from_tags(self) -> semver.VersionInfo | None:
raw_tags = list(self.tags)
best_semver: semver.Version | None = None
for tag in raw_tags:
try:
sv = semver.Version.parse(tag)
if best_semver is None or best_semver < sv:
best_semver = sv
except (ValueError, TypeError):
continue
return best_semver

def _generate_semver(self) -> semver.VersionInfo:
datestr = "unknown-date"
if self.date is not None:
datestr = (
self.date.isoformat()
.replace(":", "-")
.replace("+", "plus")
.replace(".", "-")
)
digstr = self.digest.replace(":", "-")
return semver.Version.parse(f"0.0.0-{datestr}+{digstr}")

@classmethod
def from_json(cls, inp: JSONImage | str) -> Self:
"""Much painful assertion that each field is the right type."""
Expand All @@ -157,6 +248,14 @@ def from_json(cls, inp: JSONImage | str) -> Self:
i_i = inp["id"]
if i_i and isinstance(i_i, int):
new_id = i_i
return cls(
new_obj = cls(
digest=inp["digest"], tags=new_tags, date=new_date, id=new_id
)
new_obj.apply_best_tag()
if not new_tags:
new_obj.version_class = ImageVersionClass.UNTAGGED
elif new_obj.rsp_image_tag is not None:
new_obj.version_class = ImageVersionClass.RSP
else:
new_obj.version_class = ImageVersionClass.SEMVER
return new_obj
Loading

0 comments on commit e5635d0

Please sign in to comment.