ENT-14416, CFE-4709: Retry on transient errors and fallback to cached releases.json

cf_remote/web.py

@@ -1,12 +1,16 @@
 import os
 import fcntl
 import re
+import time
+import urllib.error
+import urllib.parse
 import urllib.request
 import json
 import tempfile
 from collections import OrderedDict
 from cf_remote.utils import (
     is_different_checksum,
+    read_json,
     write_json,
     mkdir,
 )
@@ -16,15 +20,94 @@
 
 SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 
+# Attempts and seconds to wait before retrying a transient error.
+# The delay is doubled between each attempt.
+ATTEMPTS = 3
+DELAY = 2
 
-def get_json(url):
-    with urllib.request.urlopen(url) as r:
-        assert r.status >= 200 and r.status < 300
-        data = json.loads(r.read().decode(), object_pairs_hook=OrderedDict)
+# Seconds before cached JSON is considered stale and fetched again.
+MAX_AGE = 3600
+
+
+def is_transient_error(error: BaseException) -> bool:
+    """Decide whether a failed request is worth retrying
+
+    408 Request Timeout, 429 Too Many Requests and the 5xx server errors can
+    all succeed if we simply ask again. Everything else is treated as
+    permanent, retrying it would only delay the failure.
+    """
+
+    if not isinstance(error, urllib.error.HTTPError):
+        return False
+    return error.code in (408, 429) or error.code >= 500
+
+
+def urlopen_retry(url: str, attempts: int = ATTEMPTS, delay: float = DELAY):
+    """urlopen(), retrying requests which fail with a transient error"""
+
+    assert attempts >= 1
+
+    while True:
+        try:
+            return urllib.request.urlopen(url)
+        except Exception as e:
+            attempts -= 1
+            if attempts < 1 or not is_transient_error(e):
+                raise
+            log.warning(
+                "Failed to fetch '{}' ({}), retrying in {} seconds ({} attempts left)".format(
+                    url, e, delay, attempts
+                )
+            )
+            time.sleep(delay)
+            delay *= 2
+
+
+def json_cache_path(url: str) -> str:
+    # The basename alone is not unique, enterprise and community both have a
+    # releases.json, so name the file after the whole path of the URL.
+    filename = urllib.parse.urlparse(url).path.strip("/").replace("/", "_")
+    return os.path.join(cf_remote_dir("json", in_cache=True), filename)
+
+
+def is_cache_recent(path: str, max_age: int) -> bool:
+    """Whether the file was written less than max_age seconds ago"""
+
+    try:
+        return time.time() - os.path.getmtime(path) < max_age
+    except OSError:
+        return False
+
+
+def get_json(url: str, max_age: int = MAX_AGE):
+    """Get JSON from a URL, using a cached copy when possible
+
+    A cached copy younger than max_age seconds is used without contacting the
+    server at all. An older copy is only used if the server cannot be reached
+    due to a transient error, since stale release data beats no release data.
+    """
+
+    path = json_cache_path(url)
+
+    if is_cache_recent(path, max_age):
+        cached = read_json(path)
+        if cached is not None:
+            log.debug("Using recently cached '{}'".format(path))
+            return cached
+
+    try:
+        with urlopen_retry(url) as r:
+            assert r.status >= 200 and r.status < 300
+            data = json.loads(r.read().decode(), object_pairs_hook=OrderedDict)
+    except Exception as e:
+        cached = read_json(path)
+        if cached is None or not is_transient_error(e):
+            raise
+        log.warning(
+            "Failed to fetch '{}' ({}), falling back on '{}'".format(url, e, path)
+        )
+        return cached
 
-    filename = os.path.basename(url)
-    dir = cf_remote_dir("json", in_cache=True)
-    path = os.path.join(dir, filename)
     log.debug("Saving '{}' to '{}'".format(url, path))
     write_json(path, data)
 
@@ -85,7 +168,7 @@ def download_package(url, path=None, checksum=None, insecure=False):
 
     print("Downloading package: '{}'".format(path))
     fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path))
-    answer = urllib.request.urlopen(url).read()
+    answer = urlopen_retry(url).read()
     os.write(fd, answer)
     os.close(fd)
 

tests/test_web.py

