Added new PulpException replacing existing errors

Author: aKlimau

CHANGES/+add-more-pulp-exceptions.feature

@@ -0,0 +1 @@
+Add more PulpExceptions. Treat DRF APIExceptions as safe in the task runner.

pulp_file/app/tasks/synchronizing.py

@@ -117,7 +117,7 @@ def synchronize(remote_pk, repository_pk, mirror, optimize=False, url=None, **kw
     optimize = optimize or settings.FILE_SYNC_OPTIMIZATION
 
     if not remote.url:
-        raise ValueError(_("A remote must have a url specified to synchronize."))
+        raise SyncError(_("A remote must have a url specified to synchronize."))
 
     if isinstance(remote, FileGitRemote):
         first_stage = GitFirstStage(remote)

pulpcore/app/tasks/replica.py

@@ -7,11 +7,13 @@
 from django.db.models import Min
 from pulp_glue.common import __version__ as pulp_glue_version
 from pulp_glue.common.context import PluginRequirement
+from pulp_glue.common.exceptions import PulpException as GluePulpException
 
 from pulpcore.app.apps import PulpAppConfig, pulp_plugin_configs
 from pulpcore.app.models import Distribution, Repository, Task, TaskGroup, UpstreamPulp
 from pulpcore.app.replica import ReplicaContext
 from pulpcore.constants import TASK_STATES
+from pulpcore.exceptions import ExternalServiceError
 from pulpcore.tasking.tasks import dispatch
 
 
@@ -66,47 +68,52 @@ def replicate_distributions(server_pk):
         "client_key": server.client_key,
     }
 
