MGMT-19400 - create tag management flow to tag assisted installer components according to capbcoa versions

Author: omer-vishlitzky

scripts/tag_reconciler.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+
+import re
+import os
+import sys
+import requests
+import logging
+from typing import Any
+from ruamel.yaml import YAML
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("tag-reconciler")
+
+VERSIONS_FILE = "versions.yaml"
+GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
+
+yaml = YAML(typ="rt")
+yaml.default_flow_style = False
+yaml.explicit_start = False
+yaml.preserve_quotes = True
+yaml.width = 4096
+yaml.indent(mapping=2, sequence=2, offset=0)
+
+
+class TagReconciler:
+    def read_versions_file(self) -> dict[str, Any]:
+        if not os.path.exists(VERSIONS_FILE):
+            logger.error(f"{VERSIONS_FILE} does not exist")
+            sys.exit(1)
+
+        try:
+            with open(VERSIONS_FILE, "r") as f:
+                data = yaml.load(f)
+            return data or {}
+        except Exception as e:
+            logger.error(f"Error reading {VERSIONS_FILE}: {e}")
+            return {}
+
+    def tag_exists(self, repo: str, tag: str) -> bool:
+        url = f"https://api.github.com/repos/{repo}/git/refs/tags/{tag}"
+        headers = {
+            "Accept": "application/vnd.github+json",
+            "Authorization": f"token {GITHUB_TOKEN}",
+            "X-GitHub-Api-Version": "2022-11-28",
+        }
+
+        try:
+            response = requests.get(url, headers=headers)
+            return response.status_code == 200
+        except requests.RequestException as e:
+            logger.error(f"Request error checking tag {tag} for {repo}: {e}")
+            return False
+
+    def create_tag(self, repo: str, commit_sha: str, tag: str) -> bool:
+        message = f"Version {tag} - Tagged by CAPBCOA CI"
+        url = f"https://api.github.com/repos/{repo}/git/tags"
+        headers = {
+            "Accept": "application/vnd.github+json",
+            "Authorization": f"Bearer {GITHUB_TOKEN}",
+            "X-GitHub-Api-Version": "2022-11-28",
+        }
+        data = {
+            "tag": tag,
+            "message": message,
+            "object": commit_sha,
+            "type": "commit",
+        }
+
+        try:
+            response = requests.post(url, headers=headers, json=data)
+            if response.status_code != 201:
+                logger.error(
+                    f"Failed to create tag object: {response.status_code} {response.text}"
+                )
+                return False
+
+            tag_obj = response.json()
+            tag_sha = tag_obj["sha"]
+
+            ref_url = f"https://api.github.com/repos/{repo}/git/refs"
+            ref_data = {
+                "ref": f"refs/tags/{tag}",
+                "sha": tag_sha,
+            }
+
+            ref_response = requests.post(ref_url, headers=headers, json=ref_data)
+            if ref_response.status_code != 201:
+                logger.error(
+                    f"Failed to create tag reference: {ref_response.status_code}"
+                )
+                return False
+
+            logger.info(f"Created tag {tag} for {repo}")
+            return True
+
+        except requests.RequestException as e:
+            logger.error(f"Request error creating tag {tag} for {repo}: {e}")
+            return False
+
+    def extract_base_repo(self, full_repo_name: str) -> str:
+        parts = full_repo_name.split("/")
+        if len(parts) > 2:
+            return f"{parts[0]}/{parts[1]}"
+        return full_repo_name
+
+    def reconcile_tag(self, version: dict[str, Any]) -> bool:
+        version_name = version.get("name", "")
+        if not version_name:
+            logger.error("Version entry missing 'name' field")
+            return False
+
+        logger.info(f"Processing version {version_name}")
+        success = True
+
+        repositories = version.get("repositories", {})
+        if not repositories:
+            logger.warning(f"No repositories found for version {version_name}")
+            return True
+
+        for repo_name, repo_info in repositories.items():
+            commit_sha = repo_info.get("commit_sha", "")
+
+            if not re.match(r"^openshift/", repo_name) or not commit_sha:
+                logger.warning(f"Skipping repository {repo_name}")
+                continue
+
+            base_repo = self.extract_base_repo(repo_name)
+            if self.tag_exists(base_repo, version_name):
+                logger.info(f"Tag {version_name} already exists for {base_repo}")
+                continue
+            if not self.create_tag(base_repo, commit_sha, version_name):
+                logger.error(f"Failed to create tag {version_name} for {base_repo}")
+                success = False
+
+        return success
+
+    def reconcile_tags(self) -> bool:
+        versions_data = self.read_versions_file()
+        versions = versions_data.get("versions", [])
+
+        if not versions:
+            logger.info("No versions to process")
+            return True
+
+        logger.info(f"Processing {len(versions)} versions")
+        success = True
+
+        for version in versions:
+            if not self.reconcile_tag(version):
+                success = False
+
+        return success
+
+
+def main() -> None:
+    try:
+        logger.info("Starting tag reconciliation")
+        reconciler = TagReconciler()
+        success = reconciler.reconcile_tags()
+
+        if not success:
+            logger.error("Tag reconciliation completed with errors")
+            sys.exit(1)
+
+    except Exception as e:
+        logger.error(f"Error in tag reconciliation: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

scripts/test_runner.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+
+import os
+import sys
+import subprocess
+import logging
+import yaml
+from datetime import datetime
+from typing import Any
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("test-runner")
+
+RELEASE_CANDIDATES_FILE = "release-candidates.yaml"
+
+
+def read_release_candidates() -> dict[str, Any]:
+    try:
+        with open(RELEASE_CANDIDATES_FILE, "r") as f:
+            data = yaml.safe_load(f)
+        if not data or "snapshots" not in data:
+            logger.error(f"Invalid format in {RELEASE_CANDIDATES_FILE}")
+            sys.exit(1)
+        return data
+    except Exception as e:
+        logger.error(f"Error reading {RELEASE_CANDIDATES_FILE}: {e}")
+        sys.exit(1)
+
+
+def find_pending_snapshot(data: dict[str, Any]) -> tuple[int, dict[str, Any]] | None:
+    for i, snapshot in enumerate(data["snapshots"]):
+        if "metadata" in snapshot and snapshot["metadata"].get("status") == "pending":
+            return i, snapshot
+    return None, None
+
+
+def export_component_variables(snapshot: dict[str, Any]) -> None:
+    env_vars = {}
+    component_env_map = {
+        "kubernetes-sigs/cluster-api": "CAPI_VERSION",
+        "metal3-io/cluster-api-provider-metal3": "CAPM3_VERSION",
+        "openshift/assisted-service/assisted-service": "ASSISTED_SERVICE_IMAGE",
+        "openshift/assisted-image-service/assisted-image-service": "ASSISTED_IMAGE_SERVICE_IMAGE",
+        "openshift/assisted-installer-agent/assisted-installer-agent": "ASSISTED_INSTALLER_AGENT_IMAGE",
+        "openshift/assisted-installer/assisted-installer-controller": "ASSISTED_INSTALLER_CONTROLLER_IMAGE",
+        "openshift/assisted-installer/assisted-installer": "ASSISTED_INSTALLER_IMAGE",
+    }
+    for component, info in snapshot["versions"].items():
+        env_var = component_env_map.get(component)
+        if not env_var:
+            logger.warning(
+                f"No environment variable mapping for component: {component}"
+            )
+            continue
+
+        if (
+            "kubernetes-sigs/cluster-api" in component
+            or "metal3-io/cluster-api-provider-metal3" in component
+        ):
+            value = info["version"]
+        else:
+            value = info.get("image_url", "")
+
+        if value:
+            env_vars[env_var] = value
+            os.environ[env_var] = value
+            logger.info(f"Exported {env_var}={value}")
+    return env_vars
+
+
+def run_ansible_tests() -> bool:
+    try:
+        logger.info("Running Ansible tests...")
+        ansible_cmd = [
+            "ansible-playbook",
+            "test/ansible/run_test.yaml",
+            "-i",
+            "test/ansible/inventory.yaml",
+        ]
+        result = subprocess.run(
+            ansible_cmd, check=False, capture_output=False, text=True
+        )
+
+        if result.returncode == 0:
+            logger.info("Ansible tests passed successfully")
+            return True
+        else:
+            logger.error(f"Ansible tests failed with return code {result.returncode}")
+            return False
+
+    except Exception as e:
+        logger.error(f"Error running Ansible tests: {e}")
+        return False
+
+
+def update_snapshot_status(snapshot_index: int, success: bool) -> None:
+    try:
+        data = read_release_candidates()
+
+        if snapshot_index >= len(data["snapshots"]):
+            logger.error(f"Snapshot index {snapshot_index} out of range")
+            return
+
+        snapshot = data["snapshots"][snapshot_index]
+        if "metadata" not in snapshot:
+            snapshot["metadata"] = {}
+
+        snapshot["metadata"]["status"] = "successful" if success else "failed"
+        snapshot["metadata"]["tested_at"] = datetime.now().isoformat()
+
+        with open(RELEASE_CANDIDATES_FILE, "w") as f:
+            yaml.dump(data, f, default_flow_style=False)
+
+        logger.info(
+            f"Updated snapshot {snapshot_index} status to {'successful' if success else 'failed'}"
+        )
+
+    except Exception as e:
+        logger.error(f"Error updating snapshot status: {e}")
+
+
+def main() -> None:
+    try:
+        logger.info("Starting test runner")
+        data = read_release_candidates()
+
+        snapshot_index, snapshot = find_pending_snapshot(data)
+        if snapshot is None:
+            logger.info("No pending snapshots to process")
+            return
+
+        logger.info(f"Processing pending snapshot {snapshot_index}")
+        export_component_variables(snapshot)
+
+        success = run_ansible_tests()
+        update_snapshot_status(snapshot_index, success)
+
+        if success:
+            logger.info("Snapshot processed successfully")
+        else:
+            logger.error("Snapshot testing failed")
+            sys.exit(1)
+
+    except Exception as e:
+        logger.error(f"Test runner failed: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()