Retry on ResponseNotReady exception during XMLRPC calls

JIRA: RHELWF-6331

greenwave/tests/test_xmlrpc_server_proxy.py

@@ -1,17 +1,88 @@
 # SPDX-License-Identifier: GPL-2.0+
+import http.client
+import uuid
 
 import mock
 import pytest
 
 from greenwave import xmlrpc_server_proxy
 
 
+def unique_koji_url(protocol='https'):
+    """
+    Generates unique Koji API URL to bypass caching on arguments in
+    xmlrpc_server_proxy().
+    """
+    return f'{protocol}://koji-{uuid.uuid4()}.example.com/kojihub'
+
+
+@pytest.fixture
+def mock_xmlrpc_proxy():
+    with mock.patch(
+            'greenwave.xmlrpc_server_proxy.xmlrpc.client.ServerProxy') as proxy:
+        yield proxy
+
+
+def test_xmlrpc_server_proxy_call(mock_xmlrpc_proxy):
+    proxy = xmlrpc_server_proxy.get_server_proxy(unique_koji_url(), timeout=0)
+    assert mock_xmlrpc_proxy.call_count == 1
+    proxy.getBuild('fake_koji_build')
+    assert mock_xmlrpc_proxy.call_count == 1
+    proxy.proxy.getBuild.assert_called_once_with('fake_koji_build')
+
+
+def test_xmlrpc_server_proxy_failure(mock_xmlrpc_proxy):
+    proxy = xmlrpc_server_proxy.get_server_proxy(unique_koji_url(), timeout=0)
+    assert mock_xmlrpc_proxy.call_count == 1
+    proxy.proxy.getBuild.side_effect = http.client.ResponseNotReady
+    with mock.patch('greenwave.xmlrpc_server_proxy.sleep') as mock_sleep:
+        with pytest.raises(http.client.ResponseNotReady):
+            proxy.getBuild('fake_koji_build')
+        mock_sleep.assert_has_calls([
+            mock.call(1),
+            mock.call(2),
+            mock.call(4),
+        ])
+    assert mock_xmlrpc_proxy.call_count == 4
+    proxy.proxy.getBuild.assert_has_calls([
+        mock.call('fake_koji_build'),
+        mock.call('fake_koji_build'),
+        mock.call('fake_koji_build'),
+        mock.call('fake_koji_build'),
+    ])
+
+
+def test_xmlrpc_server_proxy_reconnect(mock_xmlrpc_proxy):
+    proxy = xmlrpc_server_proxy.get_server_proxy(unique_koji_url(), timeout=0)
+    assert mock_xmlrpc_proxy.call_count == 1
+    proxy.proxy.getBuild.side_effect = (
+        http.client.ResponseNotReady,
+        http.client.ResponseNotReady,
+        http.client.ResponseNotReady,
+        {},
+    )
+    with mock.patch('greenwave.xmlrpc_server_proxy.sleep') as mock_sleep:
+        proxy.getBuild('fake_koji_build')
+        mock_sleep.assert_has_calls([
+            mock.call(1),
+            mock.call(2),
+            mock.call(4),
+        ])
+    assert mock_xmlrpc_proxy.call_count == 4
+    proxy.proxy.getBuild.assert_has_calls([
+        mock.call('fake_koji_build'),
+        mock.call('fake_koji_build'),
+        mock.call('fake_koji_build'),
+        mock.call('fake_koji_build'),
+    ])
+
+
 @pytest.mark.parametrize(
     'url, expected_transport, timeout, expected_timeout',
     (
-        ('http://localhost:5000/api', xmlrpc_server_proxy.Transport, 15, 15),
-        ('https://localhost:5000/api', xmlrpc_server_proxy.SafeTransport, 15, 15),
-        ('https://localhost:5000/api', xmlrpc_server_proxy.SafeTransport, (3, 12), 12),
+        (unique_koji_url('http'), xmlrpc_server_proxy.Transport, 15, 15),
+        (unique_koji_url(), xmlrpc_server_proxy.SafeTransport, 15, 15),
+        (unique_koji_url(), xmlrpc_server_proxy.SafeTransport, (3, 12), 12),
     ),
 )
 @mock.patch('greenwave.xmlrpc_server_proxy.Transport')
