Merge branch 'Feature/Async' into async-sync-segment-workerpool

Author: chillaq

splitio/api/auth.py

@@ -4,8 +4,6 @@
 import json
 
 from splitio.api import APIException, headers_from_metadata
-from splitio.api.commons import record_telemetry
-from splitio.util.time import get_current_epoch_time_ms
 from splitio.api.client import HttpClientException
 from splitio.models.token import from_raw
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
@@ -31,6 +29,7 @@ def __init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer):
         self._sdk_key = sdk_key
         self._metadata = headers_from_metadata(sdk_metadata)
         self._telemetry_runtime_producer = telemetry_runtime_producer
+        self._client.set_telemetry_data(HTTPExceptionsAndLatencies.TOKEN, self._telemetry_runtime_producer)
 
     def authenticate(self):
         """
@@ -39,15 +38,13 @@ def authenticate(self):
         :return: Json representation of an authentication.
         :rtype: splitio.models.token.Token
         """
-        start = get_current_epoch_time_ms()
         try:
             response = self._client.get(
                 'auth',
                 'v2/auth',
                 self._sdk_key,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time_ms() - start, HTTPExceptionsAndLatencies.TOKEN, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 payload = json.loads(response.body)
                 return from_raw(payload)
@@ -59,3 +56,49 @@ def authenticate(self):
             _LOGGER.error('Exception raised while authenticating')
             _LOGGER.debug('Exception information: ', exc_info=True)
             raise APIException('Could not perform authentication.') from exc
+
+class AuthAPIAsync(object):  # pylint: disable=too-few-public-methods
+    """Async Class that uses an httpClient to communicate with the SDK Auth Service API."""
+
+    def __init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer):
+        """
+        Class constructor.
+
+        :param client: HTTP Client responsble for issuing calls to the backend.
+        :type client: HttpClient
+        :param sdk_key: User sdk key.
+        :type sdk_key: string
+        :param sdk_metadata: SDK version & machine name & IP.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
+        """
+        self._client = client
+        self._sdk_key = sdk_key
+        self._metadata = headers_from_metadata(sdk_metadata)
+        self._telemetry_runtime_producer = telemetry_runtime_producer
+        self._client.set_telemetry_data(HTTPExceptionsAndLatencies.TOKEN, self._telemetry_runtime_producer)
+
+    async def authenticate(self):
+        """
+        Perform authentication.
+
+        :return: Json representation of an authentication.
+        :rtype: splitio.models.token.Token
+        """
+        try:
+            response = await self._client.get(
+                'auth',
+                'v2/auth',
+                self._sdk_key,
+                extra_headers=self._metadata,
+            )
+            if 200 <= response.status_code < 300:
+                payload = json.loads(response.body)
+                return from_raw(payload)
+            else:
+                if (response.status_code >= 400 and response.status_code < 500):
+                    await self._telemetry_runtime_producer.record_auth_rejections()
+                raise APIException(response.body, response.status_code, response.headers)
+        except HttpClientException as exc:
+            _LOGGER.error('Exception raised while authenticating')
+            _LOGGER.debug('Exception information: ', exc_info=True)
+            raise APIException('Could not perform authentication.') from exc

splitio/api/client.py

@@ -5,6 +5,7 @@
 import abc
 
 from splitio.optional.loaders import aiohttp
+from splitio.util.time import get_current_epoch_time_ms
 
 SDK_URL = 'https://sdk.split.io/api'
 EVENTS_URL = 'https://events.split.io/api'
@@ -73,6 +74,20 @@ def get(self, server, path, apikey):
     def post(self, server, path, apikey):
         """http post request"""
 
+    def set_telemetry_data(self, metric_name, telemetry_runtime_producer):
+        """
+        Set the data needed for telemetry call
+
+        :param metric_name: metric name for telemetry
+        :type metric_name: str
+
+        :param telemetry_runtime_producer: telemetry recording instance
+        :type telemetry_runtime_producer: splitio.engine.telemetry.TelemetryRuntimeProducer
+        """
+        self._telemetry_runtime_producer = telemetry_runtime_producer
+        self._metric_name = metric_name
+
+
 class HttpClient(HttpClientBase):
     """HttpClient wrapper."""
 
@@ -116,13 +131,15 @@ def get(self, server, path, sdk_key, query=None, extra_headers=None):  # pylint:
         if extra_headers is not None:
             headers.update(extra_headers)
 
+        start = get_current_epoch_time_ms()
         try:
             response = requests.get(
                 _build_url(server, path, self._urls),
                 params=query,
                 headers=headers,
                 timeout=self._timeout
             )
+            self._record_telemetry(response.status_code, get_current_epoch_time_ms() - start)
             return HttpResponse(response.status_code, response.text, response.headers)
         except Exception as exc:  # pylint: disable=broad-except
             raise HttpClientException('requests library is throwing exceptions') from exc
