Feature: Google Drive - Presentation to Markdown / Audio|Video Transcribed (10MB Limit)

.buildkite/nightly_steps.yml

@@ -57,6 +57,8 @@ steps:
   - label: "🔨 Microsoft SQL"
     command:
       - ".buildkite/run_nigthly.sh mssql"
+    env:
+      SKIP_AARCH64: "true"
     artifact_paths:
       - "perf8-report-*/**/*"
   - label: "🔨 Jira"

Dockerfile.ftest.wolfi

@@ -1,4 +1,4 @@
-FROM docker.elastic.co/wolfi/python:3.10-dev
+FROM docker.elastic.co/wolfi/python:3.10-dev@sha256:609e3fdb2c2a219941a6ea1e81f935899d05a787af634e4d47a3a3ec9a6d3379
 USER root
 COPY . /connectors
 WORKDIR /connectors

connectors/VERSION

@@ -1 +1 @@
-8.15.0.0
+8.15.5.1

connectors/es/sink.py

@@ -197,7 +197,9 @@ async def _bulk_api_call():
             for item in res["items"]:
                 for op, data in item.items():
                     if "error" in data:
-                        self._logger.error(f"operation {op} failed, {data['error']}")
+                        self._logger.error(
+                            f"operation {op} failed for doc {data['_id']}, {data['error']}"
+                        )
 
         self._populate_stats(stats, res)
 
@@ -598,8 +600,18 @@ async def get_docs(self, generator, skip_unchanged_documents=False):
                             "doc": doc,
                         }
                     )
+                # We try raising every loop to not miss a moment when
+                # too many errors happened when downloading
+                lazy_downloads.raise_any_exception()
 
                 await asyncio.sleep(0)
+
+            # Sit and wait until an error happens
+            await lazy_downloads.join(raise_on_error=True)
+        except Exception as ex:
+            self._logger.error(f"Extractor failed with an error: {ex}")
+            lazy_downloads.cancel()
+            raise
         finally:
             # wait for all downloads to be finished
             await lazy_downloads.join()
@@ -883,9 +895,7 @@ def _extractor_task_running(self):
 
     async def cancel(self):
         if self._sink_task_running():
