fix(odin): Make check-ins resilient to startup failures and transient network errors (EDG-1032)

CHANGELOG.md

@@ -15,7 +15,9 @@ Changes are grouped as follows
 ## Unreleased
 
 ### Fixed
-* In the `unstable` package: Fixed extractors with more than 49 actionable tasks (more than 100 available actions) being unable to start: the startup request was rejected locally by a hard-coded limit of 100 available actions. The client-side limit is removed and Integrations API decides the maximum, so it can be raised without a new extractor-utils release. A request over the API's limit is rejected with a 400 whose message names the limit; the message is logged at critical level, and with `RETRY_STARTUP` (the default) the startup is retried until the limit is raised or the extractor is stopped (EDG-1049).
+* In the `unstable` package: Fixed the check-in thread dying silently when the startup request could not be built or reported (for example when the startup request is rejected). The failure is now logged at critical level and treated as fatal, and the extractor process exits with a non-zero exit code when it stops because of a fatal check-in error.
+* In the `unstable` package: Fixed transient network errors during check-ins (timeouts, connection errors, including raw `requests`/`httpx` errors from the OAuth token refresh) being treated as fatal and shutting the extractor down. They are now logged as warnings and retried at the next check-in. Server-side errors from the token endpoint are retried as well. Only 401/403 errors and bad credentials remain fatal.
+* In the `unstable` package: Fixed extractors with more than 49 actionable tasks (more than 100 available actions) being unable to start: the startup request was rejected locally by a hard-coded limit of 100 available actions. The client-side limit is removed and Integrations API decides the maximum, so it can be raised without a new extractor-utils release. A request over the API's limit is rejected with a 400 whose message names the limit; the message is logged at critical level, and with `RETRY_STARTUP` (the default) the startup is retried until the limit is raised or the extractor is stopped.
 
 ## 7.13.2
 

cognite/extractorutils/unstable/core/base.py

@@ -440,7 +440,17 @@ def _get_startup_request(self) -> StartupRequest:
         )
 
     def _run_checkin(self) -> None:
-        self._checkin_worker.run_periodic_checkin(self.cancellation_token, self._get_startup_request())
+        try:
+            self._checkin_worker.run_periodic_checkin(self.cancellation_token, self._get_startup_request())
+        except Exception as e:
+            # This runs on a daemon thread: without this, the traceback would only go to stderr and the extractor
+            # would keep running without ever checking in. Treat it as fatal instead.
+            self._logger.critical("Check-in failed unexpectedly, shutting down the extractor: %s", e, exc_info=e)
+            try:
+                self._checkin_worker.notify_fatal_error(e)
+            except Exception:
+                self._logger.exception("Fatal error handler failed")
+            self.cancellation_token.cancel()
 
     def _report_error(self, error: Error) -> None:
         self._checkin_worker.report_error(error)

cognite/extractorutils/unstable/core/checkin_worker.py

@@ -16,8 +16,11 @@
 from threading import RLock, Thread
 from time import sleep
 
+import httpx
+import requests
 from cognite.client import CogniteClient
 from cognite.client.exceptions import CogniteAPIError, CogniteAuthError, CogniteConnectionError
+from oauthlib.oauth2 import ServerError, TemporarilyUnavailableError  # type: ignore[import-untyped]
 from requests import Response
 
 from cognite.extractorutils.threading import CancellationToken
@@ -43,6 +46,23 @@
 MAX_ACTION_UPDATES_PER_CHECKIN = 100
 rng = SystemRandom()
 
