swh/core/api/__init__.py

-# Copyright (C) 2015-2023  The Software Heritage developers
+# Copyright (C) 2015-2024  The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
@@ -26,8 +26,12 @@ from deprecated import deprecated
 from flask import Flask, Request, Response, abort, request
 import requests
 import sentry_sdk
+from tenacity.before_sleep import before_sleep_log
+from tenacity.wait import wait_fixed
 from werkzeug.exceptions import HTTPException
 
+from swh.core.retry import http_retry, retry_if_exception
+
 from .negotiation import Formatter as FormatterBase
 from .negotiation import Negotiator as NegotiatorBase
 from .negotiation import negotiate as _negotiate
@@ -41,6 +45,8 @@ from .serializers import (
     msgpack_loads,
 )
 
+RETRY_WAIT_INTERVAL = 10
+
 logger = logging.getLogger(__name__)
 
 
@@ -211,6 +217,10 @@ class RPCClient(metaclass=MetaRPCClient):
       reraise_exceptions: On server errors, if any of the exception classes in
         this list has the same name as the error name, then the exception will
         be instantiated and raised instead of a generic RemoteException.
+      enable_requests_retry: If set to :const:`True`, requests sent by the client will
+        be retried when encountering specific errors. Default policy is to retry when
+        connection errors or transient remote exceptions are raised. Subclasses can
+        change that policy by overriding the :meth:`retry_policy` method.
 
     """
 
@@ -238,6 +248,11 @@ class RPCClient(metaclass=MetaRPCClient):
     extra_type_decoders: Dict[str, Callable] = {}
     """Value of `extra_decoders` passed to `json_loads` or `msgpack_loads`
     to be able to deserialize more object types."""
+    enable_requests_retry: bool = False
+    """If set to :const:`True`, requests sent by the client will be retried
+    when encountering specific errors. Default policy is to retry when connection
+    errors or transient remote exceptions are raised. Subclasses can change that
+    policy by overriding the :meth:`retry_policy` method."""
 
     def __init__(
         self,
@@ -250,12 +265,15 @@ class RPCClient(metaclass=MetaRPCClient):
         adapter_kwargs: Optional[Dict[str, Any]] = None,
         api_exception: Optional[Type[Exception]] = None,
         reraise_exceptions: Optional[List[Type[Exception]]] = None,
+        enable_requests_retry: Optional[bool] = None,
         **kwargs,
     ):
         if api_exception:
             self.api_exception = api_exception
         if reraise_exceptions:
             self.reraise_exceptions = reraise_exceptions
+        if enable_requests_retry is not None:
+            self.enable_requests_retry = enable_requests_retry
         base_url = url if url.endswith("/") else url + "/"
         self.url = base_url
 
@@ -278,6 +296,28 @@ class RPCClient(metaclass=MetaRPCClient):
 
         self.chunk_size = chunk_size
 
+        if self.enable_requests_retry:
+
+            retry = http_retry(
+                retry=self.retry_policy,
+                wait=wait_fixed(RETRY_WAIT_INTERVAL),
+                before_sleep=before_sleep_log(logger, logging.WARNING),
+            )
+            setattr(self, "_get", retry(self._get))
+            setattr(self, "_post", retry(self._post))
+
+    def retry_policy(self, retry_state):
+        return retry_if_exception(
+            retry_state,
+            lambda e: (
+                isinstance(e, TransientRemoteException)
+                or (
+                    isinstance(e, self.api_exception)
+                    and isinstance(e.args[0], requests.exceptions.ConnectionError)
+                )
+            ),
+        )
+
     def _url(self, endpoint):
         return "%s%s" % (self.url, endpoint)
 

swh/core/api/tests/test_rpc_client.py

-# Copyright (C) 2018-2022  The Software Heritage developers
+# Copyright (C) 2018-2024  The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
@@ -9,6 +9,7 @@ import pytest
 from requests.exceptions import ConnectionError
 
 from swh.core.api import (
+    RETRY_WAIT_INTERVAL,
     APIError,
     RemoteException,
     RPCClient,
@@ -16,6 +17,7 @@ from swh.core.api import (
     remote_api_endpoint,
 )
 from swh.core.api.serializers import exception_to_dict, msgpack_dumps
+from swh.core.retry import MAX_NUMBER_ATTEMPTS
 
 from .test_serializers import ExtraType, extra_decoders, extra_encoders
 
@@ -48,6 +50,7 @@ def rpc_client_class(requests_mock):
         extra_type_encoders = extra_encoders
         extra_type_decoders = extra_decoders
         reraise_exceptions = [ReraiseException]
+        enable_requests_retry = True
 
         def overridden_method(self, data):
             return "bar"
@@ -115,10 +118,11 @@ def test_client_request_too_large(rpc_client):
     assert exc_info.value.args[1].status_code == 413
 
 
-def test_client_connexion_error(rpc_client, requests_mock):
+def test_client_connexion_error(rpc_client, requests_mock, mocker):
     """
     ConnectionError should be wrapped and raised as an APIError.
     """
+    mock_sleep = mocker.patch("time.sleep")
     error_message = "unreachable host"
     requests_mock.post(
         re.compile("mock://example.com/connection_error"),
@@ -131,6 +135,14 @@ def test_client_connexion_error(rpc_client, requests_mock):
     assert type(exc_info.value.args[0]) is ConnectionError
     assert str(exc_info.value.args[0]) == error_message
 
+    # check request retries on connection errors
+    mock_sleep.assert_has_calls(
+        [
+            mocker.call(param)
+            for param in [RETRY_WAIT_INTERVAL] * (MAX_NUMBER_ATTEMPTS - 1)
+        ]
+    )
+
 
 def _exception_response(exception, status_code):
     def callback(request, context):
@@ -144,10 +156,11 @@ def _exception_response(exception, status_code):
     return callback
 
 
-def test_client_reraise_exception(rpc_client, requests_mock):
+def test_client_reraise_exception(rpc_client, requests_mock, mocker):
     """
     Exception caught server-side and whitelisted will be raised again client-side.
     """
+    mock_sleep = mocker.patch("time.sleep")
     error_message = "something went wrong"
     endpoint = "reraise_exception"
 
@@ -163,10 +176,12 @@ def test_client_reraise_exception(rpc_client, requests_mock):
         rpc_client._post(endpoint, data={})
 
     assert str(exc_info.value) == error_message
+    # no request retry for such exception
+    mock_sleep.assert_not_called()
 
 
 @pytest.mark.parametrize("status_code", [400, 500, 502, 503])
-def test_client_raise_remote_exception(rpc_client, requests_mock, status_code):
+def test_client_raise_remote_exception(rpc_client, requests_mock, status_code, mocker):
     """
     Exception caught server-side and not whitelisted will be wrapped and raised
     as a RemoteException client-side.
