feat: add a new source type named "coreos"

This commit adds a new source type named "coreos" which is capable of
fetching push items from the openshift installer JSON in GitHub.

With this, it's possible to yield `AMIPushItem` and `VHDPushItem` from
the installer, as shown in the example below:

```
pushsource-ls "coreos:https://github.com/openshift/installer/tree/release-4.22?paths=data/data/coreos/coreos-rhel-10.json"
```

The new entrypoint also support multiple paths, as shown below:

```
pushsource-ls "coreos:https://github.com/openshift/installer/tree/release-4.22?paths=data/data/coreos/coreos-rhel-9.json,data/data/coreos/coreos-rhel-10.json"
```

This is part of an effort to unblock the releases of new OpenShift
versions (greater than v4.19)

Signed-off-by: Jonathan Gangi <jgangi@redhat.com>

Assisted-by: Cursor

Author: JAVGan

docs/sources/coreos.rst

@@ -0,0 +1,31 @@
+Source: coreos
+==============
+
+The ``coreos`` push source allows the loading of content from a JSON file of OpenShift CoreOS installer from GitHub.
+
+Supported content types:
+
+* AMIs
+* VHDs
+
+
+coreos source URLs
+------------------
+
+The base form of an coreos source URL is:
+
+``coreos:github-url?paths=path1[,path2[,...]]``
+
+For example, retrieving the JSON files for the RHEL 9 and RHEL 10 CoreOS installer would look like:
+
+``coreos:https://github.com/openshift/installer/tree/master?paths=data/coreos-rhel-9.json,data/coreos-rhel-10.json``
+
+The provided GitHub URL should be the base URL of the repository,
+and should contain the tag or branch name of the repository.
+
+Python API reference
+--------------------
+
+.. autoclass:: pushsource.CoreOSSource
+   :members:
+   :special-members: __init__

src/pushsource/_impl/backend/__init__.py

@@ -5,3 +5,4 @@
 from .registry_source import RegistrySource
 from .direct import DirectSource
 from .pub_source import PubSource
+from .coreos_source import CoreOSSource

src/pushsource/_impl/backend/coreos_source/__init__.py

@@ -0,0 +1 @@
+from .coreos_source import CoreOSSource

src/pushsource/_impl/backend/coreos_source/coreos_client.py

@@ -0,0 +1,85 @@
+import json
+import logging
+import os
+import re
+import threading
+from urllib.parse import urlparse, urlunparse
+
+
+import requests
+from more_executors import Executors
+
+LOG = logging.getLogger("pushsource.coreos_client")
+
+
+class CoreOSClient(object):
+    def __init__(self, url, threads=4, timeout=60 * 60, **retry_args):
+        self._executor = (
+            Executors.thread_pool(name="pushsource-coreos-client", max_workers=threads)
+            .with_map(self._unpack_response)
+            .with_retry(**retry_args)
+            .with_timeout(timeout)
+            .with_cancel_on_shutdown()
+        )
+        self._url = url
+        self._tls = threading.local()
+
+    def shutdown(self):
+        self._executor.shutdown(True)
+
+    def _unpack_response(self, response):
+        try:
+            response.raise_for_status()
+            out = response.json()
+        except json.JSONDecodeError:
+            # do not treat invalid json as fatal error
+            out = None
+        return out
+
+    @property
+    def _session(self):
+        if not hasattr(self._tls, "session"):
+            LOG.debug(
+                "Creating requests Session for client of CoreOS service: %s", self._url
+            )
+            self._tls.session = requests.Session()
+        return self._tls.session
+
+    def _do_request(self, method, url):
+        return self._session.request(method, url)
+
+    @staticmethod
+    def convert_github_blob_to_raw(url, path):
+        """Convert github.com/.../blob/... to raw.githubusercontent.com URL."""
+        _GITHUB_BLOB_RE = re.compile(
+            r"^https?://github\.com/(?P<owner>[^/]+)/(?P<repo>[^/]+)/(blob|tree)/(?P<ref>[^/]+)(?:/(?P<path>.+))?$"
+        )
+
+        # Strip query params and fragment from the URL
+        full_url = os.path.join(url, path)
+        parsed_url = urlparse(full_url)
+        clean_url = urlunparse(parsed_url._replace(query="", fragment=""))
+
+        m = _GITHUB_BLOB_RE.match(clean_url.rstrip("/"))
+        if not m:
+            return clean_url  # already raw or non-GitHub
+        d = m.groupdict()
+
+        # Use the shorter raw form which handles branches, tags, and commit SHAs
+        # without needing to distinguish between refs/heads or refs/tags
+        raw_url = f"https://raw.githubusercontent.com/{d['owner']}/{d['repo']}/{d['ref']}/{d['path']}"
+        LOG.info("Converted URL %s to %s", parsed_url, raw_url)
+        return raw_url
+
+    def get_json_f(self, path):
+        """Returns Future[dict|list] holding json obj with VMI push items returned from CoreOS installer file.
+
+        Parameters:
+            path (str)
+                Path to the JSON file containing the CoreOS installer information.
+                Eg.: "data/data/coreos/rhcos.json"
+        """
+        url = self.convert_github_blob_to_raw(self._url, path)
+        LOG.info("Fetching VMI push items from CoreOS installer file: %s", url)
+        ft = self._executor.submit(self._do_request, method="GET", url=url)
+        return ft

src/pushsource/_impl/backend/coreos_source/coreos_source.py