@@ -152,6 +169,7 @@ def post(self, server, path, sdk_key, body, query=None, extra_headers=None):  #
         if extra_headers is not None:
             headers.update(extra_headers)
 
+        start = get_current_epoch_time_ms()
         try:
             response = requests.post(
                 _build_url(server, path, self._urls),
@@ -160,10 +178,28 @@ def post(self, server, path, sdk_key, body, query=None, extra_headers=None):  #
                 headers=headers,
                 timeout=self._timeout
             )
+            self._record_telemetry(response.status_code, get_current_epoch_time_ms() - start)
             return HttpResponse(response.status_code, response.text, response.headers)
         except Exception as exc:  # pylint: disable=broad-except
             raise HttpClientException('requests library is throwing exceptions') from exc
 
+    def _record_telemetry(self, status_code, elapsed):
+        """
+        Record Telemetry info
+
+        :param status_code: http request status code
+        :type status_code: int
+
+        :param elapsed: response time elapsed.
+        :type status_code: int
+        """
+        self._telemetry_runtime_producer.record_sync_latency(self._metric_name, elapsed)
+        if 200 <= status_code < 300:
+            self._telemetry_runtime_producer.record_successful_sync(self._metric_name, get_current_epoch_time_ms())
+            return
+        self._telemetry_runtime_producer.record_sync_error(self._metric_name, status_code)
+
+
 class HttpClientAsync(HttpClientBase):
     """HttpClientAsync wrapper."""
 
@@ -204,6 +240,7 @@ async def get(self, server, path, apikey, query=None, extra_headers=None):  # py
         headers = _build_basic_headers(apikey)
         if extra_headers is not None:
             headers.update(extra_headers)
+        start = get_current_epoch_time_ms()
         try:
             async with self._session.get(
                 _build_url(server, path, self._urls),
@@ -212,6 +249,7 @@ async def get(self, server, path, apikey, query=None, extra_headers=None):  # py
                 timeout=self._timeout
             ) as response:
                 body = await response.text()
+                await self._record_telemetry(response.status, get_current_epoch_time_ms() - start)
                 return HttpResponse(response.status, body, response.headers)
         except aiohttp.ClientError as exc:  # pylint: disable=broad-except
             raise HttpClientException('aiohttp library is throwing exceptions') from exc
@@ -237,6 +275,7 @@ async def post(self, server, path, apikey, body, query=None, extra_headers=None)
         headers = _build_basic_headers(apikey)
         if extra_headers is not None:
             headers.update(extra_headers)
+        start = get_current_epoch_time_ms()
         try:
             async with self._session.post(
                 _build_url(server, path, self._urls),
@@ -246,6 +285,23 @@ async def post(self, server, path, apikey, body, query=None, extra_headers=None)
                 timeout=self._timeout
             ) as response:
                 body = await response.text()
+                await self._record_telemetry(response.status, get_current_epoch_time_ms() - start)
                 return HttpResponse(response.status, body, response.headers)
         except aiohttp.ClientError as exc:  # pylint: disable=broad-except
-            raise HttpClientException('aiohttp library is throwing exceptions') from exc
+            raise HttpClientException('aiohttp library is throwing exceptions') from exc
+
+    async def _record_telemetry(self, status_code, elapsed):
+        """
+        Record Telemetry info
+
+        :param status_code: http request status code
+        :type status_code: int
+
+        :param elapsed: response time elapsed.
+        :type status_code: int
+        """
+        await self._telemetry_runtime_producer.record_sync_latency(self._metric_name, elapsed)
+        if 200 <= status_code < 300:
+            await self._telemetry_runtime_producer.record_successful_sync(self._metric_name, get_current_epoch_time_ms())
+            return
+        await self._telemetry_runtime_producer.record_sync_error(self._metric_name, status_code)

splitio/api/commons.py

@@ -1,31 +1,8 @@
 """Commons module."""
-from splitio.util.time import get_current_epoch_time_ms
 
 _CACHE_CONTROL = 'Cache-Control'
 _CACHE_CONTROL_NO_CACHE = 'no-cache'
 
-def record_telemetry(status_code, elapsed, metric_name, telemetry_runtime_producer):
-    """
-    Record Telemetry info
-
-    :param status_code: http request status code
-    :type status_code: int
-
-    :param elapsed: response time elapsed.
-    :type status_code: int
-
-    :param metric_name: metric name for telemetry
-    :type metric_name: str
-
-    :param telemetry_runtime_producer: telemetry recording instance
-    :type telemetry_runtime_producer: splitio.engine.telemetry.TelemetryRuntimeProducer
-    """
-    telemetry_runtime_producer.record_sync_latency(metric_name, elapsed)
-    if 200 <= status_code < 300:
-        telemetry_runtime_producer.record_successful_sync(metric_name, get_current_epoch_time_ms())
-        return
-    telemetry_runtime_producer.record_sync_error(metric_name, status_code)
-
 class FetchOptions(object):
     """Fetch Options object."""
 