-            self._logger.info(
-                f"Cancling the Sink task: {self._sink_task.name}"  # pyright: ignore
-            )
+            self._logger.info(f"Canceling the Sink task: {self._sink_task.get_name()}")
             self._sink_task.cancel()
         else:
             self._logger.debug(
@@ -894,7 +904,7 @@ async def cancel(self):
 
         if self._extractor_task_running():
             self._logger.info(
-                f"Canceling the Extractor task: {self._extractor_task.name}"  # pyright: ignore
+                f"Canceling the Extractor task: {self._extractor_task.get_name()}"
             )
             self._extractor_task.cancel()
         else:
@@ -1042,13 +1052,15 @@ async def async_bulk(
     def sink_task_callback(self, task):
         if task.exception():
             self._logger.error(
-                f"Encountered an error in the sync's {type(self._sink).__name__}: {task.get_name()}: {task.exception()}",
+                f"Encountered an error in the sync's {type(self._sink).__name__}: {task.get_name()}",
+                exc_info=task.exception(),
             )
             self.error = task.exception()
 
     def extractor_task_callback(self, task):
         if task.exception():
             self._logger.error(
-                f"Encountered an error in the sync's {type(self._extractor).__name__}: {task.get_name()}: {task.exception()}",
+                f"Encountered an error in the sync's {type(self._extractor).__name__}: {task.get_name()}",
+                exc_info=task.exception(),
             )
             self.error = task.exception()

connectors/source.py

@@ -10,7 +10,7 @@
 import importlib
 import re
 from contextlib import asynccontextmanager
-from datetime import date, datetime
+from datetime import date, datetime, time
 from decimal import Decimal
 from enum import Enum
 from functools import cache
@@ -670,7 +670,7 @@ def _serialize(value):
             elif isinstance(value, dict):
                 for key, svalue in value.items():
                     value[key] = _serialize(svalue)
-            elif isinstance(value, (datetime, date)):
+            elif isinstance(value, (datetime, date, time)):
                 value = value.isoformat()
             elif isinstance(value, Decimal128):
                 value = value.to_decimal()

connectors/sources/confluence.py

@@ -895,14 +895,20 @@ async def fetch_documents(self, api_query):
             doc = {
                 "_id": str(document["id"]),
                 "type": document["type"],
-                "_timestamp": document["history"]["lastUpdated"]["when"],
+                "_timestamp": nested_get_from_dict(
+                    document, ["history", "lastUpdated", "when"]
+                ),
                 "title": document.get("title"),
                 "ancestors": ancestor_title,
-                "space": document["space"]["name"],
-                "body": document["body"]["storage"]["value"],
+                "space": nested_get_from_dict(document, ["space", "name"]),
+                "body": nested_get_from_dict(document, ["body", "storage", "value"]),
                 "url": document_url,
-                "author": document["history"]["createdBy"][self.authorkey],
-                "createdDate": document["history"]["createdDate"],
+                "author": nested_get_from_dict(
+                    document, ["history", "createdBy", self.authorkey]
+                ),
+                "createdDate": nested_get_from_dict(
+                    document, ["history", "createdDate"]
+                ),
             }
             if self.confluence_client.index_labels:
                 doc["labels"] = document["labels"]

connectors/sources/github.py

@@ -12,7 +12,7 @@
 import aiohttp
 import fastjsonschema
 from aiohttp.client_exceptions import ClientResponseError
-from gidgethub import RateLimitExceeded, sansio
+from gidgethub import QueryError, RateLimitExceeded, sansio
 from gidgethub.abc import (
     BadGraphQLRequest,
     GraphQLAuthorizationFailure,
@@ -52,6 +52,7 @@
 RETRIES = 3
 RETRY_INTERVAL = 2
 FORBIDDEN = 403
+UNAUTHORIZED = 401
 NODE_SIZE = 100
 REVIEWS_COUNT = 45
 
@@ -673,14 +674,12 @@ def __init__(
     def set_logger(self, logger_):
         self._logger = logger_
 
-    def get_rate_limit_encountered(self, status_code, message):
-        return status_code == FORBIDDEN and "rate limit" in str(message).lower()
+    def get_rate_limit_encountered(self, status_code, rate_limit_remaining):
+        return status_code == FORBIDDEN and not int(rate_limit_remaining)
 
     async def _get_retry_after(self, resource_type):
         current_time = time.time()
-        response = await self._get_client.getitem(
-            "/rate_limit", oauth_token=self._access_token()
-        )
+        response = await self.get_github_item("/rate_limit")
         reset = nested_get_from_dict(
             response, ["resources", resource_type, "reset"], default=current_time
         )
@@ -725,6 +724,8 @@ async def _update_installation_access_token(self):
                 private_key=self.private_key,
             )
             self._installation_access_token = access_token_response["token"]
+        except RateLimitExceeded:
+            await self._put_to_sleep("core")
         except Exception:
             self._logger.exception(
                 f"Failed to get access token for installation {self._installation_id}.",
@@ -789,11 +790,20 @@ async def graphql(self, query, variables=None):
             msg = "Your Github token is either expired or revoked. Please check again."
             raise UnauthorizedException(msg) from exception
         except BadGraphQLRequest as exception:
-            if self.get_rate_limit_encountered(exception.status_code, exception):
-                await self._put_to_sleep(resource_type="graphql")
-            elif exception.status == FORBIDDEN:
+            if exception.status_code == FORBIDDEN:
                 msg = f"Provided GitHub token does not have the necessary permissions to perform the request for the URL: {url} and query: {query}."
                 raise ForbiddenException(msg) from exception
+            else:
+                raise
+        except QueryError as exception:
+            for error in exception.response.get("errors"):
+                if (
+                    error.get("type").lower() == "rate_limited"
+                    and "api rate limit exceeded" in error.get("message").lower()
+                ):
+                    await self._put_to_sleep(resource_type="graphql")
+            msg = f"Error while executing query. Exception: {exception.response.get('errors')}"
+            raise Exception(msg) from exception
         except Exception:
             raise
 
@@ -819,7 +829,7 @@ async def get_github_item(self, resource):
                 url=resource, oauth_token=self._access_token()
             )
         except ClientResponseError as exception:
-            if exception.status == 401:
+            if exception.status == UNAUTHORIZED:
                 if self.auth_method == GITHUB_APP:
                     self._logger.debug(
                         f"The access token for installation #{self._installation_id} expired, Regenerating a new token."
@@ -828,6 +838,10 @@ async def get_github_item(self, resource):
                     raise
                 msg = "Your Github token is either expired or revoked. Please check again."
                 raise UnauthorizedException(msg) from exception
+            elif self.get_rate_limit_encountered(
+                exception.status, exception.headers.get("X-RateLimit-Remaining")
+            ):
+                await self._put_to_sleep(resource_type="core")
             elif exception.status == FORBIDDEN:
                 msg = f"Provided GitHub token does not have the necessary permissions to perform the request for the URL: {resource}."
                 raise ForbiddenException(msg) from exception
@@ -861,20 +875,44 @@ async def paginated_api_call(self, query, variables, keys):
     def get_repo_details(self, repo_name):
         return repo_name.split("/")
 
+    @retryable(
+        retries=RETRIES,
+        interval=RETRY_INTERVAL,
+        strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
+        skipped_exceptions=UnauthorizedException,
+    )
     async def get_personal_access_token_scopes(self):
-        request_headers = sansio.create_headers(
-            self._get_client.requester,
-            accept=sansio.accept_format(),
-            oauth_token=self._access_token(),
-        )
-        _, headers, _ = await self._get_client._request(
-            "HEAD", self.base_url, request_headers
-        )
-        scopes = headers.get("X-OAuth-Scopes")
-        if not scopes or not scopes.strip():
-            self._logger.warning(f"Couldn't find 'X-OAuth-Scopes' in headers {headers}")
-            return set()
-        return {scope.strip() for scope in scopes.split(",")}
+        try:
+            request_headers = sansio.create_headers(
+                self._get_client.requester,
+                accept=sansio.accept_format(),
+                oauth_token=self._access_token(),
+            )
+            url = f"{self.base_url}/graphql"
+            _, headers, _ = await self._get_client._request(
+                "HEAD", url, request_headers
+            )
+            scopes = headers.get("X-OAuth-Scopes")
+            if not scopes or not scopes.strip():
+                self._logger.warning(
+                    f"Couldn't find 'X-OAuth-Scopes' in headers {headers}"
+                )
+                return set()
+            return {scope.strip() for scope in scopes.split(",")}
+        except ClientResponseError as exception:
+            if exception.status == FORBIDDEN:
+                if self.get_rate_limit_encountered(
+                    exception.status, exception.headers.get("X-RateLimit-Remaining")
+                ):
+                    await self._put_to_sleep("graphql")
+                else:
+                    msg = f"Provided GitHub token does not have the necessary permissions to perform the request for the URL: {self.base_url}."
+                    raise ForbiddenException(msg) from exception
+            elif exception.status == UNAUTHORIZED:
+                msg = "Your Github token is either expired or revoked. Please check again."
+                raise UnauthorizedException(msg) from exception
+            else:
+                raise
 
     @retryable(
         retries=RETRIES,
@@ -1279,7 +1317,6 @@ async def _get_invalid_repos_for_github_app(self):
                 self.configured_repos,
             )
         )
-
         for full_repo_name in self.configured_repos:
             if full_repo_name in invalid_repos:
                 continue
@@ -1432,7 +1469,7 @@ async def _validate_personal_access_token_scopes(self):
         if self.configuration["auth_method"] != PERSONAL_ACCESS_TOKEN:
             return
 
-        scopes = await self.github_client.get_personal_access_token_scopes()
+        scopes = await self.github_client.get_personal_access_token_scopes() or set()
         required_scopes = {"repo", "user", "read:org"}
 
         for scope in ["write:org", "admin:org"]:

connectors/sources/google.py

@@ -5,10 +5,12 @@
 #
 import json
 import os
+import urllib.parse
 from enum import Enum
 
 from aiogoogle import Aiogoogle, AuthError, HTTPError
 from aiogoogle.auth.creds import ServiceAccountCreds
+from aiogoogle.models import Request
 from aiogoogle.sessions.aiohttp_session import AiohttpSession
 
 from connectors.logger import logger
@@ -192,6 +194,41 @@ async def _call_api(google_client, method_object, kwargs):
 
         return await anext(self._execute_api_call(resource, method, _call_api, kwargs))
 
+    async def api_call_custom(self, api_service, url_path, params=None, method="GET"):
+        """
+        Make a custom API call to any Google API endpoint by specifying only the service name,
+        along with the path and parameters.
+
+        Args:
+            api_service (str): The Google service to use (e.g., "slides" or "drive").
+            url_path (str): The specific path for the API call (e.g., "/presentations/.../thumbnail").
+            params (dict): Optional dictionary of query parameters.
+            method (str): HTTP method for the request. Default is "GET".
+
+        Returns:
+            dict: The response from the Google API.
+
+        Raises:
+            HTTPError: If the API request fails.
+        """
+        # Construct the full URL with the specified Google service
+        base_url = f"https://{api_service}.googleapis.com/v1"
+        full_url = f"{base_url}{url_path}"
+
+        if params:
+            full_url += f"?{urllib.parse.urlencode(params)}"
+
+        # Define the request
+        request = Request(method=method, url=full_url)
+
+        # Perform the request
+        async with Aiogoogle(
+            service_account_creds=self.service_account_credentials
+        ) as google_client:
+            response = await google_client.as_service_account(request)
+
+        return response
+
     async def _execute_api_call(self, resource, method, call_api_func, kwargs):
         """Execute the API call with common try/except logic.
         Args: