From 045e63e279b95a7446614b383687d18e7a1b28d9 Mon Sep 17 00:00:00 2001 From: Joao Mario Lago Date: Wed, 29 Oct 2025 16:20:22 -0300 Subject: [PATCH 1/5] core:services:kraken: Reduce `install` complexity * Reduce code complexity of install function by spliting in multiple parts --- core/services/kraken/extension/extension.py | 90 ++++++++++++--------- 1 file changed, 54 insertions(+), 36 deletions(-) diff --git a/core/services/kraken/extension/extension.py b/core/services/kraken/extension/extension.py index 34cc78a8ac..bf356ea052 100644 --- a/core/services/kraken/extension/extension.py +++ b/core/services/kraken/extension/extension.py @@ -146,29 +146,18 @@ async def remove(cls, container_name: str, delete_image: bool = True) -> None: finally: cls.unlock(container_name) - async def _image_is_available_locally(self) -> bool: - try: - image_ref = f"{self.source.docker}:{self.tag}" + (f"@{self.digest}" if self.digest else "") - async with DockerCtx() as client: - await client.images.inspect(image_ref) - return True - except DockerError: - return False - - async def install( # pylint: disable=too-many-branches - self, clear_remaining_tags: bool = True, atomic: bool = False - ) -> AsyncGenerator[bytes, None]: - logger.info(f"Installing extension {self.identifier}:{self.tag}") - - # First we should make sure no other tag is running - running_ext = None + async def _disable_running_extension(self) -> Optional["Extension"]: + """Disable any currently running extension with the same identifier.""" try: running_ext = await self.from_running(self.identifier) if running_ext: await running_ext.disable() + return running_ext except (ExtensionNotRunning, ExtensionNotFound): - pass + return None + def _create_extension_settings(self) -> ExtensionSettings: + """Create and save extension settings.""" new_extension = ExtensionSettings( identifier=self.identifier, name=self.source.name, @@ -180,25 +169,57 @@ async def install( # pylint: disable=too-many-branches ) # Save in settings first, if the image fails to install it will try to fetch after in main kraken check loop self._save_settings(new_extension) + return new_extension + + def _prepare_docker_auth(self) -> Optional[str]: + """Prepare Docker authentication string from source auth credentials.""" + if self.source.auth is None: + return None + docker_auth = f"{self.source.auth.username}:{self.source.auth.password}" + return base64.b64encode(docker_auth.encode("utf-8")).decode("utf-8") + async def _image_is_available_locally(self) -> bool: try: - self.lock(self.unique_entry) + image_ref = f"{self.source.docker}:{self.tag}" + (f"@{self.digest}" if self.digest else "") + async with DockerCtx() as client: + await client.images.inspect(image_ref) + return True + except DockerError: + return False - docker_auth: Optional[str] = None - if self.source.auth is not None: - docker_auth = f"{self.source.auth.username}:{self.source.auth.password}" - docker_auth = base64.b64encode(docker_auth.encode("utf-8")).decode("utf-8") + async def _pull_docker_image(self, docker_auth: Optional[str]) -> AsyncGenerator[bytes, None]: + """Pull Docker image and yield progress updates.""" + tag = f"{self.source.docker}:{self.tag}" + (f"@{self.digest}" if self.digest else "") + async with DockerCtx() as client: + async for line in client.images.pull( + tag, repo=self.source.docker, tag=self.tag, auth=docker_auth, stream=True + ): + # TODO - Plug Error detection from docker image here + yield json.dumps(line).encode("utf-8") + # Make sure to add correct tag if a digest was used since docker messes up the tag + if self.digest: + await client.images.tag(tag, f"{self.source.docker}:{self.tag}") + + async def _clear_remaining_tags(self) -> None: + """Uninstall all other tags for this extension.""" + logger.info(f"Clearing remaining tags for extension {self.identifier}") + to_clear: List[Extension] = cast(List[Extension], await self.from_settings(self.identifier)) + to_clear = [version for version in to_clear if version.source.tag != self.tag] + await asyncio.gather(*(version.uninstall() for version in to_clear)) + + async def install(self, clear_remaining_tags: bool = True, atomic: bool = False) -> AsyncGenerator[bytes, None]: + logger.info(f"Installing extension {self.identifier}:{self.tag}") - tag = f"{self.source.docker}:{self.tag}" + (f"@{self.digest}" if self.digest else "") - async with DockerCtx() as client: - async for line in client.images.pull( - tag, repo=self.source.docker, tag=self.tag, auth=docker_auth, stream=True - ): - # TODO - Plug Error detection from docker image here - yield json.dumps(line).encode("utf-8") - # Make sure to add correct tag if a digest was used since docker messes up the tag - if self.digest: - await client.images.tag(tag, f"{self.source.docker}:{self.tag}") + # First we should make sure no other tag is running + running_ext = await self._disable_running_extension() + + self._create_extension_settings() + try: + self.lock(self.unique_entry) + + docker_auth = self._prepare_docker_auth() + async for line in self._pull_docker_image(docker_auth): + yield line except Exception as error: # In case of some external installs kraken shouldn't try to install it again so we remove from settings if atomic: @@ -223,10 +244,7 @@ async def install( # pylint: disable=too-many-branches logger.info(f"Extension {self.identifier}:{self.tag} installed") # Uninstall all other tags in case user wants to clear them if clear_remaining_tags: - logger.info(f"Clearing remaining tags for extension {self.identifier}") - to_clear: List[Extension] = cast(List[Extension], await self.from_settings(self.identifier)) - to_clear = [version for version in to_clear if version.source.tag != self.tag] - await asyncio.gather(*(version.uninstall() for version in to_clear)) + await self._clear_remaining_tags() async def update(self, clear_remaining_tags: bool) -> AsyncGenerator[bytes, None]: async for data in self.install(clear_remaining_tags): From a8d4d1798a2686dc1d480c45cee674655acc058c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Ant=C3=B4nio=20Cardoso?= Date: Fri, 28 Aug 2026 15:26:27 -0300 Subject: [PATCH 2/5] core: services: kraken: Cap start-attempt backoff with min max() floored every retry at 600s after the first failed start. --- core/services/kraken/kraken.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/core/services/kraken/kraken.py b/core/services/kraken/kraken.py index 962ed3a69b..9efb138712 100644 --- a/core/services/kraken/kraken.py +++ b/core/services/kraken/kraken.py @@ -32,9 +32,7 @@ def _extension_start_try_valid(self, extension: ExtensionSettings) -> bool: attempts, last_attempt = Extension.start_attempts.get(unique_entry, (0, 0)) maximum_delay = 600 minimum_delay = 10 - required_delay = ( - max(minimum_delay * (attempts != 0) + 2**attempts, maximum_delay) if attempts < 8 else maximum_delay - ) + required_delay = min(minimum_delay * (attempts != 0) + 2**attempts, maximum_delay) now = int(time.monotonic()) From 2c0c15c39f65da398ee8ebb20629452b9dfcb0b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Ant=C3=B4nio=20Cardoso?= Date: Fri, 28 Aug 2026 15:26:27 -0300 Subject: [PATCH 3/5] core: services: kraken: Run tagged and latest installs atomically update() and v2 from_latest/tagged install used the non-atomic default, so a failed pull was treated as success and purged the running tag. --- core/services/kraken/api/v2/routers/extension.py | 4 ++-- core/services/kraken/extension/extension.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/core/services/kraken/api/v2/routers/extension.py b/core/services/kraken/api/v2/routers/extension.py index e844445b2c..e627b7c7b0 100644 --- a/core/services/kraken/api/v2/routers/extension.py +++ b/core/services/kraken/api/v2/routers/extension.py @@ -98,7 +98,7 @@ async def install_by_identifier(identifier: str, stable: bool = True) -> Streami Install latest version of an extension by its identifier using one of the current manifests. """ extension: Extension = await Extension.from_latest(identifier, stable) - return StreamingResponse(streamer(extension.install())) + return StreamingResponse(streamer(extension.install(atomic=True))) @extension_router_v2.post("/{identifier}/{tag}/install", status_code=status.HTTP_201_CREATED) @@ -108,7 +108,7 @@ async def install_by_identifier_and_tag(identifier: str, tag: str) -> StreamingR Install a specific version of an extension by its identifier and tag using one of the current manifests. """ extension = cast(Extension, await Extension.from_manifest(identifier, tag)) - return StreamingResponse(streamer(extension.install())) + return StreamingResponse(streamer(extension.install(atomic=True))) @extension_router_v2.post("/{identifier}/{tag}/enable", status_code=status.HTTP_204_NO_CONTENT) diff --git a/core/services/kraken/extension/extension.py b/core/services/kraken/extension/extension.py index bf356ea052..0722597e57 100644 --- a/core/services/kraken/extension/extension.py +++ b/core/services/kraken/extension/extension.py @@ -247,7 +247,7 @@ async def install(self, clear_remaining_tags: bool = True, atomic: bool = False) await self._clear_remaining_tags() async def update(self, clear_remaining_tags: bool) -> AsyncGenerator[bytes, None]: - async for data in self.install(clear_remaining_tags): + async for data in self.install(clear_remaining_tags, atomic=True): yield data async def uninstall(self) -> None: From b0b53676760d191bae205c22aa94acb4cec2de4e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Ant=C3=B4nio=20Cardoso?= Date: Fri, 28 Aug 2026 15:26:27 -0300 Subject: [PATCH 4/5] core: services: kraken: Restore prior extension if a pull fails Failed pulls uninstalled the new tag and could delete the running image. Restore prior settings, refuse a sibling-image alias, and require a Docker success status on the install pull stream. --- core/services/kraken/extension/extension.py | 161 ++++++++++++++++---- 1 file changed, 134 insertions(+), 27 deletions(-) diff --git a/core/services/kraken/extension/extension.py b/core/services/kraken/extension/extension.py index 0722597e57..4fb7eb6be3 100644 --- a/core/services/kraken/extension/extension.py +++ b/core/services/kraken/extension/extension.py @@ -168,6 +168,7 @@ def _create_extension_settings(self) -> ExtensionSettings: user_permissions=self.source.user_permissions, ) # Save in settings first, if the image fails to install it will try to fetch after in main kraken check loop + # Atomic failure rolls this entry back and re-enables the previously running sibling. self._save_settings(new_extension) return new_extension @@ -178,27 +179,95 @@ def _prepare_docker_auth(self) -> Optional[str]: docker_auth = f"{self.source.auth.username}:{self.source.auth.password}" return base64.b64encode(docker_auth.encode("utf-8")).decode("utf-8") - async def _image_is_available_locally(self) -> bool: + @staticmethod + async def _inspect_or_none(client: Any, ref: str) -> Optional[Any]: try: - image_ref = f"{self.source.docker}:{self.tag}" + (f"@{self.digest}" if self.digest else "") - async with DockerCtx() as client: - await client.images.inspect(image_ref) - return True - except DockerError: + return await client.images.inspect(ref) + except DockerError as error: + if error.status == 404: + return None + raise + + async def _ensure_tagged_local_image(self, client: Any, sibling_image_id: Optional[str] = None) -> bool: + # start() and a successful pull run docker:tag. Catalog platform digests are not + # stored in RepoDigests after `docker pull repo:tag` (that records the index + # digest), so a digest match against the catalog cannot be required. + # sibling_image_id is only for the failed-pull fallback: a retag of the running + # sibling onto the new name is not the requested version. Do not pass it after a + # clean pull -- two tags can share an image Id (aliases) and still be the pull. + tag_ref = f"{self.source.docker}:{self.tag}" + tagged = await self._inspect_or_none(client, tag_ref) + image_id = tagged.get("Id") if isinstance(tagged, dict) else None + if isinstance(image_id, str) and image_id and (sibling_image_id is None or image_id != sibling_image_id): + return True + if not self.digest: + return False + digest = self.digest if self.digest.startswith("sha256:") else f"sha256:{self.digest}" + digest_ref = f"{self.source.docker}@{digest}" + info = await self._inspect_or_none(client, digest_ref) + image_id = info.get("Id") if isinstance(info, dict) else None + if not isinstance(image_id, str) or not image_id or image_id == sibling_image_id: return False + try: + await client.images.tag(digest_ref, self.source.docker, tag=self.tag) + except Exception as error: + logger.warning(f"Failed to tag {digest_ref} as {tag_ref}: {error}") + tagged = await self._inspect_or_none(client, tag_ref) + image_id = tagged.get("Id") if isinstance(tagged, dict) else None + return ( + isinstance(image_id, str) and bool(image_id) and (sibling_image_id is None or image_id != sibling_image_id) + ) + + async def _rollback_failed_install( + self, + running_ext: Optional["Extension"], + prior_settings: Optional[ExtensionSettings], + atomic: bool, + ) -> None: + try: + if prior_settings is not None: + self._save_settings(prior_settings) + elif atomic: + self._save_settings() + else: + await self.set_enabled(False) + except Exception as rollback_error: + logger.warning(f"Failed to roll back {self.identifier}:{self.tag} after pull failure: {rollback_error}") + if not running_ext: + return + try: + self.reset_start_attempt(running_ext.unique_entry) + await running_ext.enable() + except Exception as enable_error: + logger.warning( + f"Failed to re-enable {running_ext.identifier}:{running_ext.tag} after pull failure: {enable_error}" + ) async def _pull_docker_image(self, docker_auth: Optional[str]) -> AsyncGenerator[bytes, None]: """Pull Docker image and yield progress updates.""" tag = f"{self.source.docker}:{self.tag}" + (f"@{self.digest}" if self.digest else "") async with DockerCtx() as client: + pull_ok = False async for line in client.images.pull( tag, repo=self.source.docker, tag=self.tag, auth=docker_auth, stream=True ): - # TODO - Plug Error detection from docker image here yield json.dumps(line).encode("utf-8") + # Docker reports pull errors in-band; a finished stream without a success + # status is not a completed pull. + error = None + status = None + if isinstance(line, dict): + error = line.get("error") or (line.get("errorDetail") or {}).get("message") + status = line.get("status") + if error: + raise RuntimeError(str(error)) + if isinstance(status, str) and ("Downloaded newer image" in status or "Image is up to date" in status): + pull_ok = True + if not pull_ok: + raise RuntimeError("pull finished without a success status") # Make sure to add correct tag if a digest was used since docker messes up the tag - if self.digest: - await client.images.tag(tag, f"{self.source.docker}:{self.tag}") + if not await self._ensure_tagged_local_image(client): + raise RuntimeError(f"Image {self.source.docker}:{self.tag} missing after pull") async def _clear_remaining_tags(self) -> None: """Uninstall all other tags for this extension.""" @@ -207,13 +276,44 @@ async def _clear_remaining_tags(self) -> None: to_clear = [version for version in to_clear if version.source.tag != self.tag] await asyncio.gather(*(version.uninstall() for version in to_clear)) - async def install(self, clear_remaining_tags: bool = True, atomic: bool = False) -> AsyncGenerator[bytes, None]: + async def install( # pylint: disable=too-many-branches,too-many-locals,too-many-statements + self, clear_remaining_tags: bool = True, atomic: bool = True + ) -> AsyncGenerator[bytes, None]: logger.info(f"Installing extension {self.identifier}:{self.tag}") # First we should make sure no other tag is running + sibling_image_id: Optional[str] = None + sibling_id_unknown = False + prior_settings: Optional[ExtensionSettings] = None + try: + existing = self.settings + prior_settings = ExtensionSettings( + identifier=existing.identifier, + name=existing.name, + docker=existing.docker, + tag=existing.tag, + permissions=existing.permissions, + enabled=existing.enabled, + user_permissions=existing.user_permissions, + ) + except ExtensionNotFound: + pass + running_ext = await self._disable_running_extension() + if running_ext and running_ext.unique_entry != self.unique_entry: + try: + async with DockerCtx() as client: + info = await self._inspect_or_none(client, f"{running_ext.source.docker}:{running_ext.tag}") + image_id = info.get("Id") if isinstance(info, dict) else None + sibling_image_id = image_id if isinstance(image_id, str) and image_id else None + if not sibling_image_id: + sibling_id_unknown = True + except Exception: + sibling_id_unknown = True self._create_extension_settings() + + used_local_image = False try: self.lock(self.unique_entry) @@ -221,29 +321,36 @@ async def install(self, clear_remaining_tags: bool = True, atomic: bool = False) async for line in self._pull_docker_image(docker_auth): yield line except Exception as error: - # In case of some external installs kraken shouldn't try to install it again so we remove from settings - if atomic: - should_raise = False - if await self._image_is_available_locally(): - logger.info(f"Pull failed but image {self.identifier}:{self.tag} is already available locally") - else: - if not running_ext or self.unique_entry != running_ext.unique_entry: - should_raise = True - await self.uninstall() - if running_ext: - await running_ext.enable() - - if should_raise: - raise ExtensionPullFailed(f"Failed to pull extension {self.identifier}:{self.tag}") from error - # Reached only if the extensions are the same, the change is in permissions, not installation failure. - return + # In case of some external installs kraken shouldn't try to install it again so we + # remove from settings. Keep a leftover docker:tag if it is not the running sibling's + # image; otherwise roll back so we do not uninstall/delete the running image. + local_ok = False + if sibling_id_unknown: + logger.warning(f"Could not inspect running image for {self.identifier}; refusing local-image fallback") + else: + try: + async with DockerCtx() as client: + local_ok = await self._ensure_tagged_local_image(client, sibling_image_id) + except Exception as inspect_error: + logger.warning( + f"Could not verify image after pull of {self.identifier}:{self.tag}: {inspect_error}" + ) + local_ok = False + if local_ok: + logger.warning( + f"Pull failed but image {self.identifier}:{self.tag} is already available locally: {error}" + ) + used_local_image = True + else: + await self._rollback_failed_install(running_ext, prior_settings, atomic) + raise ExtensionPullFailed(f"Failed to pull extension {self.identifier}:{self.tag}: {error}") from error finally: self.unlock(self.unique_entry) self.reset_start_attempt(self.unique_entry) logger.info(f"Extension {self.identifier}:{self.tag} installed") # Uninstall all other tags in case user wants to clear them - if clear_remaining_tags: + if clear_remaining_tags and not used_local_image: await self._clear_remaining_tags() async def update(self, clear_remaining_tags: bool) -> AsyncGenerator[bytes, None]: From 02dea1d1b07ab953205342de7e285036704607ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Ant=C3=B4nio=20Cardoso?= Date: Fri, 28 Aug 2026 15:54:32 -0300 Subject: [PATCH 5/5] core: services: kraken: Stream watchdog pulls and require a success status start() used a non-streaming pull that ignored in-band Docker errors. --- core/services/kraken/extension/extension.py | 19 +++++-------------- 1 file changed, 5 insertions(+), 14 deletions(-) diff --git a/core/services/kraken/extension/extension.py b/core/services/kraken/extension/extension.py index 4fb7eb6be3..02b2a469c9 100644 --- a/core/services/kraken/extension/extension.py +++ b/core/services/kraken/extension/extension.py @@ -392,20 +392,11 @@ async def start(self) -> None: try: async with DockerCtx() as client: # Checks if image exists locally, if not tries to pull it - try: - await client.images.inspect(img_name) - except Exception: - try: - logger.info(f"Image not found locally, going to pull extension {self.identifier}:{self.tag}") - self.lock(self.unique_entry) - - tag = img_name + (f"@{self.digest}" if self.digest else "") - await client.images.pull(tag, repo=self.source.docker, tag=self.tag) - # Make sure to add correct tag if a digest was used since docker messes up the tag - if self.digest: - await client.images.tag(tag, img_name) - except Exception as error: - raise ExtensionPullFailed(f"Failed to pull extension {self.identifier}:{self.tag}") from error + if not await self._ensure_tagged_local_image(client): + logger.info(f"Image not found locally, going to pull extension {self.identifier}:{self.tag}") + self.lock(self.unique_entry) + async for _ in self._pull_docker_image(self._prepare_docker_auth()): + pass container = await client.containers.create_or_replace(name=ext.container_name(), config=config) # type: ignore await container.start()