splitio/api/events.py

@@ -4,32 +4,14 @@
 
 from splitio.api import APIException, headers_from_metadata
 from splitio.api.client import HttpClientException
-from splitio.api.commons import record_telemetry
-from splitio.util.time import get_current_epoch_time_ms
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class EventsAPI(object):  # pylint: disable=too-few-public-methods
-    """Class that uses an httpClient to communicate with the events API."""
-
-    def __init__(self, http_client, sdk_key, sdk_metadata, telemetry_runtime_producer):
-        """
-        Class constructor.
-
-        :param http_client: HTTP Client responsble for issuing calls to the backend.
-        :type http_client: HttpClient
-        :param sdk_key: sdk key.
-        :type sdk_key: string
-        :param sdk_metadata: SDK version & machine name & IP.
-        :type sdk_metadata: splitio.client.util.SdkMetadata
-        """
-        self._client = http_client
-        self._sdk_key = sdk_key
-        self._metadata = headers_from_metadata(sdk_metadata)
-        self._telemetry_runtime_producer = telemetry_runtime_producer
+class EventsAPIBase(object):  # pylint: disable=too-few-public-methods
+    """Base Class that uses an httpClient to communicate with the events API."""
 
     @staticmethod
     def _build_bulk(events):
@@ -54,6 +36,27 @@ def _build_bulk(events):
             for event in events
         ]
 
+
+class EventsAPI(EventsAPIBase):  # pylint: disable=too-few-public-methods
+    """Class that uses an httpClient to communicate with the events API."""
+
+    def __init__(self, http_client, sdk_key, sdk_metadata, telemetry_runtime_producer):
+        """
+        Class constructor.
+
+        :param http_client: HTTP Client responsble for issuing calls to the backend.
+        :type http_client: HttpClient
+        :param sdk_key: sdk key.
+        :type sdk_key: string
+        :param sdk_metadata: SDK version & machine name & IP.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
+        """
+        self._client = http_client
+        self._sdk_key = sdk_key
+        self._metadata = headers_from_metadata(sdk_metadata)
+        self._telemetry_runtime_producer = telemetry_runtime_producer
+        self._client.set_telemetry_data(HTTPExceptionsAndLatencies.EVENT, self._telemetry_runtime_producer)
+
     def flush_events(self, events):
         """
         Send events to the backend.
@@ -65,7 +68,6 @@ def flush_events(self, events):
         :rtype: bool
         """
         bulk = self._build_bulk(events)
-        start = get_current_epoch_time_ms()
         try:
             response = self._client.post(
                 'events',
@@ -74,7 +76,52 @@ def flush_events(self, events):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time_ms() - start, HTTPExceptionsAndLatencies.EVENT, self._telemetry_runtime_producer)
+            if not 200 <= response.status_code < 300:
+                raise APIException(response.body, response.status_code)
+        except HttpClientException as exc:
+            _LOGGER.error('Error posting events because an exception was raised by the HTTPClient')
+            _LOGGER.debug('Error: ', exc_info=True)
+            raise APIException('Events not flushed properly.') from exc
+
+class EventsAPIAsync(EventsAPIBase):  # pylint: disable=too-few-public-methods
+    """Async Class that uses an httpClient to communicate with the events API."""
+
+    def __init__(self, http_client, sdk_key, sdk_metadata, telemetry_runtime_producer):
+        """
+        Class constructor.
+
+        :param http_client: HTTP Client responsble for issuing calls to the backend.
+        :type http_client: HttpClient
+        :param sdk_key: sdk key.
+        :type sdk_key: string
+        :param sdk_metadata: SDK version & machine name & IP.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
+        """
+        self._client = http_client
+        self._sdk_key = sdk_key
+        self._metadata = headers_from_metadata(sdk_metadata)
+        self._telemetry_runtime_producer = telemetry_runtime_producer
+        self._client.set_telemetry_data(HTTPExceptionsAndLatencies.EVENT, self._telemetry_runtime_producer)
+
+    async def flush_events(self, events):
+        """
+        Send events to the backend.
+
+        :param events: Events bulk
+        :type events: list
+
+        :return: True if flush was successful. False otherwise
+        :rtype: bool
+        """
+        bulk = self._build_bulk(events)
+        try:
+            response = await self._client.post(
+                'events',
+                'events/bulk',
+                self._sdk_key,
+                body=bulk,
+                extra_headers=self._metadata,
+            )
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc: