RHOAIENG-26264: throttle concurrent skopeo invocations to at most 22 concurrent skopeo instances running (opendatahub-io#1165)

* RHOAIENG-26264: throttle concurrent skopeo invocations

Implements throttling for concurrent skopeo invocations in the
`scripts/update-commit-latest-env.py` script by using an asyncio.Semaphore.

This change addresses issue opendatahub-io#1158.

The script now uses a semaphore to limit the number of concurrent
`skopeo inspect` operations to 22. This helps to:
- Avoid overwhelming container registries with too many requests.
- Prevent potential rate limiting issues.
- Provide more stable execution under different network conditions.

The `get_image_vcs_ref` function was updated to accept a semaphore,
and the `inspect` function was updated to create and manage the
semaphore, passing it to the image inspection tasks.

* Revert: Revert accidental commit of test file

This commit reverts the changes made to
`manifests/base/params-latest.env` which was
accidentally included in the previous commit (feat: Throttle concurrent
skopeo invocations).

The `params-latest.env` file in that location was created
for testing purposes during development and should not be part of the
final changeset. This commit restores it to its state prior to that
accidental modification.

* Refactor: Reduce semaphore scope in get_image_vcs_ref

This commit refactors the get_image_vcs_ref function in
`scripts/update-commit-latest-env.py`.

The scope of the asyncio.Semaphore is now reduced to cover only the
`asyncio.create_subprocess_exec` call for `skopeo` and the subsequent
`process.communicate()` call.

Processing of the skopeo output (JSON parsing, label extraction, and
associated logging) has been moved outside of the semaphore block.
This change is made to improve code clarity and provide a smaller diff
for review, as per your feedback. The performance impact is expected
to be negligible for this specific script, but the change aligns with
best practices for minimizing resource holding time.

* Revert: Correct state of manifests/base/params-latest.env

This commit ensures that `manifests/base/params-latest.env` is
restored to its state prior to the recent refactoring commit
("Refactor: Reduce semaphore scope in get_image_vcs_ref").

The file `manifests/base/params-latest.env` was
unintentionally included in that refactoring commit. This commit
reverts that specific file to the version present in the preceding
commit ("Revert: Revert accidental commit of test file"), which
should represent its correct state before the recent series of changes
related to skopeo throttling.

* Revert: Ensure correct state of manifests/base/params-latest.env

This commit reverts the accidental inclusion of a test version of
`manifests/base/params-latest.env` that occurred during the
commit for a PEP 8 indentation fix ("Fix: Correct PEP 8
indentation in JSONDecodeError logging").

The file `manifests/base/params-latest.env` has been restored
to its state from the commit prior to that indentation fix,
which should be its correct, intended version.

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: jiridanek

scripts/update-commit-latest-env.py

@@ -11,7 +11,7 @@
 PROJECT_ROOT = pathlib.Path(__file__).parent.parent
 
 
-async def get_image_vcs_ref(image_url: str) -> tuple[str, str | None]:
+async def get_image_vcs_ref(image_url: str, semaphore: asyncio.Semaphore) -> tuple[str, str | None]:
     """
     Asynchronously inspects a container image's configuration using skopeo
     and extracts the 'vcs-ref' label.
@@ -32,21 +32,29 @@ async def get_image_vcs_ref(image_url: str) -> tuple[str, str | None]:
 
     logging.info(f"Starting config inspection for: {image_url}")
 
+    stdout, stderr, returncode = None, None, None
     try:
-        # Create an asynchronous subprocess
-        process = await asyncio.create_subprocess_exec(
-            *command,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE
-        )
-
-        # Wait for the command to complete and capture output
-        stdout, stderr = await process.communicate()
-
-        # Check for errors
-        if process.returncode != 0:
-            logging.error(f"Skopeo command failed for {image_url} with exit code {process.returncode}.")
-            logging.error(f"Stderr: {stderr.decode().strip()}")
+        async with semaphore:
+            logging.info(f"Semaphore acquired, starting skopeo inspect for: {image_url}")
+            # Create an asynchronous subprocess
+            process = await asyncio.create_subprocess_exec(
+                *command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE
+            )
+            # Wait for the command to complete and capture output
+            stdout, stderr = await process.communicate()
+            returncode = process.returncode
+
+        # Process the results outside the semaphore block
+        if returncode != 0:
+            logging.error(f"Skopeo command failed for {image_url} with exit code {returncode}.")
+            if stderr:
+                logging.error(f"Stderr: {stderr.decode().strip()}")
+            return image_url, None
+
+        if not stdout:
+            logging.error(f"Skopeo command returned success but stdout was empty for {image_url}.")
             return image_url, None
 
         # Decode and parse the JSON output from stdout
@@ -67,7 +75,10 @@ async def get_image_vcs_ref(image_url: str) -> tuple[str, str | None]:
         logging.error("The 'skopeo' command was not found. Please ensure it is installed and in your PATH.")
         return image_url, None
     except json.JSONDecodeError:
+        # This error can now also happen if stdout is None or not valid JSON
         logging.error(f"Failed to parse skopeo output as JSON for {image_url}.")
+        if stdout:
+            logging.debug(f"Stdout from skopeo for {image_url}: {stdout.decode(errors='replace')}")
         return image_url, None
     except Exception as e:
         logging.error(f"An unexpected error occurred while processing {image_url}: {e}")
@@ -78,7 +89,8 @@ async def inspect(images_to_inspect: typing.Iterable[str]) -> list[tuple[str, st
     """
     Main function to orchestrate the concurrent inspection of multiple images.
     """
-    tasks = [get_image_vcs_ref(image) for image in images_to_inspect]
+    semaphore = asyncio.Semaphore(22)  # Limit concurrent skopeo processes
+    tasks = [get_image_vcs_ref(image, semaphore) for image in images_to_inspect]
     return await asyncio.gather(*tasks)