@@ -0,0 +1,146 @@
+from concurrent import futures
+
+from datetime import datetime
+from more_executors import Executors
+
+from ...source import Source
+from ...helpers import (
+    force_https,
+    as_completed_with_timeout_reset,
+    list_argument,
+    try_int,
+)
+from ...model import AmiPushItem, VHDPushItem, KojiBuildInfo
+from .coreos_client import CoreOSClient
+
+
+class CoreOSSource(Source):
+    """Source for VMI push items loaded from CoreOS installer file (JSON)."""
+
+    def __init__(
+        self,
+        url,
+        paths,
+        threads=4,
+        timeout=60 * 60,
+    ):
+        """Create a new source.
+
+        Parameters:
+            url (str)
+                Base URL of for the GitHub repository containing the CoreOS installer information.
+                It should contain as well the tag or branch name of the repository.
+                Eg.: https://github.com/openshift/installer/tree/master
+            paths (list[str])
+                List of paths to the JSON files containing the CoreOS installer information.
+                Eg.: ["data/data/coreos/rhcos.json", "data/data/coreos/rhcos-rhel-10.json"]
+            threads (int)
+                Number of threads used for concurrent queries to the CoreOS installer file.
+            timeout (int)
+                Number of seconds after which an error is raised, if no progress is
+                made during queries to the CoreOS installer file.
+        """
+        self._executor = Executors.thread_pool(
+            name="pushsource-coreos", max_workers=threads
+        ).with_cancel_on_shutdown()
+        self._url = force_https(url)
+        self._client = CoreOSClient(url=self._url, threads=threads)
+        self._paths = list_argument(paths)
+        self._timeout = timeout
+
+    def _push_items_from_json(self, json_obj):
+        product_stream = json_obj.get("stream") or ""
+        product_metadata = json_obj.get("metadata") or {}
+        # Get the product name and version from the stream
+        name, version = product_stream.split("-")
+
+        # Get the release date from the metadata
+        release_date = product_metadata.get("last-modified") or ""
+        release_date = datetime.strptime(release_date, "%Y-%m-%dT%H:%M:%SZ")
+        out = []
+        vms_with_custom_meta_data = []
+        architectures = json_obj.get("architectures") or {}
+        for arch, arch_data in architectures.items():
+            images = arch_data.get("images") or {}
+            for cloud_name, image_data in images.items():
+                if cloud_name == "aws":
+                    # For AWS each region has a different AMI
+                    regions = image_data.get("regions") or {}
+                    for region, region_data in regions.items():
+                        vms_with_custom_meta_data.append(
+                            {
+                                "marketplace_name": "aws",
+                                "push_item_class": AmiPushItem,
+                                "architecture": arch,
+                                "release": region_data.get("release"),
+                                "custom_meta_data": {
+                                    "region": region,
+                                    "src": region_data.get("image"),
+                                },
+                            }
+                        )
+            # Azure images are generally in the "rhel-coreos-extensions" section
+            extensions = arch_data.get("rhel-coreos-extensions") or {}
+            for extension, extension_data in extensions.items():
+                if extension == "azure-disk":
+                    vms_with_custom_meta_data.append(
+                        {
+                            "marketplace_name": "azure",
+                            "architecture": arch,
+                            "push_item_class": VHDPushItem,
+                            "release": extension_data.get("release"),
+                            "custom_meta_data": {
+                                "src": extension_data.get("url"),
+                            },
+                        }
+                    )
+
+        for vm_item in vms_with_custom_meta_data:
+            klass = vm_item.get("push_item_class")
+            rel_klass = klass._RELEASE_TYPE
+            vm_release = vm_item.get("release") or ""
+            vm_respin = try_int(vm_release.split("-")[-1]) or 0
+            release = rel_klass(
+                arch=vm_item.get("architecture"),
+                date=release_date,
+                product=name.upper(),
+                version=version,
+                respin=vm_respin,
+            )
+            out.append(
+                klass(
+                    name=name,
+                    description=f"CoreOS image for {name} {version} on {vm_item['marketplace_name']}",
+                    **vm_item["custom_meta_data"],
+                    build_info=KojiBuildInfo(
+                        name=name,
+                        version=version,
+                        release=vm_release,
+                    ),
+                    release=release,
+                    marketplace_name=vm_item["marketplace_name"],
+                )
+            )
+
+        return out
+
+    def __iter__(self):
+        # Get all json files from the paths
+        json_fs = [self._client.get_json_f(path) for path in self._paths]
+
+        # Convert them to lists of push items
+        push_items_fs = []
+        for f in futures.as_completed(json_fs, timeout=self._timeout):
+            push_items_fs.append(
+                self._executor.submit(self._push_items_from_json, f.result())
+            )
+
+        completed_fs = as_completed_with_timeout_reset(
+            push_items_fs, timeout=self._timeout
+        )
+        for f in completed_fs:
+            for pushitem in f.result():
+                yield pushitem
+
+
+Source.register_backend("coreos", CoreOSSource)

tests/coreos/__init__.py

@@ -0,0 +1,29 @@
+import os
+import json
+from unittest import mock
+
+import pytest
+
+from pushsource._impl.backend.coreos_source.coreos_client import CoreOSClient
+
+
+@pytest.fixture
+def mock_coreos_client():
+    with mock.patch(
+        "pushsource._impl.backend.coreos_source.coreos_client.requests.Session"
+    ):
+        yield CoreOSClient(url="https://github.com/openshift/installer/tree/master")
+
+
+@pytest.fixture
+def coreos_rhel_9_data():
+    filepath = os.path.join(os.path.dirname(__file__), "data", "coreos-rhel-9.json")
+    with open(filepath, "r") as f:
+        return json.load(f)
+
+
+@pytest.fixture
+def coreos_rhel_10_data():
+    filepath = os.path.join(os.path.dirname(__file__), "data", "coreos-rhel-10.json")
+    with open(filepath, "r") as f:
+        return json.load(f)