diff --git a/fastpurge/_client.py b/fastpurge/_client.py index 5ffa8b5..4d4fd9b 100644 --- a/fastpurge/_client.py +++ b/fastpurge/_client.py @@ -1,15 +1,13 @@ import json import logging import os +import random +import time from collections import namedtuple from threading import local, Lock import requests -from requests.adapters import HTTPAdapter -from requests.exceptions import RetryError -from urllib3.util import Retry -from http import HTTPStatus try: from time import monotonic @@ -36,33 +34,47 @@ ]) -class LoggingRetry(Retry): - def __init__(self, *args, **kwargs, ): - self._logger = kwargs.pop('logger', None) - super(LoggingRetry, self).__init__(*args, **kwargs) +class FastPurgeError(RuntimeError): + """Raised when the Fast Purge API reports an error. - def new(self, **kw): - kw['logger'] = self._logger - return super(LoggingRetry, self).new(**kw) + The exception message contains a summary of the error encountered.""" - def increment(self, method, url, *args, **kwargs): - response = kwargs.get("response") - if response: - self._logger.error("An invalid status code %s was received " - "when trying to %s to %s: %s", - response.status, method, url, response.reason) - else: # pragma: no cover - self._logger.error( - "An unknown error occurred when trying to %s to %s", method, - url) - return super(LoggingRetry, self).increment(method, url, *args, - **kwargs) +def compute_backoff(failures, backoff_factor, backoff_max, jitter, + _rand=random.random): + """Delay (seconds) before retry number ``failures``. -class FastPurgeError(RuntimeError): - """Raised when the Fast Purge API reports an error. + Mirrors urllib3's exponential backoff so the ``FAST_PURGE_RETRY_*`` + tunables keep their existing meaning: the first retry (``failures == 1``) + has no delay, then ``backoff_factor * 2 ** (failures - 1)`` plus up to + ``jitter`` seconds of random jitter, capped at ``backoff_max``. + """ + if failures <= 1: + return 0.0 + delay = backoff_factor * (2 ** (failures - 1)) + if jitter: + delay += _rand() * jitter + return min(backoff_max, delay) - The exception message contains a summary of the error encountered.""" + +def parse_retry_statuses(spec): + """Parse a FAST_PURGE_RETRY_STATUSES spec into a frozenset of ints. + + ``spec`` is a comma-separated list where each item is either a single + status code (``"429"``) or an inclusive range (``"500-599"``). Whitespace + and empty items are ignored. + """ + statuses = set() + for item in spec.split(","): + item = item.strip() + if not item: + continue + if "-" in item: + low, high = item.split("-", 1) + statuses.update(range(int(low), int(high) + 1)) + else: + statuses.add(int(item)) + return frozenset(statuses) class FastPurgeClient(object): @@ -109,6 +121,16 @@ class FastPurgeClient(object): RETRY_BACKOFF_MAX = float(os.environ.get("FAST_PURGE_RETRY_BACKOFF_MAX", "150")) RETRY_BACKOFF_JITTER = float(os.environ.get("FAST_PURGE_RETRY_JITTER", "2")) + # HTTP statuses worth retrying, as an allowlist -- anything not listed + # fails fast (e.g. 401/403/404/413 are not retried pointlessly). The + # default covers rate limiting (429), all server errors (500-599), and + # 400: Akamai returns 400 "Invalid timestamp" for a stale signature (fixed + # by re-signing on retry) and for WAF rate-control, both of which recover + # on a fresh attempt. Override with FAST_PURGE_RETRY_STATUSES, a + # comma-separated list of codes and/or inclusive ranges, e.g. "429,500-599". + RETRY_STATUSES = parse_retry_statuses( + os.environ.get("FAST_PURGE_RETRY_STATUSES", "400,429,500-599")) + # Default purge type. # Akamai recommend "invalidate", so why is "delete" our default? # Here's what Akamai docs have to say: @@ -231,37 +253,25 @@ def __baseurl(self): return '{out}:{port}'.format(out=out, port=self.__port) - @property - def __retry_policy(self): - retries = getattr(self.__local, 'retries', None) - if not retries: - retries = LoggingRetry( - total=self.MAX_RETRIES, - backoff_factor=self.RETRY_BACKOFF, - backoff_max=self.RETRY_BACKOFF_MAX, - backoff_jitter=self.RETRY_BACKOFF_JITTER, - # We strictly require 201 here since that's how the server - # tells us we queued something async, as expected - status_forcelist=[status.value for status in HTTPStatus - if status.value != 201], - allowed_methods={'POST'}, - logger=LOG, - ) - self.__local.retries = retries - return retries - @property def __session(self): session = getattr(self.__local, 'session', None) if not session: session = requests.Session() session.auth = EdgeGridAuth(**self.__auth) - session.mount(self.__baseurl, - HTTPAdapter(max_retries=self.__retry_policy)) - + # Deliberately no transport-level (urllib3) retries: those resend + # the request below the requests auth layer, reusing the original + # EdgeGrid signature, so any retry past Akamai's ~30s timestamp + # window fails with "Invalid timestamp". Retries are handled in + # __start_purge, which re-issues post() so each attempt is re-signed. self.__local.session = session return session + def __backoff_time(self, failures): + return compute_backoff(failures, self.RETRY_BACKOFF, + self.RETRY_BACKOFF_MAX, + self.RETRY_BACKOFF_JITTER) + def __get_request_bodies(self, objects): out_object = {"objects": objects} out_str = json.dumps(out_object) @@ -279,16 +289,72 @@ def __get_request_bodies(self, objects): def __start_purge(self, endpoint, request_body): headers = {'Content-Type': 'application/json'} LOG.debug("POST JSON of size %s to %s", len(request_body), endpoint) + + attempts = self.MAX_RETRIES + 1 + last_reason = "no attempts made" + # One initial attempt plus up to MAX_RETRIES retries. + for failures in range(0, attempts): + delay = self.__backoff_time(failures) + if delay: + time.sleep(delay) + + try: + # A fresh post() re-prepares the request, so EdgeGridAuth signs + # it again with a current timestamp. That is what lets a retry + # succeed after Akamai's ~30s signature window -- urllib3-level + # retries resend the original request with its now-stale + # signature and can never recover. + response = self.__session.post( + endpoint, data=request_body, headers=headers) + except requests.exceptions.RequestException as error: + # Connection reset / timeout / etc: transient, always retryable. + last_reason = str(error) + LOG.error("An error occurred when trying to POST to %s: %s", + endpoint, error) + continue + + if response.status_code == 201: + # 201 is how Fast Purge acknowledges a queued async purge. + return self.__purge_from_response(response) + + # Same message as before, expanded with Retry-After and the response + # body so the real Akamai error (e.g. "Invalid timestamp") is + # visible. .get()/.text are safe even if the API omits them. + LOG.error( + "An invalid status code %s was received when trying to POST " + "to %s: %s (Retry-After=%s): %s", + response.status_code, endpoint, response.reason, + response.headers.get("Retry-After"), (response.text or "")[:500]) + + if response.status_code not in self.RETRY_STATUSES: + # Permanent error (bad credentials/permissions/request) that + # won't self-correct -- fail fast instead of burning the window. + message = ("Request to %s failed with non-retryable status %s" + % (endpoint, response.status_code)) + LOG.debug("%s", message) + raise FastPurgeError(message) + + last_reason = "too many %s error responses" % response.status_code + + message = ("Request to {endpoint} was unsuccessful after {retries} " + "retries: {reason}").format( + endpoint=endpoint, retries=self.MAX_RETRIES, + reason=last_reason) + LOG.debug("%s", message) + raise FastPurgeError(message) + + @staticmethod + def __purge_from_response(response): + # Tolerate an unexpected non-JSON 201 (e.g. a future API change): still + # treat the purge as accepted, falling back to the default estimate. try: - response = self.__session.post(endpoint, data=request_body, headers=headers) response_body = response.json() - estimated_seconds = response_body.get('estimatedSeconds', 5) - return Purge(response_body, monotonic() + estimated_seconds) - except RetryError as e: - message = "Request to {endpoint} was unsuccessful after {retries} retries: {reason}". \ - format(endpoint=endpoint, retries=self.MAX_RETRIES, reason=e.args[0].reason) - LOG.debug("%s", message) - raise FastPurgeError(message) from e + except ValueError: + LOG.warning("Purge accepted (201) with non-JSON body: %s", + (response.text or "")[:200]) + response_body = {} + estimated_seconds = response_body.get('estimatedSeconds', 5) + return Purge(response_body, monotonic() + estimated_seconds) def purge_objects(self, object_type, objects, **kwargs): """Purge a collection of objects. diff --git a/tests/test_purge.py b/tests/test_purge.py index b11ef19..b2d85de 100644 --- a/tests/test_purge.py +++ b/tests/test_purge.py @@ -1,4 +1,5 @@ import pytest +import requests import requests_mock import mock import responses @@ -9,9 +10,16 @@ from monotonic import monotonic from fastpurge import FastPurgeClient, FastPurgeError +from fastpurge._client import compute_backoff, parse_retry_statuses # pylint: disable=unused-argument +# NOTE: FastPurgeClient class attributes (MAX_RETRIES, RETRY_BACKOFF, etc.) are +# populated from environment variables once at import time, so monkeypatch.setenv +# has no effect on tests -- the class is already defined. Use +# monkeypatch.setattr(FastPurgeClient, "ATTR", value) instead, which is what the +# corresponding env var would have set had it been present before import. + @pytest.fixture def client_auth(): @@ -136,8 +144,9 @@ def test_scheme_port(client_auth, requests_mocker): def test_response_fails(client, no_thread_retries, monkeypatch): """Requests fail with a FastPurgeError if API gives unsuccessful response.""" url = 'https://fastpurge.example.com/ccu/v3/delete/cpcode/production' - # Decrease backoff, otherwise the test will run for 5 minutes - monkeypatch.setenv("FAST_PURGE_RETRY_BACKOFF", "0.001") + # Decrease backoff to avoid a multi-minute test run. + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0.001) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) responses.add(responses.POST, url, status=503, content_type="application/json", body="Error") @@ -224,3 +233,165 @@ def test_retries_on_error(client_auth): assert len(err_1.calls) == 1 assert len(err_2.calls) == 1 assert len(res.calls) == 1 + + +@responses.activate(registry=responses.registries.OrderedRegistry) +def test_resigns_each_retry(client_auth, monkeypatch): + """Each retry is re-signed, so the EdgeGrid Authorization header differs + between attempts. (urllib3-level retries reused the original signature, + which failed with "Invalid timestamp" past Akamai's ~30s window.)""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=429, body="Rate limited") + responses.add(responses.POST, url, status=429, body="Rate limited") + responses.add(responses.POST, url, status=201, + content_type="application/json", + json={'estimatedSeconds': 0.1}) + + # No real delay between attempts. + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + + assert client.purge_by_tag(['red'], network='staging').result() + + auth_headers = [call.request.headers['Authorization'] + for call in responses.calls] + assert len(auth_headers) == 3 + # A fresh signature (new timestamp + nonce) per attempt => all distinct. + assert len(set(auth_headers)) == 3 + + +@responses.activate +def test_does_not_retry_auth_errors(client_auth): + """401/403 are not retried -- bad credentials/permissions won't self-correct.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=401, body="Unauthorized") + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert 'non-retryable status 401' in str(exception) + # Failed immediately, no retries. + assert len(responses.calls) == 1 + + +@responses.activate(registry=responses.registries.OrderedRegistry) +def test_retries_connection_errors(client_auth, monkeypatch): + """Connection/network errors are retried, not just HTTP statuses.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, + body=requests.exceptions.ConnectionError("boom")) + responses.add(responses.POST, url, status=201, + content_type="application/json", + json={'estimatedSeconds': 0.1}) + + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + + assert client.purge_by_tag(['red'], network='staging').result() + assert len(responses.calls) == 2 + + +@responses.activate +def test_accepts_non_json_201(client_auth): + """A 201 with a non-JSON body is still treated as an accepted purge + (robust to an unexpected response-shape change), using the default estimate.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=201, body="OK") # not JSON + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + result = client.purge_by_tag(['red'], network='staging').result() + + assert result == [{}] # succeeded, with empty metadata + + +def test_default_retry_statuses(): + """Default allowlist: 429, 400 (EdgeGrid timestamp/WAF) and any 5xx are + retryable; other client errors are not.""" + for status in (400, 429, 500, 501, 502, 503, 504, 599): + assert status in FastPurgeClient.RETRY_STATUSES, status + for status in (201, 401, 403, 404, 405, 413, 422): + assert status not in FastPurgeClient.RETRY_STATUSES, status + + +def test_parse_retry_statuses(): + """FAST_PURGE_RETRY_STATUSES parsing: single codes, inclusive ranges, and + tolerant of whitespace / empty items.""" + assert parse_retry_statuses("429") == frozenset((429,)) + assert parse_retry_statuses("400,429") == frozenset((400, 429)) + assert parse_retry_statuses("500-504") == frozenset(range(500, 505)) + assert parse_retry_statuses(" 429 , 500-599 ") == frozenset((429,)) | frozenset(range(500, 600)) + assert parse_retry_statuses("") == frozenset() + + +@responses.activate +def test_retry_statuses_is_configurable(client_auth, monkeypatch): + """A status removed from RETRY_STATUSES is no longer retried (fails fast), + confirming the allowlist drives behavior.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=503, body="Error") + + # Narrow the allowlist so 503 is NOT retryable. + monkeypatch.setattr(FastPurgeClient, "RETRY_STATUSES", frozenset((429,))) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert 'non-retryable status 503' in str(exception) + assert len(responses.calls) == 1 # failed immediately, no retries + + +def test_compute_backoff_matches_urllib3_schedule(): + """No jitter: first retry 0, then factor*2**(n-1), capped at backoff_max.""" + delays = [compute_backoff(n, 1.5, 9, 0) for n in range(1, 6)] + assert delays == [0.0, 3.0, 6.0, 9.0, 9.0] # 3, 6, 12->9, 24->9 + + # Jitter adds up to `jitter` seconds (here forced to its max) before the cap. + assert compute_backoff(2, 1.5, 100, 2.0, _rand=lambda: 1.0) == 3.0 + 2.0 + # Neither the initial attempt (0) nor the first retry (1) ever sleeps, + # regardless of jitter. + assert compute_backoff(0, 1.5, 100, 2.0, _rand=lambda: 1.0) == 0.0 + assert compute_backoff(1, 1.5, 100, 2.0, _rand=lambda: 1.0) == 0.0 + + +@responses.activate +def test_honors_max_retries(client_auth, monkeypatch): + """A persistently failing status is attempted exactly MAX_RETRIES + 1 times.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=503, body="Error") + + monkeypatch.setattr(FastPurgeClient, "MAX_RETRIES", 2) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert len(responses.calls) == 3 # 1 initial attempt + 2 retries + assert 'too many 503 error responses' in str(exception) + + +@responses.activate +def test_connection_error_exhaustion(client_auth, monkeypatch): + """When every attempt hits a connection error, a FastPurgeError is raised + reporting the underlying error (not a 'too many N' status message).""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, + body=requests.exceptions.ConnectionError("boom")) + + monkeypatch.setattr(FastPurgeClient, "MAX_RETRIES", 2) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert len(responses.calls) == 3 # 1 initial attempt + 2 retries + assert 'boom' in str(exception)