guru
/
operator-repo
1
0
Fork 0
Code Pull Requests Activity

Add upgrade path check

This commit is contained in:
Maurizio Porrato 2023-08-03 18:23:45 +01:00
parent 667c14ac0c
commit aa96dc2403
4 changed files with 108 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
src/operator_repo/classes.py
View File

@ -68,7 +68,7 @@ class Bundle:
try:
csv_full_name = self.csv["metadata"]["name"]
name, version = csv_full_name.split(".", 1)
return name, version.lstrip('v')
return name, version.lstrip("v")
except (KeyError, ValueError) as exc:
raise InvalidBundleException(
f"CSV for {self} has invalid .metadata.name"
@ -131,7 +131,9 @@ class Bundle:
"""
for suffix in ["yaml", "yml"]:
try:
return next(self._manifests_path.glob(f"*.clusterserviceversion.{suffix}"))
return next(
self._manifests_path.glob(f"*.clusterserviceversion.{suffix}")
)
except StopIteration:
continue
raise InvalidBundleException(
@ -195,7 +197,8 @@ class Bundle:
f" and '{other.__class__.__name__}"
)
if self.csv_operator_name != other.csv_operator_name:
raise ValueError("Can't compare bundles from different operators")
# raise ValueError("Can't compare bundles from different operators")
return self.csv_operator_name < other.csv_operator_name
try:
return Version.parse(self.csv_operator_version.lstrip("v")) < Version.parse(
other.csv_operator_version.lstrip("v")
@ -314,6 +317,21 @@ class Operator:
)[-1][1]
except IndexError:
return None
except ValueError:
log.warning("%s: A bundle has non-semver compliant version: using lexical order to determine default channel", self)
try:
return sorted(
[
(
x.csv_operator_version,
x.default_channel,
)
for x in self.all_bundles()
if x.default_channel is not None
]
)[-1][1]
except IndexError:
return None
def channel_bundles(self, channel: str) -> List[Bundle]:
"""
@ -347,13 +365,11 @@ class Operator:
edges: Dict[Bundle, set[Bundle]] = {}
all_bundles_set = set(all_bundles)
for bundle in all_bundles_set:
try:
spec = bundle.csv["spec"]
except KeyError:
continue
spec = bundle.csv.get("spec", {})
replaces = spec.get("replaces")
skips = spec.get("skips", [])
for replaced_bundle_name in skips + [replaces]:
previous = set(skips) | {replaces}
for replaced_bundle_name in previous:
if replaced_bundle_name is None:
continue
if ".v" not in replaced_bundle_name:
@ -372,7 +388,11 @@ class Operator:
replaced_bundle = self.bundle(
replaced_bundle_version.lstrip("v")
)
edges.setdefault(replaced_bundle, set()).add(bundle)
if (
channel in bundle.channels
and channel in replaced_bundle.channels
):
edges.setdefault(replaced_bundle, set()).add(bundle)
except InvalidBundleException:
pass
return edges

107
src/operator_repo/cli.py
View File

@ -9,6 +9,8 @@ from itertools import chain
from pathlib import Path
from typing import Union, Dict, Any, Iterator, Tuple
from semver import Version
from operator_repo.classes import Bundle, Operator, Repo
@ -49,6 +51,8 @@ def _list(
("Channels", ", ".join(target.channels)),
("Default channel", target.default_channel),
("Container image", csv_annotations.get("containerImage", "")),
("Replaces", target.csv.get("spec", {}).get("replaces", "")),
("Skips", target.csv.get("spec", {}).get("skips", [])),
]
max_width = max([len(key) for key, _ in info])
for key, value in info:
@ -65,7 +69,9 @@ def action_list(repo_path, *what: str, recursive: bool = False) -> None:
_list(parse_target(repo, target), recursive)
def lookup_dict(data: Dict[str, Any], path: str, default: Any = None, separator: str = '.') -> Any:
def lookup_dict(
data: Dict[str, Any], path: str, default: Any = None, separator: str = "."
) -> Any:
keys = path.split(separator)
subtree = data
for key in keys:
@ -81,58 +87,83 @@ def do_check_bundle_operator_name(bundle: Bundle) -> Iterator[Tuple[str, str]]:
yield "fail", "Bundle does not define the operator name in annotations.yaml"
return
if name != bundle.csv_operator_name:
yield "fail", "Operator name from annotations.yaml does not match the name defined in the CSV"
yield "fail", f"Operator name from annotations.yaml ({name}) does not match the name defined in the CSV ({bundle.csv_operator_name})"
if name != bundle.operator_name:
yield "fail", "Operator name from annotations.yaml does not match the operator's directory name"
yield "warn", f"Operator name from annotations.yaml ({name}) does not match the operator's directory name ({bundle.operator_name})"
def do_check_bundle_image(bundle: Bundle) -> Iterator[Tuple[str, str]]:
container_image = lookup_dict(bundle.csv, "metadata.annotations.containerImage")
if container_image is None:
yield "fail", "CSV doesn't define .metadata.annotations.containerImage"
return
deployments = lookup_dict(bundle.csv, "spec.install.spec.deployments")
if deployments is None:
yield "fail", "CSV doesn't define .spec.install.spec.deployments"
return
for deployment in deployments:
containers = lookup_dict(deployment, "spec.template.spec.containers", [])
if any(container_image == x.get("image") for x in containers):
try:
container_image = lookup_dict(bundle.csv, "metadata.annotations.containerImage")
if container_image is None:
yield "fail", "CSV doesn't define .metadata.annotations.containerImage"
return
yield "fail", f"container image {container_image} not used by any deployment"
deployments = lookup_dict(bundle.csv, "spec.install.spec.deployments")
if deployments is None:
yield "fail", "CSV doesn't define .spec.install.spec.deployments"
return
for deployment in deployments:
containers = lookup_dict(deployment, "spec.template.spec.containers", [])
if any(container_image == x.get("image") for x in containers):
return
yield "fail", f"container image {container_image} not used by any deployment"
except Exception as e:
yield "fail", str(e)
def action_check_bundle(repo_path: Path, bundle: Bundle) -> None:
for result, message in chain(do_check_bundle_image(bundle), do_check_bundle_operator_name(bundle)):
print(f"{result.upper()}: {message}")
def do_check_bundle_semver(bundle: Bundle) -> Iterator[Tuple[str, str]]:
try:
_ = Version.parse(bundle.operator_version)
except ValueError:
yield "warn", f"Version from filesystem ({bundle.operator_version}) is not valid semver"
try:
_ = Version.parse(bundle.csv_operator_version)
except ValueError:
yield "warn", f"Version from CSV ({bundle.csv_operator_version}) is not valid semver"
def action_check_bundle(bundle: Bundle) -> None:
for result, message in chain(
do_check_bundle_semver(bundle),
do_check_bundle_operator_name(bundle),
do_check_bundle_image(bundle),
):
print(f"{result.upper()}: {bundle}: {message}")
def do_check_operator_upgrade(operator: Operator) -> Iterator[Tuple[str, str]]:
all_channels = operator.channels | {operator.default_channel} - {None}
# all_bundles = set(operator)
for channel in sorted(all_channels):
channel_bundles = operator.channel_bundles(channel)
channel_head = operator.head(channel)
graph = operator.update_graph(channel)
dangling_bundles = {x for x in channel_bundles if x not in graph and x != channel_head}
if dangling_bundles:
yield "fail", f"Channel {channel} has dangling bundles: {dangling_bundles}"
try:
channel_bundles = operator.channel_bundles(channel)
channel_head = operator.head(channel)
graph = operator.update_graph(channel)
dangling_bundles = {
x for x in channel_bundles if x not in graph and x != channel_head
}
if dangling_bundles:
yield "fail", f"Channel {channel} has dangling bundles: {dangling_bundles}."
except Exception as e:
yield "fail", str(e)
def action_check_operator(repo_path: Path, operator: Operator) -> None:
def action_check_operator(operator: Operator) -> None:
for result, message in do_check_operator_upgrade(operator):
print(f"{result.upper()}: {message}")
print(f"{result.upper()}: {operator}: {message}")
def action_check(repo_path: Path, *what: str) -> None:
def action_check(repo_path: Path, *what: str, recursive: bool = False) -> None:
repo = Repo(repo_path)
for target_name in what:
target = parse_target(repo, target_name)
for target in [parse_target(repo, x) for x in what] or sorted(repo):
print(f"Checking {target}")
if isinstance(target, Operator):
action_check_operator(repo_path, target)
action_check_operator(target)
if recursive:
for bundle in sorted(target):
print(f"Checking {bundle}")
action_check_bundle(bundle)
elif isinstance(target, Bundle):
action_check_bundle(repo_path, operator)
action_check_bundle(target)
def main() -> None:
@ -149,7 +180,7 @@ def main() -> None:
# list
list_parser = main_subparsers.add_parser(
"list", help="list contents of repo, operators or bundles"
"list", aliases=["ls"], help="list contents of repo, operators or bundles"
)
list_parser.add_argument(
"-R", "--recursive", action="store_true", help="descend the tree"
@ -162,7 +193,11 @@ def main() -> None:
# check_bundle
check_parser = main_subparsers.add_parser(
"check", help="check validity of an operator or bundle"
"check",
help="check validity of an operator or bundle",
)
check_parser.add_argument(
"-R", "--recursive", action="store_true", help="descend the tree"
)
check_parser.add_argument(
"target",
@ -182,10 +217,10 @@ def main() -> None:
)
log.addHandler(handler)
if args.action == "list":
if args.action in ("list", "ls"):
action_list(args.repo or Path.cwd(), *args.target, recursive=args.recursive)
elif args.action == "check":
action_check(args.repo or Path.cwd(), *args.target)
action_check(args.repo or Path.cwd(), *args.target, recursive=args.recursive)
else:
main_parser.print_help()

8
src/operator_repo/utils.py
View File

@ -7,6 +7,9 @@ from pathlib import Path
from typing import Any
import yaml
from yaml.composer import ComposerError
from operator_repo.exceptions import OperatorRepoException
log = logging.getLogger(__name__)
@ -33,7 +36,10 @@ def _load_yaml_strict(path: Path) -> Any:
log.debug("Loading %s", path)
with path.open("r") as yaml_file:
return yaml.safe_load(yaml_file)
try:
return yaml.safe_load(yaml_file)
except ComposerError:
raise OperatorRepoException(f"{path} contains multiple yaml documents")
def load_yaml(path: Path) -> Any:

3
tests/test_bundle.py
View File

@ -43,8 +43,7 @@ def test_bundle_compare(tmp_path: Path) -> None:
assert hello_bundle_1 != hello_bundle_2
assert hello_bundle_1 < hello_bundle_2
assert hello_bundle_1 != world_bundle_1
with pytest.raises(ValueError):
_ = hello_bundle_1 < world_bundle_1
assert hello_bundle_1 < world_bundle_1
def test_bundle_non_semver(tmp_path: Path) -> None: