[1.4] Recover extensions when Kraken settings are empty

core/libs/commonwealth/commonwealth/settings/managers/pykson_manager.py

@@ -125,9 +125,9 @@ def get_settings_version_from_filename(filename: pathlib.Path) -> int:
                 logger.debug(f"Using {valid_file} as settings source")
                 return
             except Exception as exception:
-                logger.debug("Invalid settings, going to try another file:", exception)
+                logger.warning(f"Invalid settings in {valid_file}, going to try another file: {exception}")
 
-        logger.debug("No valid settings found, using default settings")
+        logger.warning("No valid settings found, using default settings")
         self._settings = self.settings_type()
         self.save()
 

core/services/kraken/extension/extension.py

@@ -22,6 +22,7 @@
 from harbor.exceptions import ContainerNotFound
 from manifest import ManifestManager
 from manifest.models import ExtensionVersion
+from recovery import stamp_extension_labels
 from settings import ExtensionSettings, SettingsV2
 from utils import has_enough_disk_space
 
@@ -263,6 +264,7 @@ async def start(self) -> None:
 
         self._set_container_config_host_config(config)
         self._set_container_config_default_env_variables(config)
+        stamp_extension_labels(config, ext.identifier, ext.name, ext.permissions, ext.user_permissions or "")
 
         try:
             async with DockerCtx() as client:

core/services/kraken/harbor/container.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import AsyncGenerator, Dict, List, cast
+from typing import Any, AsyncGenerator, Dict, List, cast
 
 import psutil
 from aiodocker import Docker
@@ -85,6 +85,18 @@ async def _get_stats_from_containers(containers: List[DockerContainer]) -> Dict[
 
         return result
 
+    @staticmethod
+    async def inspect_extension_containers() -> List[Dict[str, Any]]:
+        async with DockerCtx() as client:
+            containers = await client.containers.list(all=True)  # type: ignore
+            inspects: List[Dict[str, Any]] = []
+            for container in containers:
+                names = container["Names"] or []
+                if not any(str(name).lstrip("/").startswith("extension-") for name in names):
+                    continue
+                inspects.append(await container.show())  # type: ignore
+            return inspects
+
     @staticmethod
     async def get_running_containers() -> List[ContainerModel]:
         async with DockerCtx() as client:

core/services/kraken/kraken.py

@@ -16,6 +16,7 @@
 from jobs.models import Job, JobMethod
 from manifest import ManifestManager
 from manifest.exceptions import ManifestBackendOffline
+from recovery import recovered_extension_from_inspect
 from settings import ExtensionSettings, SettingsV2
 
 
@@ -88,6 +89,46 @@ async def init_dead_extensions(self) -> None:
                         f"Dead extension {extension.identifier}:{extension.tag} could not be started: {traceback.format_exc()}"
                     )
 
+    async def recover_lost_extensions(self) -> None:
+        extensions: List[ExtensionSettings] = Extension._fetch_settings()
+        if extensions:
+            return
+
+        try:
+            inspects = await ContainerManager.inspect_extension_containers()
+        except Exception as error:
+            logger.error(f"Unable to inspect extension containers for settings recovery: {error}")
+            return
+        if not inspects:
+            return
+
+        catalog_by_docker: dict[str, tuple[str, str]] = {}
+        try:
+            entries = await asyncio.wait_for(self.manifest.fetch_consolidated(), timeout=5)
+            for entry in entries:
+                catalog_by_docker[entry.docker] = (entry.identifier, entry.name)
+        except Exception as error:
+            logger.warning(f"Catalog unavailable while recovering extensions: {error}")
+
+        recovered = []
+        for inspect in inspects:
+            data = recovered_extension_from_inspect(inspect, catalog_by_docker)
+            if data:
+                recovered.append(ExtensionSettings(**data))
+
+        if not recovered:
+            logger.error("Kraken settings have no extensions but extension containers exist and could not be recovered")
+            return
+
+        logger.error(
+            "Kraken settings were empty while extension containers still exist; recovering "
+            + ", ".join(f"{ext.identifier}:{ext.tag}" for ext in recovered)
+        )
+        Extension._manager.load()
+        Extension._settings = Extension._manager.settings
+        Extension._settings.extensions = recovered
+        Extension._manager.save()
+
     async def fetch_default_extension_data(self, url: str) -> Any:
         async with aiohttp.ClientSession() as session:
             headers = {"Accept": "application/json"}
