diff --git a/connectors/sources/confluence.py b/connectors/sources/confluence.py index ce66b05a4..41e57d422 100644 --- a/connectors/sources/confluence.py +++ b/connectors/sources/confluence.py @@ -12,7 +12,11 @@ from urllib.parse import urljoin import aiohttp -from aiohttp.client_exceptions import ClientResponseError, ServerDisconnectedError +from aiohttp.client_exceptions import ( + ClientPayloadError, + ClientResponseError, + ServerConnectionError, +) from connectors.access_control import ACCESS_CONTROL from connectors.logger import logger @@ -42,6 +46,8 @@ RETRIES = 3 RETRY_INTERVAL = 2 DEFAULT_RETRY_SECONDS = 30 +API_FAILURE_THRESHOLD = 0.10 +MIN_API_CALL = 1000 LIMIT = 50 SPACE = "space" @@ -104,6 +110,22 @@ class NotFound(Exception): pass +class SyncFailure(Exception): + pass + + +class UnauthorizedException(Exception): + pass + + +class BadRequestError(Exception): + pass + + +class Forbidden(Exception): + pass + + class ConfluenceClient: """Confluence client to handle API calls made to Confluence""" @@ -117,6 +139,8 @@ def __init__(self, configuration): self.certificate = self.configuration["ssl_ca"] self.retry_count = self.configuration["retry_count"] self.index_labels = self.configuration["index_labels"] + self.api_total_count = 0 + self.api_failed_count = 0 if self.data_source_type == CONFLUENCE_CLOUD: self.host_url = os.path.join(self.host_url, "wiki") @@ -195,6 +219,17 @@ async def _handle_client_errors(self, url, exception): await self._sleeps.sleep(retry_seconds) raise ThrottledError + elif exception.status == 400: + self._logger.error(f"Bad Request Error (400). Exception: {exception}.") + raise BadRequestError + elif exception.status == 401: + self._logger.error(f"Authentication error (401). Exception: {exception}.") + raise UnauthorizedException + elif exception.status == 403: + self._logger.error( + f"Invalid credentials provided for the request to url: {url}. Exception: {exception}" + ) + raise Forbidden elif exception.status == 404: self._logger.error(f"Getting Not Found Error for url: {url}") raise NotFound @@ -204,6 +239,39 @@ async def _handle_client_errors(self, url, exception): else: raise + async def _handle_client_payload_error(self, exception): + retry_seconds = DEFAULT_RETRY_SECONDS + response_headers = exception.headers or {} + if "Retry-After" in response_headers: + try: + retry_seconds = int(response_headers["Retry-After"]) + except (TypeError, ValueError) as exception: + self._logger.error( + f"Error while reading value of retry-after header {exception}. Using default retry time: {DEFAULT_RETRY_SECONDS} seconds" + ) + await self._sleeps.sleep(retry_seconds) + raise + + async def _handle_api_call_error(self, url, exception): + known_errors = { + ServerConnectionError, + ClientResponseError, + ClientPayloadError, + Forbidden, + UnauthorizedException, + ThrottledError, + NotFound, + InternalServerError, + } + + if isinstance(exception, tuple(known_errors)): + # Re-raising the error as it is not handled explicitly by retryable + raise + else: + self._logger.error( + f"Error encountered and the data for {url} skipped. Exception: {exception} caused the pagination to end prematurely." + ) + @retryable( retries=RETRIES, interval=RETRY_INTERVAL, @@ -223,17 +291,26 @@ async def api_call(self, url): response: Client response """ self._logger.debug(f"Making a GET call for url: {url}") - + if url != os.path.join(self.host_url, PING_URL): + self.api_total_count += 1 try: return await self._get_session().get( url=url, ssl=self.ssl_ctx, ) - except ServerDisconnectedError: + except ServerConnectionError: await self.close_session() raise except ClientResponseError as exception: await self._handle_client_errors(url=url, exception=exception) + except ClientPayloadError as exception: + await self._handle_client_payload_error(exception) + except Exception as exception: + self.api_failed_count += 1 + self._logger.debug( + f"Incrementing error count due to exception: {exception}. Failed API count: {self.api_failed_count}" + ) + raise async def paginated_api_call(self, url_name, **url_kwargs): """Make a paginated API call for Confluence objects using the passed url_name. @@ -259,9 +336,7 @@ async def paginated_api_call(self, url_name, **url_kwargs): links.get("next")[1:], ) except Exception as exception: - self._logger.warning( - f"Skipping data for type {url_name} from {url}. Exception: {exception}." - ) + await self._handle_api_call_error(url, exception) break async def paginated_api_call_for_datacenter_syncrule(self, url_name, **url_kwargs): @@ -288,13 +363,14 @@ async def paginated_api_call_for_datacenter_syncrule(self, url_name, **url_kwarg if len(json_response.get("results", [])) < LIMIT: break except Exception as exception: - self._logger.warning( - f"Skipping data for type {url_name} from {url}. Exception: {exception}." - ) + await self._handle_api_call_error(url, exception) break async def download_func(self, url): - yield await self.api_call(url) + try: + yield await self.api_call(url) + except Exception as exception: + await self._handle_api_call_error(url, exception) async def search_by_query(self, query): if self.data_source_type == CONFLUENCE_DATA_CENTER: @@ -332,6 +408,8 @@ async def fetch_server_space_permission(self, url): f"Something went wrong. Make sure you have installed Extender for running confluence datacenter/server DLS. Exception: {exception}." ) return {} + except Exception as exception: + await self._handle_api_call_error(url, exception) async def fetch_page_blog_documents(self, api_query): async for response in self.paginated_api_call( @@ -377,19 +455,27 @@ async def fetch_confluence_server_users(self): url = urljoin(self.host_url, URLS[USERS_FOR_SERVER]) while True: - url_ = url.format(start=start_at, limit=limit) - users = await self.api_call(url=url_) - response = await users.json() - if len(response.get(key)) == 0: - return - yield response.get(key) - start_at += limit + try: + url_ = url.format(start=start_at, limit=limit) + users = await self.api_call(url=url_) + response = await users.json() + if len(response.get(key)) == 0: + return + yield response.get(key) + start_at += limit + except Exception as exception: + await self._handle_api_call_error(url, exception) + break async def fetch_label(self, label_id): url = os.path.join(self.host_url, URLS[LABEL].format(id=label_id)) - label_data = await self.api_call(url=url) - labels = await label_data.json() - return [label.get("name") for label in labels["results"]] + try: + label_data = await self.api_call(url=url) + labels = await label_data.json() + return [label.get("name") for label in labels["results"]] + + except Exception as exception: + await self._handle_api_call_error(url, exception) class ConfluenceDataSource(BaseDataSource): @@ -1224,6 +1310,20 @@ async def _consumer(self): else: yield item + def check_api_exceptions_and_raise(self): + failed_percentage = round( + ( + self.confluence_client.api_failed_count + / self.confluence_client.api_total_count + ) + * 100, + 2, + ) + if failed_percentage >= API_FAILURE_THRESHOLD * 100: + msg = f"High percentage of API exceptions {failed_percentage}% is greater than or equal to the default threshold {API_FAILURE_THRESHOLD * 100}%. This calculation is done at the end of the synchronization process. Please review the logs for more details." + self._logger.error(msg) + raise SyncFailure(msg) + async def get_docs(self, filtering=None): """Executes the logic to fetch Confluence content in async manner. @@ -1274,3 +1374,6 @@ async def get_docs(self, filtering=None): async for item in self._consumer(): yield item await self.fetchers.join() + + if self.confluence_client.api_total_count > MIN_API_CALL: + self.check_api_exceptions_and_raise() diff --git a/tests/sources/test_confluence.py b/tests/sources/test_confluence.py index 087b8e4ba..65e01aaca 100644 --- a/tests/sources/test_confluence.py +++ b/tests/sources/test_confluence.py @@ -14,7 +14,11 @@ import aiohttp import pytest from aiohttp import StreamReader -from aiohttp.client_exceptions import ClientResponseError +from aiohttp.client_exceptions import ( + ClientPayloadError, + ClientResponseError, + ServerConnectionError, +) from freezegun import freeze_time from connectors.access_control import DLS_QUERY @@ -26,8 +30,12 @@ CONFLUENCE_SERVER, ConfluenceClient, ConfluenceDataSource, + Forbidden, InternalServerError, NotFound, + SyncFailure, + ThrottledError, + UnauthorizedException, ) from connectors.utils import ssl_context from tests.commons import AsyncIterator @@ -828,6 +836,25 @@ async def test_get_with_429_status(): assert result == payload +@pytest.mark.asyncio +@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0)) +async def test_call_api_with_client_payload_error(): + initial_response = ClientPayloadError(None, None) + retried_response = AsyncMock() + payload = {"value": "Test rate limit"} + retried_response = AsyncMock() + retried_response.json.return_value = payload + async with create_confluence_source() as source: + source.confluence_client._get_session().get = AsyncMock( + side_effect=[initial_response, retried_response] + ) + response = await source.confluence_client.api_call( + url="http://localhost:1000/sample" + ) + result = await response.json() + assert result == payload + + @pytest.mark.asyncio @patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0)) async def test_get_with_429_status_without_retry_after_header(): @@ -1673,3 +1700,114 @@ async def test_fetch_documents_with_html(): source.confluence_client.index_labels = True async for response, _, _, _, _ in source.fetch_documents(api_query=""): assert response == EXPECTED_PAGE + + +@pytest.mark.parametrize("exception", [ServerConnectionError, Exception]) +@pytest.mark.asyncio +@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0)) +async def test_confluence_client_api_call_with_server_disconnected_error(exception): + async with create_confluence_source() as source: + with patch.object( + ConfluenceClient, + "_get_session", + side_effect=exception, + ): + with pytest.raises(exception): + await source.confluence_client.api_call(url="abc") + + +def exception_function_params(): + return pytest.mark.parametrize( + "method_name, method_params,", + [ + ("paginated_api_call", {"url_name": "space", "api_query": "api_query"}), + ( + "paginated_api_call_for_datacenter_syncrule", + {"url_name": "space", "api_query": "api_query"}, + ), + ("fetch_confluence_server_users", {}), + ("fetch_server_space_permission", {"url": "key"}), + ("fetch_label", {"label_id": "abc"}), + ], + ) + + +@exception_function_params() +@pytest.mark.parametrize( + "exception", + [ + ServerConnectionError, + ThrottledError, + NotFound, + InternalServerError, + UnauthorizedException, + Forbidden, + ], +) +@pytest.mark.asyncio +@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0)) +async def test_api_call_with_exception_raise(exception, method_name, method_params): + async with create_confluence_source() as source: + with patch.object( + ConfluenceClient, + "api_call", + side_effect=exception, + ): + method = getattr(source.confluence_client, method_name) + with pytest.raises(exception): + if ( + method_name == "fetch_server_space_permission" + or method_name == "fetch_label" + ): + await method(**method_params) + else: + await anext(method(**method_params)) + + +@exception_function_params() +@pytest.mark.asyncio +@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0)) +async def test_api_call_with_exception(caplog, method_name, method_params): + caplog.set_level("ERROR") + async with create_confluence_source() as source: + with patch.object( + ConfluenceClient, + "api_call", + side_effect=Exception, + ): + method = getattr(source.confluence_client, method_name) + if ( + method_name == "fetch_server_space_permission" + or method_name == "fetch_label" + ): + await method(**method_params) + else: + async for _ in method(**method_params): + assert "Something went wrong" in caplog.text + + +@pytest.mark.asyncio +async def test_check_api_exceptions_percentage_above_threshold(): + async with create_confluence_source() as source: + source.confluence_client.api_total_count = 10 + source.confluence_client.api_failed_count = 2 + with pytest.raises(SyncFailure): + source.check_api_exceptions_and_raise() + + +@pytest.mark.asyncio +async def test_check_api_exceptions_percentage_below_threshold(): + async with create_confluence_source() as source: + source.confluence_client.api_total_count = 10 + source.confluence_client.api_failed_count = 0 + source.check_api_exceptions_and_raise() + + +@pytest.mark.asyncio +async def test_handle_api_call_error_known_errors(): + error = ServerConnectionError("Server disconnected") + async with create_confluence_source() as source: + with pytest.raises(Exception): + await source.confluence_client._handle_api_call_error( + "http://example.com", error + )