+# Errors that mean "could not reach CDF right now" rather than "the request or credentials are wrong". The Cognite SDK
+# only wraps some of these (e.g. connection errors from API calls, OAuth2Error from token refresh), so raw
+# requests/httpx transport errors and timeouts (notably from the OAuth token refresh) can still surface unwrapped.
+# Deliberately narrow: configuration mistakes (invalid URL, missing schema, unsupported protocol) must stay fatal.
+_TRANSIENT_ERRORS: tuple[type[BaseException], ...] = (
+    requests.exceptions.ConnectionError,
+    requests.exceptions.Timeout,
+    httpx.TimeoutException,
+    httpx.NetworkError,
+    TimeoutError,
+    ConnectionError,
+)
+
+# The identity provider reporting a problem on its side rather than with the credentials. The SDK wraps these in a
+# CogniteAuthError, with the oauthlib error as the cause.
+_TRANSIENT_TOKEN_ERRORS: tuple[type[BaseException], ...] = (ServerError, TemporarilyUnavailableError)
+
 
 def _safe_dispatch(
     dispatcher: Callable[[list[Action]], None],
@@ -184,6 +204,18 @@ def set_on_fatal_error_handler(self, on_fatal_error: Callable[[Exception], None]
         """
         self._on_fatal_error = on_fatal_error
 
+    def notify_fatal_error(self, error: Exception) -> None:
+        """
+        Invoke the fatal error handler (if any) with the given error.
+
+        Used by callers that fail outside of a check-in request, such as when building the startup request.
+
+        Arguments:
+            error (Exception): The error that should be treated as fatal.
+        """
+        if self._on_fatal_error is not None:
+            self._on_fatal_error(error)
+
     def set_retry_startup(self, retry_startup: bool) -> None:
         """
         Set whether to retry reporting startup if it fails.
@@ -480,6 +512,15 @@ def _requeue_checkin(
                 self._action_updates = action_updates + self._action_updates
 
     def _wrap_checkin_like_request(self, request: Callable[[], Response]) -> bool:
+        """
+        Run a check-in like request and classify any failure.
+
+        Transient network problems are logged and requeued without being fatal. Only authentication/authorization
+        failures (401/403, bad credentials) and unexpected errors are reported as fatal.
+
+        Returns:
+            True if the payload of the request should be requeued.
+        """
         exception_to_report: Exception | None = None
         requeue = False
         try:
@@ -493,9 +534,13 @@ def _wrap_checkin_like_request(self, request: Callable[[], Response]) -> bool:
 
         except CogniteAuthError as e:
             self._logger.error(str(e))
-            self._logger.critical("Could not get an access token. Please check your configuration.")
-            exception_to_report = e
             requeue = True
+            if isinstance(e.__cause__, _TRANSIENT_ERRORS + _TRANSIENT_TOKEN_ERRORS):
+                # The token endpoint could not be reached; this says nothing about the credentials.
+                self._log_transient_error("Could not reach the token endpoint to get an access token", e.__cause__)
+            else:
+                self._logger.critical("Could not get an access token. Please check your configuration.")
+                exception_to_report = e
 
         except CogniteAPIError as e:
             if e.code == 401:
@@ -519,6 +564,10 @@ def _wrap_checkin_like_request(self, request: Callable[[], Response]) -> bool:
                 self._logger.critical(f"Error while connecting to CDF {e!s}")
             requeue = True
 
+        except _TRANSIENT_ERRORS as e:
+            self._log_transient_error("Could not reach CDF", e)
+            requeue = True
+
         except Exception as e:
             self._logger.critical(f"Extractor could not connect to CDF {e!s}")
             exception_to_report = e
@@ -528,3 +577,8 @@ def _wrap_checkin_like_request(self, request: Callable[[], Response]) -> bool:
             self._on_fatal_error(exception_to_report)
 
         return requeue
+
+    def _log_transient_error(self, message: str, error: BaseException) -> None:
+        self._logger.warning(
+            "%s (%s: %s). The check-in will be retried at the next check-in.", message, type(error).__name__, error
+        )

cognite/extractorutils/unstable/core/runtime.py

@@ -34,6 +34,7 @@ def main() -> None:
 from multiprocessing.synchronize import Event as MpEvent
 from pathlib import Path
 from random import randint
+from threading import Event as ThreadEvent
 from threading import Thread
 from typing import Any, Generic, TypeVar
 from uuid import uuid4
@@ -68,6 +69,11 @@ def main() -> None:
 
 ExtractorType = TypeVar("ExtractorType", bound=Extractor)
 
+# Exit code the extractor process uses to tell the runtime that it stopped because of a fatal error (for example
+# bad credentials), as opposed to a requested shutdown. Kept distinct from 1 so that unrelated crashes of the child
+# process do not change the runtime's behaviour.
+FATAL_ERROR_EXIT_CODE = 3
+
 
 @dataclass
 class _RuntimeControls:
@@ -83,7 +89,13 @@ def _extractor_process_entrypoint(
 ) -> None:
     logger = logging.getLogger(f"{extractor_class.EXTERNAL_ID}.runtime")
     checkin_worker.active_revision = config.current_config_revision
-    checkin_worker.set_on_fatal_error_handler(lambda _: on_fatal_error(controls))
+    fatal_error_occurred = ThreadEvent()
+
+    def handle_fatal_error(_: Exception) -> None:
+        fatal_error_occurred.set()
+        on_fatal_error(controls)
+
+    checkin_worker.set_on_fatal_error_handler(handle_fatal_error)
     checkin_worker.set_on_revision_change_handler(lambda _: on_revision_changed(controls))
     checkin_worker.set_retry_startup(extractor_class.RETRY_STARTUP)
     extractor = extractor_class._init_from_runtime(config, checkin_worker)
@@ -97,8 +109,18 @@ def _extractor_process_entrypoint(
             extractor.run()
 
     except Exception:
+        if fatal_error_occurred.is_set():
+            # The fatal error cancelled the extractor, which can make running tasks raise while shutting down. Don't
+            # mistake that for a crash, or the runtime would restart an extractor that can never succeed.
+            logger.exception("Extractor raised while stopping because of a fatal error")
+            sys.exit(FATAL_ERROR_EXIT_CODE)
         logger.exception("Extractor crashed, will attempt restart")
         controls.message_queue.put(RuntimeMessage.RESTART)
+        return
+
+    if fatal_error_occurred.is_set():
+        logger.critical("Extractor stopped because of a fatal error")
+        sys.exit(FATAL_ERROR_EXIT_CODE)
 
 
 def on_revision_changed(controls: _RuntimeControls) -> None:
@@ -598,6 +620,11 @@ def _main_runtime(self, args: Namespace) -> None:
                     case _:
                         assert_never(message)
 
+            elif process.exitcode == FATAL_ERROR_EXIT_CODE:
+                self.logger.critical("Extractor stopped because of a fatal error, shutting down runtime")
+                self._cancellation_token.cancel()
+                sys.exit(1)
+
             else:
                 self.logger.info("Shutting down runtime")
                 self._cancellation_token.cancel()