@@ -36,12 +107,14 @@ def test_get_server_proxy(
 
 def test_get_server_proxy_cached():
     """Server proxy objects are cached"""
-    s1 = xmlrpc_server_proxy.get_server_proxy('https://localhost:5000/api', timeout=None)
-    s2 = xmlrpc_server_proxy.get_server_proxy('https://localhost:5000/api', timeout=None)
+    url1 = unique_koji_url()
+    s1 = xmlrpc_server_proxy.get_server_proxy(url1, timeout=None)
+    s2 = xmlrpc_server_proxy.get_server_proxy(url1, timeout=None)
     assert s1 is s2
 
-    s3 = xmlrpc_server_proxy.get_server_proxy('https://localhost:5000/api', timeout=10)
+    s3 = xmlrpc_server_proxy.get_server_proxy(url1, timeout=10)
     assert s1 is not s3
 
-    s4 = xmlrpc_server_proxy.get_server_proxy('https://localhost:5001/api', timeout=None)
+    url2 = unique_koji_url()
+    s4 = xmlrpc_server_proxy.get_server_proxy(url2, timeout=None)
     assert s1 is not s4

greenwave/xmlrpc_server_proxy.py

@@ -3,9 +3,58 @@
 """
 Provides an "xmlrpc.client.ServerProxy" object with a timeout on the socket.
 """
+import http.client
+import logging
 import urllib.parse
 import xmlrpc.client
 from functools import lru_cache
+from time import sleep
+
+log = logging.getLogger(__name__)
+
+RETRY_ON_EXCEPTIONS = (
+    http.client.ResponseNotReady,
+)
+MAX_RETRIES = 3
+
+
+class XmlRpcServerProxy:
+    def __init__(self, uri, timeout):
+        self.uri = uri
+        self.timeout = timeout
+        self.reconnect()
+
+    def reconnect(self):
+        parsed_uri = urllib.parse.urlparse(self.uri)
+        if parsed_uri.scheme == 'https':
+            transport = SafeTransport(timeout=self.timeout)
+        else:
+            transport = Transport(timeout=self.timeout)
+
+        self.proxy = xmlrpc.client.ServerProxy(self.uri, transport=transport)
+
+    def __getattr__(self, name):
+        return XmlRpcMethod(self, name)
+
+
+class XmlRpcMethod:
+    def __init__(self, proxy, name):
+        self.proxy = proxy
+        self.name = name
+
+    def __call__(self, *args):
+        retry_counter = 0
+        while True:
+            try:
+                return getattr(self.proxy.proxy, self.name)(*args)
+            except RETRY_ON_EXCEPTIONS as e:
+                retry_counter += 1
+                if retry_counter > MAX_RETRIES:
+                    raise
+
+                log.warning("Retrying XMLRPC call %r on error: %s", self.name, e)
+                sleep(2 ** (retry_counter - 1))
+                self.proxy.reconnect()
 
 
 @lru_cache(maxsize=None)
@@ -20,16 +69,10 @@ def get_server_proxy(uri, timeout):
         timeout (int): The timeout to set on the transport socket.
 
     Returns:
-        xmlrpc.client.ServerProxy: An instance of :py:class:`xmlrpc.client.ServerProxy` with
+        XmlRpcServerProxy: Wrapper for :py:class:`xmlrpc.client.ServerProxy` with
             a socket timeout set.
     """
-    parsed_uri = urllib.parse.urlparse(uri)
-    if parsed_uri.scheme == 'https':
-        transport = SafeTransport(timeout=timeout)
-    else:
-        transport = Transport(timeout=timeout)
-
-    return xmlrpc.client.ServerProxy(uri, transport=transport)
+    return XmlRpcServerProxy(uri, timeout)
 
 
 class Transport(xmlrpc.client.Transport):