-    task_group = TaskGroup.current()
-    supported_replicators = []
-    # Load all the available replicators
-    for config in pulp_plugin_configs():
-        if config.replicator_classes:
-            for replicator_class in config.replicator_classes:
-                req = PluginRequirement(config.label, specifier=replicator_class.required_version)
-                if ctx.has_plugin(req):
-                    replicator = replicator_class(ctx, task_group, tls_settings, server)
-                    supported_replicators.append(replicator)
-
-    distro_repo_pairs = []
-    for replicator in supported_replicators:
-        distros = replicator.upstream_distributions(q=server.q_select)
-        distro_names = []
-        for distro in distros:
-            # Create remote
-            remote = replicator.create_or_update_remote(upstream_distribution=distro)
-            if not remote:
-                # The upstream distribution is not serving any content,
-                # let if fall through the cracks and be cleanup below.
-                continue
-            # Check if there is already a repository
-            repository = replicator.create_or_update_repository(remote=remote)
-            if not repository:
-                # No update occurred because server.policy==LABELED and there was
-                # an already existing local repository with the same name
-                continue
-
-            # Dispatch a sync task if needed
-            if replicator.requires_syncing(distro):
-                replicator.sync(repository, remote)
-
-            # Get or create a distribution
-            replicator.create_or_update_distribution(repository, distro)
-
-            # Add name to the list of known distribution names
-            distro_names.append(distro["name"])
-            distro_repo_pairs.append((distro["name"], str(repository.pk)))
-
-        replicator.remove_missing(distro_names)
+    try:
+        task_group = TaskGroup.current()
+        supported_replicators = []
+        # Load all the available replicators
+        for config in pulp_plugin_configs():
+            if config.replicator_classes:
+                for replicator_class in config.replicator_classes:
+                    req = PluginRequirement(
+                        config.label, specifier=replicator_class.required_version
+                    )
+                    if ctx.has_plugin(req):
+                        replicator = replicator_class(ctx, task_group, tls_settings, server)
+                        supported_replicators.append(replicator)
+
+        distro_repo_pairs = []
+        for replicator in supported_replicators:
+            distro_names = []
+            distros = replicator.upstream_distributions(q=server.q_select)
+            for distro in distros:
+                # Create remote
+                remote = replicator.create_or_update_remote(upstream_distribution=distro)
+                if not remote:
+                    # The upstream distribution is not serving any content,
+                    # let if fall through the cracks and be cleanup below.
+                    continue
+                # Check if there is already a repository
+                repository = replicator.create_or_update_repository(remote=remote)
+                if not repository:
+                    # No update occurred because server.policy==LABELED and there was
+                    # an already existing local repository with the same name
+                    continue
+
+                # Dispatch a sync task if needed
+                if replicator.requires_syncing(distro):
+                    replicator.sync(repository, remote)
+
+                # Get or create a distribution
+                replicator.create_or_update_distribution(repository, distro)
+
+                # Add name to the list of known distribution names
+                distro_names.append(distro["name"])
+                distro_repo_pairs.append((distro["name"], str(repository.pk)))
+
+            replicator.remove_missing(distro_names)
+    except GluePulpException as e:
+        raise ExternalServiceError(service_name=server.base_url, details=str(e))
 
     dispatch(
         finalize_replication,

pulpcore/app/tasks/repository.py

@@ -7,11 +7,11 @@
 
 from asgiref.sync import sync_to_async
 from django.db import transaction
-from rest_framework.serializers import ValidationError
 
 from pulpcore.app import models
 from pulpcore.app.models import ProgressReport
 from pulpcore.app.util import get_domain
+from pulpcore.exceptions.base import RepositoryVersionDeleteError
 
 log = getLogger(__name__)
 
@@ -44,12 +44,7 @@ def delete_version(pk):
             return
 
         if version.repository.versions.complete().count() <= 1:
-            raise ValidationError(
-                _(
-                    "Cannot delete repository version. Repositories must have at least one "
-                    "repository version."
-                )
-            )
+            raise RepositoryVersionDeleteError()
 
         log.info(
             "Deleting and squashing version {num} of repository '{repo}'".format(

pulpcore/content/handler.py

@@ -61,6 +61,9 @@
 from pulpcore.cache import AsyncContentCache  # noqa: E402
 from pulpcore.exceptions import (  # noqa: E402
     DigestValidationError,
+    HttpResponseError,
+    RemoteConnectionError,
+    SslConnectionError,
     UnsupportedDigestValidationError,
 )
 from pulpcore.metrics import artifacts_size_counter  # noqa: E402
@@ -915,13 +918,13 @@ async def _match_and_stream(self, path, request):
                             save_artifact=save_artifact,
                             repository=repository,
                         )
-                    except ClientResponseError as ce:
+                    except (ClientResponseError, HttpResponseError) as e:
 
                         class Error(HTTPError):
-                            status_code = ce.status
+                            status_code = e.status
 
                         reason = _("Error while fetching from upstream remote({url}): {r}").format(
-                            url=url, r=ce.message
+                            url=url, r=e.message
                         )
                         raise Error(reason=reason)
 
@@ -962,6 +965,9 @@ async def _stream_content_artifact(self, request, response, content_artifact):
             ClientResponseError,
             UnsupportedDigestValidationError,
             ClientConnectionError,
+            HttpResponseError,
+            SslConnectionError,
+            RemoteConnectionError,
         )
 
         protection_time = settings.REMOTE_CONTENT_FETCH_FAILURE_COOLDOWN

pulpcore/download/http.py

@@ -6,7 +6,12 @@
 
 from pulpcore.exceptions import (
     DigestValidationError,
+    DnsDomainNameException,
+    HttpResponseError,
+    ProxyAuthenticationError,
+    RemoteConnectionError,
     SizeValidationError,
+    SslConnectionError,
     TimeoutException,
 )
 
@@ -236,6 +241,7 @@ async def run(self, extra_data=None):
             aiohttp.ClientPayloadError,
             aiohttp.ClientResponseError,
             aiohttp.ServerDisconnectedError,
+            DnsDomainNameException,
             TimeoutError,
             TimeoutException,
             DigestValidationError,
@@ -269,9 +275,21 @@ async def download_wrapper():
                             e.message,
                         )
                     )
-                    raise e
+                    raise ProxyAuthenticationError(self.proxy)
 
