Fix repo syncs and repo backups in our scripts

Author: anfimovdm

scripts/backup_beta_repos.py

@@ -5,42 +5,43 @@
 import pprint
 import typing
 
-from alws.utils.pulp_client import PulpClient
+from alws.utils import pulp_client as pulp_cli
 from scripts.utils.pulp import get_pulp_params
 
-
 PROG_NAME = "backup_beta_repositories"
 # We search using startswith
 PLATFORM_REPO_MAP = {
     "almalinux_9": "AlmaLinux-9-beta-AlmaLinux-9",
     "almalinux_10": "eabdullin1-almalinux10-beta-almalinux-10",
+    "almalinux_10_epel": "eabdullin1-epel-al-almalinux-10",
 }
 
 ReposType = typing.List[typing.Dict[str, typing.Any]]
 
+
 async def find_pulp_repos(
-        name_starts: str,
-        pulp_client: typing.Optional[PulpClient] = None
+    name_starts: str, pulp_client: typing.Optional[pulp_cli.PulpClient] = None
 ) -> ReposType:
     if not pulp_client:
         host, user, password = get_pulp_params()
-        pulp_client = PulpClient(host, user, password)
+        pulp_client = pulp_cli.PulpClient(host, user, password)
     repositories = await pulp_client.get_rpm_repositories_by_params(
-        {"name__startswith": name_starts})
+        {"name__startswith": name_starts}
+    )
     repositories = [r for r in repositories if "backup" not in r["name"]]
 
     return repositories
 
 
 async def create_pulp_backup_repos(
-        repos: ReposType,
-        dry_run: bool = False,
-        pulp_client: typing.Optional[PulpClient] = None
+    repos: ReposType,
+    dry_run: bool = False,
+    pulp_client: typing.Optional[pulp_cli.PulpClient] = None,
 ) -> typing.Dict[str, typing.Dict[str, typing.Any]]:
     logger = logging.getLogger(PROG_NAME)
     if not pulp_client:
         host, user, password = get_pulp_params()
-        pulp_client = PulpClient(host, user, password)
+        pulp_client = pulp_cli.PulpClient(host, user, password)
 
     result = {}
     for repo in repos:
@@ -51,8 +52,9 @@ async def create_pulp_backup_repos(
         backup_repo = await pulp_client.get_rpm_repository(backup_repo_name)
         if not backup_repo:
             url, href = await pulp_client.create_rpm_repository(
-                backup_repo_name, create_publication=True,
-                base_path_start="backups"
+                backup_repo_name,
+                create_publication=True,
+                base_path_start="backups",
             )
             logger.info("Backup repository URL: %s, href: %s", url, href)
             backup_repo = await pulp_client.get_rpm_repository(backup_repo_name)
@@ -64,7 +66,8 @@ async def _main(repos_to_backup, dry_run: bool = False):
     logger = logging.getLogger(PROG_NAME)
     logger.debug("Acquiring Pulp connection data and creating client")
     host, user, password = get_pulp_params()
-    pulp_client = PulpClient(host, user, password)
+    pulp_cli.PULP_SEMAPHORE = asyncio.Semaphore(10)
+    pulp_client = pulp_cli.PulpClient(host, user, password)
     logger.info("Searching for all beta repositories")
     repos = []
     for repo_to_backup in repos_to_backup:
@@ -82,13 +85,9 @@ async def _main(repos_to_backup, dry_run: bool = False):
         version_href = repo["latest_version_href"]
         logger.debug("Version href: %s", version_href)
         pkgs = await pulp_client.get_rpm_packages(
-            include_fields=fields,
-            repository_version=version_href,
-            limit=1000
-        )
-        modules = await pulp_client.get_modules(
-            repository_version=version_href
+            include_fields=fields, repository_version=version_href, limit=1000
         )
+        modules = await pulp_client.get_modules(repository_version=version_href)
         for entity in itertools.chain(pkgs, modules):
             hrefs.append(entity["pulp_href"])
         backup_repo = backup_repos[repo["name"]]
@@ -109,7 +108,8 @@ async def _main(repos_to_backup, dry_run: bool = False):
                 )
             remove_tasks.append(
                 pulp_client.modify_repository(
-                    repo["pulp_href"], add=[], remove=hrefs)
+                    repo["pulp_href"], add=[], remove=hrefs
+                )
             )
             if not repo.get("autopublish", False):
                 publications_tasks.append(
@@ -132,16 +132,27 @@ async def _main(repos_to_backup, dry_run: bool = False):
 
 def main():
     parser = argparse.ArgumentParser(PROG_NAME)
-    parser.add_argument("-v", "--verbose", action="store_true", default=False,
-                        help="Enable verbose output")
-    parser.add_argument("-d", "--dry-run", action="store_true", default=False,
-                        help="Output everything that will happen, "
-                             "but do not create/modify anything")
+    parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        default=False,
+        help="Enable verbose output",
+    )
+    parser.add_argument(
+        "-d",
+        "--dry-run",
+        action="store_true",
+        default=False,
+        help="Output everything that will happen, "
+        "but do not create/modify anything",
+    )
     for platform in PLATFORM_REPO_MAP:
         parser.add_argument(
-            f"--{platform}", f"--{platform.replace('_', '-')}",
+            f"--{platform}",
+            f"--{platform.replace('_', '-')}",
             help=f"Create backups of {platform} beta repos",
-            action="store_true"
+            action="store_true",
         )
     args = parser.parse_args()
 