@@ -0,0 +1,163 @@
+import os
+import urllib.error
+from email.message import Message
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from cf_remote.web import (
+    get_json,
+    is_transient_error,
+    json_cache_path,
+    urlopen_retry,
+)
+
+
+def http_error(code: int) -> urllib.error.HTTPError:
+    return urllib.error.HTTPError("url", code, "reason", Message(), None)
+
+
+class FakeResponse:
+    """Just enough of an HTTPResponse for get_json()"""
+
+    def __init__(self, payload: str) -> None:
+        self.status = 200
+        self.payload = payload
+
+    def read(self) -> bytes:
+        return self.payload.encode()
+
+    def __enter__(self) -> "FakeResponse":
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        return None
+
+
+def test_is_transient_error() -> None:
+    assert is_transient_error(http_error(408))
+    assert is_transient_error(http_error(429))
+    assert is_transient_error(http_error(502))
+    assert is_transient_error(http_error(503))
+
+    assert not is_transient_error(http_error(400))
+    assert not is_transient_error(http_error(404))
+    assert not is_transient_error(urllib.error.URLError("connection refused"))
+
+
+def test_urlopen_retry_transient_error() -> None:
+    calls = 0
+
+    def urlopen(_: str) -> str:
+        nonlocal calls
+        calls += 1
+        if calls < 3:
+            raise http_error(502)
+        return "response"
+
+    with patch("urllib.request.urlopen", urlopen):
+        assert urlopen_retry("url", delay=0) == "response"
+    assert calls == 3
+
+
+def test_urlopen_retry_out_of_attempts() -> None:
+    calls = 0
+
+    def urlopen(_: str) -> str:
+        nonlocal calls
+        calls += 1
+        raise http_error(502)
+
+    with patch("urllib.request.urlopen", urlopen):
+        try:
+            urlopen_retry("url", attempts=3, delay=0)
+            assert False
+        except urllib.error.HTTPError:
+            pass
+    assert calls == 3
+
+
+def test_urlopen_retry_permanent_error() -> None:
+    calls = 0
+
+    def urlopen(_: str) -> str:
+        nonlocal calls
+        calls += 1
+        raise http_error(404)
+
+    with patch("urllib.request.urlopen", urlopen):
+        try:
+            urlopen_retry("url", delay=0)
+            assert False
+        except urllib.error.HTTPError:
+            pass
+    assert calls == 1
+
+
+ENTERPRISE_URL = "https://cfengine.com/release-data/enterprise/releases.json"
+COMMUNITY_URL = "https://cfengine.com/release-data/community/releases.json"
+
+
+@pytest.fixture
+def cache_dir(tmp_path, monkeypatch) -> Path:
+    """Keep the JSON cache of these tests out of the real one"""
+
+    # get_json() uses the default delay, so don't wait for the retries
+    monkeypatch.setattr("time.sleep", lambda _: None)
+    directory = tmp_path / "cf-remote"
+    monkeypatch.setenv("CF_REMOTE_DIR", str(directory))
+    return directory / "json"
+
+
+def test_json_cache_path_is_unique_per_edition(cache_dir: Path) -> None:
+    assert json_cache_path(ENTERPRISE_URL) != json_cache_path(COMMUNITY_URL)
+
+
+def test_get_json_caches(cache_dir: Path) -> None:
+    calls = 0
+
+    def urlopen(_: str) -> FakeResponse:
+        nonlocal calls
+        calls += 1
+        return FakeResponse('{"version": "3.27.1"}')
+
+    with patch("urllib.request.urlopen", urlopen):
+        assert get_json(ENTERPRISE_URL) == {"version": "3.27.1"}
+        assert calls == 1
+
+        # Recently cached, so the server is left alone
+        assert get_json(ENTERPRISE_URL) == {"version": "3.27.1"}
+        assert calls == 1
+
+        # Cache too old to be trusted
+        assert get_json(ENTERPRISE_URL, max_age=0) == {"version": "3.27.1"}
+        assert calls == 2
+
+
+def test_get_json_falls_back_on_stale_cache(cache_dir: Path) -> None:
+    def urlopen(_: str) -> FakeResponse:
+        return FakeResponse('{"version": "3.27.1"}')
+
+    with patch("urllib.request.urlopen", urlopen):
+        get_json(ENTERPRISE_URL)
+
+    def urlopen_502(_: str) -> FakeResponse:
+        raise http_error(502)
+
+    with patch("urllib.request.urlopen", urlopen_502):
+        assert get_json(ENTERPRISE_URL, max_age=0) == {"version": "3.27.1"}
+
+
+def test_get_json_without_cache_raises(cache_dir: Path) -> None:
+    def urlopen(_: str) -> FakeResponse:
+        raise http_error(502)
+
+    with patch("urllib.request.urlopen", urlopen):
+        try:
+            get_json(ENTERPRISE_URL)
+            assert False
+        except urllib.error.HTTPError:
+            pass
+
+    assert not os.path.exists(json_cache_path(ENTERPRISE_URL))