@@ -104,23 +145,33 @@ def is_install_default_ext_job_created(self, identifier: str) -> bool:
 
     async def setup_default_extensions(self) -> None:
         extensions: List[ExtensionSettings] = Extension._fetch_settings()
-        for ext in [
-            ext
-            for ext in DEFAULT_EXTENSIONS
-            if not any(ext["identifier"] == extension.identifier for extension in extensions)
-        ]:
+        for ext in DEFAULT_EXTENSIONS:
+            if any(ext["identifier"] == extension.identifier for extension in extensions):
+                continue
             job_id = f'__default_install_{ext["identifier"]}'
-            if not self.is_install_default_ext_job_created(job_id):
-                data = await self.fetch_default_extension_data(ext["url"])
-                job = Job(
-                    id=job_id,
-                    route="v2.0/extension",
-                    method=JobMethod.POST,
-                    body=data,
-                    retries=1,
-                )
-                JobsManager.add(job)
-                logger.info(f"Created job to install default extension {ext['identifier']}")
+            if self.is_install_default_ext_job_created(job_id):
+                continue
+            data = await self.fetch_default_extension_data(ext["url"])
+            recovered = next((extension for extension in extensions if extension.docker == data.get("docker")), None)
+            if recovered:
+                if recovered.identifier != ext["identifier"]:
+                    logger.info(
+                        f"Rebinding recovered extension {recovered.identifier} to default identifier {ext['identifier']}"
+                    )
+                    recovered.identifier = ext["identifier"]
+                    if data.get("name"):
+                        recovered.name = data["name"]
+                    Extension._manager.save()
+                continue
+            job = Job(
+                id=job_id,
+                route="v2.0/extension",
+                method=JobMethod.POST,
+                body=data,
+                retries=1,
+            )
+            JobsManager.add(job)
+            logger.info(f"Created job to install default extension {ext['identifier']}")
 
     async def kill_invalid_extensions(self) -> None:
         extensions: List[ExtensionSettings] = Extension._fetch_settings()
@@ -143,6 +194,10 @@ async def kill_dangling_containers(self) -> None:
             return
 
         extensions: List[ExtensionSettings] = Extension._fetch_settings()
+        if not extensions:
+            if any(container.name[1:].startswith("extension-") for container in containers):
+                logger.error("Skipping dangling extension cleanup because settings have no extensions")
+            return
 
         for container in containers:
             container_name = container.name[1:]

core/services/kraken/main.py

@@ -32,6 +32,7 @@
     server = Server(config)
     jobs.set_base_host(f"http://{args.host}:{args.port}")
 
+    loop.run_until_complete(kraken.recover_lost_extensions())
     loop.create_task(kraken.start_cleaner_task())
     loop.create_task(kraken.start_starter_task())
     loop.create_task(jobs.start())

core/services/kraken/recovery.py