scripts/bootstrap_repositories.py

@@ -1,3 +1,4 @@
+import asyncio
 import os
 import sys
 import typing
@@ -147,7 +148,9 @@ async def get_repository(
 async def get_remote(repo_info: dict, remote_sync_policy: str):
     async with open_async_session(key=get_async_db_key()) as db:
         remote_payload = repo_info.copy()
-        remote_payload["name"] = f'{repo_info["name"]}-{repo_info["arch"]}'
+        remote_payload["name"] = (
+            f'{repo_info["name"]}-{repo_info["arch"]}-{remote_sync_policy}'
+        )
         remote_payload.pop("type", None)
         remote_payload.pop("debug", False)
         remote_payload.pop("production", False)
@@ -231,6 +234,43 @@ async def add_repositories_to_platform(
         )
 
 
+async def sync_repositories(repo_sync_list: list, pulp_client: PulpClient):
+    sync_tasks = []
+    publish_tasks = []
+
+    async def sync_repo(repo_, remote_, policy):
+        logging.info('Syncing repository %s from %s', repo_, remote_)
+        try:
+            await pulp_client.sync_rpm_repo_from_remote(
+                repo_,
+                remote_,
+                policy,
+                wait_for_result=True,
+            )
+        except Exception:
+            logging.exception('Cannot sync repository %s', repo_)
+
+    async def publish_repo(repo_):
+        logging.info('Publishing repo %s', repo_)
+        try:
+            await pulp_client.create_rpm_publication(repo['repo_href'])
+        except Exception:
+            logging.exception('Cannot publish repository %s', repo_)
+
+    for repo in repo_sync_list:
+        sync_tasks.append(
+            sync_repo(
+                repo['repo_href'],
+                repo['remote_href'],
+                repo['sync_policy'],
+            )
+        )
+        publish_tasks.append(publish_repo(repo['repo_href']))
+    async with asyncio.Semaphore(5):
+        await asyncio.gather(*sync_tasks)
+        await asyncio.gather(*publish_tasks)
+
+
 def main():
     pulp_host = os.environ["PULP_HOST"]
     pulp_user = os.environ["PULP_USER"]
@@ -302,6 +342,7 @@ def main():
                     repo["pulp_href"],
                 )
 
+        repos_to_sync = []
         for repo_info in repositories_data:
             logger.info(
                 "Creating repository from the following data: %s",
@@ -336,7 +377,7 @@ def main():
             remote = sync(get_remote(repo_info, remote_sync_policy))
             pulp_remote = sync(
                 pulp_client.get_rpm_remote(
-                    f'{repo_info["name"]}-{repo_info["arch"]}',
+                    f'{repo_info["name"]}-{repo_info["arch"]}-{remote_sync_policy}',
                 )
             )
             if pulp_remote['pulp_href'] != remote.pulp_href:
@@ -354,22 +395,16 @@ def main():
             if args.no_sync:
                 logger.info("Synchronization from remote is disabled, skipping")
                 continue
-            try:
-                logger.info("Syncing %s from %s...", repository, remote)
-                sync(
-                    pulp_client.sync_rpm_repo_from_remote(
-                        repository.pulp_href,
-                        remote.pulp_href,
-                        sync_policy=repo_sync_policy,
-                        wait_for_result=True,
-                    )
-                )
-                sync(pulp_client.create_rpm_publication(repository.pulp_href))
-                logger.info("Repository %s sync is completed", repository)
-            except Exception as e:
-                logger.info(
-                    "Repository %s sync is failed: \n%s", repository, str(e)
-                )
+            logger.info("Appending %s to sync from %s...", repository, remote)
+            repo_sync_info = {
+                'repo_href': repository.pulp_href,
+                'remote_href': remote.pulp_href,
+                'sync_policy': repo_sync_policy,
+            }
+            repos_to_sync.append(repo_sync_info)
+        if repos_to_sync and not args.no_sync:
+            sync(sync_repositories(repos_to_sync, pulp_client))
+            return
         sync(add_repositories_to_platform(platform_data, repository_ids))
         sync(add_owner_id())