FFM-7177 Authentication Retries (#67)

* FFM-7177 Add check for retryable codes

* FFM-7177 Use retrying package

* FFM-7177 Improve code parsing

* FFM-7177 Change to tenacity package as retrying is no longer available

* FFM-7177 Tweak

* FFM-7177 Tweak

* FFM-7177 Tweak

* FFM-7177 Add backoff

* FFM-7177 Tweak

* FFM-7177 Fix logging

* FFM-7177 Fix logging

* FFM-7177 Tidy up

* FFM-7177 Tidy up

* FFM-7177 Tidy up

* FFM-7177 Tidy up

* FFM-7177 Tidy up

* FFM-7177 Remove annotation and use function directly

* FFM-7177 Remove annotation and use function directly

* FFM-7177 Serve defaults and add max retry

* FFM-7177 Fix retry code

* FFM-7177 Don't start poller or streaming if auth failed

* FFM-7177 Mark client as initialized if auth fails

* FFM-7177 Fix 200 response

* FFM-7177 Tidyup

* FFM-7177 Fix nested condigftion

* FFM-7177 Doc update for max auth retries

* FFM-7177 Use set for lookup

* FFM-7177 Remove todo comment

* FFM-7177 Lint

* FFM-7177 Lint

* FFM-7177 Fix retry handler

* FFM-7177 Fix retry handler

* FFM-7177 Fix retry handler

Author: erdirowlands

docs/further_reading.md

@@ -23,6 +23,7 @@ You can pass the configuration in as options when the SDK client is created.
 | enableStream    | with_stream_enabled(True),                               | Enable streaming mode.                                                                                                                           | true                                 |
 | enableAnalytics | with_analytics_enabled(True)                             | Enable analytics.  Metrics data is posted every 60s                                                                                              | true                                 |
 | pollInterval    | with_poll_interval(120)                                  | When running in stream mode, the interval in seconds that we poll for changes.                                                                   | 60                                   |
+| maxAuthRetries  | with_max_auth_retries(10)                                | The number of retry attempts to make if client authentication fails on a retryable HTTP error                                                    | 10                                   |
 
 ## Logging Configuration
 The SDK provides a logger that wraps the standard python logging package.  You can import and use it with:

featureflags/__init__.py

@@ -2,4 +2,4 @@
 
 __author__ = """Enver Bisevac"""
 __email__ = "enver.bisevac@harness.io"
-__version__ = '1.1.12'
+__version__ = '1.1.13'

featureflags/api/client.py

@@ -13,6 +13,7 @@ class Client:
     cookies: Dict[str, str] = attr.ib(factory=dict, kw_only=True)
     headers: Dict[str, str] = attr.ib(factory=dict, kw_only=True)
     timeout: float = attr.ib(5.0, kw_only=True)
+    max_auth_retries: int
 
     def get_headers(self) -> Dict[str, str]:
         """Get headers to be used in all endpoints"""
@@ -35,6 +36,9 @@ def with_cookies(self, cookies: Dict[str, str]) -> "Client":
     def get_timeout(self) -> float:
         return self.timeout
 
+    def get_max_auth_retries(self) -> int:
+        return self.max_auth_retries
+
     def with_timeout(self, timeout: float) -> "Client":
         """
         Get a new client matching this one with a new timeout (in seconds)

featureflags/api/default/authenticate.py

@@ -1,17 +1,24 @@
 from typing import Any, Dict, Optional, Union
+from featureflags.util import log
 
 import httpx
 
 from featureflags.api.client import Client
 from featureflags.api.types import Response
 from featureflags.models.authentication_request import AuthenticationRequest
 from featureflags.models.authentication_response import AuthenticationResponse
+from tenacity import retry_if_result, wait_exponential, \
+    stop_after_attempt, Retrying, retry_all
+
+
+class UnrecoverableAuthenticationException(Exception):
+    pass
 
 
 def _get_kwargs(
-    *,
-    client: Client,
-    json_body: AuthenticationRequest,
+        *,
+        client: Client,
+        json_body: AuthenticationRequest,
 ) -> Dict[str, Any]:
     url = "{}/client/auth".format(client.base_url)
 
@@ -30,33 +37,17 @@ def _get_kwargs(
 
 
 def _parse_response(
-    *, response: httpx.Response
+        *, response: httpx.Response
 ) -> Optional[Union[AuthenticationResponse, None]]:
     if response.status_code == 200:
-        response_200 = AuthenticationResponse.from_dict(response.json())
-
-        return response_200
-    if response.status_code == 401:
-        response_401 = None
-
-        return response_401
-    if response.status_code == 403:
-        response_403 = None
-
-        return response_403
-    if response.status_code == 404:
-        response_404 = None
-
-        return response_404
-    if response.status_code == 500:
-        response_500 = None
-
-        return response_500
-    return None
+        return AuthenticationResponse.from_dict(response.json())
+    else:
+        raise UnrecoverableAuthenticationException(
+            f'Authentication failed on an unrecoverable error: {response}')
 
 
 def _build_response(
-    *, response: httpx.Response
+        *, response: httpx.Response
 ) -> Response[Union[AuthenticationResponse, None]]:
     return Response(
         status_code=response.status_code,
@@ -67,26 +58,50 @@ def _build_response(
 
 
 def sync_detailed(
-    *,
-    client: Client,
-    json_body: AuthenticationRequest,
+        *,
+        client: Client,
+        json_body: AuthenticationRequest,
 ) -> Response[Union[AuthenticationResponse, None]]:
     kwargs = _get_kwargs(
         client=client,
         json_body=json_body,
     )
+    max_auth_retries = client.get_max_auth_retries()
+    response = _post_request(kwargs, max_auth_retries)
+    return _build_response(response=response)
 
-    response = httpx.post(
-        **kwargs,
-    )
 
-    return _build_response(response=response)
+def handle_http_result(response):
+    code = response.status_code
+    if code in {408, 425, 429, 500, 502, 503, 504}:
+        return True
+    else:
+        log.error(
+            f'SDK_AUTH_2001: Authentication failed with HTTP code #{code} and '
+            'will not attempt to reconnect')
+        return False
+
+
+def _post_request(kwargs, max_auth_retries):
+    retryer = Retrying(
+        wait=wait_exponential(multiplier=1, min=4, max=10),
+        retry=retry_all(
+            retry_if_result(lambda response: response.status_code != 200),
+            retry_if_result(handle_http_result)
+        ),
+        before_sleep=lambda retry_state: log.warning(
+            f'SDK_AUTH_2002: Authentication attempt #'
+            f'{retry_state.attempt_number} '
+            f'got {retry_state.outcome.result()} Retrying...'),
+        stop=stop_after_attempt(max_auth_retries),
+    )
+    return retryer(httpx.post, **kwargs)
 
 
 def sync(
-    *,
-    client: Client,
-    json_body: AuthenticationRequest,
+        *,
+        client: Client,
+        json_body: AuthenticationRequest,
 ) -> Optional[Union[AuthenticationResponse, None]]:
     """Used to retrieve all target segments for certain account id."""
 
@@ -97,9 +112,9 @@ def sync(
 
 
 async def asyncio_detailed(
-    *,
-    client: Client,
-    json_body: AuthenticationRequest,
+        *,
+        client: Client,
+        json_body: AuthenticationRequest,
 ) -> Response[Union[AuthenticationResponse, None]]:
     kwargs = _get_kwargs(
         client=client,
@@ -113,9 +128,9 @@ async def asyncio_detailed(
 
 
 async def asyncio(
-    *,
-    client: Client,
-    json_body: AuthenticationRequest,
+        *,
+        client: Client,
+        json_body: AuthenticationRequest,
 ) -> Optional[Union[AuthenticationResponse, None]]:
     """Used to retrieve all target segments for certain account id."""
 

featureflags/client.py

@@ -3,6 +3,7 @@
 import threading
 from typing import Any, Callable, Dict, Optional
 
+from tenacity import RetryError
 from jwt import decode
 
 from featureflags.analytics import AnalyticsService
@@ -11,6 +12,7 @@
 
 from .api.client import AuthenticatedClient, Client
 from .api.default.authenticate import AuthenticationRequest
+from .api.default.authenticate import UnrecoverableAuthenticationException
 from .api.default.authenticate import sync as authenticate
 from .config import Config, default_config
 from .evaluations.auth_target import Target
@@ -52,47 +54,61 @@ def __init__(
 
         self.run()
 
-
     def run(self):
-        self.authenticate()
-
-        streaming_event = threading.Event()
-        polling_event = threading.Event()
-
-        self._polling_processor = PollingProcessor(
-            client=self._client,
-            config=self._config,
-            environment_id=self._environment_id,
-            #  PollingProcessor is responsible for doing the initial
-            #  flag/group fetch and cache. So we allocate it the responsibility
-            #  for setting the Client is_initialized variable.
-            wait_for_initialization=self._initialized,
-            ready=polling_event,
-            stream_ready=streaming_event,
-            repository=self._repository
-        )
-        self._polling_processor.start()
+        try:
+            self.authenticate()
+            streaming_event = threading.Event()
+            polling_event = threading.Event()
 
-        if self._config.enable_stream:
-            self._stream = StreamProcessor(
-                repository=self._repository,
+            self._polling_processor = PollingProcessor(
                 client=self._client,
-                environment_id=self._environment_id,
-                api_key=self._sdk_key,
-                token=self._auth_token,
                 config=self._config,
-                ready=streaming_event,
-                poller=polling_event,
-                cluster=self._cluster,
-            )
-            self._stream.start()
-
-        if self._config.enable_analytics:
-            self._analytics = AnalyticsService(
-                config=self._config,
-                client=self._client,
-                environment=self._environment_id
+                environment_id=self._environment_id,
+                #  PollingProcessor is responsible for doing the initial
+                #  flag/group fetch and cache. So we allocate it the
+                #  responsibility
+                #  for setting the Client is_initialized variable.
+                wait_for_initialization=self._initialized,
+                ready=polling_event,
+                stream_ready=streaming_event,
+                repository=self._repository
             )
+            self._polling_processor.start()
+
+            if self._config.enable_stream:
+                self._stream = StreamProcessor(
+                    repository=self._repository,
+                    client=self._client,
+                    environment_id=self._environment_id,
+                    api_key=self._sdk_key,
+                    token=self._auth_token,
+                    config=self._config,
+                    ready=streaming_event,
+                    poller=polling_event,
+                    cluster=self._cluster,
+                )
+                self._stream.start()
+
+            if self._config.enable_analytics:
+                self._analytics = AnalyticsService(
+                    config=self._config,
+                    client=self._client,
+                    environment=self._environment_id
+                )
+
+        except RetryError:
+            log.error(
+                "Authentication failed and max retries have been exceeded - "
+                "defaults will be served.")
+            # Mark the client as initialized in case wait_for_initialization
+            # is called. The SDK has already logged that authentication
+            # failed and defaults will be returned.
+            self._initialized.set()
+        except UnrecoverableAuthenticationException:
+            log.error(
+                "Authentication failed - defaults will be served.")
+            # Same again, just mark the client as initailized.
+            self._initialized.set()
 
     def wait_for_initialization(self):
         log.debug("Waiting for initialization to finish")
@@ -107,7 +123,8 @@ def get_environment_id(self):
     def authenticate(self):
         client = Client(
             base_url=self._config.base_url,
-            events_url=self._config.events_url
+            events_url=self._config.events_url,
+            max_auth_retries=self._config.max_auth_retries
         )
         body = AuthenticationRequest(api_key=self._sdk_key)
         response = authenticate(client=client, json_body=body)
@@ -125,7 +142,8 @@ def authenticate(self):
             token=self._auth_token,
             params={
                 'cluster': self._cluster
-            }
+            },
+            max_auth_retries=self._config.max_auth_retries
         )
         # Additional headers used to track usage
         additional_headers = {

featureflags/config.py

@@ -25,7 +25,8 @@ def __init__(
             cache: Cache = None,
             store: object = None,
             enable_stream: bool = True,
-            enable_analytics: bool = True
+            enable_analytics: bool = True,
+            max_auth_retries: int = 10
     ):
         self.base_url = base_url
         self.events_url = events_url
@@ -38,6 +39,7 @@ def __init__(
         self.store = store
         self.enable_stream = enable_stream
         self.enable_analytics = enable_analytics
+        self.max_auth_retries = max_auth_retries
 
 
 default_config = Config()
@@ -76,3 +78,10 @@ def func(config: Config) -> None:
         config.pull_interval = value
 
     return func
+
+
+def with_max_auth_retries(value: int) -> Callable:
+    def func(config: Config) -> None:
+        config.max_auth_retries = value
+
+    return func