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

Author: omer-vishlitzky

.gitignore

@@ -25,3 +25,4 @@ go.work
 *.swp
 *.swo
 *~
+scripts/__pycache__/*

scripts/ansible_test_runner.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+
+import os
+import sys
+import subprocess
+import logging
+import yaml
+from datetime import datetime
+from shared_types import ReleaseCandidates, SnapshotInfo
+
+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() -> ReleaseCandidates:
+    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: ReleaseCandidates) -> tuple[str, SnapshotInfo] | None:
+    for snapshot in data["snapshots"]:
+        if "metadata" in snapshot and snapshot["metadata"].get("status") == "pending":
+            return snapshot["metadata"].get("generated_at"), snapshot
+    return None, None
+
+def export_component_variables(snapshot: SnapshotInfo) -> 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_IMAGE",
+        "openshift/assisted-image-service": "ASSISTED_IMAGE_SERVICE_IMAGE",
+        "openshift/assisted-installer-agent": "ASSISTED_INSTALLER_AGENT_IMAGE",
+        "openshift/assisted-installer-controller": "ASSISTED_INSTALLER_CONTROLLER_IMAGE",
+        "openshift/assisted-installer": "ASSISTED_INSTALLER_IMAGE",
+    }
+    for component in snapshot["components"]:
+        repo = component["repository"]
+        env_var = component_env_map.get(repo)
+        if not env_var:
+            logger.warning(
+                f"No environment variable mapping for component: {component}"
+            )
+            continue
+
+        if (
+            repo == "kubernetes-sigs/cluster-api"
+            or repo == "metal3-io/cluster-api-provider-metal3"
+        ):
+            value = component["ref"]
+        else:
+            value = component["image_url"]
+
+        if value:
+            env_vars[env_var] = value
+            os.environ[env_var] = value
+            logger.info(f"Exported {env_var}={value}")
+
+
+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(timestamp: str, success: bool) -> None:
+    try:
+        data: ReleaseCandidates = read_release_candidates()
+        
+        for snapshot in data["snapshots"]:
+            if snapshot.get("metadata", {}).get("generated_at") == timestamp:
+                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 from {timestamp} status to {'successful' if success else 'failed'}")
+                return
+                
+        logger.error(f"Snapshot with timestamp {timestamp} not found")
+    except Exception as e:
+        logger.error(f"Error updating snapshot status: {e}")
+
+
+def main() -> None:
+    try:
+        logger.info("Starting test runner")
+        data: ReleaseCandidates = read_release_candidates()
+
+        timestamp, snapshot = find_pending_snapshot(data)
+        if snapshot is None:
+            logger.info("No pending snapshots to process")
+            return
+
+        logger.info(f"Processing pending snapshot from {timestamp}")
+        export_component_variables(snapshot)
+
+        success = run_ansible_tests()
+        update_snapshot_status(timestamp, 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()

scripts/github_auth.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+
+import os
+import logging
+from github import Github, GithubIntegration
+
+logger = logging.getLogger("github-auth")
+
+def get_github_client():
+    """
+    Creates a PyGithub client using GitHub App authentication
+    
+    Environment variables:
+        GITHUB_APP_ID: The ID of your GitHub App
+        GITHUB_APP_INSTALLATION_ID: The installation ID for your GitHub App
+        GITHUB_APP_PRIVATE_KEY: Private key .pem file
+    """
+    try:
+        app_id = os.environ["GITHUB_APP_ID"]
+        installation_id = os.environ["GITHUB_APP_INSTALLATION_ID"]
+        private_key = os.environ["GITHUB_APP_PRIVATE_KEY"]
+        
+        if not all([app_id, installation_id, private_key]):
+            logger.error("Missing GitHub App credentials in environment")
+            raise ValueError("""
+                Missing GitHub App credentials.
+                Ensure GITHUB_APP_ID, GITHUB_APP_INSTALLATION_ID, and GITHUB_APP_PRIVATE_KEY are set.
+            """)
+            
+        integration = GithubIntegration(
+            int(app_id),
+            private_key
+        )
+        access_token = integration.get_access_token(int(installation_id))
+        github_client = Github(access_token.token)
+        logger.debug("Created GitHub client with new token")
+        return github_client
+    except Exception as e:
+        logger.error(f"Error creating GitHub client: {e}")
+        raise

scripts/shared_types.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+
+from typing import TypedDict
+
+class RepositoryInfo(TypedDict):
+    repository: str
+    ref: str
+    image_url: str | None
+
+class SnapshotMetadata(TypedDict):
+    generated_at: str
+    status: str
+    tested_at: str
+
+class SnapshotInfo(TypedDict):
+    metadata: SnapshotMetadata
+    components: list[RepositoryInfo]
+
+class ReleaseCandidates(TypedDict):
+    snapshots: list[SnapshotInfo]
+
+class VersionEntry(TypedDict):
+    name: str
+    components: list[RepositoryInfo]
+
+class VersionsFile(TypedDict):
+    components: list[VersionEntry]
+
+class RepositoryConfig(TypedDict, total=False):
+    images: list[str] | None
+    version_prefix: str

scripts/tag_reconciler.py

@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+
+import re
+import os
+import sys
+import logging
+from ruamel.yaml import YAML
+from github_auth import get_github_client
+from shared_types import RepositoryInfo, VersionEntry, VersionsFile
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("tag-reconciler")
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+PROJECT_ROOT = os.path.dirname(SCRIPT_DIR)
+VERSIONS_FILE = os.path.join(PROJECT_ROOT, "versions.yaml")
+
+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 __init__(self):
+        self.github_client = get_github_client()
+    
+    def read_versions_file(self) -> VersionsFile:
+        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:
+        try:
+            github_repo = self.github_client.get_repo(repo)
+            github_repo.get_git_ref(f"tags/{tag}")
+            return True
+        except Exception as e:
+            logger.error(f"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"
+        
+        try:
+            github_repo = self.github_client.get_repo(repo)
+            tag_obj = github_repo.create_git_tag(
+                tag=tag,
+                message=message,
+                object=commit_sha,
+                type="commit"
+            )
+            
+            github_repo.create_git_ref(f"refs/tags/{tag}", tag_obj.sha)
+            logger.info(f"Created tag {tag} for {repo}")
+            return True
+        except Exception as e:
+            logger.error(f"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: VersionEntry) -> bool:
+        version_name: str = version["name"]
+        if not version_name:
+            logger.error("Version entry missing 'name' field")
+            return False
+
+        logger.info(f"Processing version {version_name}")
+        success = True
+
+        components = version["components"]
+        if not components:
+            logger.warning(f"No repositories found for version {version_name}")
+            return True
+
+        for component in components:
+            repo_name: str = component["repository"]
+            ref: str = component["ref"]
+
+            if not re.match(r"^openshift/", repo_name) or not ref:
+                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, ref, 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: VersionsFile = 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_entry in versions:
+            if not self.reconcile_tag(version_entry):
+                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()