@@ -0,0 +1,127 @@
+import json
+import re
+from typing import Any, Dict, Optional, Tuple
+
+LABEL_IDENTIFIER = "blueos.extension.identifier"
+LABEL_NAME = "blueos.extension.name"
+LABEL_PERMISSIONS = "blueos.extension.permissions"
+LABEL_USER_PERMISSIONS = "blueos.extension.user_permissions"
+
+_HOST_CONFIG_KEYS = (
+    "Binds",
+    "NetworkMode",
+    "Privileged",
+    "ExtraHosts",
+    "PortBindings",
+    "Devices",
+    "CapAdd",
+)
+
+_CONTAINER_NAME_SANITIZE = re.compile("[^a-zA-Z0-9]")
+
+
+def split_image_name(image: str) -> Optional[Tuple[str, str]]:
+    if not image or image.startswith("sha256:"):
+        return None
+    repo, sep, tag = image.rpartition(":")
+    if not sep or not repo or "/" in tag:
+        return None
+    return repo, tag
+
+
+def container_name_for(docker: str, tag: str) -> str:
+    return "extension-" + _CONTAINER_NAME_SANITIZE.sub("", f"{docker}{tag}")
+
+
+def permissions_from_inspect(inspect: Dict[str, Any]) -> str:
+    config = inspect.get("Config") or {}
+    host = inspect.get("HostConfig") or {}
+    result: Dict[str, Any] = {}
+
+    exposed = config.get("ExposedPorts")
+    if exposed:
+        result["ExposedPorts"] = exposed
+
+    hostconfig: Dict[str, Any] = {}
+    for key in _HOST_CONFIG_KEYS:
+        value = host.get(key)
+        if value in (None, "", [], {}, False):
+            continue
+        if key == "NetworkMode" and value == "default":
+            continue
+        hostconfig[key] = value
+    if hostconfig:
+        result["HostConfig"] = hostconfig
+
+    return json.dumps(result)
+
+
+def _identifier_and_name(
+    labels: Dict[str, Any],
+    docker: str,
+    catalog_by_docker: Optional[Dict[str, Tuple[str, str]]],
+) -> Tuple[str, str]:
+    identifier = labels.get(LABEL_IDENTIFIER) or ""
+    display_name = labels.get(LABEL_NAME) or ""
+    if not identifier and catalog_by_docker and docker in catalog_by_docker:
+        identifier, catalog_name = catalog_by_docker[docker]
+        display_name = display_name or catalog_name
+    if not identifier:
+        identifier = docker.replace("/", ".")
+    if not display_name:
+        display_name = docker.rsplit("/", 1)[-1]
+    return identifier, display_name
+
+
+def recovered_extension_from_inspect(
+    inspect: Dict[str, Any],
+    catalog_by_docker: Optional[Dict[str, Tuple[str, str]]] = None,
+) -> Optional[Dict[str, Any]]:
+    names = inspect.get("Name") or ""
+    if isinstance(names, list):
+        name = names[0] if names else ""
+    else:
+        name = str(names)
+    container_name = name.lstrip("/")
+    if not container_name.startswith("extension-"):
+        return None
+
+    config = inspect.get("Config") or {}
+    image = config.get("Image") or ""
+    split = split_image_name(image)
+    if not split:
+        return None
+    docker, tag = split
+    if container_name_for(docker, tag) != container_name:
+        return None
+
+    labels = config.get("Labels") or {}
+    identifier, display_name = _identifier_and_name(labels, docker, catalog_by_docker)
+    permissions = labels.get(LABEL_PERMISSIONS) or ""
+    user_permissions = labels.get(LABEL_USER_PERMISSIONS) or ""
+
+    if not permissions:
+        permissions = permissions_from_inspect(inspect)
+
+    return {
+        "identifier": identifier,
+        "name": display_name,
+        "docker": docker,
+        "tag": tag,
+        "permissions": permissions,
+        "enabled": True,
+        "user_permissions": user_permissions,
+    }
+
+
+def stamp_extension_labels(
+    config: Dict[str, Any], identifier: str, name: str, permissions: str, user_permissions: str
+) -> None:
+    labels = config.get("Labels")
+    if not isinstance(labels, dict):
+        labels = {}
+        config["Labels"] = labels
+    labels[LABEL_IDENTIFIER] = identifier
+    labels[LABEL_NAME] = name
+    labels[LABEL_PERMISSIONS] = permissions or ""
+    labels[LABEL_USER_PERMISSIONS] = user_permissions or ""