Add products exporter

Resolves: AlmaLinux/build-system#354

Author: anfimovdm

alws/crud/products.py

@@ -105,7 +105,7 @@ async def create_product(
         )
     task_results = await asyncio.gather(*repo_tasks)
 
-    for repo_name, repo_url, arch, pulp_href, is_debug in task_results:
+    for repo_name, repo_url, arch, pulp_href, export_path, is_debug in task_results:
         repo = models.Repository(
             name=repo_name,
             url=repo_url,
@@ -114,6 +114,7 @@ async def create_product(
             type=arch,
             debug=is_debug,
             production=True,
+            export_path=export_path,
         )
         product.repositories.append(repo)
         items_to_insert.append(repo)

alws/dependencies.py

@@ -1,11 +1,16 @@
+from contextlib import asynccontextmanager
+from typing import AsyncGenerator
+
+from fastapi_sqla import open_async_session
 from redis import asyncio as aioredis
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from alws.config import settings
 
 __all__ = ['get_redis', 'get_async_db_key']
 
 
-async def get_redis() -> aioredis.Redis:
+async def get_redis() -> AsyncGenerator[aioredis.Redis, None]:
     client = aioredis.from_url(settings.redis_url)
     try:
         yield client
@@ -15,3 +20,9 @@ async def get_redis() -> aioredis.Redis:
 
 def get_async_db_key() -> str:
     return "async"
+
+
+@asynccontextmanager
+async def get_async_db_session() -> AsyncGenerator[AsyncSession, None]:
+    async with open_async_session(key=get_async_db_key()) as session:
+        yield session

alws/utils/copr.py

@@ -92,7 +92,7 @@ async def create_product_repo(
     platform_name: str,
     arch: str,
     is_debug: bool,
-) -> typing.Tuple[str, str, str, str, bool]:
+) -> typing.Tuple[str, str, str, str, str, bool]:
 
     debug_suffix = '-debug' if is_debug else ''
     repo_name = (
@@ -104,7 +104,8 @@ async def create_product_repo(
         create_publication=True,
         base_path_start='copr',
     )
-    return repo_name, repo_url, arch, repo_href, is_debug
+    export_path = f"{product_name}/{platform_name}/{'debug/' if is_debug else ''}{arch}/"
+    return repo_name, repo_url, arch, repo_href, export_path, is_debug
 
 
 async def create_product_sign_key_repo(

alws/utils/exporter.py

@@ -1,7 +1,7 @@
 import os
-import typing
-import urllib
+import urllib.parse
 from pathlib import Path
+from typing import List, Union
 
 import aiofiles
 import aiohttp
@@ -12,14 +12,11 @@
 
 
 async def fs_export_repository(
-    repository_ids: typing.List[int], db: AsyncSession
+    repository_ids: List[int],
+    session: AsyncSession,
 ):
-    export_task = await repo_exporter.create_pulp_exporters_to_fs(
-        db, repository_ids
-    )
-    export_data = await repo_exporter.execute_pulp_exporters_to_fs(
-        db, export_task
-    )
+    export_task = await repo_exporter.create_pulp_exporters_to_fs(session, repository_ids)
+    export_data = await repo_exporter.execute_pulp_exporters_to_fs(session, export_task)
     export_paths = list(export_data.keys())
     for repo_elem, repo_data in export_data.items():
         repo_url = urllib.parse.urljoin(repo_data, 'repodata/')
@@ -32,17 +29,16 @@ async def fs_export_repository(
 
 
 async def get_repodata_file_links(base_url: str):
-    async with aiohttp.ClientSession() as session:
+    async with aiohttp.ClientSession(raise_for_status=True) as session:
         async with session.get(base_url) as response:
-            response.raise_for_status()
             content = await response.text()
             doc = document_fromstring(content)
             children_urls = [base_url + a.get('href') for a in doc.xpath('//a')]
             return children_urls
 
 
-async def download_file(url: str, dest: str):
-    async with aiohttp.ClientSession() as session:
+async def download_file(url: str, dest: Union[str, Path]):
+    async with aiohttp.ClientSession(raise_for_status=True) as session:
         async with session.get(url) as response:
             content = await response.content.read()
         async with aiofiles.open(dest, 'wb') as f:

alws/utils/pulp_client.py

@@ -11,13 +11,15 @@
     Any,
     Dict,
     List,
+    Optional,
 )
 
 import aiohttp
 from aiohttp.client_exceptions import ClientResponseError
 from aiohttp_retry import ExponentialRetry, RetryClient
 from fastapi import status
 
+from alws.config import settings
 from alws.constants import UPLOAD_FILE_CHUNK_SIZE
 from alws.utils.file_utils import hash_content, hash_file
 from alws.utils.ids import get_random_unique_version
@@ -31,7 +33,7 @@ def __init__(
         host: str,
         username: str,
         password: str,
-        semaphore: asyncio.Semaphore = None,
+        semaphore: Optional[asyncio.Semaphore] = None,
     ):
         self.semaphore = semaphore
         self._host = host
@@ -875,8 +877,7 @@ async def list_filesystem_exporters(self):
         result = await self.request("GET", endpoint)
         if result["count"] > 0:
             return result["results"]
-        else:
-            return []
+        return []
 
     async def get_filesystem_exporter(self, fse_pulp_href: str):
         return await self.request("GET", fse_pulp_href)
@@ -1050,3 +1051,12 @@ async def request(
                     exc.message += f": {str(response_json)}"
                 raise exc
             return response_json
+
+
+def get_pulp_client(semaphore: Optional[asyncio.Semaphore] = None) -> PulpClient:
+    return PulpClient(
+        host=settings.pulp_host,
+        username=settings.pulp_user,
+        password=settings.pulp_password,
+        semaphore=semaphore,
+    )

scripts/add_export_paths_to_product_repos.py

@@ -0,0 +1,48 @@
+import os
+import re
+import sys
+
+sys.path.append(os.path.dirname(os.path.dirname(__file__)))
+
+from fastapi_sqla import open_session
+from sqlalchemy import select
+from sqlalchemy.orm import joinedload
+
+from alws.models import Product
+from alws.utils.fastapi_sqla_setup import sync_setup
+
+
+def main():
+    sync_setup()
+    with open_session() as session:
+        for product in (
+            session.execute(
+                select(Product)
+                .where(Product.is_community.is_(True))
+                .options(joinedload(Product.repositories), joinedload(Product.platforms))
+            )
+            .scalars()
+            .unique()
+            .all()
+        ):
+            if not product.repositories:
+                continue
+            platform_names = '|'.join((platform.name for platform in product.platforms))
+            platform_pattern = re.compile(
+                rf'-({platform_names})-(\w+)(-debug|)-dr$',
+                flags=re.IGNORECASE,
+            )
+            for repo in product.repositories:
+                if repo.arch == 'sign_key':
+                    continue
+                regex_result = platform_pattern.search(repo.name)
+                if not regex_result:
+                    continue
+                platform, *_ = regex_result.groups()
+                repo.export_path = (
+                    f"{product.name}/{platform}/{'debug/' if repo.debug else ''}{repo.arch}/"
+                )
+
+
+if __name__ == '__main__':
+    main()

scripts/exporters/__init__.py

@@ -0,0 +1,182 @@
+import asyncio
+import logging
+import re
+import shutil
+import sys
+import urllib.parse
+from pathlib import Path
+from typing import List, Literal, Optional, Tuple
+
+from plumbum import local
+from sqlalchemy import select
+
+from alws.config import settings
+from alws.dependencies import get_async_db_session
+from alws.models import Repository
+from alws.utils.exporter import download_file, get_repodata_file_links
+from alws.utils.pulp_client import get_pulp_client
+
+
+class BasePulpExporter:
+    def __init__(
+        self,
+        repodata_cache_dir: str,
+        logger_name: str = '',
+        log_file_path: Path = Path('/tmp/exporter.log'),
+        verbose: bool = False,
+        export_method: Literal['write', 'hardlink', 'symlink'] = 'hardlink',
+        export_path: str = settings.pulp_export_path,
+    ):
+        self.pulp_client = get_pulp_client()
+        self.export_method = export_method
+        self.export_path = export_path
+        self.createrepo_c = local["createrepo_c"]
+
+        self.repodata_cache_dir = Path(repodata_cache_dir).expanduser().absolute()
+        self.checksums_cache_dir = self.repodata_cache_dir.joinpath('checksums')
+        for dir_path in (self.repodata_cache_dir, self.checksums_cache_dir):
+            if dir_path.exists():
+                continue
+            dir_path.mkdir()
+
+        self.logger = logging.getLogger(logger_name)
+        Path(log_file_path).parent.mkdir(exist_ok=True)
+        logging.basicConfig(
+            format="%(asctime)s %(levelname)-8s %(message)s",
+            level=logging.DEBUG if verbose else logging.INFO,
+            datefmt="%Y-%m-%d %H:%M:%S",
+            handlers=[
+                logging.FileHandler(filename=log_file_path, mode="a"),
+                logging.StreamHandler(stream=sys.stdout),
+            ],
+        )
+
+    def regenerate_repo_metadata(self, repo_path: str):
+        partial_path = re.sub(str(settings.pulp_export_path), "", str(repo_path)).strip("/")
+        repodata_path = Path(repo_path, "repodata")
+        repo_repodata_cache = self.repodata_cache_dir.joinpath(partial_path)
+        cache_repodata_dir = repo_repodata_cache.joinpath("repodata")
+        self.logger.info('Repodata cache dir: %s', cache_repodata_dir)
+        args = [
+            "--update",
+            "--keep-all-metadata",
+            "--cachedir",
+            self.checksums_cache_dir,
+        ]
+        if repo_repodata_cache.exists():
+            args.extend(["--update-md-path", cache_repodata_dir])
+        args.append(repo_path)
+        self.logger.info('Starting createrepo_c')
+        _, stdout, _ = self.createrepo_c.run(args=args)
+        self.logger.info(stdout)
+        self.logger.info('createrepo_c is finished')
+        # Cache newly generated repodata into folder for future re-use
+        if not repo_repodata_cache.exists():
+            repo_repodata_cache.mkdir(parents=True)
+        else:
+            # Remove previous repodata before copying new ones
+            if cache_repodata_dir.exists():
+                shutil.rmtree(cache_repodata_dir)
+
+        shutil.copytree(repodata_path, cache_repodata_dir)
+
+    async def create_filesystem_exporters(
+        self,
+        repository_ids: List[int],
+        get_publications: bool = False,
+    ):
+        async def get_exporter_data(repository: Repository) -> Tuple[str, dict]:
+            export_path = str(Path(self.export_path, repository.export_path, "Packages"))
+            exporter_name = (
+                f"{repository.name}-{repository.arch}-debug"
+                if repository.debug
+                else f"{repository.name}-{repository.arch}"
+            )
+            fs_exporter_href = await self.pulp_client.create_filesystem_exporter(
+                exporter_name,
+                export_path,
+                export_method=self.export_method,
+            )
+
+            repo_latest_version = await self.pulp_client.get_repo_latest_version(
+                repository.pulp_href
+            )
+            if not repo_latest_version:
+                raise ValueError('cannot find latest repo version')
+            repo_exporter_dict = {
+                "repo_id": repository.id,
+                "repo_url": repository.url,
+                "repo_latest_version": repo_latest_version,
+                "exporter_name": exporter_name,
+                "export_path": export_path,
+                "exporter_href": fs_exporter_href,
+            }
+            if get_publications:
+                publications = await self.pulp_client.get_rpm_publications(
+                    repository_version_href=repo_latest_version,
+                    include_fields=["pulp_href"],
+                )
+                if publications:
+                    publication_href = publications[0].get("pulp_href")
+                    repo_exporter_dict["publication_href"] = publication_href
+            return fs_exporter_href, repo_exporter_dict
+
+        async with get_async_db_session() as session:
+            query = select(Repository).where(Repository.id.in_(repository_ids))
+            result = await session.execute(query)
+            repositories = list(result.scalars().all())
+
+        results = await asyncio.gather(*(get_exporter_data(repo) for repo in repositories))
+
+        return list(dict(results).values())
+
+    async def download_repodata(self, repodata_path, repodata_url):
+        file_links = await get_repodata_file_links(repodata_url)
+        for link in file_links:
+            file_name = Path(link).name
+            if file_name.endswith('..'):
+                continue
+            self.logger.info("Downloading repodata from %s", link)
+            await download_file(link, Path(repodata_path, file_name))
+
+    async def _export_repository(self, exporter: dict) -> Optional[str]:
+        self.logger.info(
+            "Exporting repository using following data: %s",
+            str(exporter),
+        )
+        export_path = exporter["export_path"]
+        href = exporter["exporter_href"]
+        repository_version = exporter["repo_latest_version"]
+        try:
+            await self.pulp_client.export_to_filesystem(href, repository_version)
+        except Exception:
+            self.logger.exception(
+                "Cannot export repository via %s",
+                str(exporter),
+            )
+            return
+        parent_dir = Path(export_path).parent
+        if not parent_dir.exists():
+            self.logger.info(
+                "Repository %s directory is absent",
+                exporter["exporter_name"],
+            )
+            return
+
+        repodata_path = parent_dir.joinpath("repodata").absolute()
+        repodata_url = urllib.parse.urljoin(exporter["repo_url"], "repodata/")
+        if repodata_path.exists():
+            shutil.rmtree(repodata_path)
+        repodata_path.mkdir()
+        self.logger.info('Downloading repodata from %s', repodata_url)
+        try:
+            await self.download_repodata(repodata_path, repodata_url)
+        except Exception as e:
+            self.logger.exception("Cannot download repodata file: %s", str(e))
+
+        return export_path
+
+    async def export_repositories(self, repo_ids: List[int]) -> List[str]:
+        exporters = await self.create_filesystem_exporters(repo_ids)
+        results = await asyncio.gather(*(self._export_repository(e) for e in exporters))
+        return [path for path in results if path]