-            return await download_wrapper()
+            try:
+                return await download_wrapper()
+            except aiohttp.ClientResponseError as e:
+                raise HttpResponseError(url=self.url, status=e.status, message=e.message)
+            except aiohttp.ClientConnectorSSLError as e:
+                raise SslConnectionError(url=self.url, details=str(e))
+            except (
+                aiohttp.ClientConnectorError,
+                aiohttp.ClientOSError,
+                aiohttp.ClientPayloadError,
+                aiohttp.ServerDisconnectedError,
+            ) as e:
+                raise RemoteConnectionError(url=self.url, details=str(e))
 
     async def _run(self, extra_data=None):
         """
@@ -296,10 +314,13 @@ async def _run(self, extra_data=None):
         }
         if extra_data and extra_data.get("request_kwargs"):
             request_kwargs.update(extra_data["request_kwargs"])
-        async with self.session.get(self.url, **request_kwargs) as response:
-            self.raise_for_status(response)
-            to_return = await self._handle_response(response)
-            await response.release()
+        try:
+            async with self.session.get(self.url, **request_kwargs) as response:
+                self.raise_for_status(response)
+                to_return = await self._handle_response(response)
+                await response.release()
+        except aiohttp.ClientConnectorDNSError:
+            raise DnsDomainNameException(self.url)
         if self._close_session_on_finalize:
             await self.session.close()
         return to_return

pulpcore/exceptions/__init__.py

@@ -18,6 +18,11 @@
     ReplicateError,
     SyncError,
     PublishError,
+    TaskConfigurationError,
+    TaskTimeoutException,
+    HttpResponseError,
+    SslConnectionError,
+    RemoteConnectionError,
 )
 from .validation import (
     DigestValidationError,

pulpcore/exceptions/base.py

@@ -326,3 +326,107 @@ class ReplicateError(PulpException):
 
     def __str__(self):
         return f"[{self.error_code}] " + _("Replication failed")
+
+
+class TaskConfigurationError(PulpException):
+    """
+    Raised when a task is incorrectly configured.
+    """
+
+    error_code = "PLP0023"
+
+    def __init__(self, task_name, message):
+        """
+        :param task_name: the fully qualified name of the task function
+        :type task_name: str
+        :param message: description of the configuration error
+        :type message: str
+        """
+        self.task_name = task_name
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _(
+            "Task type '{task_name}' is misconfigured: {message}"
+        ).format(task_name=self.task_name, message=self.message)
+
+
+class TaskTimeoutException(PulpException):
+    """
+    Raised when an immediate task exceeds its execution timeout.
+    """
+
+    error_code = "PLP0024"
+
+    def __init__(self, task_name, task_pk, timeout_seconds):
+        """
+        :param task_name: the fully qualified name of the task function
+        :type task_name: str
+        :param task_pk: the unique task identifier
+        :type task_pk: str
+        :param timeout_seconds: the timeout value that was exceeded
+        :type timeout_seconds: int
+        """
+        self.task_name = task_name
+        self.task_pk = task_pk
+        self.timeout_seconds = timeout_seconds
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _(
+            "Immediate task {task_pk} (type: {task_name}) timed out after {timeout} seconds."
+        ).format(task_pk=self.task_pk, task_name=self.task_name, timeout=self.timeout_seconds)
+
+
+class HttpResponseError(PulpException):
+    """
+    Raised when a remote server returns an HTTP error response after retries are exhausted.
+    """
+
+    error_code = "PLP0025"
+
+    def __init__(self, url, status, message):
+        super().__init__()
+        self.url = url
+        self.status = status
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _(
+            "HTTP error {status} when downloading {url}: {message}"
+        ).format(url=self.url, status=self.status, message=self.message)
+
+
+class SslConnectionError(PulpException):
+    """
+    Raised when an SSL/TLS connection fails after retries are exhausted.
+    """
+
+    error_code = "PLP0026"
+
+    def __init__(self, url, details):
+        super().__init__()
+        self.url = url
+        self.details = details
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("SSL connection failed for {url}: {details}").format(
+            url=self.url, details=self.details
+        )
+
+
+class RemoteConnectionError(PulpException):
+    """
+    Raised when a connection to a remote server fails after retries are exhausted.
+    """
+
+    error_code = "PLP0027"
+
+    def __init__(self, url, details):
+        super().__init__()
+        self.url = url
+        self.details = details
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("Connection failed for {url}: {details}").format(
+            url=self.url, details=self.details
+        )

pulpcore/exceptions/validation.py

@@ -96,14 +96,12 @@ class UnsupportedDigestValidationError(ValidationError):
 
     error_code = "PLP0020"
 
-    def __init__(self, digest_name=None):
-        self.digest_name = digest_name
+    def __init__(self, message=None):
+        self.message = message
 
     def __str__(self):
-        if self.digest_name:
-            return f"[{self.error_code}] " + _(
-                "Checksum type '{digest}' is not supported or enabled."
-            ).format(digest=self.digest_name)
+        if self.message:
+            return f"[{self.error_code}] {self.message}"
         return f"[{self.error_code}] " + _("Unsupported checksum type.")