@@ -181,6 +196,7 @@ def test_client_raise_remote_exception(rpc_client, requests_mock, status_code):
             status_code=status_code,
         ),
     )
+    mock_sleep = mocker.patch("time.sleep")
 
     with pytest.raises(RemoteException) as exc_info:
         rpc_client._post(endpoint, data={})
@@ -189,8 +205,17 @@ def test_client_raise_remote_exception(rpc_client, requests_mock, status_code):
     assert str(exc_info.value.args[0]["message"]) == error_message
     if status_code in (502, 503):
         assert isinstance(exc_info.value, TransientRemoteException)
+        # check request retry on transient remote exception
+        mock_sleep.assert_has_calls(
+            [
+                mocker.call(param)
+                for param in [RETRY_WAIT_INTERVAL] * (MAX_NUMBER_ATTEMPTS - 1)
+            ]
+        )
     else:
         assert not isinstance(exc_info.value, TransientRemoteException)
+        # no request retry on other remote exceptions
+        mock_sleep.assert_not_called()
 
 
 @pytest.mark.parametrize(

swh/core/cli/db.py

@@ -6,6 +6,7 @@
 
 import logging
 from os import environ
+from subprocess import CalledProcessError
 from typing import Any, Dict, List, Optional, Tuple
 import warnings
 
@@ -340,9 +341,17 @@ def initialize_one(package, module, backend_class, flavor, dbname, cfg):
 
     logger.debug("db_init %s flavor=%s dbname=%s", module, flavor, dbname)
     dbmodule = f"{package}:{cfg['cls']}"
-    initialized, dbversion, dbflavor = populate_database_for_package(
-        dbmodule, dbname, flavor
-    )
+    try:
+        initialized, dbversion, dbflavor = populate_database_for_package(
+            dbmodule, dbname, flavor
+        )
+    except CalledProcessError as exc:
+        click.secho("Error during database setup", fg="red", bold=True)
+        click.echo(str(exc))
+        click.echo("Process output:")
+        click.echo(exc.stderr)
+        raise click.Abort()
+
     if dbversion is not None:
         click.secho(
             "ERROR: the database version has been populated by sql init scripts. "

swh/core/db/tests/data/README

@@ -23,3 +23,6 @@ test:
 
 will define 2 'swh.core.db' powered backends initialized from (resp.)
 './postgresql/sql' and './cli2/sql'.
+
+The 'test:fail' backend is expected to fail when trying to initialize the
+database (missing the pgcrypto extension, so the psql execution should fail).

swh/core/db/tests/data/test/fail/sql/30-schema.sql

+-- origin table
+create table if not exists origin (
+  id       bigserial not null,
+  url      text not null,
+  hash     text not null
+);

swh/core/db/tests/data/test/fail/sql/40-funcs.sql

+create or replace function hash_sha1(text)
+    returns text
+    language sql strict immutable
+as $$
+    select encode(public.digest($1, 'sha1'), 'hex')
+$$;

swh/core/db/tests/test_cli.py

@@ -3,8 +3,6 @@
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
-from subprocess import CalledProcessError
-
 import pytest
 from pytest_postgresql import factories
 import yaml
@@ -117,8 +115,7 @@ def test_cli_swh_db_initialization_fail_without_creation_first(
     result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", conninfo])
     # Fails because we cannot connect to an inexisting db
     assert result.exit_code == 1, f"Unexpected output: {result.output}"
-    assert isinstance(result.exception, CalledProcessError)
-    assert b'FATAL:  database "inexisting-db" does not exist' in result.exception.stderr
+    assert 'FATAL:  database "inexisting-db" does not exist' in result.output
 
 
 def test_cli_swh_db_initialization_fail_without_extension(
@@ -136,10 +133,8 @@ def test_cli_swh_db_initialization_fail_without_extension(
     # Fails as the function `public.digest` is not installed, init-admin calls is needed
     # first (the next tests show such behavior)
     assert result.exit_code == 1, f"Unexpected output: {result.output}"
-    assert isinstance(result.exception, CalledProcessError)
     assert (
-        b"ERROR:  function public.digest(text, unknown) does not exist"
-        in result.exception.stderr
+        "ERROR:  function public.digest(text, unknown) does not exist" in result.output
     )
 
 
@@ -263,6 +258,30 @@ def test_cli_swh_db_create_and_init_db_new_api(
         assert len(origins) == 1
 
 
+def test_cli_swh_db_init_report_sqlsh_error(
+    cli_runner,
+    postgresql,
+    mock_get_entry_points,
+    mocker,
+    tmp_path,
+):
+    """Create a db then initializing it should be ok for a "new style" datastore"""
+    module_name = "test:fail"
+
+    conninfo = craft_conninfo(postgresql)
+
+    # This initializes the schema and data
+    result = cli_runner.invoke(swhdb, ["init-admin", module_name, "--dbname", conninfo])
+    assert_result(result)
+
+    result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", conninfo])
+    assert result.exit_code == 1
+    assert (
+        "test/fail/sql/40-funcs.sql:6: "
+        "ERROR:  function public.digest(text, unknown) does not exist"
+    ) in result.output
+
+
 @pytest.mark.init_version(version=2)
 def test_cli_swh_db_upgrade_new_api(
     request,